1080 lines
38 KiB
Python
1080 lines
38 KiB
Python
"""AI/Tech Regime Monitor v2.
|
|
|
|
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, relative
|
|
strength, and sourced fundamental observations).
|
|
|
|
Daily snapshots are the point-in-time record. The first v2 run rewrites the
|
|
latest ``REBUILD_SESSIONS`` trading sessions once; ordinary runs thereafter only
|
|
upsert the latest trading date. Fundamental observations are never replayed
|
|
before their effective date.
|
|
"""
|
|
|
|
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_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 = "v2"
|
|
REBUILD_SESSIONS = 400
|
|
MIN_COVERAGE = 75.0
|
|
SOURCE_MAX_LAG_DAYS = 7
|
|
|
|
QUADRANT_STATE_DIVIDER = 60.0
|
|
QUADRANT_WARNING_DIVIDER = 60.0
|
|
QUADRANT_MARGIN = 5.0
|
|
|
|
HY_OAS_MILD = 3.5
|
|
HY_OAS_ELEVATED = 5.0
|
|
HY_OAS_STRESSED = 7.0
|
|
HY_OAS_REFERENCE_YEARS = 10.0
|
|
|
|
STATE_WEIGHTS = {
|
|
"price": 40.0,
|
|
"breadth": 25.0,
|
|
"credit": 20.0,
|
|
"volatility": 15.0,
|
|
}
|
|
WARNING_WEIGHTS = {
|
|
"breadth_divergence": 50.0,
|
|
"relative_strength": 30.0,
|
|
"capex": 12.0,
|
|
"earnings_reaction": 8.0,
|
|
}
|
|
|
|
# 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")
|
|
GNSD_STATES = ("yes", "no", "mixed")
|
|
_CAPEX_STATE_SCORES = {"raising": 0.0, "holding": 0.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 band_for(score: float) -> str:
|
|
if score < 30:
|
|
return "stable"
|
|
if score < 60:
|
|
return "watch"
|
|
if score < 80:
|
|
return "elevated"
|
|
return "breaking"
|
|
|
|
|
|
def _under_200(closes: list[float]) -> float | None:
|
|
sma200 = _sma(closes, 200)
|
|
if sma200 is None:
|
|
return None
|
|
return 100.0 if closes[-1] < sma200 else 0.0
|
|
|
|
|
|
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(closes: list[float]) -> float | None:
|
|
if len(closes) < 30:
|
|
return None
|
|
peak = max(closes[-252:])
|
|
if peak <= 0:
|
|
return None
|
|
dd_pct = (peak - closes[-1]) / peak * 100.0
|
|
return _clamp(dd_pct * 5.0)
|
|
|
|
|
|
def p3_drawdown(smh: list[float], qqq: list[float]) -> float | None:
|
|
vals = [v for v in (_drawdown(smh), _drawdown(qqq)) if v is not None]
|
|
return max(vals) if vals else None
|
|
|
|
|
|
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:
|
|
if vix is None:
|
|
return None
|
|
return _clamp((vix - 15.0) / 15.0 * 100.0)
|
|
|
|
|
|
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 stress: 70% named absolute anchors + 30% upper-tail percentile."""
|
|
if not oas_values:
|
|
return None
|
|
latest = oas_values[-1]
|
|
absolute = _oas_absolute_score(latest)
|
|
if len(oas_values) < 30:
|
|
return round(absolute, 2)
|
|
less = sum(1 for v in oas_values if v < latest)
|
|
equal = sum(1 for v in oas_values if v == latest)
|
|
percentile = (less + 0.5 * equal) / len(oas_values) * 100.0
|
|
relative = _clamp((percentile - 50.0) / 45.0 * 100.0)
|
|
return round(absolute * 0.7 + relative * 0.3, 2)
|
|
|
|
|
|
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]) -> 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) if rounded is not None and coverage >= MIN_COVERAGE else None,
|
|
"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, years: float) -> list[float]:
|
|
if not series:
|
|
return []
|
|
start = as_of - timedelta(days=int(365.25 * years))
|
|
return [v for d, v in series if start <= d <= as_of]
|
|
|
|
|
|
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 _fundamental_scores_asof(overrides: dict, config: dict, as_of: date) -> tuple[float | None, float | None, dict]:
|
|
effective = _fundamental_effective_date(overrides)
|
|
if effective is None or as_of < effective:
|
|
return None, None, {"effective_date": effective.isoformat() if effective else None, "age_days": None}
|
|
age = (as_of - effective).days
|
|
stale = age > int(config.get("fundamental_staleness_days", 80))
|
|
f1 = overrides.get("f1_score")
|
|
f3 = overrides.get("f3_score")
|
|
return (
|
|
None if stale or f1 is None else _clamp(float(f1)),
|
|
None if stale or f3 is None else _clamp(float(f3)),
|
|
{"effective_date": effective.isoformat(), "age_days": age, "stale": stale},
|
|
)
|
|
|
|
|
|
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,
|
|
) -> dict:
|
|
"""Compute the complete v2 State/Warning snapshot as of one trading date."""
|
|
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_REFERENCE_YEARS)
|
|
credit_score = f2_credit_spreads(oas_window)
|
|
|
|
divergence = _value_asof(divergence_series, as_of)
|
|
relative_strength = p4_relative_strength(smh, spy)
|
|
f1, f3, fundamental_meta = _fundamental_scores_asof(overrides, 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": "capex",
|
|
"label": "Hyperscaler capex revisions",
|
|
"score": round(f1, 1) if f1 is not None else None,
|
|
"sensors": [_sensor("F1", "Capex guidance cuts", f1)],
|
|
},
|
|
{
|
|
"id": "earnings_reaction",
|
|
"label": "Good news, stock down",
|
|
"score": round(f3, 1) if f3 is not None else None,
|
|
"sensors": [_sensor("F3", "Abnormal earnings reaction", f3)],
|
|
},
|
|
]
|
|
|
|
state = _score_pillars(state_pillars, STATE_WEIGHTS)
|
|
warning = _score_pillars(warning_pillars, WARNING_WEIGHTS)
|
|
|
|
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,
|
|
"date": as_of.isoformat(),
|
|
"state": state,
|
|
"warning": warning,
|
|
"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": fundamental_meta.get("effective_date"),
|
|
"fundamentals_age_days": fundamental_meta.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,
|
|
},
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 v2 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": "mixed",
|
|
"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
|
|
if stored.get("methodology") != METHODOLOGY:
|
|
return default
|
|
capex = _normalise_capex_states(stored.get("capex"), names)
|
|
reaction = str(stored.get("good_news_stock_down", "mixed")).strip().lower()
|
|
if reaction not in GNSD_STATES:
|
|
reaction = "mixed"
|
|
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 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(),
|
|
})
|
|
await update_setting(db, KEY_FUNDAMENTALS, json.dumps(current))
|
|
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("Regime 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("Regime 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_v2: 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_v2 = _parse_v2(row.breakdown_json)
|
|
if existing_v2 is not None and not rewrite_existing_v2:
|
|
return False, existing_v2
|
|
row.total_score = float(state_score or 0.0)
|
|
row.band = state_band or "unavailable"
|
|
row.breakdown_json = payload
|
|
return True, result
|
|
|
|
|
|
def _parse_v2(raw: str) -> dict | None:
|
|
try:
|
|
parsed = json.loads(raw)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return parsed if parsed.get("methodology") == METHODOLOGY else None
|
|
|
|
|
|
async def _latest_v2_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_v2(row.breakdown_json)
|
|
if parsed is not None:
|
|
return row, parsed
|
|
return None
|
|
|
|
|
|
async def update_regime_monitor(db: AsyncSession, rebuild_sessions: int = REBUILD_SESSIONS) -> dict:
|
|
config = await get_regime_config(db)
|
|
overrides = await get_fundamental_overrides(db)
|
|
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("Regime 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)
|
|
oas_series = await _fetch_fred_series(
|
|
"BAMLH0A0HYM2", end - timedelta(days=int(365.25 * 13)), 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("Regime monitor: fixed-basket breadth skipped: %s", exc)
|
|
breadth, breadth_counts, divergence = {}, {}, {}
|
|
|
|
latest_v2 = await _latest_v2_row(db)
|
|
rebuilding = latest_v2 is None and bool(leader_series)
|
|
if rebuilding:
|
|
dates = [d for d, _ in leader_series[-max(1, rebuild_sessions):]]
|
|
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)
|
|
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,
|
|
)
|
|
written, latest_result = await _upsert_snapshot(
|
|
db,
|
|
computed,
|
|
rewrite_existing_v2=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_v2(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_v2_row(db)
|
|
if latest is None:
|
|
return {"available": False, "reason": "v2 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
|
|
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_v2(row.breakdown_json)
|
|
if data is None:
|
|
continue
|
|
state, warning = data.get("state") or {}, data.get("warning") 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,
|
|
"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 = "mixed"
|
|
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"),
|
|
}
|
|
await update_setting(db, KEY_FUNDAMENTALS, json.dumps(result))
|
|
logger.info(json.dumps({
|
|
"event": "regime_fundamentals_refreshed",
|
|
"f1": result["f1_score"],
|
|
"f3": result["f3_score"],
|
|
"effective_date": result["effective_date"],
|
|
}))
|
|
return result
|