src/tools/analytics/anomaly.py

"""analyze_anomaly — flag unusual rows with Isolation Forest (§0.9).

Second ML analytical tool. Scores rows across one or more numeric columns with a
scikit-learn IsolationForest (fit OFF the event loop via `run_fit`), flags the
outliers, and returns a summary + the flagged rows + a scatter with the anomaly
points highlighted ("sepaket").

Follows docs/tools/ANALYTICAL_TOOLS_CONVENTIONS.md: compute is synchronous except
for the `await run_fit` offload; `data` is never self-fetched (Pattern A); errors
escape to the invoker's never-throw seam; output is deterministic (fixed
random_state). Mirrors analyze_forecast's shape so the ML tools stay uniform.
"""

from __future__ import annotations

import numpy as np
import pandas as pd
from pydantic import BaseModel, Field

from src.tools.analytics.charts import anomaly_scatter_chart
from src.tools.analytics.descriptive import ColumnNotFoundError
from src.tools.analytics.temporal import _clean
from src.tools.ml.executor import run_fit
from src.tools.spec_builder import build_description

# IsolationForest is meaningless on a handful of rows; below this we raise.
_MIN_ROWS = 10
# Cap the returned anomaly SAMPLE (the count stays exact) so a tool output can't
# balloon to tens of thousands of tokens — the list is illustrative, not the result.
# The full total is reported alongside as `anomalies_total`. A23: was 200; a bounded
# top-by-score sample is the right payload for an LLM-facing tool (defense-in-depth
# with the assembler's own cap).
_MAX_ANOMALY_ROWS = 50
# Fixed so the same data always flags the same rows (family-B determinism promise).
_RANDOM_STATE = 42
# Above this share of rows flagged, the result is not "a few anomalies" — the notion
# has collapsed and the answer is noise. The reserved `quality` block reads `weak` off
# it. NOT a naive low ceiling: A20 reverted one because `contamination="auto"` flags a
# large slice of CLEAN data. Re-measured 2026-09-21 — clean gaussian noise flags 0.14–0.22
# (mean 0.17, p95 0.22), a genuine 30%-outlier frame flags 0.30; the demo's degenerate
# run flagged 0.86. 0.5 clears both the noise floor and a real high-outlier finding while
# still catching "most rows flagged".
_ANOMALY_RATE_CEILING = 0.5
# Internal column names for the highlighted scatter (never collide with user data).
_FLAG_COL = "__is_anomaly__"
_INDEX_COL = "__row__"


class NotEnoughDataError(ValueError):
    """Too few valid rows to fit an anomaly model (error_code NOT_ENOUGH_DATA)."""


class EmptyColumnSelectionError(ValueError):
    """column_ids was empty (error_code EMPTY_COLUMN_SELECTION)."""


class AnomalyInput(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] = Field(
        ..., description="One or more NUMERIC columns to score rows for anomalies."
    )
    label_columns: list[str] | None = Field(
        None, description="Identifier columns (e.g. unit/equipment no.) to carry into each "
        "flagged record so the anomaly can be NAMED. Never fed to the model — pass the id "
        "columns whenever the question asks 'which X is unusual'."
    )
    contamination: float | None = Field(
        None, description="Expected fraction of anomalies (0-0.5); omit for automatic."
    )


DESCRIPTION = build_description(
    summary="Flag unusual/outlier rows across numeric columns with an Isolation Forest.",
    use_when="the user asks to find anomalies, outliers, or 'weird'/'unusual' records "
    "— 'ada anomali', 'outlier', 'data aneh', 'yang nggak wajar'.",
    dont_use_when=[
        "the user wants a plain distribution/spread of one column -> analyze_descriptive",
        "the user wants movement over time or a projection -> analyze_trend / analyze_forecast",
        "the user wants what DRIVES a metric -> analyze_importance",
    ],
    output="a summary (row/anomaly counts, rate) + the flagged rows (named by any "
    "label_columns given) + a quality verdict on the flag rate + a scatter with "
    "anomaly points highlighted.",
    examples=[
        "ada anomali nggak di data bulan ini?",
        "detect outliers in the sensor readings",
        "unit mana yang angkanya nggak wajar?",
    ],
)


def _fit_iforest(x: np.ndarray, contamination, random_state: int):
    """Fit an IsolationForest. Imported lazily so app startup doesn't pay for
    scikit-learn unless an anomaly analysis runs. Runs inside run_fit's thread."""
    from sklearn.ensemble import IsolationForest

    model = IsolationForest(contamination=contamination, random_state=random_state)
    model.fit(x)
    return model


def _flag_rate_quality(anomaly_rate: float) -> dict[str, object]:
    """Reserved `quality` block from the flag rate (conventions §3).

    Split out so both branches are unit-testable without a model run — the fit can't be
    coaxed above a 0.5 flag rate through `contamination` (sklearn caps it there), so the
    weak branch has no organic fixture.
    """
    weak = anomaly_rate > _ANOMALY_RATE_CEILING
    return {
        "metric": "anomaly_rate",
        "value": round(anomaly_rate, 4),
        "floor": _ANOMALY_RATE_CEILING,
        "weak": weak,
        "note": (
            f"{anomaly_rate:.0%} of rows flagged — above half the data, so this is "
            "noise, not a few outliers"
            if weak
            else f"{anomaly_rate:.0%} of rows flagged as anomalous"
        ),
    }


def _anomaly_chart(
    df: pd.DataFrame, column_ids: list[str], is_anomaly: pd.Series
) -> dict[str, object]:
    """Scatter with anomalies highlighted: first two numeric columns, or row-index
    vs the single column when only one was scored."""
    if len(column_ids) >= 2:
        chart_df = df.assign(**{_FLAG_COL: is_anomaly.to_numpy()})
        return anomaly_scatter_chart(chart_df, column_ids[0], column_ids[1], _FLAG_COL)
    col = column_ids[0]
    chart_df = pd.DataFrame(
        {_INDEX_COL: range(len(df)), col: df[col].to_numpy(), _FLAG_COL: is_anomaly.to_numpy()}
    )
    return anomaly_scatter_chart(chart_df, _INDEX_COL, col, _FLAG_COL)


async def analyze_anomaly(
    df: pd.DataFrame,
    column_ids: list[str],
    contamination: float | None = None,
    label_columns: list[str] | None = None,
) -> dict[str, object]:
    """Flag anomalous rows across `column_ids` with an Isolation Forest.

    Async ONLY so the model fit can be offloaded via `run_fit`; everything else is
    synchronous compute. `data` is materialised upstream (Pattern A) and arrives as
    `df`.

    `label_columns` are identifier columns carried into each flagged record so the
    anomaly can be NAMED (e.g. `eqm_no`). They are NEVER fed to the model — only the
    numeric `column_ids` are scored.

    Returns a dict with: method, column_ids, contamination, n_rows, n_anomalies,
    anomaly_rate, anomalies_total (= n_anomalies) / anomalies_shown (how many the
    capped sample actually carries), `anomalies` (the flagged rows, top by score,
    capped at _MAX_ANOMALY_ROWS), a reserved `quality` block (flag-rate verdict), and
    `charts` (the reserved auto-chart key).

    Raises (all wrapped by the invoker's never-throw seam):
        ColumnNotFoundError, EmptyColumnSelectionError, NotEnoughDataError.
    """
    if not column_ids:
        raise EmptyColumnSelectionError("column_ids must name at least one numeric column")
    labels = list(label_columns) if label_columns else []
    missing = [c for c in column_ids + labels if c not in df.columns]
    if missing:
        raise ColumnNotFoundError(f"columns not found: {missing}")

    # Coerce to numeric; a row is scorable only if every selected column is numeric.
    features = df[column_ids].apply(pd.to_numeric, errors="coerce")
    valid = features.notna().all(axis=1)
    scored = features[valid]
    if scored.shape[0] < _MIN_ROWS:
        raise NotEnoughDataError(
            f"need >= {_MIN_ROWS} rows with numeric values in all of {column_ids} "
            f"to detect anomalies, got {scored.shape[0]}"
        )

    contam = "auto" if contamination is None else contamination
    x = scored.to_numpy(dtype=float)
    model = await run_fit(_fit_iforest, x, contam, _RANDOM_STATE, label="analyze_anomaly")

    flags = model.predict(x) == -1  # sklearn: -1 anomaly, 1 normal
    scores = -model.score_samples(x)  # higher = more anomalous

    is_anomaly = pd.Series(False, index=df.index)
    is_anomaly.loc[scored.index] = flags

    # Flagged rows, most anomalous first, capped.
    flagged_idx = scored.index[flags]
    flagged_scores = scores[flags]
    order = np.argsort(flagged_scores)[::-1]
    anomalies: list[dict[str, object]] = []
    for pos in order[:_MAX_ANOMALY_ROWS]:
        idx = flagged_idx[pos]
        # Identifiers FIRST so a reader sees WHICH row before its numbers. Pulled from
        # the original df (labels are non-numeric and never entered `scored`).
        record: dict[str, object] = {c: _clean(df.at[idx, c]) for c in labels}
        record.update({c: _clean(scored.at[idx, c]) for c in column_ids})
        record["anomaly_score"] = round(float(flagged_scores[pos]), 4)
        anomalies.append(record)

    anomaly_rate = float(flags.mean()) if scored.shape[0] else 0.0
    return {
        "method": "isolation_forest",
        "column_ids": column_ids,
        "contamination": contam,
        "n_rows": int(scored.shape[0]),
        "n_anomalies": int(flags.sum()),
        "anomaly_rate": anomaly_rate,
        # `anomalies` is a bounded top-by-score SAMPLE, not the full set. These two
        # keep it honest: total says how many there really are, shown says how many
        # the sample carries. When total > shown the caller knows it's truncated.
        "anomalies_total": int(flags.sum()),
        "anomalies_shown": len(anomalies),
        "anomalies": anomalies,
        # Reserved `quality` block (conventions §3): the tool owns its own verdict; CK7
        # surfaces `weak`. Isolation Forest has no accuracy score, so the honest signal
        # is the flag RATE — an unsupervised detector that flags most of the data found
        # nothing. `weak` fires only ABOVE _ANOMALY_RATE_CEILING (0.5), clear of the
        # ~0.22 clean-data floor A20 measured, so a normal run is never marked weak; the
        # note reports the rate plainly either way (A20's fallback, kept as the message).
        "quality": _flag_rate_quality(anomaly_rate),
        "charts": [_anomaly_chart(df, column_ids, is_anomaly)],
    }