benchmarks/anomaly_detector_gap.py

"""
benchmarks/anomaly_detector_gap.py — Quantifies the gap between AgriFlow's TWO
price-anomaly detectors on the project's own real data.

BACKGROUND
----------
Two detectors exist in the codebase, both in the production path:

  matching_engine/engine.py:62  detect_price_anomaly(node, historical_median,
      historical_std) -- z_score = |price - median| / std, flag if > 3.0
      (PRICE_ANOMALY_SIGMA). This is what actually gates matching: called at
      engine.py:416 and :429 inside run_matching()'s D3 preprocessing, and a
      flagged node is DROPPED from the surplus/deficit pool entirely (not just
      down-weighted).

  analysis/price_anomaly.py    detect_anomalies() -- S-H-ESD: deseasonalise
      (rolling-median trend + monthly seasonal), then robust MAD on the
      residual, with a persistence filter. This is validated against the
      literature (Hochenbaum/Vallis/Kejariwal 2017) and is what feeds the
      dashboard anomaly panel / API -- NOT the matching engine.

detect_price_anomaly's std is non-robust: a real outlier inflates std, which
shrinks the z-score of ALL points (including the outlier itself and future
genuine anomalies) -- the classic "masking effect" in robust statistics. This
script measures whether that mechanism is real on THIS project's actual price
data, and quantifies it two ways:

  PART 1 -- "As shipped" gap on real data.
      The production historical_prices dict is NOT a rolling window -- it is a
      single STATIC (median, std) constant per commodity, hand-set in
      sample_data/historical_price_stats.csv (mirrored into db/schema.sql's
      historical_prices table). This part runs the ACTUAL shipped function,
      detect_price_anomaly(), against every real observed price in
      sample_data/price_history/*.csv using the ACTUAL shipped static
      constants, and compares the flag set against the already-validated
      S-H-ESD persistent-anomaly set (ground truth for "this was a real
      event"). Reports recall (of genuine persistent anomalies, how many does
      the static 3-sigma catch) and precision-adjacent flag volume.

  PART 2 -- Masking mechanism, isolated, on real price levels.
      Simulates the scenario the code's OWN docstrings claim ("historical_prices:
      dict commodity_code -> (median, std) 30-day rolling", engine.py:352,370)
      but the shipped loaders never actually implement: a rolling 30-day window
      recomputed from real data. Calls the ACTUAL detect_price_anomaly()
      function (not a re-implementation) with median/std computed from a
      window seeded with real cabai_rawit/beras_medium price levels, sweeping
      the number of contaminating outliers already present in that window
      (0..~40% of the window), and reports at what contamination fraction the
      REAL genuine spike stops being flagged. Cross-checked against the
      MAD-based rule (median +/- k*1.4826*MAD, same k=3.0) at every
      contamination level.

Both parts are seeded (--seed, default 2026) for reproducibility.

Usage:
    python benchmarks/anomaly_detector_gap.py
"""

from __future__ import annotations

import argparse
import os
import random
import statistics
import sys
from datetime import datetime
from typing import Dict, List, Tuple

if sys.platform == "win32":
    try:
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stderr.reconfigure(encoding="utf-8")
    except (AttributeError, OSError):
        pass

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if ROOT not in sys.path:
    sys.path.insert(0, ROOT)

from matching_engine.engine import PRICE_ANOMALY_SIGMA, detect_price_anomaly
from matching_engine.models import Commodity, DemandNode, Kabupaten, Tier
from analysis.price_anomaly import CITY_NAMES, load_series, scan_all

DEFAULT_SEED = 2026
PRICE_HISTORY_DIR = os.path.join(ROOT, "sample_data", "price_history")
HISTORICAL_STATS_CSV = os.path.join(ROOT, "sample_data", "historical_price_stats.csv")


def load_static_stats() -> Dict[str, Tuple[float, float, str]]:
    """Read the ACTUAL shipped static (median, std, source) per commodity."""
    out = {}
    with open(HISTORICAL_STATS_CSV, encoding="utf-8") as fh:
        import csv
        for row in csv.DictReader(fh):
            out[row["commodity_code"]] = (
                float(row["median_idr_per_kg"]),
                float(row["std_idr_per_kg"]),
                row["source"],
            )
    return out


def _dummy_node(price: float):
    """
    Minimal SupplyNode-like stand-in: detect_price_anomaly only reads
    node.price_per_kg, so a lightweight namespace avoids constructing full
    Kabupaten/Commodity graphs for a million-point scan.
    """
    class _N:
        pass
    n = _N()
    n.price_per_kg = price
    return n


# =============================================================================
# PART 1 — "AS SHIPPED": static constants vs real observed prices
# =============================================================================

def run_as_shipped_gap(seed: int) -> None:
    print("=" * 82)
    print("  PART 1 — 'As shipped': static (median, std) vs real PIHPS Jatim prices")
    print("=" * 82)
    print(f"  historical_price_stats.csv is a STATIC constant per commodity "
          f"(not a rolling window,\n  despite engine.py's docstring claiming "
          f"'30-day rolling') -- this reproduces exactly what\n  production does "
          f"today: detect_price_anomaly() called with the SAME fixed (median, std)\n"
          f"  against every real observed price.\n")

    static_stats = load_static_stats()
    ground_truth = scan_all(PRICE_HISTORY_DIR, window=30, k=3.0, persist=2)
    ground_truth_persistent = [a for a in ground_truth if a["persistent"]]

    header = (f"  {'commodity':<15}{'source':<38}{'n_obs':>8}{'static_3σ_flags':>17}"
              f"{'MAD_persistent':>16}{'overlap':>9}{'recall':>9}")
    print(header)
    print("  " + "-" * (len(header) - 2))

    total_obs = 0
    total_static_flags = 0
    total_mad_persistent = 0
    total_overlap = 0

    for commodity, (median, std, source) in sorted(static_stats.items()):
        gt_comm = [a for a in ground_truth_persistent if a["commodity_code"] == commodity]
        if not gt_comm and not any(True for _ in []):
            pass  # commodity may still have price series even with 0 persistent anomalies

        n_obs = 0
        static_flag_dates = set()  # (city_id, date)
        for city_id in CITY_NAMES:
            series = load_series(commodity, city_id, PRICE_HISTORY_DIR)
            if not series:
                continue
            for date, price in series:
                n_obs += 1
                if detect_price_anomaly(_dummy_node(price), median, std):
                    static_flag_dates.add((city_id, date))

        if n_obs == 0:
            continue  # commodity has no real price series (e.g. cabai_merah, tomat, ...)

        gt_dates = {(a["city_id"], a["date"]) for a in gt_comm}
        overlap = static_flag_dates & gt_dates
        recall = (len(overlap) / len(gt_dates) * 100.0) if gt_dates else float("nan")

        total_obs += n_obs
        total_static_flags += len(static_flag_dates)
        total_mad_persistent += len(gt_dates)
        total_overlap += len(overlap)

        recall_str = f"{recall:5.1f}%" if gt_dates else "n/a"
        print(f"  {commodity:<15}{source:<38}{n_obs:>8}{len(static_flag_dates):>17}"
              f"{len(gt_dates):>16}{len(overlap):>9}{recall_str:>9}")

    print()
    overall_recall = (total_overlap / total_mad_persistent * 100.0) if total_mad_persistent else 0.0
    print(f"  TOTAL: {total_obs} real observations across 7 commodities x 8 IHK cities.")
    print(f"  Static 3σ flagged {total_static_flags} points total.")
    print(f"  S-H-ESD (robust, validated) flagged {total_mad_persistent} PERSISTENT genuine events.")
    print(f"  Static 3σ caught {total_overlap}/{total_mad_persistent} of those "
          f"({overall_recall:.1f}% recall) on the SAME date+city.")
    print(f"\n  Caveat: telur_ayam and daging_ayam static constants are labelled "
          f"'SYNTHETIC' in\n  historical_price_stats.csv (never calibrated against "
          f"the real PIHPS series at all) --\n  their recall numbers reflect a "
          f"made-up threshold, not a stale-but-real one.\n")


# =============================================================================
# PART 2 — Masking mechanism, isolated, on real price levels
# =============================================================================

def run_masking_mechanism(seed: int, window: int = 30, trials: int = 200) -> None:
    print("=" * 82)
    print("  PART 2 — Masking mechanism (rolling window, real price levels)")
    print("=" * 82)
    print(f"  Simulates the 'rolling {window}-day' design the docstring promises "
          f"(engine.py:352)\n  but the shipped loader never implements. Calls the "
          f"ACTUAL detect_price_anomaly()\n  function with median/std computed "
          f"from a window built on REAL median price levels\n  (not an arbitrary "
          f"Rp30,000 placeholder), sweeping how many contaminating outliers\n"
          f"  are already sitting in that window when a genuine 3x spike arrives.\n")

    # Real median price levels, PIHPS Jatim (Surabaya, 3578), for grounding.
    scenarios = [
        ("cabai_rawit (volatile, MAPE 23.2% per forecasting validation)", 41_000.0),
        ("beras_medium (low-volatility staple)", 14_000.0),
    ]

    for label, base_price in scenarios:
        print(f"  --- {label}: base price Rp{base_price:,.0f}/kg ---")
        header = (f"  {'contam_n':>9}{'contam_%':>10}{'mean_std':>12}"
                  f"{'3σ_flags_spike':>16}{'MAD_flags_spike':>17}")
        print(header)

        for contam_n in range(0, int(window * 0.5) + 1, 2):
            contam_frac = contam_n / window * 100.0
            rng = random.Random(seed + contam_n)
            spike3sigma_hits = 0
            mad_hits = 0
            stds = []

            for _ in range(trials):
                # Build a window of `window` observations: base price with mild
                # noise, `contam_n` of them replaced by contaminating spikes
                # (2-3x base, random sign/magnitude), and ONE genuine test spike
                # (3x base) at a fixed slot -- this is the point we ask both
                # detectors to catch.
                win = [base_price * rng.gauss(1.0, 0.03) for _ in range(window)]
                contam_positions = rng.sample(range(window - 1), k=min(contam_n, window - 1))
                for p in contam_positions:
                    mult = rng.uniform(2.0, 3.0) * rng.choice([1, -0.4])
                    win[p] = max(1.0, base_price * abs(mult))
                test_spike_price = base_price * 3.0
                win[-1] = test_spike_price  # genuine anomaly under test

                median = statistics.median(win)
                std = statistics.pstdev(win)
                stds.append(std)

                # ACTUAL shipped function — not a re-implementation.
                if detect_price_anomaly(_dummy_node(test_spike_price), median, std):
                    spike3sigma_hits += 1

                # MAD-equivalent at the SAME k, matching analysis/price_anomaly's
                # spread estimator (robust) instead of engine.py's std (non-robust).
                abs_devs = [abs(x - median) for x in win]
                mad = statistics.median(abs_devs)
                if mad > 0:
                    mad_z = abs(test_spike_price - median) / (1.4826 * mad)
                    if mad_z > PRICE_ANOMALY_SIGMA:
                        mad_hits += 1
                else:
                    mad_hits += 1  # flat window + spike -> trivially anomalous

            mean_std = statistics.mean(stds)
            print(f"  {contam_n:>9}{contam_frac:>9.1f}%{mean_std:>12,.0f}"
                  f"{spike3sigma_hits:>10}/{trials:<5}{mad_hits:>10}/{trials:<5}")
        print()


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Quantify the gap between engine.py's z-score detector and "
                     "analysis/price_anomaly's MAD detector, on AgriFlow's own real data.",
    )
    parser.add_argument("--seed", type=int, default=DEFAULT_SEED)
    parser.add_argument("--trials", type=int, default=200)
    args = parser.parse_args()

    print(f"AgriFlow — Anomaly Detector Gap Analysis  (seed={args.seed})")
    print(f"Run: {datetime.now().isoformat(timespec='seconds')}\n")

    run_as_shipped_gap(args.seed)
    run_masking_mechanism(args.seed, trials=args.trials)


if __name__ == "__main__":
    main()