backtest_indonesia.py

"""
backtest_indonesia.py
=====================
Historical replay / evaluation harness for the Indonesia weather-risk stack.

WHAT THIS IS FOR
----------------
The Optuna sweep (train_kaggle.py) evaluates the RL AGENT's hyperparameters.
This harness answers a different, more operational question: "fed real
historical weather for Indonesian zones, does the deterministic scoring
stack (climatology anomalies -> ZoneObs signals -> crop_risk_scorer alert
levels) actually fire on the days when the climate says something bad was
happening -- and with how much lead time?" It is the missing link between
"the modules pass unit tests" and "the system demonstrably works as a
weather modelling / prediction / planning tool for Indonesia".

PIPELINE PER REPLAY STEP (per zone, per date)
---------------------------------------------
  1. obs: live mode -> era5_data_pipeline.fetch_zone_obs (Open-Meteo archive
     for historical dates -- no credentials needed; ERA5/CDS and GEE are
     deliberately bypassed by forcing OPENMETEO_LIVE so the replay does not
     depend on paid/configured services).
     synthetic mode -> deterministic make_synthetic_zone_obs with planted
     event blocks (offline, CI-friendly).
  2. anomalies: climatology.apply_climatology_anomalies with a PINNED
     climatology whose period ends BEFORE the replay window starts
     (end_year = replay_start.year - 1). This is the no-look-ahead
     guarantee; fetching anomalies through cfg.use_climatology_anomalies
     would instead use the most-recent years and leak the replayed period.
  3. forecast: timesfm_wrapper 'baseline' backend (persistence /
     climatology-reverting, derived from the obs only -- no look-ahead).
     A real NWP hindcast archive would be the strict upgrade; the baseline
     keeps the replay honest and reproducible.
  4. score: crop_risk_scorer.compute_risk_score -> alert level per day.

GROUND TRUTH
------------
Default (proxy): climatological percentiles from the SAME pinned climatology:
  drought day: obs.precip_30d_mm < mean_30d - 0.84 * std_30d
  flood day:   obs.precip_7d_mm  > mean_7d  + 1.28 * std_7d
L1 (optional): --impact-labels PATH loads impact_labels JSON (e.g.
  impact_labels_java_v1.json). When a zone-day has an L1 drought/flood
  event, that overrides the proxy flags for that day only. Days without
  L1 coverage keep the proxy. This is the documented production path
  toward BNPB/provincial catalogues without inventing live APIs.

PRODUCT EMIT (optional)
-----------------------
--emit-product-alerts: after each compute_risk_score, call
product_alert_service.emit_product_alert (idempotent, WARNING+ or
hazard gate). Trusted backend path only; uses LocalTransport by default.

SUGGESTED LIVE DEMO WINDOWS (historically documented events):
  * 2023 El Nino + positive-IOD dry season, Java:
      --zones karawang_rice,indramayu_rice --start 2023-07-01 --end 2023-11-30
  * 2020-21 La Nina wet season (Jan 2021 Java floods):
      --zones karawang_rice --start 2020-12-01 --end 2021-02-28
"""

from __future__ import annotations

import argparse
import json
import logging
import sys
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional, Sequence, Tuple

import zone_observation as _zo

assert _zo.SCHEMA_VERSION == 3, (
    f"backtest_indonesia: zone_observation schema mismatch "
    f"(expected 3, got {_zo.SCHEMA_VERSION})"
)

from zone_observation import (
    AlertLevel,
    DataSource,
    ForecastConfig,
    ZoneObs,
    make_synthetic_zone_obs,
)

from climatology import (
    ZoneClimatology,
    apply_climatology_anomalies,
    get_zone_climatology,
)
from crop_risk_scorer import compute_risk_score
from indonesia_zones import INDONESIA_ZONES, get_zone, register_indonesia_zones

logger = logging.getLogger(__name__)

_ALERT_POSITIVE = (AlertLevel.ADVISORY, AlertLevel.WARNING, AlertLevel.CRITICAL)

# Ground-truth percentile cut-points (Gaussian approx, see module docstring).
_DROUGHT_Z = 0.84   # ~20th percentile of the 30-day window aggregate
_FLOOD_Z = 1.28     # ~90th percentile of the 7-day window aggregate

# Event runs separated by a single quiet day are merged (monsoon hazards
# are persistent; a 1-day lull is not a new event).
_RUN_MERGE_GAP_DAYS = 1
# An alert this many days before an event run's start counts as early warning.
_LEAD_WINDOW_DAYS = 21


# ---------------------------------------------------------------------------
# Per-day record + metrics
# ---------------------------------------------------------------------------

@dataclass
class DayRecord:
    date: str                 # ISO date
    zone_id: str
    alert: bool
    alert_level: str
    drought_risk: float
    flood_risk: float
    event_drought: bool
    event_flood: bool
    precip_30d_mm: float
    precip_anomaly_idx: float
    source: str
    gt_source: str = "proxy"          # "proxy" | "l1" | "proxy+l1_miss"
    product_emit: str = "skipped"     # outcome_code or skipped/disabled

    @property
    def event(self) -> bool:
        return self.event_drought or self.event_flood


@dataclass
class BacktestMetrics:
    """Day-level confusion + event-level detection/lead time."""
    n_days: int = 0
    tp: int = 0
    fp: int = 0
    fn: int = 0
    tn: int = 0
    n_event_runs: int = 0
    n_runs_detected: int = 0
    mean_lead_days: Optional[float] = None
    drought_recall: Optional[float] = None
    flood_recall: Optional[float] = None

    @property
    def precision(self) -> Optional[float]:
        return self.tp / (self.tp + self.fp) if (self.tp + self.fp) else None

    @property
    def recall(self) -> Optional[float]:
        return self.tp / (self.tp + self.fn) if (self.tp + self.fn) else None

    @property
    def f1(self) -> Optional[float]:
        p, r = self.precision, self.recall
        if p is None or r is None or (p + r) == 0:
            return None
        return 2 * p * r / (p + r)

    @property
    def event_detection_rate(self) -> Optional[float]:
        return (self.n_runs_detected / self.n_event_runs
                if self.n_event_runs else None)

    def to_dict(self) -> Dict[str, Any]:
        d = asdict(self)
        d["precision"] = self.precision
        d["recall"] = self.recall
        d["f1"] = self.f1
        d["event_detection_rate"] = self.event_detection_rate
        return d


def compute_metrics(records: Sequence[DayRecord]) -> BacktestMetrics:
    """Day-level confusion + event-run detection with lead time.

    Day-level:  TP = alert on an event day, FP = alert on a non-event day,
                FN = missed event day, TN = quiet day correctly quiet.
    Run-level:  contiguous event days (merged across gaps <=
                _RUN_MERGE_GAP_DAYS) form one event run. A run is DETECTED
                if any alert fires inside the run or in the
                _LEAD_WINDOW_DAYS before its first day. Lead time = days
                from that first qualifying alert to the run start (0 for
                alerts landing on day 1 of the run).
    """
    m = BacktestMetrics(n_days=len(records))
    for r in records:
        if r.event and r.alert:
            m.tp += 1
        elif r.event and not r.alert:
            m.fn += 1
        elif not r.event and r.alert:
            m.fp += 1
        else:
            m.tn += 1

    drought_days = [r for r in records if r.event_drought]
    flood_days = [r for r in records if r.event_flood]
    if drought_days:
        m.drought_recall = sum(1 for r in drought_days if r.alert) / len(drought_days)
    if flood_days:
        m.flood_recall = sum(1 for r in flood_days if r.alert) / len(flood_days)

    # --- Build event runs ---
    runs: List[Tuple[int, int]] = []  # (start_idx, end_idx) inclusive
    i = 0
    n = len(records)
    while i < n:
        if records[i].event:
            j = i
            while j + 1 < n and (
                records[j + 1].event
                or (j + 2 < n and records[j + 2].event)  # peek over 1 gap day
            ):
                if records[j + 1].event:
                    j += 1
                elif j + 2 < n and records[j + 2].event and (j + 2) - (j + 1) <= _RUN_MERGE_GAP_DAYS:
                    j += 2
                else:
                    break
            runs.append((i, j))
            i = j + 1
        else:
            i += 1

    m.n_event_runs = len(runs)
    leads: List[int] = []
    dates = [datetime.fromisoformat(r.date) for r in records]
    for (s, e) in runs:
        first_alert_idx: Optional[int] = None
        for k in range(max(0, s - _LEAD_WINDOW_DAYS), e + 1):
            if records[k].alert:
                first_alert_idx = k
                break
        if first_alert_idx is not None:
            m.n_runs_detected += 1
            leads.append(max(0, (dates[s] - dates[first_alert_idx]).days))
    if leads:
        m.mean_lead_days = sum(leads) / len(leads)
    return m


def compute_product_l1_metrics(records: Sequence[Dict[str, Any]]) -> Dict[str, Any]:
    """
    Primary product metrics: EMITTED vs L1 event days (not ADVISORY vs proxy).

    Uses record dicts (as written to result JSON). A day is:
      product_positive = product_emit == 'EMITTED'
      l1_event = gt_source == 'l1' and (event_drought or event_flood)
    """
    n = len(records)
    tp = fp = fn = tn = 0
    n_l1 = 0
    n_emitted = 0
    for r in records:
        emitted = r.get("product_emit") == "EMITTED"
        l1 = r.get("gt_source") == "l1" and (
            r.get("event_drought") or r.get("event_flood")
        )
        if l1:
            n_l1 += 1
        if emitted:
            n_emitted += 1
        if l1 and emitted:
            tp += 1
        elif l1 and not emitted:
            fn += 1
        elif not l1 and emitted:
            fp += 1
        else:
            tn += 1

    def _div(a: int, b: int) -> Optional[float]:
        return a / b if b else None

    p = _div(tp, tp + fp)
    r = _div(tp, tp + fn)
    f1 = (2 * p * r / (p + r)) if (p and r and (p + r)) else None
    return {
        "n_days": n,
        "n_l1_event_days": n_l1,
        "n_emitted": n_emitted,
        "tp": tp,
        "fp": fp,
        "fn": fn,
        "tn": tn,
        "precision": p,
        "recall": r,
        "f1": f1,
        "emit_rate": _div(n_emitted, n),
        "l1_coverage": _div(n_l1, n),
    }


# ---------------------------------------------------------------------------
# Replay engine
# ---------------------------------------------------------------------------

def _classify_events(obs: ZoneObs, clim: ZoneClimatology) -> Tuple[bool, bool]:
    """Proxy ground truth from the pinned climatology (see module docstring)."""
    event_drought = False
    event_flood = False
    if obs.precip_30d_mm > 0.0:
        mean30, std30 = clim.window_precip_stats(obs.valid_time, 30)
        event_drought = obs.precip_30d_mm < (mean30 - _DROUGHT_Z * std30)
    if obs.precip_7d_mm > 0.0:
        mean7, std7 = clim.window_precip_stats(obs.valid_time, 7)
        event_flood = obs.precip_7d_mm > (mean7 + _FLOOD_Z * std7)
    return event_drought, event_flood


def replay_zone(
    zone_id: str,
    start: datetime,
    end: datetime,
    step_days: int = 3,
    mode: str = "synthetic",
    climatology_years: int = 10,
    planted_events: Optional[Dict[str, Tuple[datetime, datetime]]] = None,
    impact_store: Any = None,
    emit_product: bool = False,
    emission_ledger: Any = None,
    transport: Any = None,
) -> List[DayRecord]:
    """Replay one zone over [start, end] at `step_days` resolution.

    mode='live':      obs via era5_data_pipeline (Open-Meteo archive for
                      historical dates), forecast via baseline backend.
    mode='synthetic': deterministic synthetic obs; `planted_events` maps
                      'drought'/'flood' -> (start, end) blocks during which
                      the synthetic generator's event flag is forced on.
    impact_store:     optional ImpactLabelStore; L1 flags override proxy
                      when present for that zone-day.
    emit_product:     if True, call product_alert_service after each score
                      (requires emission_ledger + transport).
    """
    if start.tzinfo is None:
        start = start.replace(tzinfo=timezone.utc)
    if end.tzinfo is None:
        end = end.replace(tzinfo=timezone.utc)

    z = get_zone(zone_id)

    # Pinned climatology: strictly before the replay window (no look-ahead).
    clim = get_zone_climatology(
        zone_id, z.lat, z.lon,
        years=climatology_years,
        end_year=start.year - 1,
        prefer_real=(mode == "live"),
    )
    logger.info(
        "replay %s: climatology source=%s period=%d-%d",
        zone_id, clim.source, clim.period_start_year, clim.period_end_year,
    )

    if mode == "live":
        import era5_data_pipeline as edp
        from timesfm_wrapper import create_forecast_backend
        forecast_backend = create_forecast_backend(mode="baseline", horizon_days=14)
        cfg = ForecastConfig(force_data_source=DataSource.OPENMETEO_LIVE)

    records: List[DayRecord] = []
    vt = start
    while vt <= end:
        if mode == "live":
            dr = (vt - timedelta(days=35), vt)
            obs = edp.fetch_zone_obs(zone_id, dr, cfg)
            forecast = forecast_backend.forecast(obs)
        else:
            flag: Dict[str, bool] = {}
            for hazard, blk in (planted_events or {}).items():
                if blk[0] <= vt <= blk[1]:
                    flag[hazard] = True
            obs = make_synthetic_zone_obs(
                zone_id,
                seed=_zo._stable_seed(f"{zone_id}|{vt.date().isoformat()}"),
                **flag,
            )
            # The factory draws its own random valid_time from the seed;
            # the replay clock is authoritative -- override via the
            # codebase's to_dict/from_dict idiom (never mutate).
            _d = obs.to_dict()
            _d.pop("_schema_version", None)
            _d["valid_time"] = vt.isoformat()
            obs = ZoneObs.from_dict(_d)
            from zone_observation import make_synthetic_forecast_result
            forecast = make_synthetic_forecast_result(
                zone_id, valid_time=vt,
                seed=_zo._stable_seed(f"f|{zone_id}|{vt.date().isoformat()}"),
                **flag,
            )

        obs = apply_climatology_anomalies(obs, clim)
        risk = compute_risk_score(obs, forecast, ForecastConfig())
        ev_drought, ev_flood = _classify_events(obs, clim)
        gt_source = "proxy"

        # L1 override: when store has a label for this zone-day, use it.
        if impact_store is not None:
            try:
                l1_d, l1_f = impact_store.labels_for_day(zone_id, vt.date())
                if l1_d or l1_f:
                    ev_drought, ev_flood = l1_d, l1_f
                    gt_source = "l1"
                else:
                    gt_source = "proxy+l1_miss"
            except Exception as e:
                logger.warning("impact_store lookup failed: %s", e)

        product_emit = "disabled"
        if emit_product and emission_ledger is not None and transport is not None:
            try:
                import asyncio
                from product_alert_service import emit_product_alert

                async def _one():
                    return await emit_product_alert(
                        transport, risk,
                        ledger=emission_ledger,
                        valid_time=vt,
                    )

                try:
                    loop = asyncio.get_running_loop()
                except RuntimeError:
                    loop = None
                if loop and loop.is_running():
                    # Nested running loop (e.g. notebook): schedule carefully
                    import concurrent.futures
                    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                        product_emit = pool.submit(lambda: asyncio.run(_one())).result().outcome_code
                else:
                    product_emit = asyncio.run(_one()).outcome_code
            except Exception as e:
                logger.warning("product emit failed: %s", e)
                product_emit = "TRANSPORT_FAILED"

        records.append(DayRecord(
            date=vt.date().isoformat(),
            zone_id=zone_id,
            alert=risk.alert_level in _ALERT_POSITIVE,
            alert_level=risk.alert_level.value,
            drought_risk=round(risk.drought_risk, 4),
            flood_risk=round(risk.flood_risk, 4),
            event_drought=ev_drought,
            event_flood=ev_flood,
            precip_30d_mm=round(obs.precip_30d_mm, 1),
            precip_anomaly_idx=round(obs.precip_anomaly_idx, 3),
            source=obs.source.value,
            gt_source=gt_source,
            product_emit=product_emit,
        ))
        vt += timedelta(days=step_days)
    return records


def run_backtest(
    zone_ids: Sequence[str],
    start: datetime,
    end: datetime,
    step_days: int = 3,
    mode: str = "synthetic",
    climatology_years: int = 10,
    planted_events: Optional[Dict[str, Tuple[datetime, datetime]]] = None,
    impact_labels_path: Optional[str] = None,
    emit_product_alerts: bool = False,
) -> Dict[str, Any]:
    """Replay several zones; return per-zone + overall metrics and records."""
    if mode == "live":
        register_indonesia_zones()

    impact_store = None
    if impact_labels_path:
        from impact_labels import load_impact_events
        loaded = load_impact_events(impact_labels_path)
        if not loaded.success:
            raise RuntimeError(
                f"impact labels load failed: {loaded.outcome_code} {loaded.data}"
            )
        impact_store = loaded.data["store"]
        logger.info(
            "L1 impact labels: %s events_loaded=%s",
            loaded.outcome_code, loaded.data.get("events_loaded"),
        )

    emission_ledger = None
    transport = None
    if emit_product_alerts:
        from product_alert_service import EmissionLedger
        from node_transport import LocalTransport
        emission_ledger = EmissionLedger()
        transport = LocalTransport()

    all_records: List[DayRecord] = []
    per_zone: Dict[str, Any] = {}
    for zid in zone_ids:
        recs = replay_zone(
            zid, start, end, step_days=step_days, mode=mode,
            climatology_years=climatology_years,
            planted_events=planted_events,
            impact_store=impact_store,
            emit_product=emit_product_alerts,
            emission_ledger=emission_ledger,
            transport=transport,
        )
        all_records.extend(recs)
        per_zone[zid] = {
            "metrics": compute_metrics(recs).to_dict(),
            "n_records": len(recs),
            "sources": sorted({r.source for r in recs}),
        }
        logger.info("zone %s: %s", zid, per_zone[zid]["metrics"])

    overall = compute_metrics(all_records)
    return {
        "mode": mode,
        "window": [start.date().isoformat(), end.date().isoformat()],
        "step_days": step_days,
        "climatology_years": climatology_years,
        "overall": overall.to_dict(),
        "per_zone": per_zone,
        "records": [asdict(r) for r in all_records],
    }


def _print_report(result: Dict[str, Any]) -> None:
    o = result["overall"]

    def _f(x: Optional[float]) -> str:
        return f"{x:.3f}" if isinstance(x, float) else "  -  "

    print(f"\n=== backtest [{result['mode']}] "
          f"{result['window'][0]} -> {result['window'][1]} "
          f"(step {result['step_days']}d) ===")
    print(f"days={o['n_days']}  TP={o['tp']} FP={o['fp']} FN={o['fn']} TN={o['tn']}")
    print(f"precision={_f(o['precision'])}  recall={_f(o['recall'])}  f1={_f(o['f1'])}")
    print(f"event runs: {o['n_runs_detected']}/{o['n_event_runs']} detected "
          f"(rate={_f(o['event_detection_rate'])})  "
          f"mean lead={_f(o['mean_lead_days'])}d")
    print(f"drought recall={_f(o['drought_recall'])}  "
          f"flood recall={_f(o['flood_recall'])}")
    for zid, zr in result["per_zone"].items():
        zm = zr["metrics"]
        print(f"  {zid:24s} n={zr['n_records']:3d} src={','.join(zr['sources'])} "
              f"P={_f(zm['precision'])} R={_f(zm['recall'])} "
              f"runs={zm['n_runs_detected']}/{zm['n_event_runs']}")


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Backtest the Indonesia weather-risk stack over a "
                    "historical window (default invocation with NO arguments "
                    "runs the offline self-test instead).",
    )
    p.add_argument("--mode", choices=["synthetic", "live"], default="live",
                   help="'live' = real Open-Meteo archive data (network); "
                        "'synthetic' = deterministic offline replay.")
    p.add_argument("--zones", default="karawang_rice",
                   help="Comma-separated zone ids from indonesia_zones.")
    p.add_argument("--start", required=True, help="YYYY-MM-DD")
    p.add_argument("--end", required=True, help="YYYY-MM-DD")
    p.add_argument("--step-days", type=int, default=3)
    p.add_argument("--climatology-years", type=int, default=10)
    p.add_argument("--out", default=None, help="Optional JSON output path.")
    p.add_argument(
        "--impact-labels", default=None,
        help="Path to impact_labels JSON (L1). Overrides proxy GT when "
             "zone-day is labeled.",
    )
    p.add_argument(
        "--emit-product-alerts", action="store_true",
        help="After each score, idempotently emit product alerts via "
             "product_alert_service (LocalTransport).",
    )
    return p.parse_args(argv)


def _main(argv: Optional[List[str]] = None) -> int:
    args = _parse_args(argv)
    logging.basicConfig(level=logging.INFO,
                        format="%(levelname)s %(name)s: %(message)s")
    start = datetime.fromisoformat(args.start).replace(tzinfo=timezone.utc)
    end = datetime.fromisoformat(args.end).replace(tzinfo=timezone.utc)
    result = run_backtest(
        zone_ids=[z.strip() for z in args.zones.split(",") if z.strip()],
        start=start, end=end,
        step_days=args.step_days,
        mode=args.mode,
        climatology_years=args.climatology_years,
        impact_labels_path=args.impact_labels,
        emit_product_alerts=args.emit_product_alerts,
    )
    _print_report(result)
    # Product emit + L1 primary metrics
    recs = result.get("records") or []
    if args.emit_product_alerts and recs:
        from collections import Counter
        c = Counter(r.get("product_emit", "skipped") for r in recs)
        print("\nproduct_emit counts:", dict(c))
    if args.impact_labels and recs:
        from collections import Counter
        c = Counter(r.get("gt_source", "proxy") for r in recs)
        print("gt_source counts:", dict(c))
    if args.emit_product_alerts and args.impact_labels and recs:
        pm = compute_product_l1_metrics(recs)
        result["product_l1_metrics"] = pm

        def _f(x: Optional[float]) -> str:
            return f"{x:.3f}" if x is not None else "  -  "

        print("\n--- product vs L1 (primary product skill) ---")
        print(
            f"n={pm['n_days']}  l1_days={pm['n_l1_event_days']}  "
            f"emitted={pm['n_emitted']}  "
            f"emit_rate={_f(pm['emit_rate'])}  l1_coverage={_f(pm['l1_coverage'])}"
        )
        print(
            f"TP={pm['tp']} FP={pm['fp']} FN={pm['fn']} TN={pm['tn']}  "
            f"P={_f(pm['precision'])} R={_f(pm['recall'])} F1={_f(pm['f1'])}"
        )
    if args.out:
        with open(args.out, "w") as f:
            json.dump(result, f, indent=2)
        print(f"\nWrote {args.out}")
    return 0


# ---------------------------------------------------------------------------
# Self-test (python backtest_indonesia.py) -- fully offline
# ---------------------------------------------------------------------------

def _self_test() -> int:
    logging.basicConfig(level=logging.WARNING)
    print("backtest_indonesia.py self-test (offline, synthetic)\n")
    failures: List[str] = []

    def _assert(cond: bool, msg: str) -> None:
        if not cond:
            failures.append(msg)
            print(f"  FAIL: {msg}")

    def _rec(day: int, alert: bool = False, event: bool = False,
             drought: bool = False, flood: bool = False) -> DayRecord:
        d = (datetime(2024, 1, 1, tzinfo=timezone.utc) + timedelta(days=day))
        return DayRecord(
            date=d.date().isoformat(), zone_id="t",
            alert=alert, alert_level="advisory" if alert else "none",
            drought_risk=0.0, flood_risk=0.0,
            event_drought=(drought or (event and not flood)),
            event_flood=flood,
            precip_30d_mm=0.0, precip_anomaly_idx=0.0, source="synthetic",
        )

    # 1. Metrics on a fabricated series with known expected values.
    #    alerts: 5,6,7 (early), 25 (isolated), 40,41 (inside run 2)
    #    events: 10-14 (run 1), 40-42 (run 2)
    recs: List[DayRecord] = []
    for day in range(50):
        alert = day in (5, 6, 7, 25, 40, 41)
        event = (10 <= day <= 14) or (40 <= day <= 42)
        recs.append(_rec(day, alert=alert, event=event, drought=event))
    m = compute_metrics(recs)
    _assert(m.tp == 2 and m.fn == 6 and m.fp == 4 and m.tn == 38,
            f"confusion wrong: tp={m.tp} fn={m.fn} fp={m.fp} tn={m.tn}")
    _assert(abs((m.precision or 0) - 2 / 6) < 1e-9, f"precision {m.precision}")
    _assert(abs((m.recall or 0) - 2 / 8) < 1e-9, f"recall {m.recall}")
    _assert(m.n_event_runs == 2, f"runs={m.n_event_runs}")
    _assert(m.n_runs_detected == 2, f"detected={m.n_runs_detected}")
    _assert(abs((m.mean_lead_days or 0) - 10.0) < 1e-9,
            f"mean lead {m.mean_lead_days} (expected 10: 5d + 15d)")
    print(f"  Metrics OK: P={m.precision:.3f} R={m.recall:.3f} "
          f"runs {m.n_runs_detected}/{m.n_event_runs} lead={m.mean_lead_days}d")

    # 2. Run merging across a 1-day lull: events 10,11,13 = ONE run.
    recs2 = [_rec(day, event=day in (10, 11, 13), drought=True) for day in range(30)]
    m2 = compute_metrics(recs2)
    _assert(m2.n_event_runs == 1, f"1-day lull should merge: runs={m2.n_event_runs}")
    print(f"  Run-merge OK (runs={m2.n_event_runs})")

    # 3. Event classifier: planted drought reads as drought event vs climatology
    clim = get_zone_climatology("karawang_rice", -6.30, 107.30, years=5,
                                end_year=2020, prefer_real=False)
    vt = datetime(2021, 8, 15, tzinfo=timezone.utc)

    def _obs_at(flag: str, seed: int) -> ZoneObs:
        o = make_synthetic_zone_obs("karawang_rice", seed=seed, **{flag: True})
        _d = o.to_dict()
        _d.pop("_schema_version", None)
        _d["valid_time"] = vt.isoformat()
        return ZoneObs.from_dict(_d)

    dry_obs = _obs_at("drought", 11)
    ev_d, ev_f = _classify_events(dry_obs, clim)
    _assert(ev_d and not ev_f, f"planted drought misclassified: d={ev_d} f={ev_f}")
    wet_obs = _obs_at("flood", 12)
    ev_d2, ev_f2 = _classify_events(wet_obs, clim)
    _assert(ev_f2 and not ev_d2, f"planted flood misclassified: d={ev_d2} f={ev_f2}")
    print("  Event classifier OK (drought/flood classified correctly)")

    # 4. Synthetic end-to-end: 5-month replay with a planted drought block.
    start = datetime(2021, 6, 1, tzinfo=timezone.utc)
    end = datetime(2021, 10, 31, tzinfo=timezone.utc)
    planted = {"drought": (datetime(2021, 8, 1, tzinfo=timezone.utc),
                           datetime(2021, 9, 10, tzinfo=timezone.utc))}
    result = run_backtest(["karawang_rice"], start, end, step_days=5,
                          mode="synthetic", climatology_years=5,
                          planted_events=planted)
    o = result["overall"]
    _assert(o["n_days"] > 25, f"too few replay days: {o['n_days']}")
    _assert((o["drought_recall"] or 0) >= 0.9,
            f"planted drought should be caught: drought_recall={o['drought_recall']}")
    _assert(o["n_event_runs"] >= 1, "no event runs found")
    _assert(o["n_runs_detected"] >= 1, "planted drought run not detected")
    json.dumps(result)  # whole result must be JSON-serialisable
    print(f"  End-to-end OK: drought_recall={o['drought_recall']:.2f} "
          f"runs={o['n_runs_detected']}/{o['n_event_runs']} "
          f"P={o['precision'] if o['precision'] is not None else float('nan'):.2f}")
    _print_report(result)

    print()
    if failures:
        print(f"FAILED  {len(failures)} test(s):")
        for f in failures:
            print(f"  - {f}")
        return 1
    print("All 4 test groups passed.")
    return 0


if __name__ == "__main__":
    if len(sys.argv) > 1:
        sys.exit(_main())
    else:
        sys.exit(_self_test())