signal-platform/app/services/event_study_service.py

"""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


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,
) -> dict[date, float]:
    """Warning score per session, from the monitor's own sensor definitions.

    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.
    """
    tickers = config["tickers"]
    smh_full = prices.get(tickers["leaders"][0], [])
    spy_full = prices.get(tickers["market"], [])
    out: dict[date, float] = {}
    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)
    return out


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 = _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
    )

    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}."
    )
    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,
        "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"],
        "warned": metrics["events_warned"],
        "false_alarms_per_year": metrics["false_alarms_per_year"],
    }))
    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