"""AI/Tech Risk Monitor v4.

The monitor is a risk thermometer, not a probability or trading rule.  It keeps
two deliberately separate outputs:

* State: current structural stress (price, breadth, credit, volatility).
* Warning: deterioration/divergence that may precede State (breadth divergence,
  relative strength, credit impulse).

* Fundamental context: a categorical channel (supportive / neutral / adverse /
  unknown) with an evidence-quality grade, derived by fixed rules from the
  sourced hyperscaler capex and earnings-reaction observations.

Both scores are quantitative and daily.  The fundamental channel is deliberately
**not** a term in either: the three are read together by confluence, because
adding a slow categorical judgement to a fast continuous score manufactures
precision by summing unlike things, and any fusion weight would be a policy
preference presented as a measurement until there is enough point-in-time
history to fit one.  A missing observation therefore stays ``unknown`` instead of
silently redistributing its weight onto the technical sensors.

Daily snapshots are the point-in-time record.  The first run under a new
``METHODOLOGY`` rewrites every session inside ``REBUILD_LOOKBACK_DAYS`` once;
ordinary runs thereafter only upsert the latest trading date.  The overlay is
still gated by its effective date so a rebuild cannot stamp today's observation
onto historical snapshots.
"""

from __future__ import annotations

import hashlib
import json
import logging
import os
from datetime import date, datetime, timedelta, timezone
from pathlib import Path

import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.exceptions import ProviderError, ValidationError
from app.models.regime_fundamental_observation import RegimeFundamentalObservation
from app.models.regime_snapshot import RegimeSnapshot
from app.providers.alpaca import AlpacaOHLCVProvider
from app.services import breadth_service, settings_store
from app.services.admin_service import update_setting
from app.services.sentiment_provider_service import _resolve as resolve_llm_config

logger = logging.getLogger(__name__)

_CA_BUNDLE = os.environ.get("SSL_CERT_FILE", "")

KEY_CONFIG = "regime_monitor_config"
KEY_FUNDAMENTALS = "regime_fundamental_overrides"

METHODOLOGY = "v4"
# Snapshots are reseeded on a methodology bump, but fundamental observations are
# collected by hand/LLM and carried across it when the format is compatible.
# EVERY methodology sharing the categorical format must be listed: this is checked
# against the *stored* blob, so omitting the current one discards the observation
# on its first write, which leaves fetched_at null and locked false -- and then
# update_regime_monitor refreshes it via the LLM on every single run, forever.
# "v5" is listed although no v5 scoring exists: a v5 was briefly built (a weighted
# fundamental modifier on Warning) and reverted, so a development box can have
# that string sitting in its settings blob. Keeping it costs nothing; omitting it
# costs the failure above.
CATEGORICAL_FUNDAMENTAL_METHODOLOGIES = frozenset({"v2", "v3", "v4", "v5"})

# Bumped when a fix changes what historical rows *should* contain without
# changing the live formula, so stored history needs one reseed.  Deliberately
# not METHODOLOGY: that partitions the history API and discards the cached event
# study, neither of which is warranted here -- the study recomputes its Warning
# series from source rather than reading snapshots, so a reseed cannot stale it.
# Snapshots written before this marker existed carry no key and read as 1.
# Deliberately NOT bumped for v4: a METHODOLOGY change already forces a full
# reseed (every stored row fails _parse_snapshot, so _latest_snapshot_row returns
# None and rebuilding is True). Bumping both would imply the reseed was
# revision-driven.
SENSOR_REVISION = 2
MIN_COVERAGE = 75.0
SOURCE_MAX_LAG_DAYS = 7

# Bands are per axis: the two scores have genuinely different realized ranges,
# so one shared set made Warning's top bands unreachable (v2 Warning never
# exceeded 64.9 in 408 sessions while State reached 91.2).  Thresholds are round
# numbers chosen so each band covers a sane share of history, not percentile
# fits -- percentile-derived bands would drift on every rebuild and silently
# rewrite what past snapshots meant.
#
# v4 moved State's top band 80 -> 65, and only that one.  With credit calm it
# scores 0.0 (not None) and still holds its full 20 points, so price + breadth +
# volatility at *literal maximum* summed to exactly 80.0 -- the old threshold, to
# the decimal, with nothing to spare.  A 2022-style AI/tech drawdown with calm
# credit computes to 70.3-74.0 depending on whether a death cross has formed, so
# at 80 the case this monitor exists to measure could not print the top band.
# 65 clears it under either assumption.  Realized shares over the 408 sessions to
# 2026-07-24, reported not fitted: State 78.9/13.0/4.7/3.4%, Warning 69/20/8/3%.
# The v4 breaking share (3.4%) matches v3's, which was arrived at independently.
STATE_BANDS = (20.0, 50.0, 65.0)
WARNING_BANDS = (20.0, 40.0, 60.0)

QUADRANT_STATE_DIVIDER = 50.0
QUADRANT_WARNING_DIVIDER = 40.0
QUADRANT_MARGIN = 5.0

HY_OAS_MILD = 3.5
HY_OAS_ELEVATED = 5.0
HY_OAS_STRESSED = 7.0
# ICE restricted FRED to a rolling 3-year window for BAMLH0A0HYM2 in April 2026
# ("Starting in April 2026, this series will only include 3 years of
# observations"), so v2's 10-year reference window silently became 3.  Against 3
# years of uniformly tight spreads (2.59-4.61 over the calibration window) the
# blended upper-tail percentile saturated at an OAS of ~4.5 and scored 20 points
# of stress at 3.5 -- the level these anchors call "mild".  The anchors already
# encode the long-run distribution, so the credit *level* is now purely anchored
# and credit *dynamics* live in W3 on the Warning axis where they belong.
# Calendar days, and it must cover the oldest date a rebuild replays -- not just
# W3's lookback.  REBUILD_SESSIONS is 400 *trading* sessions (~579 calendar
# days), so a 400-calendar-day fetch left the oldest ~180 days of a rebuild with
# no OAS at all: C1 and W3 both returned None, State landed at 80% coverage and
# Warning at exactly MIN_COVERAGE, and *both still published bands* -- a series
# that looks homogeneous while its oldest rows were scored without credit.
# Widening only prepends older observations; C1 reads [-1] and W3 reads [-21], so
# live scores are unchanged and this needs no methodology bump.  Stays under
# ICE's ~3-year cap so FRED still honours the request.
HY_OAS_WINDOW_DAYS = 700

# A rebuild replays every session inside this window.  Bounded by calendar days
# rather than a session count because the binding constraint is the OAS fetch:
# each replayed row needs W3's 20-business-day lookback (~28 calendar days)
# inside HY_OAS_WINDOW_DAYS, so replaying further back would recreate the exact
# credit gap a reseed exists to close.  672 days is ~464 trading sessions, which
# comfortably covers the 400-session series the v3 cutover wrote.
REBUILD_LOOKBACK_DAYS = HY_OAS_WINDOW_DAYS - 28
W3_OAS_LOOKBACK = 20
W3_OAS_FULL_SCALE_PCT = 35.0

# Drawdown anchors (drawdown %, stress score).  v2 used a bare ``dd_pct * 5``,
# which pegged at a 20% drawdown -- the 90th percentile of the observed
# distribution -- so 39 of 408 sessions sat at exactly 100 with no resolution
# left during the part of a selloff that matters most.  These anchors keep
# headroom past the observed 36% maximum.
P3_DRAWDOWN_ANCHORS = (
    (0.0, 0.0), (4.0, 10.0), (8.0, 25.0), (16.0, 50.0), (28.0, 78.0), (40.0, 100.0),
)

# Trend-break depth (% below the 200-DMA, stress score). v4; see _under_200 for
# why the crossing gets a floor of 20 rather than starting at 0. Calibrated to
# sit alongside P3 rather than swamp it -- the 200-DMA lags, so a 20% drawdown
# typically coincides with ~10% below the average, where this reads ~61 against
# P3's ~59. On the population the P1_SCORE_CAP rule actually names -- sessions
# with State >= 40 -- P1 is the sole price argmax on 17 of 47 (36.2%), against
# P2's 16 and P3's 14, so it informs the pillar without owning it and no cap
# was needed.
P1_TREND_BREAK_ANCHORS = (
    (0.0, 20.0), (3.0, 35.0), (8.0, 55.0), (15.0, 75.0), (25.0, 100.0),
)

# VIX level anchors (v4). Full scale at 55 rather than at 2020's ~82: anchoring
# the top at a once-in-a-generation print would make VIX 50 -- a genuine crisis
# -- read only ~70. A typical correction (25-35) now reads 38-67 where v3 read
# 66.7-100. The anchors encode the long-run distribution as constants, the same
# argument the credit level uses.
P5_VIX_ANCHORS = (
    (15.0, 0.0), (20.0, 20.0), (25.0, 38.0), (30.0, 55.0), (40.0, 80.0), (55.0, 100.0),
)

STATE_WEIGHTS = {
    "price": 40.0,
    "breadth": 25.0,
    "credit": 20.0,
    "volatility": 15.0,
}
# Fundamentals left the score in v3.  At 12 + 8 points they could not reach the
# event study's alarm threshold even when both pegged at 100, so the LLM read was
# decorative; it is now a separate qualitative overlay.  Credit *impulse* takes
# their place because the OAS level is pinned at zero below the 3.5 anchor while
# its rate of change is not.
WARNING_WEIGHTS = {
    "breadth_divergence": 45.0,
    "relative_strength": 30.0,
    "credit_impulse": 25.0,
}

# The sourced fundamental read is a **separate channel**, never a term in either
# score. It is reported as a categorical state beside State and Warning, and the
# three are read together by confluence rather than added up.
#
# Two things had to be true at once and only this shape gets both.
#
# **v3's reason for removing it was wrong.** v3 argued that F1+F3, at 12+8 of 100
# Warning points, "could not change any published conclusion" because pegged they
# produced a Warning of exactly 20.0. That holds only when every technical sensor
# reads exactly zero. Weighted, those points added +10 to +20 across the
# realistic range and moved the technical score needed to reach the 40 quadrant
# divider from 40 to 25. So the observation was not inert, and demoting it to
# decoration was not justified by that argument.
#
# **But no weight is measurable either.** A weighted modifier was built (v5,
# reverted) and its size could not be derived from anything: with ~10 correction
# events and essentially no fundamental history, any fusion weight is a policy
# preference presented as a measurement. Adding a slow categorical judgement to a
# fast continuous score also manufactures precision by summing unlike things, and
# it forces a missing observation to silently redistribute its weight onto the
# technical sensors -- the opposite of leaving it unknown.
#
# So the read gets a channel, not a coefficient. Revisit only with enough
# point-in-time history to test whether the state improves prediction
# *conditional on* Warning; a fitted model then has something to fit.
FUNDAMENTAL_STATES = ("supportive", "neutral", "adverse", "unknown")
EVIDENCE_QUALITY = ("complete", "partial", "stale", "manual", "unavailable")

# Fixed at the v2 launch.  These are liquid S&P 500/Nasdaq AI, semiconductor,
# infrastructure, cloud, and enterprise-software names that the platform's
# normal universe sync already stores.
DEFAULT_BREADTH_BASKET = [
    "AAPL", "MSFT", "NVDA", "AMZN", "META", "GOOGL", "AVGO", "AMD",
    "ORCL", "CRM", "NOW", "PLTR", "ANET", "DELL", "SMCI", "MU",
    "QCOM", "INTC", "AMAT", "LRCX", "KLAC", "SNPS", "CDNS", "ADI",
    "TXN", "IBM", "CSCO", "PANW", "CRWD", "VRT",
]

DEFAULT_CONFIG: dict = {
    "tickers": {
        "leaders": ["SMH"],
        "confirm": ["QQQ"],
        "market": "SPY",
        "hyperscalers": ["GOOGL", "AMZN", "META", "MSFT"],
    },
    "breadth_basket": DEFAULT_BREADTH_BASKET,
    "basket_asof": "2026-07-15",
    "fundamental_staleness_days": 80,
}

CAPEX_STATES = ("raising", "holding", "cutting", "unknown")
# "mixed" is a genuinely observed mixed reaction; "unknown" is nobody looked or
# the extraction failed. They were the same value until 2026-08-13, so a failed
# LLM parse silently became neutral *evidence* -- an observation of normality
# manufactured out of a parse error. Same distinction the capex map already made
# with its own "unknown", and the same one the whole channel is built on.
GNSD_STATES = ("yes", "no", "mixed", "unknown")
# v2 scored raising and holding identically at 0, so in a capex boom the reading
# was pinned at 0 and could not express the raising -> holding deceleration that
# is the actual early warning. Display-only in v3, but it should still describe.
_CAPEX_STATE_SCORES = {"raising": 0.0, "holding": 50.0, "cutting": 100.0}
_GNSD_SCORES = {"yes": 100.0, "no": 0.0}

Series = list[tuple[date, float]]


# ---------------------------------------------------------------------------
# Pure numeric helpers and sensors
# ---------------------------------------------------------------------------

def _clamp(x: float, lo: float = 0.0, hi: float = 100.0) -> float:
    return max(lo, min(hi, x))


def _sma(values: list[float], window: int) -> float | None:
    if len(values) < window:
        return None
    return sum(values[-window:]) / window


def _mean(values: list[float]) -> float | None:
    return sum(values) / len(values) if values else None


def _blend(leader: float | None, confirm: float | None, leader_weight: float = 2.0) -> float | None:
    parts: list[tuple[float, float]] = []
    if leader is not None:
        parts.append((leader, leader_weight))
    if confirm is not None:
        parts.append((confirm, 1.0))
    if not parts:
        return None
    return sum(v * w for v, w in parts) / sum(w for _, w in parts)


def _interpolate(x: float, anchors: tuple[tuple[float, float], ...]) -> float:
    """Piecewise-linear lookup, flat outside the first and last anchor."""
    if x <= anchors[0][0]:
        return anchors[0][1]
    for (x0, y0), (x1, y1) in zip(anchors, anchors[1:]):
        if x <= x1:
            return y0 + (y1 - y0) * (x - x0) / (x1 - x0)
    return anchors[-1][1]


def band_for(score: float, bands: tuple[float, float, float] = STATE_BANDS) -> str:
    watch, elevated, breaking = bands
    if score < watch:
        return "stable"
    if score < elevated:
        return "watch"
    if score < breaking:
        return "elevated"
    return "breaking"


def _under_200(closes: list[float]) -> float | None:
    """Trend break graded by depth below the 200-DMA, not a bare yes/no.

    Through v3 this returned 0 or 100, so P1 printed 100 the moment SMH and QQQ
    were both under their average -- and because the price pillar takes
    ``max(P1, P2, P3)``, that pinned the pillar and stopped P3's anchored ladder
    resolving anything for the whole of a selloff. It pegged on 46 of the 408
    sessions to 2026-07-24; under this table, none.

    The step at the crossing (0 -> 20) is deliberate: the break itself is a
    genuine binary event and deserves a floor. Only the depth past it is graded.
    """
    sma200 = _sma(closes, 200)
    if sma200 is None or sma200 <= 0:
        return None
    pct_below = (sma200 - closes[-1]) / sma200 * 100.0
    if pct_below <= 0:
        return 0.0
    return _clamp(_interpolate(pct_below, P1_TREND_BREAK_ANCHORS))


def p1_trend_break(smh: list[float], qqq: list[float], leader_weight: float = 2.0) -> float | None:
    return _blend(_under_200(smh), _under_200(qqq), leader_weight)


def _death_cross(closes: list[float]) -> float | None:
    sma50 = _sma(closes, 50)
    sma200 = _sma(closes, 200)
    if sma50 is None or sma200 is None or len(closes) < 221 or sma200 == 0:
        return None
    gap_pct = (sma50 / sma200 - 1.0) * 100.0
    severity = 0.0 if gap_pct >= 0 else _clamp(-gap_pct * 20.0)
    sma200_past = _sma(closes[:-20], 200)
    if sma200_past:
        slope_pct = (sma200 / sma200_past - 1.0) * 100.0
        if slope_pct >= 0:
            severity *= 0.5
    return severity


def p2_death_cross(smh: list[float], qqq: list[float], leader_weight: float = 2.0) -> float | None:
    return _blend(_death_cross(smh), _death_cross(qqq), leader_weight)


def drawdown_pct(closes: list[float]) -> float | None:
    """Percentage below the trailing 52-week closing high."""
    if len(closes) < 30:
        return None
    peak = max(closes[-252:])
    if peak <= 0:
        return None
    return (peak - closes[-1]) / peak * 100.0


def _drawdown(closes: list[float]) -> float | None:
    dd_pct = drawdown_pct(closes)
    return None if dd_pct is None else _clamp(_interpolate(dd_pct, P3_DRAWDOWN_ANCHORS))


def p3_drawdown(smh: list[float], qqq: list[float], leader_weight: float = 2.0) -> float | None:
    """Anchored drawdown stress on the same 2:1 leader/confirm blend P1 and P2 use.

    v2 took ``max()`` here, which meant the more volatile leader always won and
    the price pillar reduced to this one sensor: its realized share of State was
    65% against a nominal 40% weight.  Blending brings that back to 40%.
    """
    return _blend(_drawdown(smh), _drawdown(qqq), leader_weight)


def p4_relative_strength(smh: list[float], spy: list[float], lookback: int = 60) -> float | None:
    """Stress-only SMH/SPY rollover: flat/outperformance=0, -10%=100."""
    if len(smh) < lookback + 1 or len(spy) < lookback + 1:
        return None
    if spy[-1] == 0 or spy[-lookback - 1] == 0:
        return None
    now = smh[-1] / spy[-1]
    past = smh[-lookback - 1] / spy[-lookback - 1]
    if past == 0:
        return None
    chg_pct = (now / past - 1.0) * 100.0
    return _clamp(-chg_pct * 10.0)


def p5_volatility(vix: float | None) -> float | None:
    """VIX level against named anchors, so it keeps resolving past a 30 print.

    v3 used ``(vix - 15) / 15``, which reached 100 at VIX 30 -- the same
    saturation v3 itself had just removed from P3. VIX 30 is a bad week, 50 is a
    crisis and 82 was March 2020, and all three scored identically. In the 408
    sessions to 2026-07-24 that flattened five distinct April-2025 prints
    (52.33, 46.98, 45.31, 40.72, 38.57) into a single 100.
    """
    if vix is None:
        return None
    return _clamp(_interpolate(vix, P5_VIX_ANCHORS))


def breadth_level_score(pct_above_200: float | None) -> float | None:
    """Broad >=60%=healthy; <=20%=full breadth stress; linear between."""
    if pct_above_200 is None:
        return None
    return _clamp((60.0 - pct_above_200) / 40.0 * 100.0)


def _oas_absolute_score(value: float) -> float:
    if value <= HY_OAS_MILD:
        return 0.0
    if value <= HY_OAS_ELEVATED:
        return (value - HY_OAS_MILD) / (HY_OAS_ELEVATED - HY_OAS_MILD) * 50.0
    if value < HY_OAS_STRESSED:
        return 50.0 + (value - HY_OAS_ELEVATED) / (HY_OAS_STRESSED - HY_OAS_ELEVATED) * 50.0
    return 100.0


def f2_credit_spreads(oas_values: list[float]) -> float | None:
    """HY OAS level against named absolute anchors (3.5 mild / 5.0 / 7.0).

    v2 blended 70% of this with a 30% upper-tail percentile over the available
    history. That leg was always a second, noisier estimate of what the anchors
    already encode -- and once the usable window shrank to 3 uniformly tight
    years it saturated far below any real stress level. Removed rather than
    repaired: see ``HY_OAS_WINDOW_DAYS``.
    """
    if not oas_values:
        return None
    return round(_oas_absolute_score(oas_values[-1]), 2)


def w3_credit_impulse(
    oas_values: list[float], lookback: int = W3_OAS_LOOKBACK
) -> float | None:
    """HY OAS rate of change: widening only, relative so it works at any level.

    The credit *level* (C1) sits at zero for as long as spreads stay under the
    3.5 mild anchor -- 2.77 as of the v3 cutover -- so it contributes nothing to
    State in a calm tape.  The rate of change still does, and spread widening is
    a classic lead, which is what Warning is for.  Relative rather than absolute
    because +0.5pp means something very different at 2.7 than at 8.0.
    """
    if len(oas_values) < lookback + 1:
        return None
    past = oas_values[-lookback - 1]
    if past <= 0:
        return None
    change_pct = (oas_values[-1] / past - 1.0) * 100.0
    return _clamp(change_pct / W3_OAS_FULL_SCALE_PCT * 100.0)


def warning_sensor_scores(
    divergence: float | None,
    leader_closes: list[float],
    market_closes: list[float],
    oas_window: list[float],
) -> dict[str, float | None]:
    """The three Warning sensors, by pillar id.

    Single definition so the live monitor and the event study cannot drift apart
    -- in v2 the study re-derived the score from ``WARNING_WEIGHTS`` by hand and
    would have silently kept measuring the old construct through this change.
    """
    return {
        "breadth_divergence": divergence,
        "relative_strength": p4_relative_strength(leader_closes, market_closes),
        "credit_impulse": w3_credit_impulse(oas_window),
    }


def score_warning_sensors(sensors: dict[str, float | None]) -> float | None:
    """Weighted Warning score, renormalised over the sensors that are available."""
    live = [
        (float(score), float(WARNING_WEIGHTS[key]))
        for key, score in sensors.items()
        if score is not None and key in WARNING_WEIGHTS
    ]
    if not live:
        return None
    return sum(s * w for s, w in live) / sum(w for _, w in live)


def _capex_signal(capex: dict[str, str] | None, names: list[str]) -> str:
    """Categorical read of hyperscaler capex direction. Never an average.

    Averaging is what this must not do: it would let two ``cutting`` reads and
    two ``unknown`` ones land on "neutral", presenting missing evidence as
    evidence of normality. Any cut is adverse on partial evidence; only a fully
    known, uniformly rising basket is supportive.
    """
    states = [str((capex or {}).get(name, "unknown")).strip().lower() for name in names]
    known = [state for state in states if state in ("raising", "holding", "cutting")]
    if not known:
        return "unknown"
    if "cutting" in known:
        return "adverse"
    if "holding" in known:
        return "neutral"
    return "supportive"


def _reaction_signal(good_news_stock_down: str | None) -> str:
    """Good earnings being sold is a late-cycle tell; not being sold is healthy.

    Anything that is not one of the three observed categories -- including the
    explicit ``"unknown"`` an extraction failure now writes -- falls through to
    ``unknown`` rather than to ``mixed``. A parse error is not a reading.
    """
    return {
        "yes": "adverse",
        "no": "supportive",
        "mixed": "neutral",
    }.get(str(good_news_stock_down or "").strip().lower(), "unknown")


def combine_fundamental_signals(capex_signal: str, reaction_signal: str) -> str:
    """Confluence, not arithmetic: precedence over the two categorical reads.

    ``unknown`` is deliberately unreachable by combination -- it survives only
    when *nothing* was observed. A single adverse read carries, because partial
    evidence of deterioration is still evidence of deterioration; supportive
    requires every observed signal to agree.
    """
    signals = (capex_signal, reaction_signal)
    if "adverse" in signals:
        return "adverse"
    observed = [signal for signal in signals if signal != "unknown"]
    if not observed:
        return "unknown"
    return "supportive" if all(signal == "supportive" for signal in observed) else "neutral"


def _usable_context(observed: bool, pending: bool, stale: bool, state: str) -> bool:
    """Whether a fundamental reading may count as evidence.

    One definition, called by both the point-in-time record and the live
    reading, because they publish the same field name to the same consumers and
    a second copy would drift. Distinct from `available`, which is about timing
    alone: an observation whose extraction failed on everything is effective and
    fresh, and still knows nothing.
    """
    return observed and not pending and not stale and state != "unknown"


def _evidence_quality(
    capex: dict[str, str] | None,
    good_news_stock_down: str | None,
    names: list[str],
    *,
    observed: bool,
    stale: bool,
    source: str | None,
) -> str:
    """How much to trust the state above, as one field the reader can act on.

    Ordered by what an operator most needs to know: nothing collected beats
    everything else, then a reading too old to be current, then a hand override,
    then completeness.
    """
    if not observed:
        return "unavailable"
    if stale:
        return "stale"
    if str(source or "").strip().lower() == "manual":
        return "manual"
    known = sum(
        1
        for name in names
        if str((capex or {}).get(name, "unknown")).strip().lower() != "unknown"
    )
    # `bool(names)` matters: with an empty basket `known == len(names)` is
    # vacuously true, so nothing observed would grade as complete.
    complete = (
        bool(names)
        and known == len(names)
        and _reaction_signal(good_news_stock_down) != "unknown"
    )
    return "complete" if complete else "partial"


def _sensor(sensor_id: str, label: str, score: float | None, **details: object) -> dict:
    return {
        "id": sensor_id,
        "label": label,
        "score": round(score, 1) if score is not None else None,
        "available": score is not None,
        "details": details,
    }


def _score_pillars(
    pillars: list[dict],
    weights: dict[str, float],
    bands: tuple[float, float, float] = STATE_BANDS,
) -> dict:
    expected = sum(max(0.0, float(w)) for w in weights.values())
    available_weight = sum(
        max(0.0, float(weights.get(p["id"], 0.0)))
        for p in pillars
        if p.get("score") is not None
    )
    coverage = available_weight / expected * 100.0 if expected else 0.0
    score = None
    if available_weight:
        score = sum(
            float(p["score"]) * float(weights.get(p["id"], 0.0))
            for p in pillars
            if p.get("score") is not None
        ) / available_weight

    rows: list[dict] = []
    for pillar in pillars:
        row = dict(pillar)
        weight = float(weights.get(row["id"], 0.0))
        row["weight"] = weight
        row["available"] = row.get("score") is not None
        row["contribution"] = (
            round(float(row["score"]) * weight / available_weight, 2)
            if row["available"] and available_weight
            else 0.0
        )
        rows.append(row)

    rounded = round(score, 1) if score is not None else None
    return {
        "score": rounded,
        "band": (
            band_for(rounded, bands)
            if rounded is not None and coverage >= MIN_COVERAGE
            else None
        ),
        "bands": {"watch": bands[0], "elevated": bands[1], "breaking": bands[2]},
        "coverage": round(coverage, 1),
        "minimum_coverage": MIN_COVERAGE,
        "available_pillars": [p["id"] for p in rows if p["available"]],
        "pillars": rows,
    }


# ---------------------------------------------------------------------------
# Point-in-time helpers
# ---------------------------------------------------------------------------

def _closes_asof(series: Series, as_of: date) -> list[float]:
    return [v for d, v in series if d <= as_of]


def _item_asof(series: Series | None, as_of: date) -> tuple[date, float] | None:
    if not series:
        return None
    chosen: tuple[date, float] | None = None
    for item in series:
        if item[0] <= as_of:
            chosen = item
        else:
            break
    return chosen


def _value_asof(series: Series | None, as_of: date) -> float | None:
    item = _item_asof(series, as_of)
    return item[1] if item else None


def _window_asof(series: Series | None, as_of: date, days: int) -> list[float]:
    if not series:
        return []
    start = as_of - timedelta(days=days)
    return [v for d, v in series if start <= d <= as_of]


def _coverage_days(series: Series | None, as_of: date) -> int | None:
    """Span of history actually available at ``as_of``.

    Recorded in every snapshot because the v2 credit percentile degraded from a
    10-year to a 3-year reference silently when the upstream licence changed --
    nothing asserted the window it claimed, so nothing noticed for months.
    """
    dates = [d for d, _ in series or [] if d <= as_of]
    return (as_of - dates[0]).days if dates else None


def _next_weekday(d: date) -> date:
    candidate = d + timedelta(days=1)
    while candidate.weekday() >= 5:
        candidate += timedelta(days=1)
    return candidate


def _parse_date(value: object) -> date | None:
    if not value:
        return None
    try:
        return date.fromisoformat(str(value)[:10])
    except ValueError:
        return None


def _fundamental_effective_date(overrides: dict) -> date | None:
    explicit = _parse_date(overrides.get("effective_date"))
    if explicit:
        return explicit
    fetched = _parse_date(overrides.get("fetched_at"))
    return _next_weekday(fetched) if fetched else None


def _overlay_timing(
    overrides: dict, config: dict, as_of: date
) -> tuple[date | None, bool, int | None, bool]:
    """Shared effective-date arithmetic: (effective, pending, age_days, stale)."""
    effective = _fundamental_effective_date(overrides)
    pending = effective is None or as_of < effective
    age = None if pending else (as_of - effective).days
    stale = bool(age is not None and age > int(config.get("fundamental_staleness_days", 80)))
    return effective, pending, age, stale


def fundamental_context(overrides: dict, config: dict, as_of: date) -> dict:
    """Point-in-time fundamental channel. Never a term in State or Warning.

    Called an "overlay" until 2026-08-12, which undersold it: it is the third
    channel of the model, read alongside the two scores by confluence rather than
    decorating them. The categorical ``state`` is what a reader and the chart
    consume; ``evidence_quality`` is how far to trust it.

    Both are derived from the stored categorical facts by fixed rules, not from
    an LLM's numeric judgement. The LLM's job is extraction and explanation --
    find the capex guidance, classify it, cite it -- and the rules turn those
    facts into a state, so the same observation always yields the same category.

    The effective-date gate stays even though nothing is scored from this: the
    rebuild replays historical dates, and stamping today's read onto 2024
    snapshots would be plain lookahead in the stored record.

    This is the *record*. For "what do we know right now", use
    ``current_observation`` -- do not add a bypass flag here, because this runs
    for every replayed date during a rebuild.
    """
    effective, pending, age, stale = _overlay_timing(overrides, config, as_of)
    names = list(config["tickers"]["hyperscalers"])
    capex = None if pending else overrides.get("capex")
    reaction = None if pending else overrides.get("good_news_stock_down")
    observed = not pending and bool(overrides.get("fetched_at"))

    capex_signal = _capex_signal(capex, names) if observed else "unknown"
    reaction_signal = _reaction_signal(reaction) if observed else "unknown"
    state = combine_fundamental_signals(capex_signal, reaction_signal)
    return {
        "state": state,
        "evidence_quality": _evidence_quality(
            capex, reaction, names,
            observed=observed, stale=stale, source=overrides.get("source"),
        ),
        "capex_signal": capex_signal,
        "reaction_signal": reaction_signal,
        # Two different questions, and conflating them is a trap:
        #
        # `available` is about *timing* -- there is an effective, non-stale record
        # to display. `usable` is about *content* -- it also actually says
        # something. A collected observation whose extraction failed on every
        # hyperscaler is available (show it, with its date) but not usable: it
        # knows nothing, so it must never count as evidence.
        #
        # The distinction is load-bearing for the event study. Coverage is
        # measured in sessions with usable context, and if repeated extraction
        # failures counted, they would slowly accumulate "exposure" until the
        # fundamental rows flipped to measurable 0/8 -- a failed result reported
        # for a channel that never knew anything, which is the exact confusion
        # coverage-matching exists to prevent.
        "available": not pending and not stale,
        "usable": _usable_context(observed, pending, stale, state),
        "pending": pending,
        "stale": stale,
        "effective_date": effective.isoformat() if effective else None,
        "age_days": age,
        "capex": capex,
        "good_news_stock_down": reaction,
        "capex_stress": None if pending else overrides.get("f1_score"),
        "earnings_stress": None if pending else overrides.get("f3_score"),
        "reasoning": None if pending else overrides.get("reasoning"),
        "source": overrides.get("source"),
        "fetched_at": overrides.get("fetched_at"),
    }


def current_observation(overrides: dict, config: dict, as_of: date) -> dict:
    """The observation as it stands now, for the live reading only.

    Same shape as ``fundamental_context``, but the effective date is *reported*
    rather than used to blank the content.  A refresh stamps
    ``_next_weekday(today)``, so gating the live card hid a just-collected read
    for one day -- three over a weekend -- and refreshing appeared to do
    nothing.  Nothing here is scored, so showing it early cannot leak into a
    published number; the stored snapshot keeps the gate.
    """
    effective, pending, age, stale = _overlay_timing(overrides, config, as_of)
    # The default override carries "unknown" placeholders for every
    # hyperscaler.  Those are the absence of an observation, not an observation
    # of absence, and must never be presented as collected.  ``fetched_at`` is
    # the collection timestamp and is the only field written on every path that
    # produces real content (LLM refresh and manual save both stamp it).
    observed = bool(overrides.get("fetched_at"))
    names = list(config["tickers"]["hyperscalers"])
    capex_signal = _capex_signal(overrides.get("capex"), names) if observed else "unknown"
    reaction_signal = (
        _reaction_signal(overrides.get("good_news_stock_down")) if observed else "unknown"
    )
    state = combine_fundamental_signals(capex_signal, reaction_signal)
    return {
        "observed": observed,
        "state": state,
        "evidence_quality": _evidence_quality(
            overrides.get("capex"), overrides.get("good_news_stock_down"), names,
            observed=observed, stale=stale, source=overrides.get("source"),
        ),
        "capex_signal": capex_signal,
        "reaction_signal": reaction_signal,
        # Same shape as the record means the same *fields*, not just the same
        # ones this function happens to need: the frontend types both payloads
        # identically, so an omission here is an undefined at runtime that
        # TypeScript cannot catch across a trusted server boundary.
        #
        # Note this is stricter than the `available` directly below: a pending
        # observation is the freshest thing we have and worth showing, but it is
        # not yet in force, so it is not yet evidence.
        "usable": _usable_context(observed, pending, stale, state),
        # Live availability is about usefulness, not effectiveness: a pending
        # observation is the freshest thing we have -- but nothing collected is
        # never available.
        "available": observed and not stale,
        "pending": pending,
        "stale": stale,
        "effective_date": effective.isoformat() if effective else None,
        "age_days": age,
        "capex": overrides.get("capex") if observed else None,
        "good_news_stock_down": overrides.get("good_news_stock_down") if observed else None,
        "capex_stress": overrides.get("f1_score") if observed else None,
        "earnings_stress": overrides.get("f3_score") if observed else None,
        "reasoning": overrides.get("reasoning") if observed else None,
        "source": overrides.get("source"),
        "fetched_at": overrides.get("fetched_at"),
    }


def _basket_hash(symbols: list[str]) -> str:
    canonical = ",".join(sorted({s.strip().upper() for s in symbols if s.strip()}))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:12]


def _mapping_series(values: dict[date, float]) -> Series:
    return sorted(values.items(), key=lambda item: item[0])


def _compute_index(
    prices: dict[str, Series],
    vix_series: Series | None,
    oas_series: Series | None,
    overrides: dict,
    config: dict,
    as_of: date,
    breadth_series: Series | None = None,
    divergence_series: Series | None = None,
    breadth_counts: dict[date, int] | None = None,
    observations: list[dict] | None = None,
) -> dict:
    """Compute the complete State/Warning snapshot as of one trading date.

    ``observations`` is the point-in-time fundamental series and is authoritative
    when supplied; ``overrides`` is the single-slot fallback for callers that
    predate the table (the calibration harness). Either way the reading is scored
    into the same ``fundamental_context`` -- only where it is read from differs,
    so the live monitor and the event study cannot report different states.
    """
    tickers = config["tickers"]
    smh = _closes_asof(prices.get(tickers["leaders"][0], []), as_of)
    qqq = _closes_asof(prices.get(tickers["confirm"][0], []), as_of)
    spy = _closes_asof(prices.get(tickers["market"], []), as_of)

    p1 = p1_trend_break(smh, qqq)
    p2 = p2_death_cross(smh, qqq)
    p3 = p3_drawdown(smh, qqq)
    price_values = [v for v in (p1, p2, p3) if v is not None]
    price_score = max(price_values) if price_values else None

    breadth_item = _item_asof(breadth_series, as_of)
    breadth_pct = breadth_item[1] if breadth_item else None
    breadth_score = breadth_level_score(breadth_pct)
    vix_item = _item_asof(vix_series, as_of)
    vix_score = p5_volatility(vix_item[1] if vix_item else None)
    oas_item = _item_asof(oas_series, as_of)
    oas_window = _window_asof(oas_series, as_of, HY_OAS_WINDOW_DAYS)
    credit_score = f2_credit_spreads(oas_window)

    divergence = _value_asof(divergence_series, as_of)
    sensors = warning_sensor_scores(divergence, smh, spy, oas_window)
    relative_strength = sensors["relative_strength"]
    credit_impulse = sensors["credit_impulse"]
    observation = (
        observation_asof(observations, as_of) if observations is not None else overrides
    ) or {}
    context = fundamental_context(observation, config, as_of)

    state_pillars = [
        {
            "id": "price",
            "label": "Price structure",
            "score": round(price_score, 1) if price_score is not None else None,
            "sensors": [
                _sensor("P1", "Trend break (200-DMA)", p1),
                _sensor("P2", "Death cross + slope", p2),
                _sensor("P3", "Drawdown from 52w high", p3),
            ],
        },
        {
            "id": "breadth",
            "label": "Breadth level",
            "score": round(breadth_score, 1) if breadth_score is not None else None,
            "sensors": [_sensor("B1", "% basket above 200-DMA", breadth_score, pct_above_200=breadth_pct)],
        },
        {
            "id": "credit",
            "label": "Credit level",
            "score": round(credit_score, 1) if credit_score is not None else None,
            "sensors": [_sensor("C1", "HY option-adjusted spread", credit_score, oas=oas_item[1] if oas_item else None)],
        },
        {
            "id": "volatility",
            "label": "Volatility level",
            "score": round(vix_score, 1) if vix_score is not None else None,
            "sensors": [_sensor("V1", "VIX level", vix_score, vix=vix_item[1] if vix_item else None)],
        },
    ]
    warning_pillars = [
        {
            "id": "breadth_divergence",
            "label": "Breadth divergence",
            "score": round(divergence, 1) if divergence is not None else None,
            "sensors": [_sensor("W1", "Price holding while breadth narrows", divergence)],
        },
        {
            "id": "relative_strength",
            "label": "SMH/SPY rollover",
            "score": round(relative_strength, 1) if relative_strength is not None else None,
            "sensors": [_sensor("W2", "60-session relative-strength deterioration", relative_strength)],
        },
        {
            "id": "credit_impulse",
            "label": "Credit impulse",
            "score": round(credit_impulse, 1) if credit_impulse is not None else None,
            "sensors": [
                _sensor(
                    "W3",
                    f"HY OAS {W3_OAS_LOOKBACK}-session widening",
                    credit_impulse,
                    oas=oas_item[1] if oas_item else None,
                )
            ],
        },
    ]

    state = _score_pillars(state_pillars, STATE_WEIGHTS, STATE_BANDS)
    warning = _score_pillars(warning_pillars, WARNING_WEIGHTS, WARNING_BANDS)

    price_item = _item_asof(prices.get(tickers["leaders"][0]), as_of)
    dated_sources = {
        "price": price_item[0] if price_item else None,
        "breadth": breadth_item[0] if breadth_item else None,
        "vix": vix_item[0] if vix_item else None,
        "credit": oas_item[0] if oas_item else None,
    }
    source_ages = {
        key: (as_of - d).days for key, d in dated_sources.items() if d is not None
    }
    stale_inputs = [key for key, age in source_ages.items() if age > SOURCE_MAX_LAG_DAYS]
    basket = list(config["breadth_basket"])
    basket_count = None
    if breadth_counts and breadth_item:
        basket_count = breadth_counts.get(breadth_item[0])

    return {
        "methodology": METHODOLOGY,
        # Not part of the history filter -- only the reseed trigger.
        "sensor_revision": SENSOR_REVISION,
        "date": as_of.isoformat(),
        "state": state,
        "warning": warning,
        "fundamental_context": context,
        "quadrant_config": {
            "state_divider": QUADRANT_STATE_DIVIDER,
            "warning_divider": QUADRANT_WARNING_DIVIDER,
            "margin": QUADRANT_MARGIN,
        },
        "basket": {
            "symbols": basket,
            "hash": _basket_hash(basket),
            "basket_asof": config["basket_asof"],
            "members_available": basket_count,
            "members_expected": len(basket),
            "history_kind": "forward" if as_of >= date.fromisoformat(config["basket_asof"]) else "retrospective",
        },
        "inputs": {
            "vix": round(vix_item[1], 2) if vix_item else None,
            "vix_date": vix_item[0].isoformat() if vix_item else None,
            "hy_oas": round(oas_item[1], 2) if oas_item else None,
            "hy_oas_date": oas_item[0].isoformat() if oas_item else None,
            "breadth_pct_above_200": round(breadth_pct, 1) if breadth_pct is not None else None,
            "breadth_date": breadth_item[0].isoformat() if breadth_item else None,
            "fundamentals_fetched_at": overrides.get("fetched_at"),
            "fundamentals_effective_date": context.get("effective_date"),
            "fundamentals_age_days": context.get("age_days"),
        },
        "data_quality": {
            "minimum_coverage": MIN_COVERAGE,
            "oldest_market_input_age_days": max(source_ages.values()) if source_ages else None,
            "stale_inputs": stale_inputs,
            "inputs_fresh": not stale_inputs,
            # Upstream history spans, so a provider silently truncating a series
            # shows up in the record instead of quietly reshaping a sensor.
            "credit_history_days": _coverage_days(oas_series, as_of),
            "vix_history_days": _coverage_days(vix_series, as_of),
        },
    }


# ---------------------------------------------------------------------------
# Configuration and fundamental storage
# ---------------------------------------------------------------------------

def _normalise_basket(symbols: list[str]) -> list[str]:
    cleaned = [str(s).strip().upper().replace(".", "-") for s in symbols if str(s).strip()]
    if len(cleaned) != len(set(cleaned)):
        raise ValidationError("Breadth basket symbols must be unique")
    if not 20 <= len(cleaned) <= 100:
        raise ValidationError("Breadth basket must contain between 20 and 100 symbols")
    return cleaned


async def get_regime_config(db: AsyncSession) -> dict:
    cfg = json.loads(json.dumps(DEFAULT_CONFIG))
    raw = await settings_store.get_value(db, KEY_CONFIG)
    if raw:
        try:
            stored = json.loads(raw)
            if isinstance(stored.get("breadth_basket"), list):
                cfg["breadth_basket"] = _normalise_basket(stored["breadth_basket"])
            if stored.get("basket_asof"):
                cfg["basket_asof"] = str(stored["basket_asof"])
            if stored.get("fundamental_staleness_days") is not None:
                cfg["fundamental_staleness_days"] = int(stored["fundamental_staleness_days"])
        except (TypeError, ValueError, ValidationError):
            logger.warning("Corrupt %s; using defaults", KEY_CONFIG)
    return cfg


async def update_regime_config(db: AsyncSession, updates: dict) -> dict:
    cfg = await get_regime_config(db)
    if "breadth_basket" in updates:
        basket = _normalise_basket(updates["breadth_basket"])
        if basket != cfg["breadth_basket"]:
            cfg["breadth_basket"] = basket
            cfg["basket_asof"] = date.today().isoformat()
    if "fundamental_staleness_days" in updates:
        days = int(updates["fundamental_staleness_days"])
        if not 30 <= days <= 180:
            raise ValidationError("Fundamental staleness must be between 30 and 180 days")
        cfg["fundamental_staleness_days"] = days
    await update_setting(db, KEY_CONFIG, json.dumps(cfg))
    return cfg


async def get_fundamental_overrides(db: AsyncSession) -> dict:
    names = DEFAULT_CONFIG["tickers"]["hyperscalers"]
    default = {
        "methodology": METHODOLOGY,
        "f1_score": None,
        "f3_score": None,
        "capex": {name: "unknown" for name in names},
        "good_news_stock_down": "unknown",
        "locked": False,
        "reasoning": None,
        "fetched_at": None,
        "effective_date": None,
        "source": "default",
    }
    raw = await settings_store.get_value(db, KEY_FUNDAMENTALS)
    if not raw:
        return default
    try:
        stored = json.loads(raw)
    except (TypeError, ValueError):
        return default
    # The guard rejects pre-v2 blobs, where f1/f3 were arbitrary numbers with no
    # categorical source. v2 and v3 share the categorical format and both derive
    # f1/f3 from it below, so a methodology bump must not discard a live
    # observation -- only the capex *scale* changed, and that is recomputed.
    if stored.get("methodology") not in CATEGORICAL_FUNDAMENTAL_METHODOLOGIES:
        return default
    capex = _normalise_capex_states(stored.get("capex"), names)
    reaction = str(stored.get("good_news_stock_down", "unknown")).strip().lower()
    if reaction not in GNSD_STATES:
        reaction = "unknown"
    return {
        **default,
        **stored,
        "methodology": METHODOLOGY,
        "f1_score": _score_capex_states(capex, names),
        "f3_score": _GNSD_SCORES.get(reaction),
        "capex": capex,
        "good_news_stock_down": reaction,
    }


def _normalise_capex_states(
    raw: object,
    names: list[str],
    *,
    strict: bool = False,
) -> dict[str, str]:
    values = raw if isinstance(raw, dict) else {}
    if strict and set(values) != set(names):
        raise ValidationError(
            f"Capex override must contain exactly: {', '.join(names)}"
        )
    out: dict[str, str] = {}
    for name in names:
        state = str(values.get(name, "unknown")).strip().lower()
        if state not in CAPEX_STATES:
            if strict:
                raise ValidationError(f"Invalid capex state for {name}: {state}")
            state = "unknown"
        out[name] = state
    return out


def _score_capex_states(capex: dict[str, str], names: list[str]) -> float | None:
    scores = [_CAPEX_STATE_SCORES[capex[name]] for name in names if capex[name] in _CAPEX_STATE_SCORES]
    score = _mean(scores) if len(scores) >= 3 else None
    return round(score, 1) if score is not None else None


async def record_fundamental_observation(db: AsyncSession, observation: dict) -> None:
    """Append the observation to the point-in-time series, keyed on effective date.

    Upsert rather than insert: re-saving on the same effective date is a
    correction to that day's reading, not a second observation of it.

    Silently does nothing without an effective date or a ``fetched_at``. Those
    are the default placeholder blob -- the absence of an observation, which must
    never enter the series as though someone had looked.

    Deliberately does **not** commit. ``update_regime_monitor`` calls this inside
    a run that owns its transaction and commits once after the snapshot loop;
    committing here would take that boundary away from it. The two override
    writers commit for themselves.
    """
    effective = _parse_date(observation.get("effective_date"))
    fetched_raw = observation.get("fetched_at")
    if effective is None or not fetched_raw:
        return
    try:
        fetched = datetime.fromisoformat(str(fetched_raw))
    except ValueError:
        fetched = datetime.now(timezone.utc)
    if fetched.tzinfo is None:
        fetched = fetched.replace(tzinfo=timezone.utc)

    existing = await db.execute(
        select(RegimeFundamentalObservation).where(
            RegimeFundamentalObservation.effective_date == effective
        )
    )
    row = existing.scalar_one_or_none()
    payload = {
        "f1_score": observation.get("f1_score"),
        "f3_score": observation.get("f3_score"),
        "capex_json": json.dumps(observation.get("capex") or {}),
        "good_news_stock_down": str(observation.get("good_news_stock_down") or "unknown")[:10],
        "reasoning": observation.get("reasoning"),
        "source": str(observation.get("source") or "unknown")[:30],
        "fetched_at": fetched,
    }
    if row is None:
        db.add(RegimeFundamentalObservation(
            effective_date=effective,
            created_at=datetime.now(timezone.utc),
            **payload,
        ))
    else:
        for key, value in payload.items():
            setattr(row, key, value)


async def get_fundamental_observations(db: AsyncSession) -> list[dict]:
    """The whole observation series, oldest first, for point-in-time scoring."""
    result = await db.execute(
        select(RegimeFundamentalObservation).order_by(
            RegimeFundamentalObservation.effective_date.asc()
        )
    )
    out: list[dict] = []
    for row in result.scalars().all():
        try:
            capex = json.loads(row.capex_json)
        except (TypeError, ValueError):
            capex = {}
        out.append({
            "effective_date": row.effective_date,
            "f1_score": row.f1_score,
            "f3_score": row.f3_score,
            "capex": capex,
            "good_news_stock_down": row.good_news_stock_down,
            "reasoning": row.reasoning,
            "source": row.source,
            "fetched_at": row.fetched_at.isoformat() if row.fetched_at else None,
        })
    return out


def observation_asof(observations: list[dict] | None, as_of: date) -> dict | None:
    """Latest observation effective on or before ``as_of``.

    This *is* the effective-date gate now. The settings-blob version had to
    recompute it per call because there was only ever one observation to gate;
    with a series, "which reading was live that day" is just a lookup.
    """
    chosen: dict | None = None
    for observation in observations or []:
        if observation["effective_date"] <= as_of:
            chosen = observation
        else:
            break
    return chosen


async def set_fundamental_overrides(
    db: AsyncSession,
    capex: dict[str, str] | None = None,
    good_news_stock_down: str | None = None,
    locked: bool | None = None,
) -> dict:
    current = await get_fundamental_overrides(db)
    observation_changed = capex is not None or good_news_stock_down is not None
    if capex is not None:
        names = DEFAULT_CONFIG["tickers"]["hyperscalers"]
        normalised = _normalise_capex_states(capex, names, strict=True)
        current["capex"] = normalised
        current["f1_score"] = _score_capex_states(normalised, names)
    if good_news_stock_down is not None:
        reaction = good_news_stock_down.strip().lower()
        if reaction not in GNSD_STATES:
            raise ValidationError(f"Invalid good-news-stock-down state: {reaction}")
        current["good_news_stock_down"] = reaction
        current["f3_score"] = _GNSD_SCORES.get(reaction)
    if locked is not None:
        current["locked"] = bool(locked)
    elif observation_changed:
        current["locked"] = True
    if observation_changed:
        now = datetime.now(timezone.utc)
        current.update({
            "methodology": METHODOLOGY,
            "source": "manual",
            "reasoning": None,
            "fetched_at": now.isoformat(),
            "effective_date": _next_weekday(now.date()).isoformat(),
        })
    # The blob (what the live card reads) and the series row (what the
    # point-in-time replay reads) are the same observation. Committed together:
    # `update_setting` commits internally, so using it here would leave a window
    # where a failure publishes the reading to the card but not to the record,
    # and the two would disagree permanently with nothing to detect it.
    await settings_store.upsert_setting(db, KEY_FUNDAMENTALS, json.dumps(current))
    if observation_changed:
        await record_fundamental_observation(db, current)
    await db.commit()
    return current


# ---------------------------------------------------------------------------
# External data fetching
# ---------------------------------------------------------------------------

def _price_symbols(config: dict) -> list[str]:
    tickers = config["tickers"]
    symbols = list(tickers["leaders"]) + list(tickers["confirm"]) + [tickers["market"]]
    return list(dict.fromkeys(s for s in symbols if s))


async def _fetch_prices(config: dict, start: date, end: date) -> dict[str, Series]:
    if not settings.alpaca_api_key or not settings.alpaca_api_secret:
        return {}
    provider = AlpacaOHLCVProvider(settings.alpaca_api_key, settings.alpaca_api_secret)
    out: dict[str, Series] = {}
    for symbol in _price_symbols(config):
        try:
            bars = await provider.fetch_ohlcv(symbol, start, end)
            out[symbol] = sorted(((b.date, float(b.close)) for b in bars), key=lambda item: item[0])
        except Exception as exc:
            logger.warning("Risk monitor: price fetch failed for %s: %s", symbol, exc)
    return out


async def _fetch_fred_series(series_id: str, start: date, end: date) -> Series | None:
    if not settings.fred_api_key:
        return None
    verify = _CA_BUNDLE if (_CA_BUNDLE and Path(_CA_BUNDLE).exists()) else True
    params = {
        "series_id": series_id,
        "api_key": settings.fred_api_key,
        "file_type": "json",
        "observation_start": start.isoformat(),
        "observation_end": end.isoformat(),
    }
    try:
        async with httpx.AsyncClient(timeout=30, verify=verify) as client:
            response = await client.get(
                "https://api.stlouisfed.org/fred/series/observations", params=params
            )
            response.raise_for_status()
            payload = response.json()
    except Exception as exc:
        logger.warning("Risk monitor: FRED fetch failed for %s: %s", series_id, exc)
        return None

    out: Series = []
    for observation in payload.get("observations", []):
        value = observation.get("value")
        if value in (None, ".", ""):
            continue
        try:
            out.append((date.fromisoformat(observation["date"]), float(value)))
        except (TypeError, ValueError):
            continue
    return sorted(out, key=lambda item: item[0])


# ---------------------------------------------------------------------------
# Snapshot persistence and reads
# ---------------------------------------------------------------------------

async def _upsert_snapshot(
    db: AsyncSession,
    result: dict,
    *,
    rewrite_existing: bool,
) -> tuple[bool, dict]:
    snapshot_date = date.fromisoformat(result["date"])
    existing = await db.execute(select(RegimeSnapshot).where(RegimeSnapshot.date == snapshot_date))
    row = existing.scalar_one_or_none()
    state_score = (result.get("state") or {}).get("score")
    state_band = (result.get("state") or {}).get("band")
    payload = json.dumps(result)
    if row is None:
        db.add(RegimeSnapshot(
            date=snapshot_date,
            total_score=float(state_score or 0.0),
            band=state_band or "unavailable",
            breakdown_json=payload,
            created_at=datetime.now(timezone.utc),
        ))
    else:
        existing_parsed = _parse_snapshot(row.breakdown_json)
        if existing_parsed is not None and not rewrite_existing:
            return False, existing_parsed
        row.total_score = float(state_score or 0.0)
        row.band = state_band or "unavailable"
        row.breakdown_json = payload
    return True, result


def _snapshot_revision(snapshot: dict) -> int:
    """Sensor revision of a stored snapshot; pre-marker rows read as 1."""
    try:
        return int(snapshot.get("sensor_revision") or 1)
    except (TypeError, ValueError):
        return 1


def _context_from_legacy_overlay(overlay: dict) -> dict:
    """Rebuild the categorical channel from a pre-rename snapshot's overlay.

    The channel was called ``fundamental_overlay`` until 2026-08-12 and stored
    the same underlying facts -- the capex map, the earnings reaction, the
    effective date. The rename shipped without a methodology bump (no score
    changed), so those rows are still served and were never reseeded: reading
    only the new key would turn every one of them into ``unknown`` and silently
    discard real recorded evidence -- historical Path colours, and any exposure
    the event study could legitimately count.

    Derived, not guessed. The hyperscaler list comes from the overlay's own
    capex keys, which is exactly the basket that was observed at the time rather
    than today's configured one.
    """
    capex = overlay.get("capex") or {}
    reaction = overlay.get("good_news_stock_down")
    names = list(capex)
    pending = bool(overlay.get("pending"))
    stale = bool(overlay.get("stale"))
    observed = not pending and bool(overlay.get("fetched_at"))

    capex_signal = _capex_signal(capex, names) if observed else "unknown"
    reaction_signal = _reaction_signal(reaction) if observed else "unknown"
    state = combine_fundamental_signals(capex_signal, reaction_signal)
    return {
        **overlay,
        "state": state,
        "evidence_quality": _evidence_quality(
            capex, reaction, names,
            observed=observed, stale=stale, source=overlay.get("source"),
        ),
        "capex_signal": capex_signal,
        "reaction_signal": reaction_signal,
        "usable": _usable_context(observed, pending, stale, state),
    }


def _parse_snapshot(raw: str) -> dict | None:
    try:
        parsed = json.loads(raw)
    except (TypeError, ValueError):
        return None
    if parsed.get("methodology") != METHODOLOGY:
        return None
    # Normalise here rather than at each call site: every reader of a stored
    # snapshot goes through this function, so a legacy row cannot reach one of
    # them un-adapted.
    if "fundamental_context" not in parsed and "fundamental_overlay" in parsed:
        parsed["fundamental_context"] = _context_from_legacy_overlay(
            parsed["fundamental_overlay"] or {}
        )
    return parsed


async def _latest_snapshot_row(db: AsyncSession) -> tuple[RegimeSnapshot, dict] | None:
    result = await db.execute(
        select(RegimeSnapshot).order_by(RegimeSnapshot.date.desc()).limit(1000)
    )
    for row in result.scalars().all():
        parsed = _parse_snapshot(row.breakdown_json)
        if parsed is not None:
            return row, parsed
    return None


async def update_regime_monitor(
    db: AsyncSession, rebuild_lookback_days: int = REBUILD_LOOKBACK_DAYS
) -> dict:
    config = await get_regime_config(db)
    overrides = await get_fundamental_overrides(db)
    # Carries the pre-v5 single-slot observation into the series on first run, so
    # a deployment does not lose the live reading. A no-op once recorded, and a
    # no-op for the placeholder blob (no fetched_at).
    await record_fundamental_observation(db, overrides)
    if _fundamentals_stale(overrides, config) and not overrides.get("locked"):
        try:
            overrides = await refresh_fundamental_overrides(db, config=config)
        except Exception as exc:
            logger.warning("Risk monitor: fundamentals refresh skipped: %s", exc)

    end = date.today()
    prices = await _fetch_prices(config, end - timedelta(days=1200), end)
    leader = config["tickers"]["leaders"][0]
    leader_series = prices.get(leader, [])
    if not leader_series:
        return {"available": False, "reason": "no benchmark price data"}
    latest_date = leader_series[-1][0]

    vix_series = await _fetch_fred_series("VIXCLS", end - timedelta(days=1200), end)
    # Asking for 13 years was misleading once the licence capped the series at 3;
    # the level needs the latest point and W3 needs its lookback, nothing more.
    oas_series = await _fetch_fred_series(
        "BAMLH0A0HYM2", end - timedelta(days=HY_OAS_WINDOW_DAYS), end
    )

    basket = config["breadth_basket"]
    try:
        breadth, breadth_counts = await breadth_service.compute_breadth_details(
            db, basket, window=200, min_tickers=20
        )
        divergence = breadth_service.compute_divergence_series(breadth, leader_series)
    except Exception as exc:
        logger.warning("Risk monitor: fixed-basket breadth skipped: %s", exc)
        breadth, breadth_counts, divergence = {}, {}, {}

    latest_snapshot = await _latest_snapshot_row(db)
    # A stored series written under an older sensor revision is reseeded once.
    # Without this, raising HY_OAS_WINDOW_DAYS would only ever reach newly
    # computed rows: routine runs touch the latest date alone, so every older row
    # would keep the credit gap indefinitely.
    rebuilding = bool(leader_series) and (
        latest_snapshot is None
        or _snapshot_revision(latest_snapshot[1]) < SENSOR_REVISION
    )
    if rebuilding:
        floor = end - timedelta(days=rebuild_lookback_days)
        dates = [d for d, _ in leader_series if d >= floor] or [latest_date]
    else:
        # Routine PIT rule: only the latest trading date may be inserted/updated.
        dates = [latest_date]

    breadth_series = _mapping_series(breadth)
    divergence_series = _mapping_series(divergence)
    # Loaded once, after any refresh, so a reseed scores each replayed date with
    # the observation that was effective on it rather than with today's.
    observations = await get_fundamental_observations(db)
    latest_result: dict | None = None
    snapshots_written = 0
    for snapshot_date in dates:
        computed = _compute_index(
            prices,
            vix_series,
            oas_series,
            overrides,
            config,
            snapshot_date,
            breadth_series,
            divergence_series,
            breadth_counts,
            observations=observations,
        )
        written, latest_result = await _upsert_snapshot(
            db,
            computed,
            # True for *every* replayed date on a reseed, or it would write one
            # row and leave the rest at the old revision.
            rewrite_existing=rebuilding or snapshot_date == latest_date,
        )
        snapshots_written += int(written)
    await db.commit()

    logger.info(json.dumps({
        "event": "regime_monitor_updated",
        "methodology": METHODOLOGY,
        "date": latest_result.get("date") if latest_result else None,
        "state": ((latest_result or {}).get("state") or {}).get("score"),
        "warning": ((latest_result or {}).get("warning") or {}).get("score"),
        "snapshots_written": snapshots_written,
    }))
    return latest_result or {"available": False, "reason": "no data"}


async def _result_at_or_before(
    db: AsyncSession,
    target: date,
    basket_hash: str | None = None,
) -> dict | None:
    result = await db.execute(
        select(RegimeSnapshot.breakdown_json)
        .where(RegimeSnapshot.date <= target)
        .order_by(RegimeSnapshot.date.desc())
        .limit(1000)
    )
    for raw in result.scalars().all():
        parsed = _parse_snapshot(raw)
        parsed_hash = ((parsed or {}).get("basket") or {}).get("hash")
        if parsed is not None and (basket_hash is None or parsed_hash == basket_hash):
            return parsed
    return None


def _delta(current: dict, previous: dict | None) -> float | None:
    if not previous:
        return None
    if current.get("available_pillars") != previous.get("available_pillars"):
        return None
    a, b = current.get("score"), previous.get("score")
    return round(a - b, 1) if a is not None and b is not None else None


async def get_regime_monitor(db: AsyncSession) -> dict:
    latest = await _latest_snapshot_row(db)
    if latest is None:
        return {"available": False, "reason": "not computed yet"}
    row, result = latest
    basket_hash = (result.get("basket") or {}).get("hash")
    previous_7 = await _result_at_or_before(
        db, row.date - timedelta(days=7), basket_hash
    )
    previous_30 = await _result_at_or_before(
        db, row.date - timedelta(days=30), basket_hash
    )

    for key in ("state", "warning"):
        block = result.get(key) or {}
        block["trend"] = {
            "delta_7": _delta(block, (previous_7 or {}).get(key)),
            "delta_30": _delta(block, (previous_30 or {}).get(key)),
        }
        result[key] = block

    snapshot_age = (date.today() - row.date).days
    quality = result.get("data_quality") or {}
    quality["snapshot_age_days"] = snapshot_age
    quality["is_fresh"] = bool(quality.get("inputs_fresh")) and snapshot_age <= 4
    result["data_quality"] = quality

    # The snapshot's `fundamental_context` is the point-in-time record; the
    # reader also wants the current observation even when it is not effective
    # until the next session, or refreshing it looks like it did nothing.
    config = await get_regime_config(db)
    overrides = await get_fundamental_overrides(db)
    live = current_observation(overrides, config, date.today())
    # Deliberately reads the *snapshot's* record, not the live one: this is how
    # the reader tells "shown here" from "in the stored record".
    live["observed_in_snapshot"] = bool(
        (result.get("fundamental_context") or {}).get("available")
    )
    # `fundamental_context` is the stored channel and stays the snapshot's;
    # `fundamental_live` is what we know right now. Collapsing the two under one
    # key is what made a just-collected observation look like it had been
    # backdated into history.
    result["fundamental_live"] = live
    result["available"] = True
    return result


async def get_regime_history(db: AsyncSession, days: int = 800) -> list[dict]:
    cutoff = date.today() - timedelta(days=days)
    result = await db.execute(
        select(RegimeSnapshot)
        .where(RegimeSnapshot.date >= cutoff)
        .order_by(RegimeSnapshot.date.asc())
    )
    out: list[dict] = []
    for row in result.scalars().all():
        data = _parse_snapshot(row.breakdown_json)
        if data is None:
            continue
        state, warning = data.get("state") or {}, data.get("warning") or {}
        context = data.get("fundamental_context") or {}
        out.append({
            "date": row.date.isoformat(),
            "state": state.get("score") if state.get("band") is not None else None,
            "warning": warning.get("score") if warning.get("band") is not None else None,
            # The third channel, carried per point so the Path view can colour a
            # dot by the fundamental context that was on the record that day.
            # Rows written before the channel existed carry nothing, which reads
            # as "unknown" -- correct, since nothing was observed then either.
            "fundamental_state": context.get("state") or "unknown",
            "evidence_quality": context.get("evidence_quality") or "unavailable",
            "state_coverage": state.get("coverage"),
            "warning_coverage": warning.get("coverage"),
            "basket_hash": (data.get("basket") or {}).get("hash"),
        })
    if not out:
        return out
    latest_hash = out[-1]["basket_hash"]
    if latest_hash is None:
        return out
    return [point for point in out if point["basket_hash"] == latest_hash]


# ---------------------------------------------------------------------------
# Grounded fundamental extraction
# ---------------------------------------------------------------------------

_CAPEX_PROMPT = """\
You are a markets analyst. Search the web for the MOST RECENT (last reported \
quarter) capital-expenditure (capex) guidance from these hyperscalers: {names}.

For each name, classify forward capex/AI-infrastructure guidance vs. the prior \
quarter as exactly one of: "raising", "holding", "cutting", "unknown".

Also judge the recent good-news-stock-down dynamic across these names and the \
semiconductor sector after earnings/revenue beats. Answer "yes", "no", or "mixed".

Respond ONLY with JSON (no markdown):
{{"capex": {{ {example} }}, "good_news_stock_down": "yes|no|mixed", \
"reasoning": "<2-3 sourced sentences>"}}
"""


def _fundamentals_stale(overrides: dict, config: dict) -> bool:
    fetched = overrides.get("fetched_at")
    if not fetched:
        return True
    try:
        timestamp = datetime.fromisoformat(fetched)
    except (TypeError, ValueError):
        return True
    if timestamp.tzinfo is None:
        timestamp = timestamp.replace(tzinfo=timezone.utc)
    return datetime.now(timezone.utc) - timestamp > timedelta(
        days=int(config.get("fundamental_staleness_days", 80))
    )


def _strip_fences(text: str) -> str:
    clean = (text or "").strip()
    if clean.startswith("```"):
        clean = clean.split("\n", 1)[1] if "\n" in clean else clean[3:]
    if clean.endswith("```"):
        clean = clean[:-3]
    return clean.strip()


def _extract_responses_text(response: object) -> str:
    for item in getattr(response, "output", []) or []:
        if getattr(item, "type", None) == "message" and getattr(item, "content", None):
            for block in item.content:
                if getattr(block, "text", None):
                    return block.text
    return ""


async def _call_llm_json(cfg: dict, prompt: str) -> dict:
    provider, model, api_key = cfg["provider"], cfg["model"], cfg["api_key"]
    base_url = cfg.get("base_url")
    if provider == "gemini":
        from google import genai
        from google.genai import types

        client = genai.Client(api_key=api_key)
        response = await client.aio.models.generate_content(
            model=model,
            contents=prompt,
            config=types.GenerateContentConfig(
                tools=[types.Tool(google_search=types.GoogleSearch())],
                response_mime_type="application/json",
            ),
        )
        return json.loads(_strip_fences(response.text))

    from openai import AsyncOpenAI

    verify = _CA_BUNDLE if (_CA_BUNDLE and Path(_CA_BUNDLE).exists()) else True
    client = AsyncOpenAI(
        api_key=api_key,
        base_url=base_url or None,
        http_client=httpx.AsyncClient(verify=verify),
    )
    if provider in ("openai", "xai"):
        tool = "web_search_preview" if provider == "openai" else "web_search"
        response = await client.responses.create(
            model=model,
            tools=[{"type": tool}],
            instructions="Respond with valid JSON only, no markdown fences.",
            input=prompt,
        )
        return json.loads(_strip_fences(_extract_responses_text(response)))

    response = await client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )
    return json.loads(_strip_fences(response.choices[0].message.content))


async def refresh_fundamental_overrides(
    db: AsyncSession, config: dict | None = None, force: bool = False
) -> dict:
    current = await get_fundamental_overrides(db)
    if current.get("locked") and not force:
        return current

    config = config or await get_regime_config(db)
    llm = await resolve_llm_config(db)
    if not llm.get("api_key"):
        raise ProviderError(f"No API key configured for LLM provider '{llm.get('provider')}'")

    names = config["tickers"]["hyperscalers"]
    example = ", ".join(f'"{name}": "holding"' for name in names)
    parsed = await _call_llm_json(
        llm, _CAPEX_PROMPT.format(names=", ".join(names), example=example)
    )
    raw_capex = parsed.get("capex", {}) if isinstance(parsed, dict) else {}
    capex = _normalise_capex_states(raw_capex, names)
    f1 = _score_capex_states(capex, names)
    reaction = str(parsed.get("good_news_stock_down", "")).strip().lower()
    if reaction not in GNSD_STATES:
        reaction = "unknown"
    f3 = _GNSD_SCORES.get(reaction)
    now = datetime.now(timezone.utc)
    result = {
        "methodology": METHODOLOGY,
        "f1_score": f1,
        "f3_score": f3,
        "capex": capex,
        "good_news_stock_down": reaction or None,
        "reasoning": parsed.get("reasoning") if isinstance(parsed, dict) else None,
        "fetched_at": now.isoformat(),
        "effective_date": _next_weekday(now.date()).isoformat(),
        "locked": False,
        "source": llm.get("provider"),
    }
    # One transaction: see set_fundamental_overrides on why these two writes must
    # not be able to land separately.
    await settings_store.upsert_setting(db, KEY_FUNDAMENTALS, json.dumps(result))
    await record_fundamental_observation(db, result)
    await db.commit()
    logger.info(json.dumps({
        "event": "regime_fundamentals_refreshed",
        "f1": result["f1_score"],
        "f3": result["f3_score"],
        "effective_date": result["effective_date"],
    }))
    return result