feat: macro gauge DB + regime triggers + date-aware instrument snapshot

DB:
- New table macro_gauge_snapshots (daily snapshot of all 28+ gauges + dominant + scores)
- save_macro_gauge_snapshot / get_macro_gauge_snapshot_at / get_macro_gauge_history
- Auto-save once per calendar day on every macro-regime fetch (not just force=True)

API:
- GET /api/market/macro-gauges/at?date=YYYY-MM-DD — nearest snapshot ≤ date
- GET /api/market/macro-gauges/history?days=N

Detector (_check_macro_gauges in Eco Desk):
- Regime transition events (goldilocks→stagflation etc.) with severity scoring
- Yield curve inversion / désinversion (slope_10y3m sign change)
- DXY shock (% change over lookback window)
- Credit stress (HYG drop threshold)
- Gold/Copper ratio regime crossings

InstrumentDashboard:
- macroAtDate state: fetches /api/market/macro-gauges/at when crosshair date ≠ last date
- RegimeCard uses historical macro regime when on a past date
- MacroGaugePanel: full breakdown of all gauges by bloc (liquidité, crédit, énergie...)
  visible only when on a historical date — shows value + change_pct + regime scores bar

AIDesks: added fundamental + sentiment to AIDesk type

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OpenSquared
2026-06-26 09:34:35 +02:00
parent a12d7a1ef3
commit 85eb864584
5 changed files with 513 additions and 5 deletions

View File

@@ -238,6 +238,20 @@ def init_db():
except Exception:
pass
c.execute("""CREATE TABLE IF NOT EXISTS macro_gauge_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
snapshot_date TEXT NOT NULL,
gauges_json TEXT NOT NULL DEFAULT '{}',
dominant TEXT,
regime_scores_json TEXT DEFAULT '{}',
created_at TEXT DEFAULT (datetime('now')),
UNIQUE(snapshot_date)
)""")
try:
c.execute("CREATE INDEX IF NOT EXISTS idx_mgs_date ON macro_gauge_snapshots(snapshot_date DESC)")
except Exception:
pass
c.execute("""CREATE TABLE IF NOT EXISTS geo_alert_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
@@ -1028,7 +1042,18 @@ def init_db():
"active": 1,
"system_prompt": "Tu analyses les surprises économiques des données macro US (FRED).",
"instruments": json.dumps(["SPY","TLT","GLD","EURUSD=X","USO","HYG"]),
"config": json.dumps({"z_threshold": 1.5, "days": 7}),
"config": json.dumps({
"z_threshold": 1.5,
"days": 7,
"lookback_days": 7,
"gauge_signals": {
"regime_transition": {"enabled": True},
"yield_curve_inversion": {"enabled": True, "threshold": 0.0},
"dxy_shock": {"enabled": True, "pct_threshold": 2.0},
"credit_stress": {"enabled": True, "pct_threshold": -1.5},
"gold_copper_ratio": {"enabled": True, "fear_threshold": 700, "growth_threshold": 500},
},
}),
},
{
"name": "Fundamental Desk",
@@ -1570,6 +1595,85 @@ def get_macro_regime_history(days: int = 15) -> List[Dict[str, Any]]:
return result
# ── Macro gauge snapshots (daily persistence of all 28+ gauges) ──────────────
def save_macro_gauge_snapshot(
snapshot_date: str,
gauges: Dict[str, Any],
dominant: Optional[str] = None,
regime_scores: Optional[Dict[str, Any]] = None,
) -> bool:
"""Save (or replace) a full gauge snapshot for a given date. Returns True if new."""
conn = get_conn()
try:
existing = conn.execute(
"SELECT id FROM macro_gauge_snapshots WHERE snapshot_date=?", (snapshot_date,)
).fetchone()
conn.execute(
"INSERT OR REPLACE INTO macro_gauge_snapshots "
"(snapshot_date, gauges_json, dominant, regime_scores_json, created_at) "
"VALUES (?, ?, ?, ?, datetime('now'))",
(snapshot_date, json.dumps(gauges), dominant or "",
json.dumps(regime_scores or {}))
)
conn.commit()
return existing is None
finally:
conn.close()
def get_macro_gauge_snapshot_at(date: str) -> Optional[Dict[str, Any]]:
"""Return the most recent snapshot whose date is <= `date`."""
conn = get_conn()
try:
row = conn.execute(
"SELECT * FROM macro_gauge_snapshots WHERE snapshot_date <= ? "
"ORDER BY snapshot_date DESC LIMIT 1",
(date[:10],)
).fetchone()
if not row:
return None
d = dict(row)
d["gauges"] = json.loads(d.pop("gauges_json", "{}"))
d["regime_scores"] = json.loads(d.pop("regime_scores_json", "{}"))
return d
finally:
conn.close()
def get_macro_gauge_history(days: int = 30) -> List[Dict[str, Any]]:
"""Return gauge snapshots for the last N days, most recent first."""
conn = get_conn()
try:
rows = conn.execute(
"SELECT * FROM macro_gauge_snapshots "
"WHERE snapshot_date >= date('now', ?) "
"ORDER BY snapshot_date DESC",
(f"-{days} days",)
).fetchall()
result = []
for r in rows:
d = dict(r)
d["gauges"] = json.loads(d.pop("gauges_json", "{}"))
d["regime_scores"] = json.loads(d.pop("regime_scores_json", "{}"))
result.append(d)
return result
finally:
conn.close()
def macro_gauge_snapshot_exists_today() -> bool:
today = datetime.utcnow().strftime("%Y-%m-%d")
conn = get_conn()
try:
row = conn.execute(
"SELECT id FROM macro_gauge_snapshots WHERE snapshot_date=?", (today,)
).fetchone()
return row is not None
finally:
conn.close()
def log_geo_alert(geo_score: int, top_patterns: List[Dict[str, Any]], news_count: int, run_id: str):
"""Append a geo alert snapshot tied to a scoring run."""
conn = get_conn()