feat: curve regime

This commit is contained in:
OpenSquared
2026-07-25 09:23:26 +02:00
parent 3addea6850
commit ad7ab35d1d
8 changed files with 177 additions and 31 deletions

View File

@@ -395,3 +395,34 @@ _CATEGORY_TO_ASSET_BIAS_KEY = {
def _macro_bias_for_category(asset_bias: Dict[str, str], category: Optional[str]) -> Optional[str]:
key = _CATEGORY_TO_ASSET_BIAS_KEY.get(category or "")
return asset_bias.get(key) if key else None
async def refresh_watchlist_curve_regimes() -> int:
"""Classify every Cockpit Watchlist instrument and cache the result — called from
services.wavelet_scheduler's periodic refresh pass, same cadence as the wavelet cache,
so the Dashboard's Curve Regime card never needs a live classify call per page view and
always agrees with what Instrument Analysis's Curve Regime tab would compute right now.
Auto quick-adds a Watchlist ticker into the Instrument Analysis catalog first if it
isn't there yet (services.instrument_service.quick_add_instrument_from_watchlist) —
classify_instrument_regime needs a full snapshot (price history + indicators), which
only exists for catalog-registered instruments. Returns the number classified."""
from services.database import get_instruments_watchlist, save_curve_regime_cache
from services.instrument_service import quick_add_instrument_from_watchlist, get_snapshot
count = 0
for item in get_instruments_watchlist():
ticker = item["ticker"]
try:
added = quick_add_instrument_from_watchlist(ticker)
snapshot = await get_snapshot(added["id"], period="1y", interval="1d")
if "error" in snapshot:
logger.warning(f"[curve_regime] Skipping '{ticker}': snapshot error: {snapshot['error']}")
continue
classification = classify_instrument_regime(added["id"], snapshot)
save_curve_regime_cache(ticker, classification)
count += 1
except Exception as e:
logger.warning(f"[curve_regime] Skipping '{ticker}': {e}")
continue
logger.info(f"[curve_regime] Refreshed {count}/{len(get_instruments_watchlist())} Watchlist instruments")
return count

View File

@@ -310,6 +310,20 @@ def init_db():
"ALTER TABLE instrument_overrides ADD COLUMN name TEXT",
"ALTER TABLE instrument_overrides ADD COLUMN yf_ticker TEXT",
"ALTER TABLE instrument_overrides ADD COLUMN category TEXT",
# Curve Regime — one row per Watchlist ticker, overwritten every refresh pass (see
# services/wavelet_scheduler.py, same cadence as the wavelet cache). Instrument-level
# regime (services.curve_regime), NOT the global macro regime shown elsewhere. Feeds
# the Dashboard's "Curve Regime" card so it never needs a live classify call per view.
"""CREATE TABLE IF NOT EXISTS curve_regime_cache (
ticker TEXT PRIMARY KEY,
computed_at TEXT DEFAULT (datetime('now')),
regime_key TEXT,
regime_label TEXT,
score REAL,
direction TEXT,
profile_json TEXT,
has_options_data INTEGER DEFAULT 0
)""",
]:
try:
c.execute(_sql)
@@ -3721,6 +3735,48 @@ def get_wavelet_decomposition_cache(ticker: str) -> Optional[Dict]:
return d
def save_curve_regime_cache(ticker: str, classification: Dict) -> None:
top = classification.get("top_match") or {}
conn = get_conn()
conn.execute(
"INSERT INTO curve_regime_cache (ticker, computed_at, regime_key, regime_label, score, direction, profile_json, has_options_data) "
"VALUES (?, datetime('now'), ?, ?, ?, ?, ?, ?) "
"ON CONFLICT(ticker) DO UPDATE SET computed_at=excluded.computed_at, regime_key=excluded.regime_key, "
"regime_label=excluded.regime_label, score=excluded.score, direction=excluded.direction, "
"profile_json=excluded.profile_json, has_options_data=excluded.has_options_data",
(
ticker.upper(), top.get("key"), top.get("label"), top.get("score"),
(classification.get("profile") or {}).get("direction"),
json.dumps(classification), 1 if classification.get("has_options_data") else 0,
),
)
conn.commit()
conn.close()
def get_all_curve_regime_cache() -> List[Dict]:
"""One row per Watchlist ticker, in Watchlist sort order — a data=None placeholder for
any ticker never successfully classified yet, same "always show every Watchlist
instrument" convention as get_latest_wavelet_state_by_instrument()."""
conn = get_conn()
rows = conn.execute("SELECT * FROM curve_regime_cache").fetchall()
conn.close()
by_ticker = {r["ticker"]: dict(r) for r in rows}
out: List[Dict] = []
for item in get_instruments_watchlist():
ticker = item["ticker"].upper()
row = by_ticker.get(ticker)
if row:
row["profile"] = json.loads(row.pop("profile_json")).get("profile") if row.get("profile_json") else None
out.append(row)
else:
out.append({
"ticker": ticker, "computed_at": None, "regime_key": None, "regime_label": None,
"score": None, "direction": None, "profile": None, "has_options_data": 0,
})
return out
# ── AI Chat widget — persisted conversation ────────────────────────────────────
def save_chat_message(session_id: str, role: str, content: str) -> None:

View File

@@ -249,10 +249,12 @@ def rolling_causal_bands(
n = len(values)
out_dates: list[str] = []
out_original: list[float] = []
out_mean: list[float] = []
band_series: list[list[float]] = [[] for _ in range(num_levels)]
band_labels = [f"bande {i + 1}" for i in range(num_levels)]
band_failed = [False] * num_levels
last_tip: list[float] | None = None
last_tip_mean = 0.0
recomputations = 0
for t in range(start_idx, n):
@@ -264,12 +266,21 @@ def rolling_causal_bands(
window_dates = dates[window_start:t + 1]
result = band_decompose(window_values, window_dates, num_levels, wavelet)
last_tip = [band["series"][-1] for band in result["bands"]]
# band_decompose de-means each trailing window before decomposing it (see its
# own x_mean/xc) — reconstructing a day's price needs THAT window's mean added
# back, not one fixed constant for the whole causal run (unlike band_decompose's
# own single-window "mean" key): a lookback window slides forward every step, so
# its mean drifts along with any real trend. Losing this here (never captured)
# meant the frontend's mean+sum(bands) reconstruction was silently missing the
# entire price level — bands.sum() alone centers on ~0, not the real price.
last_tip_mean = result["mean"]
band_labels = [band["label"] for band in result["bands"]]
for i, band in enumerate(result["bands"]):
band_failed[i] = band_failed[i] or band.get("reconstruction_failed", False)
recomputations += 1
out_dates.append(dates[t])
out_original.append(values[t])
out_mean.append(last_tip_mean)
for i in range(num_levels):
band_series[i].append(last_tip[i])
@@ -288,6 +299,9 @@ def rolling_causal_bands(
return {
"dates": out_dates,
"original": out_original,
"mean": out_mean, # per-day series (the trailing window's mean drifts) — NOT a
# single scalar like band_decompose's own "mean" key; callers
# must add mean[i] (not a bare `mean`) to bands.sum() per day
"bands": bands,
"wavelet": wavelet,
"lookback": lookback,
@@ -442,6 +456,7 @@ def rolling_causal_bands_ssq(
n = len(values)
out_dates: list[str] = []
out_original: list[float] = []
out_mean: list[float] = []
band_series: list[list[float]] = [[] for _ in range(num_levels)]
energy_series: list[list[float]] = [[] for _ in range(num_levels)]
ridge_series: list[float | None] = []
@@ -450,6 +465,7 @@ def rolling_causal_bands_ssq(
last_tip: list[float] | None = None
last_energy_tip: list[float] | None = None
last_ridge_tip: float | None = None
last_tip_mean = 0.0
recomputations = 0
for t in range(start_idx, n):
@@ -463,12 +479,17 @@ def rolling_causal_bands_ssq(
last_tip = [band["series"][-1] for band in result["bands"]]
last_energy_tip = [band["energy"][-1] for band in result["bands"]]
last_ridge_tip = result["ridge_period_days"][-1]
# See rolling_causal_bands's identical fix for why this is a per-day series,
# not a single scalar: each trailing window is de-meaned independently before
# decomposition, and that mean drifts as the window slides forward.
last_tip_mean = result["mean"]
band_labels = [band["label"] for band in result["bands"]]
for i, band in enumerate(result["bands"]):
band_failed[i] = band_failed[i] or band.get("reconstruction_failed", False)
recomputations += 1
out_dates.append(dates[t])
out_original.append(values[t])
out_mean.append(last_tip_mean)
for i in range(num_levels):
band_series[i].append(last_tip[i])
energy_series[i].append(last_energy_tip[i])
@@ -490,6 +511,7 @@ def rolling_causal_bands_ssq(
return {
"dates": out_dates,
"original": out_original,
"mean": out_mean, # per-day series — see rolling_causal_bands
"bands": bands,
"wavelet": wavelet,
"lookback": lookback,

View File

@@ -39,18 +39,27 @@ def set_settings(enabled: bool, refresh_minutes: float) -> None:
def run_refresh_pass() -> int:
"""One pass over the whole Watchlist: Saxo-first price fetch + wavelet recompute (see
services.wavelet_signals.scan_watchlist_wavelet_signals) the same work the daily cycle's
own wavelet step does, just callable on its own cadence. Shared by the periodic loop and
the manual 'refresh now' button. Returns the number of signal rows written."""
services.wavelet_signals.scan_watchlist_wavelet_signals), then Curve Regime
classification per instrument (services.curve_regime.refresh_watchlist_curve_regimes,
reusing that same wavelet cache) — the same work the daily cycle's own wavelet step
does, just callable on its own cadence. Shared by the periodic loop and the manual
'refresh now' button. Returns the number of signal rows written."""
import asyncio
from .wavelet_signals import compute_and_save_wavelet_signals
from .curve_regime import refresh_watchlist_curve_regimes
run_id = f"refresh-{uuid.uuid4().hex[:10]}"
try:
results = compute_and_save_wavelet_signals(run_id)
logger.info(f"[Wavelet Scheduler] Refresh pass complete: {len(results)} signal rows ({run_id})")
return len(results)
except Exception as e:
logger.warning(f"[Wavelet Scheduler] Refresh pass failed: {e}")
return 0
logger.warning(f"[Wavelet Scheduler] Wavelet refresh failed: {e}")
results = []
try:
n_regimes = asyncio.run(refresh_watchlist_curve_regimes())
logger.info(f"[Wavelet Scheduler] Curve Regime refresh complete: {n_regimes} instruments")
except Exception as e:
logger.warning(f"[Wavelet Scheduler] Curve Regime refresh failed: {e}")
return len(results)
def _loop(stop: threading.Event):