src/tools/analytics/data_quality.py

"""analyze_data_quality — per-column data-quality profile (§0.9, renamed from analyze_profile).

Rename + activate of the former `analyze_profile` (`quality.py`): profiles each
column's health — dtype, inferred type, completeness (null count/rate), cardinality
(distinct count/rate, constant flag), and — for numeric columns — min/max/mean plus
an IQR-based outlier count; for non-numeric columns the most frequent value. Now
also returns an issues summary and ships a **missing-rate bar chart** as a package.

Follows docs/tools/ANALYTICAL_TOOLS_CONVENTIONS.md: pure synchronous compute (no
model fit → stays sync), Pattern A `data`, errors escape to the invoker's never-throw
seam. Deterministic.
"""

from __future__ import annotations

import pandas as pd
from pydantic import BaseModel, Field

from src.tools.analytics.charts import LAYOUT_PRESET, _axis, envelope
from src.tools.analytics.descriptive import ColumnNotFoundError
from src.tools.spec_builder import build_description


def _clean(value: object) -> object:
    """Convert numpy/pandas scalars to plain Python so the output is JSON-clean.

    `top_value` (most frequent value) can be a `pandas.Timestamp` when profiling a
    datetime column — neither `Timestamp` nor numpy scalars are JSON-safe.
    """
    if isinstance(value, pd.Timestamp):
        return value.isoformat()
    if hasattr(value, "item"):
        return value.item()
    return value


def _profile_one(series: pd.Series) -> dict[str, object]:
    """Build the quality profile for a single column."""
    total = len(series)
    non_null = series.dropna()
    nn = len(non_null)
    distinct = int(series.nunique(dropna=True))

    is_bool = pd.api.types.is_bool_dtype(series)
    is_datetime = pd.api.types.is_datetime64_any_dtype(series)
    # bool is technically numeric in pandas; treat it as its own type.
    is_numeric = pd.api.types.is_numeric_dtype(series) and not is_bool

    if is_bool:
        inferred = "boolean"
    elif is_datetime:
        inferred = "datetime"
    elif is_numeric:
        inferred = "numeric"
    else:
        inferred = "categorical"

    out: dict[str, object] = {
        "dtype": str(series.dtype),
        "inferred_type": inferred,
        "count": int(total),
        "null_count": int(series.isna().sum()),
        "null_rate": float(series.isna().mean()) if total else 0.0,
        "distinct_count": distinct,
        "distinct_rate": (distinct / nn) if nn else 0.0,  # over non-null values
        "is_constant": distinct <= 1,
    }

    if is_numeric and nn > 0:
        out["min"] = _clean(non_null.min())
        out["max"] = _clean(non_null.max())
        out["mean"] = _clean(non_null.mean())
        # IQR rule: values outside [Q1 - 1.5*IQR, Q3 + 1.5*IQR] are outliers.
        # Needs enough points for stable quartiles.
        if nn >= 4:
            q1 = non_null.quantile(0.25)
            q3 = non_null.quantile(0.75)
            iqr = q3 - q1
            lower, upper = q1 - 1.5 * iqr, q3 + 1.5 * iqr
            out["outlier_count"] = int(((non_null < lower) | (non_null > upper)).sum())
        else:
            out["outlier_count"] = None
    elif not is_numeric and nn > 0:
        counts = non_null.value_counts()
        out["top_value"] = _clean(counts.index[0])
        out["top_freq"] = int(counts.iloc[0])

    return out


class DataQualityInput(BaseModel):
    # `data` FIRST — property order follows field order (planner-prompt readability).
    data: str = Field(..., description="Placeholder ${t<id>} of the upstream table (Pattern A).")
    column_ids: list[str] | None = Field(
        None, description="Columns to profile; omit to profile every column."
    )


DESCRIPTION = build_description(
    summary="Per-column data-quality profile — dtype, completeness (nulls), cardinality, "
    "constants, and IQR outliers for numeric columns.",
    use_when="the question is about the HEALTH of the data, not its statistics — missing "
    "values, data types, duplicates/unique, constant columns, outliers, 'is this clean "
    "enough to analyze' — 'kualitas', 'data kosong', 'tipe data', 'duplikat', 'outlier'.",
    dont_use_when=[
        "the user wants statistics like mean/median/std/skew -> analyze_descriptive",
        "it groups or compares -> analyze_aggregate / analyze_ab_test",
    ],
    output="a per-column profile (types, null rate, distinct, outliers), an issues summary "
    "(constant + high-null columns), and a bar chart of missing rate per column.",
    examples=[
        "is this dataset clean enough to analyze?",
        "which columns have a lot of missing values?",
        "data types and unique counts per column?",
    ],
)

# Columns with at least this null rate are surfaced in the issues summary.
_HIGH_NULL_THRESHOLD = 0.2


def _missing_rate_chart(columns: dict[str, dict[str, object]]) -> dict[str, object]:
    """Bar chart of null rate per column (the headline data-quality signal)."""
    cols = list(columns)
    null_rates = [round(float(columns[c]["null_rate"]), 4) for c in cols]
    title = "Missing rate by column"
    layout = {
        **LAYOUT_PRESET,
        "title": {"text": title},
        "showlegend": False,
        "xaxis": _axis("column"),
        "yaxis": _axis("null rate"),
    }
    traces = [{"type": "bar", "x": cols, "y": null_rates, "name": "null rate"}]
    return envelope("bar", title, traces, layout)


def analyze_data_quality(
    df: pd.DataFrame,
    column_ids: list[str] | None = None,
) -> dict[str, object]:
    """Per-column data-quality profile.

    `data` is materialised upstream (Pattern A) and arrives as `df`.

    Returns a dict with: n_rows, n_columns, `columns` ({column: profile}),
    `duplicate_rows` ({count, rate} — full duplicate rows over the profiled columns),
    `issues` (constant_columns + high_null_columns + duplicate_row_count), and `charts`
    (the reserved auto-chart key).

    Raises (wrapped by the invoker's never-throw seam):
        ColumnNotFoundError: if any column_id is absent from df.
    """
    cols = list(column_ids) if column_ids is not None else list(df.columns)
    missing = [c for c in cols if c not in df.columns]
    if missing:
        raise ColumnNotFoundError(f"columns not found: {missing}")

    columns = {col: _profile_one(df[col]) for col in cols}

    # Full-row duplicates over the profiled columns — `use_when` advertises "duplikat",
    # so it must be computed, not implied by distinct_rate. `keep="first"` counts the
    # redundant copies (rows beyond the first of each group); over a column subset this
    # is a candidate-key duplicate check. Always reported (0 when none), never absent.
    n_rows = int(len(df))
    dup_count = int(df[cols].duplicated().sum())
    duplicate_rows = {"count": dup_count, "rate": (dup_count / n_rows) if n_rows else 0.0}

    issues = {
        "constant_columns": [c for c, p in columns.items() if p["is_constant"]],
        "high_null_columns": [
            c for c, p in columns.items() if p["null_rate"] >= _HIGH_NULL_THRESHOLD
        ],
        "duplicate_row_count": dup_count,
    }

    return {
        "n_rows": n_rows,
        "n_columns": len(cols),
        "columns": columns,
        "duplicate_rows": duplicate_rows,
        "issues": issues,
        "charts": [_missing_rate_chart(columns)],
    }