"""Compact chronological validation for the Regime Monitor warning score. The study calls its outcome a 10% correction, uses the first 70% of sessions to freeze an 80th-percentile warning threshold, and reports alarm episodes only on the final 30%. It is still labelled exploratory while the fixed breadth basket is reconstructed before its freeze date. """ from __future__ import annotations import json import logging from datetime import date, datetime, timedelta, timezone from sqlalchemy.ext.asyncio import AsyncSession from app.services import breadth_service, settings_store from app.services import regime_monitor_service as rms from app.services.admin_service import update_setting logger = logging.getLogger(__name__) KEY_REPORT = "regime_event_study" EVENT_THRESHOLD_PCT = 10.0 EVENT_COOLDOWN_DAYS = 40 DRAWDOWN_LOOKBACK = 252 HORIZON_DAYS = 20 WARN_PERCENTILE = 80.0 TRAIN_FRACTION = 0.70 # Below this many holdout corrections, recall is one event away from a very # different headline and should not be read as a property of the score. MIN_EVENTS_FOR_CONFIDENCE = 8 SENSOR_MISMATCH_TOLERANCE = 0.10 def _median(values: list[float]) -> float | None: if not values: return None ordered = sorted(values) middle = len(ordered) // 2 return ( float(ordered[middle]) if len(ordered) % 2 else (ordered[middle - 1] + ordered[middle]) / 2.0 ) def _percentile(values: list[float], pct: float) -> float | None: ordered = sorted(v for v in values if v is not None) if not ordered: return None position = (len(ordered) - 1) * pct / 100.0 lower = int(position) upper = min(lower + 1, len(ordered) - 1) return ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower) def detect_events( closes: list[float], dates: list[date], threshold_pct: float = EVENT_THRESHOLD_PCT, lookback: int = DRAWDOWN_LOOKBACK, cooldown: int = EVENT_COOLDOWN_DAYS, ) -> list[dict]: """Rising-edge corrections from the trailing 52-week high.""" events: list[dict] = [] previous_drawdown = 0.0 last_event = -10**9 for index, close in enumerate(closes): high = max(closes[max(0, index - lookback + 1): index + 1]) drawdown = (high - close) / high * 100.0 if high > 0 else 0.0 if ( drawdown >= threshold_pct and previous_drawdown < threshold_pct and index - last_event >= cooldown ): events.append({ "date": dates[index].isoformat(), "index": index, "depth_pct": round(drawdown, 1), }) last_event = index previous_drawdown = drawdown return events def alarm_episodes( indicator: dict[date, float], dates: list[date], threshold: float, start_index: int = 1, ) -> list[int]: """Indices where the warning crosses upward; it must reset below first.""" alarms: list[int] = [] was_high = False if start_index > 0: previous = indicator.get(dates[start_index - 1]) was_high = previous is not None and previous >= threshold for index in range(start_index, len(dates)): value = indicator.get(dates[index]) if value is None: continue high = value >= threshold if high and not was_high: alarms.append(index) was_high = high return alarms def evaluate_alarms( alarm_indices: list[int], event_indices: list[int], dates: list[date], horizon: int = HORIZON_DAYS, ) -> dict: """Event recall, episode false alarms, and lead time for one holdout.""" leads: list[float] = [] per_event: list[dict] = [] warned = 0 for event_index in event_indices: matching = [ alarm for alarm in alarm_indices if 0 < event_index - alarm <= horizon ] lead = max((event_index - alarm for alarm in matching), default=None) if lead is not None: warned += 1 leads.append(float(lead)) per_event.append({ "date": dates[event_index].isoformat(), "warned": lead is not None, "lead_days": lead, }) false_alarms = sum( 1 for alarm in alarm_indices if not any(0 < event - alarm <= horizon for event in event_indices) ) return { "events": len(event_indices), "events_warned": warned, "events_missed": len(event_indices) - warned, "alarm_episodes": len(alarm_indices), "false_alarms": false_alarms, "median_lead_days": _median(leads), "per_event": per_event, } def _warning_series( prices: dict[str, rms.Series], breadth_divergence: dict[date, float], dates: list[date], config: dict, oas_series: rms.Series | None = None, ) -> tuple[dict[date, float], dict[date, int]]: """Warning score per session plus how many sensors backed it. v2 re-derived this by hand from ``WARNING_WEIGHTS`` and so would have kept measuring the old construct after a scoring change. Since v3 dropped fundamentals from the score, this is now exactly the live Warning score rather than a technical-only approximation of it. The sensor count matters because the score renormalises over whatever is available: a session backed by two sensors is not drawn from the same distribution as one backed by three, and the frozen threshold assumes it is. """ tickers = config["tickers"] smh_full = prices.get(tickers["leaders"][0], []) spy_full = prices.get(tickers["market"], []) out: dict[date, float] = {} backing: dict[date, int] = {} for session in dates: sensors = rms.warning_sensor_scores( breadth_divergence.get(session), rms._closes_asof(smh_full, session), rms._closes_asof(spy_full, session), rms._window_asof(oas_series, session, rms.HY_OAS_WINDOW_DAYS), ) score = rms.score_warning_sensors(sensors) if score is not None: out[session] = round(score, 2) backing[session] = sum(1 for value in sensors.values() if value is not None) return out, backing def _reliability( dates: list[date], split: int, backing: dict[date, int], events_detected: int, events_in_holdout: int, ) -> dict: """How far the headline metrics can actually be trusted. Two things repeatedly invite over-reading this report: * The holdout carries only the corrections that fall in the last 30% of the sample. A "2/4" is one event away from "3/4", and in practice the events that flip are decided by where the frozen threshold happens to land rather than by whether the score saw anything. * The score renormalises over available sensors, so a training window that predates a sensor's history freezes a threshold on a different construct than the holdout is measured against. """ expected = len(rms.WARNING_WEIGHTS) train = [backing[d] for d in dates[:split] if d in backing] holdout = [backing[d] for d in dates[split:] if d in backing] train_full = sum(1 for n in train if n == expected) / len(train) if train else 0.0 holdout_full = sum(1 for n in holdout if n == expected) / len(holdout) if holdout else 0.0 return { "events_detected": events_detected, "events_in_holdout": events_in_holdout, "minimum_events": MIN_EVENTS_FOR_CONFIDENCE, "underpowered": events_in_holdout < MIN_EVENTS_FOR_CONFIDENCE, "sensors_expected": expected, "train_full_sensor_share": round(train_full * 100, 1), "holdout_full_sensor_share": round(holdout_full * 100, 1), "sensor_coverage_mismatch": abs(train_full - holdout_full) > SENSOR_MISMATCH_TOLERANCE, } async def run_event_study( db: AsyncSession, threshold_pct: float = EVENT_THRESHOLD_PCT, horizon: int = HORIZON_DAYS, ) -> dict: config = await rms.get_regime_config(db) end = date.today() start = end - timedelta(days=5 * 365 + 30) prices = await rms._fetch_prices(config, start, end) leader = config["tickers"]["leaders"][0] benchmark = sorted(prices.get(leader, []), key=lambda item: item[0]) if len(benchmark) < 500: return {"available": False, "reason": "insufficient benchmark history"} dates = [d for d, _ in benchmark] closes = [value for _, value in benchmark] breadth, _ = await breadth_service.compute_breadth_details( db, config["breadth_basket"], window=200, min_tickers=20 ) divergence = breadth_service.compute_divergence_series(breadth, benchmark) oas_series = await rms._fetch_fred_series("BAMLH0A0HYM2", start, end) warning, backing = _warning_series(prices, divergence, dates, config, oas_series) # The credit sensor cannot reach back as far as the price history does (the # upstream series is capped at ~3 years), so the earlier part of the sample # scores on W1+W2 alone via renormalisation. Report where W3 starts rather # than letting the threshold quietly straddle two sensor sets. credit_from = oas_series[0][0].isoformat() if oas_series else None split = max(1, min(len(dates) - 1, int(len(dates) * TRAIN_FRACTION))) train_values = [warning[d] for d in dates[:split] if d in warning] warn_threshold = _percentile(train_values, WARN_PERCENTILE) if warn_threshold is None: return {"available": False, "reason": "insufficient warning history"} all_events = detect_events(closes, dates, threshold_pct) holdout_events = [event["index"] for event in all_events if event["index"] >= split] alarms = alarm_episodes(warning, dates, warn_threshold, start_index=split) metrics = evaluate_alarms(alarms, holdout_events, dates, horizon) holdout_sessions = max(1, len(dates) - split) metrics["false_alarms_per_year"] = round( metrics["false_alarms"] / (holdout_sessions / 252.0), 2 ) reliability = _reliability(dates, split, backing, len(all_events), len(holdout_events)) basket_asof = date.fromisoformat(config["basket_asof"]) retrospective = dates[split] < basket_asof evaluation = "exploratory" if retrospective else "holdout" lead_text = ( f"median lead {metrics['median_lead_days']:.0f} sessions" if metrics["median_lead_days"] is not None else "no successful warning lead" ) summary = ( f"{evaluation.capitalize()} chronological test: warning episodes preceded " f"{metrics['events_warned']}/{metrics['events']} 10% corrections; " f"{metrics['events_missed']} missed, {metrics['false_alarms_per_year']:.1f} " f"false alarms/year, {lead_text}. " f"{metrics['events']} of {reliability['events_detected']} detected corrections " f"fall in the test period" + ( f"; too few to read recall as a property of the score." if reliability["underpowered"] else "." ) ) per_event = metrics.pop("per_event") report = { "available": True, "methodology": rms.METHODOLOGY, "generated_at": datetime.now(timezone.utc).isoformat(), "evaluation": evaluation, "summary": summary, "params": { "benchmark": leader, "outcome": "10% correction from trailing 52-week high", "event_threshold_pct": threshold_pct, "event_cooldown_days": EVENT_COOLDOWN_DAYS, "horizon_days": horizon, "train_fraction": TRAIN_FRACTION, "warn_percentile": WARN_PERCENTILE, "warn_threshold": round(warn_threshold, 1), "credit_sensor_from": credit_from, "basket_hash": rms._basket_hash(config["breadth_basket"]), "basket_asof": config["basket_asof"], }, "sample": { "start": dates[0].isoformat(), "end": dates[-1].isoformat(), "train_end": dates[split - 1].isoformat(), "test_start": dates[split].isoformat(), "sessions": len(dates), "holdout_sessions": holdout_sessions, }, "metrics": metrics, "reliability": reliability, "events": per_event, "recent_breadth": [ {"date": d.isoformat(), "breadth": breadth[d], "warning": warning.get(d)} for d in dates[-90:] if d in breadth ], } logger.info(json.dumps({ "event": "regime_event_study_complete", "evaluation": evaluation, "events": metrics["events"], "events_detected": reliability["events_detected"], "warned": metrics["events_warned"], "false_alarms_per_year": metrics["false_alarms_per_year"], "underpowered": reliability["underpowered"], "sensor_coverage_mismatch": reliability["sensor_coverage_mismatch"], })) return report async def run_and_store(db: AsyncSession) -> dict: report = await run_event_study(db) await update_setting(db, KEY_REPORT, json.dumps(report)) return report async def get_event_study_report(db: AsyncSession) -> dict | None: setting = await settings_store.get_setting(db, KEY_REPORT) if setting is None: return None try: report = json.loads(setting.value) except (TypeError, ValueError): return None return report if report.get("methodology") == rms.METHODOLOGY else None