"""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} 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)], }