PolyWeather/web/services/ops/health.py

"""Ops health / source health / training accuracy service functions."""
from __future__ import annotations

import os
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional

from fastapi import Request
import requests as _requests

from src.utils.runtime_secrets import get_runtime_secret
from web.services.observation_freshness import (
    build_observation_freshness,
    canonical_observation_source_code,
)
import web.routes as legacy_routes

_DEB_VERSION_BACKTEST_SAMPLE_LIMIT = 400


# ── tiny helpers (also in ops_api.py, copied for self-contained module) ──────

def _sf(value: Any) -> Optional[float]:
    try:
        if value is None or value == "":
            return None
        return float(value)
    except (TypeError, ValueError):
        return None


def _round_metric(value: Optional[float], digits: int = 1) -> Optional[float]:
    return None if value is None else round(float(value), digits)


# ── AMSC AWOS ────────────────────────────────────────────────────────

def _build_amsc_awos_headers() -> dict[str, str]:
    headers = {
        "Accept": "application/json, text/plain, */*",
        "Referer": os.getenv("AMSC_AWOS_REFERER", "https://www.amsc.net.cn/"),
        "User-Agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36"
        ),
    }
    cookie = get_runtime_secret("POLYWEATHER_AMSC_COOKIE")
    session_id = get_runtime_secret("POLYWEATHER_AMSC_SESSION_ID")
    if session_id:
        headers["sessionId"] = session_id
        headers["app"] = "AMS"
    elif cookie:
        headers["Cookie"] = cookie
    return headers


def _check_amsc_awos_health(timeout: int = 8) -> dict[str, Any]:
    import time as _time

    from src.data_collection.amsc_awos_sources import _amsc_parse_wind_plate_payload

    amsc_base = str(os.getenv("AMSC_AWOS_BASE_URL") or "").strip()
    if not amsc_base:
        return {"ok": False, "error": "not configured"}

    credential_configured = bool(
        get_runtime_secret("POLYWEATHER_AMSC_COOKIE")
        or get_runtime_secret("POLYWEATHER_AMSC_SESSION_ID")
    )
    try:
        t0 = _time.perf_counter()
        response = _requests.get(
            f"{amsc_base}?cccc=ZSPD",
            timeout=timeout,
            verify=False,
            headers=_build_amsc_awos_headers(),
        )
        latency_ms = round((_time.perf_counter() - t0) * 1000)
        try:
            payload = response.json() if response.content else {}
        except ValueError:
            payload = {}
        parsed = _amsc_parse_wind_plate_payload(
            payload if isinstance(payload, dict) else {},
            city_key="shanghai",
            icao="ZSPD",
        )
        points = (
            ((parsed or {}).get("runway_obs") or {}).get("point_temperatures")
            if isinstance(parsed, dict)
            else []
        )
        point_count = len(points or [])
        ok = bool(response.ok and parsed and point_count > 0)
        result: dict[str, Any] = {
            "ok": ok,
            "status": response.status_code,
            "latency_ms": latency_ms,
            "credential_configured": credential_configured,
            "points": point_count,
        }
        if isinstance(parsed, dict):
            result["sample_city"] = "shanghai"
            result["observation_time_local"] = parsed.get("observation_time_local")
        if not ok:
            result["error"] = "empty_or_unauthorized_response"
        return result
    except Exception as exc:
        return {
            "ok": False,
            "credential_configured": credential_configured,
            "error": str(exc)[:100],
        }


# ── Source health helpers ─────────────────────────────────────────────

def _safe_source_text(value: Any) -> str:
    return str(value or "").strip()


def _source_observed_at(source: dict[str, Any]) -> Any:
    for key in (
        "observed_at",
        "obs_time",
        "report_time",
        "observation_time",
        "observation_time_local",
        "time",
        "timestamp",
    ):
        value = source.get(key)
        if _safe_source_text(value):
            return value
    return None


def _source_code(source: dict[str, Any], fallback: str = "") -> str:
    for key in ("source_code", "source", "provider_code", "network_provider"):
        value = _safe_source_text(source.get(key))
        if value:
            return canonical_observation_source_code(value)
    return canonical_observation_source_code(fallback)


def _source_label(source: dict[str, Any], code: str, fallback: str = "") -> str:
    for key in ("source_label", "station_label", "station_name", "label", "provider_label"):
        value = _safe_source_text(source.get(key))
        if value:
            return value
    return fallback or code.upper()


def _source_health_entry(
    *,
    city: str,
    role: str,
    source: dict[str, Any],
    fallback_code: str = "",
    fallback_label: str = "",
) -> dict[str, Any] | None:
    if not isinstance(source, dict) or not source:
        return None

    code = _source_code(source, fallback_code)
    label = _source_label(source, code, fallback_label)
    observed_at = _source_observed_at(source)
    freshness = source.get("freshness") if isinstance(source.get("freshness"), dict) else None
    if freshness:
        status = _safe_source_text(freshness.get("freshness_status")) or "unknown"
        age_sec = freshness.get("age_sec")
        observed_at = freshness.get("observed_at") or freshness.get("observed_at_local") or observed_at
        expected_next = freshness.get("expected_next_update_at")
        reason = freshness.get("freshness_reason")
    else:
        age_min = source.get("obs_age_min")
        try:
            age_min_int = int(age_min) if age_min is not None else None
        except Exception:
            age_min_int = None
        freshness = build_observation_freshness(
            source_code=code,
            source_label=label,
            observed_at=observed_at,
            age_min=age_min_int,
        )
        status = str(freshness.get("freshness_status") or "unknown")
        age_sec = freshness.get("age_sec")
        expected_next = freshness.get("expected_next_update_at")
        reason = freshness.get("freshness_reason")

    return {
        "city": city,
        "role": role,
        "source_code": code,
        "source_label": label,
        "station_code": source.get("station_code") or source.get("icao"),
        "station_label": source.get("station_label") or source.get("station_name"),
        "status": status,
        "reason": reason,
        "age_sec": age_sec,
        "age_min": round(float(age_sec) / 60, 1) if isinstance(age_sec, (int, float)) else None,
        "observed_at": observed_at,
        "expected_next_update_at": expected_next,
        "temp": source.get("temp"),
    }


def _expected_city_source_codes(city: str, payload: dict[str, Any]) -> list[str]:
    city_key = str(city or "").strip().lower().replace(" ", "")
    expected: list[str] = []
    if city_key in {"ankara", "istanbul"}:
        expected.append("mgm")
    if city_key == "amsterdam":
        expected.append("knmi")
    if city_key in {"telaviv", "telavivyafo"}:
        expected.append("ims")
    provider = canonical_observation_source_code(payload.get("official_network_source"))
    if provider and provider not in {"metar", "none"}:
        expected.append(provider)
    return list(dict.fromkeys(expected))


def _collect_city_source_health(city: str, entry: dict[str, Any] | None) -> dict[str, Any]:
    payload = (entry or {}).get("payload") if isinstance(entry, dict) else {}
    if not isinstance(payload, dict):
        payload = {}

    sources: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()

    def add(role: str, source: Any, fallback_code: str = "", fallback_label: str = "") -> None:
        item = _source_health_entry(
            city=city,
            role=role,
            source=source if isinstance(source, dict) else {},
            fallback_code=fallback_code,
            fallback_label=fallback_label,
        )
        if not item:
            return
        key = (str(item.get("role")), str(item.get("source_code")))
        if key in seen:
            return
        seen.add(key)
        sources.append(item)

    add("settlement", payload.get("current"), fallback_label="Settlement")
    add("airport_metar", payload.get("airport_current"), fallback_code="metar", fallback_label="METAR")
    add("airport_primary", payload.get("airport_primary"), fallback_label="Airport station")

    official = payload.get("official") if isinstance(payload.get("official"), dict) else {}
    add("official_airport_primary", official.get("airport_primary"), fallback_label="Official airport station")
    add("official_airport_primary", official.get("airport_primary_current"), fallback_label="Official airport station")

    mgm = payload.get("mgm") if isinstance(payload.get("mgm"), dict) else {}
    if mgm:
        mgm_current = mgm.get("current") if isinstance(mgm.get("current"), dict) else {}
        add(
            "official_network",
            {
                **mgm_current,
                "source_code": "mgm",
                "source_label": "MGM",
                "obs_time": mgm.get("obs_time") or mgm_current.get("time"),
            },
            fallback_code="mgm",
            fallback_label="MGM",
        )

    for nearby in payload.get("official_nearby") or payload.get("mgm_nearby") or []:
        if isinstance(nearby, dict):
            add("nearby_official", nearby, fallback_label="Nearby official")

    expected_codes = _expected_city_source_codes(city, payload)
    present_codes = {str(item.get("source_code") or "") for item in sources}
    for code in expected_codes:
        if code and code not in present_codes:
            sources.append(
                {
                    "city": city,
                    "role": "expected_source",
                    "source_code": code,
                    "source_label": code.upper(),
                    "status": "missing",
                    "reason": "expected_source_not_present_in_cached_detail",
                    "age_sec": None,
                    "age_min": None,
                    "observed_at": None,
                    "expected_next_update_at": None,
                    "temp": None,
                }
            )

    priority = {"stale": 4, "missing": 4, "delayed": 3, "unknown": 2, "expected_wait": 1, "fresh": 0}
    worst = max(sources, key=lambda item: priority.get(str(item.get("status") or ""), 2), default=None)
    full_age_sec = (
        round(max(0.0, __import__("time").time() - float(entry.get("updated_at_ts") or 0.0)), 1)
        if entry
        else None
    )
    return {
        "city": city,
        "cache_exists": bool(entry),
        "cache_updated_at": entry.get("updated_at") if entry else None,
        "cache_age_sec": full_age_sec,
        "source_count": len(sources),
        "worst_status": str(worst.get("status") if worst else "missing"),
        "sources": sources,
    }


# ── Ops endpoint functions ────────────────────────────────────────────

def get_ops_source_health(
    request: Request,
    cities: str = "",
    limit: int = 80,
) -> dict[str, Any]:
    from web.services.ops_api import _require_ops

    _require_ops(request)
    requested = legacy_routes._normalize_city_list(cities) if cities else []
    if requested:
        selected = requested
    else:
        all_cities = getattr(legacy_routes, "CITIES", {})
        selected = list(all_cities.keys()) if isinstance(all_cities, dict) else []
    safe_limit = max(1, min(int(limit or 80), 200))
    rows = []
    for city in selected[:safe_limit]:
        entry = legacy_routes._CACHE_DB.get_city_cache("full", city)
        if not entry:
            entry = legacy_routes._CACHE_DB.get_city_cache("panel", city)
        rows.append(_collect_city_source_health(city, entry))

    status_counts: dict[str, int] = {}
    for row in rows:
        for source in row.get("sources") or []:
            status = str(source.get("status") or "unknown")
            status_counts[status] = status_counts.get(status, 0) + 1

    return {
        "checked_at": __import__("datetime").datetime.utcnow().isoformat() + "Z",
        "cities": rows,
        "status_counts": status_counts,
        "total_cities": len(rows),
    }


def get_ops_observation_collector_status(
    request: Request,
    limit: int = 200,
) -> dict[str, Any]:
    from web.services.ops_api import _require_ops, ObservationCollectorStatusRepository

    _require_ops(request)
    safe_limit = max(1, min(int(limit or 200), 500))
    return ObservationCollectorStatusRepository().load_snapshot(limit=safe_limit)


def get_ops_health_check(request: Request) -> dict[str, Any]:
    from web.services.ops_api import _require_ops

    _require_ops(request)
    import os as _os
    import requests as _r
    import time as _time
    import urllib3 as _urllib3

    _urllib3.disable_warnings(_urllib3.exceptions.InsecureRequestWarning)

    results: dict[str, dict] = {}
    timeout = 8

    # Supabase
    supabase_url = str(_os.getenv("SUPABASE_URL") or "").strip().rstrip("/")
    supabase_key = str(_os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip()
    if supabase_url and supabase_key:
        try:
            r = _r.get(
                f"{supabase_url}/rest/v1/",
                headers={
                    "apikey": supabase_key,
                    "Authorization": f"Bearer {supabase_key}",
                },
                timeout=timeout,
            )
            results["supabase"] = {
                "ok": r.ok,
                "status": r.status_code,
                "latency_ms": round(r.elapsed.total_seconds() * 1000),
            }
        except Exception as e:
            results["supabase"] = {"ok": False, "error": str(e)[:100]}
    else:
        results["supabase"] = {"ok": False, "error": "not configured"}

    # Open-Meteo
    try:
        t0 = _time.perf_counter()
        r = _r.get(
            "https://api.open-meteo.com/v1/forecast?latitude=52.52&longitude=13.41&daily=temperature_2m_max&timezone=auto&forecast_days=1",
            timeout=timeout,
        )
        results["open_meteo"] = {
            "ok": r.ok,
            "status": r.status_code,
            "latency_ms": round((_time.perf_counter() - t0) * 1000),
        }
    except Exception as e:
        results["open_meteo"] = {"ok": False, "error": str(e)[:100]}

    # METAR (aviationweather)
    try:
        t0 = _time.perf_counter()
        r = _r.get(
            "https://aviationweather.gov/api/data/metar?ids=KJFK&format=json",
            timeout=timeout,
        )
        results["metar"] = {
            "ok": r.ok,
            "status": r.status_code,
            "latency_ms": round((_time.perf_counter() - t0) * 1000),
        }
    except Exception as e:
        results["metar"] = {"ok": False, "error": str(e)[:100]}

    # KNMI
    knmi_key = str(_os.getenv("KNMI_API_KEY") or "").strip()
    if knmi_key:
        try:
            t0 = _time.perf_counter()
            r = _r.get(
                "https://api.dataplatform.knmi.nl/open-data/v1/datasets/10-minute-in-situ-meteorological-observations/versions/1.0/files?maxKeys=1",
                headers={"Authorization": knmi_key},
                timeout=timeout,
            )
            results["knmi"] = {
                "ok": r.ok,
                "status": r.status_code,
                "latency_ms": round((_time.perf_counter() - t0) * 1000),
            }
        except Exception as e:
            results["knmi"] = {"ok": False, "error": str(e)[:100]}
    else:
        results["knmi"] = {"ok": False, "error": "not configured"}

    # MADIS (NOAA)
    try:
        t0 = _time.perf_counter()
        r = _r.head(
            "https://madis-data.ncep.noaa.gov/madisPublic1/data/LDAD/hfmetar/netCDF/",
            timeout=timeout,
            allow_redirects=True,
        )
        results["madis"] = {
            "ok": r.ok,
            "status": r.status_code,
            "latency_ms": round((_time.perf_counter() - t0) * 1000),
        }
    except Exception as e:
        results["madis"] = {"ok": False, "error": str(e)[:100]}

    # Telegram Bot
    bot_token = str(_os.getenv("TELEGRAM_BOT_TOKEN") or "").strip()
    if bot_token:
        try:
            t0 = _time.perf_counter()
            r = _r.get(
                f"https://api.telegram.org/bot{bot_token}/getMe", timeout=timeout
            )
            results["telegram"] = {
                "ok": r.ok,
                "status": r.status_code,
                "latency_ms": round((_time.perf_counter() - t0) * 1000),
            }
        except Exception as e:
            results["telegram"] = {"ok": False, "error": str(e)[:100]}
    else:
        results["telegram"] = {"ok": False, "error": "not configured"}

    # JMA (Japan Meteorological Agency)
    try:
        t0 = _time.perf_counter()
        r = _r.get("https://www.jma.go.jp/bosai/forecast/", timeout=timeout)
        results["jma"] = {
            "ok": r.ok,
            "status": r.status_code,
            "latency_ms": round((_time.perf_counter() - t0) * 1000),
        }
    except Exception as e:
        results["jma"] = {"ok": False, "error": str(e)[:100]}

    # MGM (Turkish State Meteorological Service)
    try:
        t0 = _time.perf_counter()
        r = _r.get(
            "https://servis.mgm.gov.tr/web/sondurumlar?istno=17130&_=1",
            timeout=timeout,
            headers={"Origin": "https://www.mgm.gov.tr"},
        )
        results["mgm"] = {
            "ok": r.ok,
            "status": r.status_code,
            "latency_ms": round((_time.perf_counter() - t0) * 1000),
        }
    except Exception as e:
        results["mgm"] = {"ok": False, "error": str(e)[:100]}

    # FMI (Finnish Meteorological Institute)
    try:
        t0 = _time.perf_counter()
        r = _r.get(
            "https://opendata.fmi.fi/wfs?service=WFS&version=2.0.0&request=GetCapabilities",
            timeout=timeout,
        )
        results["fmi"] = {
            "ok": r.ok,
            "status": r.status_code,
            "latency_ms": round((_time.perf_counter() - t0) * 1000),
        }
    except Exception as e:
        results["fmi"] = {"ok": False, "error": str(e)[:100]}

    # KMA (Korea Meteorological Administration)
    try:
        t0 = _time.perf_counter()
        r = _r.get(
            "https://www.weather.go.kr/wgis-nuri/js/info/sfc.geojson", timeout=timeout
        )
        results["kma"] = {
            "ok": r.ok,
            "status": r.status_code,
            "latency_ms": round((_time.perf_counter() - t0) * 1000),
        }
    except Exception as e:
        results["kma"] = {"ok": False, "error": str(e)[:100]}

    # HKO (Hong Kong Observatory)
    try:
        t0 = _time.perf_counter()
        r = _r.get(
            "https://data.weather.gov.hk/weatherAPI/hko_data/regional-weather/latest_1min_temperature.csv",
            timeout=timeout,
        )
        results["hko"] = {
            "ok": r.ok,
            "status": r.status_code,
            "latency_ms": round((_time.perf_counter() - t0) * 1000),
        }
    except Exception as e:
        results["hko"] = {"ok": False, "error": str(e)[:100]}

    # Singapore MSS (data.gov.sg)
    try:
        t0 = _time.perf_counter()
        r = _r.get(
            "https://api.data.gov.sg/v1/environment/air-temperature", timeout=timeout
        )
        results["singapore_mss"] = {
            "ok": r.ok,
            "status": r.status_code,
            "latency_ms": round((_time.perf_counter() - t0) * 1000),
        }
    except Exception as e:
        results["singapore_mss"] = {"ok": False, "error": str(e)[:100]}

    # CWA (Taiwan Central Weather Administration)
    cwa_key = str(
        _os.getenv("CWA_API_KEY")
        or _os.getenv("CWA_OPEN_DATA_AUTH")
        or _os.getenv("CWA_OPEN_DATA_API_KEY")
        or ""
    ).strip()
    if cwa_key:
        try:
            t0 = _time.perf_counter()
            r = _r.get(
                f"https://opendata.cwa.gov.tw/api/v1/rest/datastore/O-A0003-001?Authorization={cwa_key}&limit=1",
                timeout=timeout,
            )
            results["cwa"] = {
                "ok": r.ok,
                "status": r.status_code,
                "latency_ms": round((_time.perf_counter() - t0) * 1000),
            }
        except Exception as e:
            results["cwa"] = {"ok": False, "error": str(e)[:100]}
    else:
        results["cwa"] = {"ok": False, "error": "not configured"}

    # AMOS (Korea runway sensors)
    try:
        t0 = _time.perf_counter()
        r = _r.get(
            "https://global.amo.go.kr/amosobsnew/AmosRealTimeImage.do", timeout=timeout
        )
        results["amos"] = {
            "ok": r.ok,
            "status": r.status_code,
            "latency_ms": round((_time.perf_counter() - t0) * 1000),
        }
    except Exception as e:
        results["amos"] = {"ok": False, "error": str(e)[:100]}

    # AMSC AWOS (China mainland airports)
    results["amsc_awos"] = _check_amsc_awos_health(timeout=timeout)

    # NOAA WRH (US settlement verification)
    try:
        t0 = _time.perf_counter()
        r = _r.get("https://www.weather.gov/wrh/timeseries?site=KJFK", timeout=timeout)
        results["noaa_wrh"] = {
            "ok": r.ok,
            "status": r.status_code,
            "latency_ms": round((_time.perf_counter() - t0) * 1000),
        }
    except Exception as e:
        results["noaa_wrh"] = {"ok": False, "error": str(e)[:100]}

    all_ok = all(v.get("ok") for v in results.values())
    return {
        "ok": all_ok,
        "checked_at": __import__("datetime").datetime.utcnow().isoformat() + "Z",
        "services": results,
    }


# ── Training accuracy helpers ─────────────────────────────────────────

def _evaluate_deb_records(
    records: List[Dict[str, Any]],
    *,
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
) -> Dict[str, Any]:
    from src.analysis.settlement_rounding import apply_city_settlement

    hits = 0
    total = 0
    errors: List[float] = []
    signed_errors: List[float] = []
    cities = set()
    dates = set()
    for row in records:
        target_date = str(row.get("target_date") or "").strip()
        if start_date and target_date < start_date:
            continue
        if end_date and target_date > end_date:
            continue
        city = str(row.get("city") or "").strip().lower()
        prediction = _sf(row.get("deb_prediction"))
        actual = _sf(row.get("actual_high"))
        if not city or not target_date or prediction is None or actual is None:
            continue
        try:
            pred_bucket = apply_city_settlement(city, prediction)
            actual_bucket = apply_city_settlement(city, actual)
        except Exception:
            continue
        if pred_bucket is None or actual_bucket is None:
            continue
        total += 1
        if pred_bucket == actual_bucket:
            hits += 1
        errors.append(abs(prediction - actual))
        signed_errors.append(prediction - actual)
        cities.add(city)
        dates.add(target_date)

    return {
        "start_date": start_date,
        "end_date": end_date,
        "samples": total,
        "hits": hits,
        "hit_rate": _round_metric((hits / total * 100) if total else None, 1),
        "mae": _round_metric((sum(errors) / len(errors)) if errors else None, 2),
        "bias": _round_metric((sum(signed_errors) / len(signed_errors)) if signed_errors else None, 2),
        "city_count": len(cities),
        "date_count": len(dates),
    }


def _build_city_deb_rows(
    city_id: str,
    city_rows: Dict[str, Dict[str, Any]],
    today_str: str,
) -> List[Dict[str, Any]]:
    rows: List[Dict[str, Any]] = []
    for target_date, record in sorted((city_rows or {}).items()):
        if target_date >= today_str or not isinstance(record, dict):
            continue
        rows.append(
            {
                "city": city_id,
                "target_date": target_date,
                "actual_high": record.get("actual_high"),
                "deb_prediction": record.get("deb_prediction"),
            }
        )
    return rows


def _deb_recent_bias_direction(bias: Optional[float]) -> str:
    if bias is None:
        return "unknown"
    if bias <= -0.5:
        return "under"
    if bias >= 0.5:
        return "over"
    return "neutral"


def _build_deb_recent_trust_strategy(
    recent_7d: Dict[str, Any],
    recent_14d: Dict[str, Any],
) -> Dict[str, Any]:
    window_key = "recent_14d" if int(recent_14d.get("samples") or 0) >= int(recent_7d.get("samples") or 0) else "recent_7d"
    metrics = recent_14d if window_key == "recent_14d" else recent_7d
    samples = int(metrics.get("samples") or 0)
    hit_rate = _sf(metrics.get("hit_rate"))
    mae = _sf(metrics.get("mae"))
    bias = _sf(metrics.get("bias"))

    if samples < 3 or hit_rate is None or mae is None:
        trust_tier = "insufficient"
        recommendation = "insufficient"
        reason = f"{window_key}: only {samples} settled DEB samples."
    elif hit_rate >= 67.0 and mae <= 1.25:
        trust_tier = "high"
        recommendation = "primary"
        reason = f"{window_key}: {hit_rate:.1f}% hit rate, MAE {mae:.2f}°."
    elif hit_rate >= 34.0 and mae <= 1.75:
        trust_tier = "medium"
        recommendation = "supporting"
        reason = f"{window_key}: {hit_rate:.1f}% hit rate, MAE {mae:.2f}°."
    else:
        trust_tier = "low"
        recommendation = "context_only"
        reason = f"{window_key}: {hit_rate:.1f}% hit rate, MAE {mae:.2f}°."

    return {
        "trust_tier": trust_tier,
        "recommendation": recommendation,
        "bias_direction": _deb_recent_bias_direction(bias),
        "reason": reason,
    }


def _build_city_deb_recent_strategy(
    city_id: str,
    city_rows: Dict[str, Dict[str, Any]],
    today_str: str,
) -> Optional[Dict[str, Any]]:
    today_date = datetime.strptime(today_str, "%Y-%m-%d").date()
    recent_7_start = (today_date - timedelta(days=7)).isoformat()
    recent_14_start = (today_date - timedelta(days=14)).isoformat()
    end_date = (today_date - timedelta(days=1)).isoformat()
    rows = _build_city_deb_rows(city_id, city_rows, today_str)
    if not rows:
        return None
    recent_7d = _evaluate_deb_records(rows, start_date=recent_7_start, end_date=end_date)
    recent_14d = _evaluate_deb_records(rows, start_date=recent_14_start, end_date=end_date)
    return {
        "recent_7d": recent_7d,
        "recent_14d": recent_14d,
        **_build_deb_recent_trust_strategy(recent_7d, recent_14d),
    }


def _build_city_deb_accuracy(city_id: str, city_rows: Dict[str, Dict[str, Any]], today_str: str) -> Optional[Dict[str, Any]]:
    rows = _build_city_deb_rows(city_id, city_rows, today_str)
    metrics = _evaluate_deb_records(rows)
    if not metrics["samples"]:
        return None
    total = int(metrics["samples"])
    hits = int(metrics["hits"])
    mae = float(metrics["mae"] or 0.0)
    hit_rate = float(metrics["hit_rate"] or 0.0)
    return {
        "hit_rate": hit_rate,
        "mae": mae,
        "total_days": total,
        "hits": hits,
        "details_str": f"过去{total}天 WU命中 {hits}/{total} ({hit_rate:.0f}%) | MAE: {mae:.1f}°",
    }


def _build_city_mu_accuracy(city_id: str, city_rows: Dict[str, Dict[str, Any]], today_str: str) -> Optional[Dict[str, Any]]:
    from src.analysis.settlement_rounding import apply_city_settlement

    errors: List[float] = []
    hits = 0
    total = 0
    brier_scores: List[float] = []
    for target_date, record in sorted((city_rows or {}).items()):
        if target_date >= today_str or not isinstance(record, dict):
            continue
        actual = _sf(record.get("actual_high"))
        mu_value = _sf(record.get("mu"))
        if actual is None or mu_value is None:
            continue
        total += 1
        errors.append(abs(mu_value - actual))
        if apply_city_settlement(city_id, mu_value) == apply_city_settlement(city_id, actual):
            hits += 1
        prob_snapshot = record.get("prob_snapshot") or []
        if isinstance(prob_snapshot, list):
            actual_bucket = apply_city_settlement(city_id, actual)
            score = 0.0
            used = False
            for entry in prob_snapshot:
                if not isinstance(entry, dict):
                    continue
                predicted_p = _sf(entry.get("p")) or 0.0
                outcome = 1.0 if entry.get("v") == actual_bucket else 0.0
                score += (predicted_p - outcome) ** 2
                used = True
            if used:
                brier_scores.append(score)

    if not total:
        return None
    mae = sum(errors) / len(errors)
    hit_rate = hits / total * 100
    brier = (sum(brier_scores) / len(brier_scores)) if brier_scores else None
    details_parts = [
        f"μ准确率: 过去{total}天",
        f"WU命中 {hits}/{total} ({hit_rate:.0f}%)",
        f"MAE: {mae:.1f}°",
    ]
    if brier is not None:
        details_parts.append(f"Brier: {brier:.3f}")
    return {
        "mae": mae,
        "hit_rate": hit_rate,
        "brier_score": brier,
        "total_days": total,
        "hits": hits,
        "details_str": " | ".join(details_parts),
    }


def _build_deb_historical_summary(accuracy_data: List[Dict[str, Any]]) -> Dict[str, Any]:
    deb_rows = [row for row in accuracy_data if row.get("deb")]
    if not deb_rows:
        return {
            "city_count": 0,
            "avg_hit_rate": None,
            "weighted_hit_rate": None,
            "avg_mae": None,
            "avg_days_per_city": 0,
            "sample_days": 0,
            "hits": 0,
        }
    sample_days = sum(int(row["deb"].get("total_days") or 0) for row in deb_rows)
    hits = sum(int(row["deb"].get("hits") or 0) for row in deb_rows)
    return {
        "city_count": len(deb_rows),
        "avg_hit_rate": _round_metric(
            sum(float(row["deb"].get("hit_rate") or 0.0) for row in deb_rows) / len(deb_rows),
            1,
        ),
        "weighted_hit_rate": _round_metric((hits / sample_days * 100) if sample_days else None, 1),
        "avg_mae": _round_metric(
            sum(float(row["deb"].get("mae") or 0.0) for row in deb_rows) / len(deb_rows),
            2,
        ),
        "avg_days_per_city": round(sample_days / len(deb_rows)) if deb_rows else 0,
        "sample_days": sample_days,
        "hits": hits,
    }


def _build_deb_usable_recent_summary(
    accuracy_data: List[Dict[str, Any]],
) -> Dict[str, Any]:
    usable_rows = []
    recommendations = {"primary": 0, "supporting": 0}
    for row in accuracy_data:
        recent = row.get("deb_recent") if isinstance(row.get("deb_recent"), dict) else None
        if not recent:
            continue
        recommendation = str(recent.get("recommendation") or "").strip().lower()
        if recommendation not in {"primary", "supporting"}:
            continue
        usable_rows.append(row)
        recommendations[recommendation] += 1

    def build_window(window_key: str) -> Dict[str, Any]:
        samples = 0
        hits = 0
        weighted_mae = 0.0
        city_count = 0
        for row in usable_rows:
            recent = row.get("deb_recent") if isinstance(row.get("deb_recent"), dict) else {}
            metrics = recent.get(window_key) if isinstance(recent.get(window_key), dict) else {}
            row_samples = int(metrics.get("samples") or 0)
            if row_samples <= 0:
                continue
            row_hits = int(metrics.get("hits") or 0)
            row_mae = _sf(metrics.get("mae"))
            samples += row_samples
            hits += row_hits
            city_count += 1
            if row_mae is not None:
                weighted_mae += row_mae * row_samples
        return {
            "window": window_key,
            "city_count": city_count,
            "samples": samples,
            "hits": hits,
            "hit_rate": _round_metric((hits / samples * 100) if samples else None, 1),
            "avg_mae": _round_metric((weighted_mae / samples) if samples else None, 2),
            "recommendations": dict(recommendations),
        }

    recent_7d = build_window("recent_7d")
    if int(recent_7d.get("samples") or 0) > 0:
        return recent_7d
    return build_window("recent_14d")


def _flatten_training_history(history: Dict[str, Dict[str, Dict[str, Any]]], today_str: str) -> List[Dict[str, Any]]:
    rows: List[Dict[str, Any]] = []
    for city, city_rows in (history or {}).items():
        if not isinstance(city_rows, dict):
            continue
        for target_date, record in city_rows.items():
            if not isinstance(record, dict):
                continue
            target_text = str(target_date or "").strip()
            if not target_text or target_text >= today_str:
                continue
            rows.append(
                {
                    "city": str(city or "").strip().lower(),
                    "target_date": target_text,
                    "actual_high": record.get("actual_high"),
                    "deb_prediction": record.get("deb_prediction"),
                    "mu": record.get("mu"),
                    "prob_snapshot": record.get("prob_snapshot"),
                }
            )
    rows.sort(key=lambda row: (row["target_date"], row["city"]))
    return rows


def _build_training_accuracy_payload(
    history: Dict[str, Dict[str, Dict[str, Any]]],
    city_registry: Dict[str, Dict[str, Any]],
    *,
    today_str: Optional[str] = None,
) -> Dict[str, Any]:
    from src.analysis.deb_evaluation import backtest_deb_versions

    today = today_str or datetime.now(timezone.utc).strftime("%Y-%m-%d")
    accuracy_data: List[Dict[str, Any]] = []
    for city_id, info in (city_registry or {}).items():
        city_rows = history.get(city_id) or history.get(str(city_id).strip().lower()) or {}
        deb_payload = _build_city_deb_accuracy(city_id, city_rows, today)
        deb_recent = _build_city_deb_recent_strategy(city_id, city_rows, today) if deb_payload else None
        mu_payload = _build_city_mu_accuracy(city_id, city_rows, today)
        if deb_payload or mu_payload:
            accuracy_data.append(
                {
                    "city_id": city_id,
                    "name": (info or {}).get("name") or city_id,
                    "deb": deb_payload,
                    "deb_recent": deb_recent,
                    "mu": mu_payload,
                }
            )

    accuracy_data.sort(
        key=lambda row: max(
            row["deb"]["total_days"] if row.get("deb") else 0,
            row["mu"]["total_days"] if row.get("mu") else 0,
        ),
        reverse=True,
    )

    all_rows = _flatten_training_history(history, today)
    today_date = datetime.strptime(today, "%Y-%m-%d").date()
    recent_7_start = (today_date - timedelta(days=7)).isoformat()
    recent_14_start = (today_date - timedelta(days=14)).isoformat()
    end_date = (today_date - timedelta(days=1)).isoformat()
    version_rows = all_rows[-_DEB_VERSION_BACKTEST_SAMPLE_LIMIT:]
    versions = backtest_deb_versions(
        version_rows,
        min_train_samples=2,
    ).get("versions", {})

    return {
        "accuracy": accuracy_data,
        "deb_summary": {
            "historical": _build_deb_historical_summary(accuracy_data),
            "usable_recent": _build_deb_usable_recent_summary(accuracy_data),
            "recent_7d": _evaluate_deb_records(
                all_rows,
                start_date=recent_7_start,
                end_date=end_date,
            ),
            "recent_14d": _evaluate_deb_records(
                all_rows,
                start_date=recent_14_start,
                end_date=end_date,
            ),
            "versions": versions,
        },
    }


def get_ops_training_accuracy(request: Request) -> Dict[str, Any]:
    from src.analysis.deb_algorithm import load_history
    from src.data_collection.city_registry import CITY_REGISTRY

    history_file = os.path.join(
        os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))),
        "data",
        "daily_records.json",
    )
    history = load_history(history_file)
    return _build_training_accuracy_payload(history, CITY_REGISTRY)


# ── Truth history ─────────────────────────────────────────────────────

def get_ops_truth_history(
    request: Request,
    city: str = "",
    date_from: str = "",
    date_to: str = "",
    limit: int = 200,
) -> Dict[str, Any]:
    from web.services.ops_api import _require_ops

    _require_ops(request)

    truth_history = legacy_routes.TruthRecordRepository().load_all()
    normalized_city = str(city or "").strip().lower()
    normalized_from = str(date_from or "").strip()
    normalized_to = str(date_to or "").strip()
    max_limit = max(1, min(int(limit or 200), 1000))

    rows = []
    for row_city, by_date in truth_history.items():
        if normalized_city and row_city != normalized_city:
            continue
        if not isinstance(by_date, dict):
            continue
        for target_date, payload in by_date.items():
            if normalized_from and str(target_date) < normalized_from:
                continue
            if normalized_to and str(target_date) > normalized_to:
                continue
            if not isinstance(payload, dict):
                continue
            rows.append(
                {
                    "city": row_city,
                    "display_name": str(
                        (legacy_routes.CITY_REGISTRY.get(row_city) or {}).get("name")
                        or row_city
                    ),
                    "target_date": str(target_date),
                    "actual_high": payload.get("actual_high"),
                    "settlement_source": payload.get("settlement_source"),
                    "settlement_station_code": payload.get("settlement_station_code"),
                    "settlement_station_label": payload.get("settlement_station_label"),
                    "truth_version": payload.get("truth_version"),
                    "updated_by": payload.get("updated_by"),
                    "truth_updated_at": payload.get("truth_updated_at"),
                    "is_final": payload.get("is_final"),
                }
            )

    rows.sort(
        key=lambda item: (str(item["target_date"]), str(item["city"])), reverse=True
    )
    filtered_count = len(rows)
    rows = rows[:max_limit]
    available_cities = [
        {
            "city": city_id,
            "name": str(info.get("name") or city_id),
        }
        for city_id, info in sorted(
            legacy_routes.CITY_REGISTRY.items(),
            key=lambda item: str(item[1].get("name") or item[0]),
        )
    ]
    return {
        "items": rows,
        "available_cities": available_cities,
        "filters": {
            "city": normalized_city or None,
            "date_from": normalized_from or None,
            "date_to": normalized_to or None,
            "limit": max_limit,
        },
        "filtered_count": filtered_count,
    }