From 01709e5edfa118f449b939ff7167059670439146 Mon Sep 17 00:00:00 2001 From: OpenSquared Date: Thu, 2 Jul 2026 16:46:38 +0200 Subject: [PATCH] feat: causal lab --- backend/main.py | 14 ++ backend/routers/eco.py | 37 +++++ backend/services/causal_graphs.py | 85 ++++++++++ backend/services/guidance_sync.py | 248 ++++++++++++++++++++++++++++++ 4 files changed, 384 insertions(+) create mode 100644 backend/services/guidance_sync.py diff --git a/backend/main.py b/backend/main.py index a01c6ca..2176b03 100644 --- a/backend/main.py +++ b/backend/main.py @@ -94,6 +94,13 @@ def startup(): _log.info("[Startup] Causal graph templates seeded") except Exception as _e: _log.warning(f"[Startup] Causal graph seed failed: {_e}") + # Guidance sync au démarrage (crée les guidance events manquants) + try: + from services.guidance_sync import sync_guidance as _gs + _gr = _gs() + _log.info(f"[Startup] Guidance sync done: {_gr}") + except Exception as _e: + _log.warning(f"[Startup] Guidance sync failed: {_e}") # Auto-bootstrap désactivé — utiliser les boutons dans Cycle Actions / Timeline # Start auto-cycle scheduler if enabled from services.auto_cycle import start_scheduler @@ -131,6 +138,13 @@ def startup(): r = calendar_sync() src = r.get("source") or ("ff_fallback" if r.get("_fallback") else "fxstreet") print(f"[Calendar] Sync done — {src}, upserted={r.get('total_upserted', '?')}", flush=True) + # Guidance sync — crée/met à jour les guidance events après chaque calendar sync + try: + from services.guidance_sync import sync_guidance + gr = sync_guidance() + print(f"[Guidance] Sync done — {gr}", flush=True) + except Exception as _ge: + print(f"[Guidance] Sync error: {_ge}", flush=True) except Exception as e: print(f"[Calendar] Sync loop error: {e}", flush=True) try: diff --git a/backend/routers/eco.py b/backend/routers/eco.py index 150f04c..b139aef 100644 --- a/backend/routers/eco.py +++ b/backend/routers/eco.py @@ -673,6 +673,43 @@ def ff_purge_fxstreet() -> Dict[str, Any]: conn.close() +@router.post("/guidance-sync") +def guidance_sync_endpoint() -> Dict[str, Any]: + """ + Déclenche manuellement la synchronisation des guidance events (MACRO_RATE_GUIDANCE). + Crée/met à jour un market_event de type guidance pour chaque decision de taux à venir + détectée dans ff_calendar. Appelé automatiquement après chaque calendar-sync. + """ + try: + from services.guidance_sync import sync_guidance + result = sync_guidance() + return {"status": "ok", **result} + except Exception as e: + import traceback + return {"status": "error", "error": str(e), "trace": traceback.format_exc()} + + +@router.get("/guidance-sync/status") +def guidance_sync_status() -> Dict[str, Any]: + """Liste les guidance market_events actuellement en DB.""" + from services.database import get_conn + conn = get_conn() + try: + rows = conn.execute(""" + SELECT me.id, me.name, me.start_date, me.end_date, + me.expected_value, me.sub_type, + COUNT(cea.id) as n_analyses + FROM market_events me + LEFT JOIN causal_event_analyses cea ON cea.market_event_id = me.id + WHERE me.origin = 'guidance_sync' + GROUP BY me.id + ORDER BY me.end_date ASC + """).fetchall() + return {"count": len(rows), "events": [dict(r) for r in rows]} + finally: + conn.close() + + @router.get("/ff-stats") def ff_stats() -> Dict[str, Any]: """Quick inventory of ff_calendar table.""" diff --git a/backend/services/causal_graphs.py b/backend/services/causal_graphs.py index 2b39ff4..65e3c2f 100644 --- a/backend/services/causal_graphs.py +++ b/backend/services/causal_graphs.py @@ -900,6 +900,91 @@ BUILT_IN_TEMPLATES = [ }, }, }, + + # ═══════════════════════════════════════════════════════════════════════════════ + # 13. MACRO_RATE_GUIDANCE (auto-créé par guidance_sync — pas dans REGIME_SLUGS) + # Anticipation d'une décision de taux avant l'event : repricing progressif OIS + # ═══════════════════════════════════════════════════════════════════════════════ + { + "name": "Macro Rate Guidance", + "slug": "MACRO_RATE_GUIDANCE", + "category": "central_bank", + "sub_type": "guidance", + "heuristic_ver": 2, + "calibration_json": {"lag_days": 0, "half_life_days": 30, "absorption_days": 60, "decay_type": "linear"}, + "instruments": ["EURUSD", "TLT", "USDJPY", "SP500", "XAUUSD", "GBPUSD", "EEM", "QQQ"], + "description": "Anticipation marché d'une décision de taux — pricing-in progressif avant l'event (OIS forwards, carry FX, duration bonds)", + "ai_rationale": "Quand le consensus se forme autour d'un cut/hike, les OIS forwards repricing graduellement. Le carry FX shift avant la décision. TLT monte si cut attendu. Signal = forecast - previous en bps. Signal=0 = marché neutre (status quo).", + "graph_json": { + "nodes": [ + _n("guidance_signal", "Expected Δ Rate", "input", 400, 50, unit="bps", + description="Positif=hike attendu, Négatif=cut attendu (bps)"), + _n("ois_forward", "OIS Fwd Shift", "observable", 180, 200, + formula="guidance_signal * {{coef_signal_ois}}", unit="bps"), + _n("usd_carry", "USD Carry Shift", "observable", 60, 350, + formula="ois_forward * {{coef_ois_carry}}", unit="%"), + _n("rate_expectation", "Rate Path Exp.", "latent", 400, 350, + formula="ois_forward * {{coef_ois_re}}"), + _n("risk_premium", "Risk Premium", "latent", 700, 350, + formula="guidance_signal * {{coef_signal_rp}}", + description="Négatif si cut (risk-on), positif si hike (risk-off)"), + # Outputs row 1 + _n("eurusd", "EUR/USD", "market_asset", 60, 530, formula="-usd_carry * {{coef_carry_eurusd}}", unit="pips", instrument="EURUSD"), + _n("tlt", "TLT Bond", "market_asset", 240, 530, formula="-rate_expectation * {{coef_re_tlt}}", unit="pts", instrument="TLT"), + _n("usdjpy", "USD/JPY", "market_asset", 420, 530, formula="usd_carry * {{coef_carry_usdjpy}}", unit="pips", instrument="USDJPY"), + _n("sp500", "S&P 500", "market_asset", 600, 530, formula="-risk_premium * {{coef_rp_sp}} + rate_expectation * {{coef_re_sp}}", unit="pts", instrument="SP500"), + # Outputs row 2 + _n("xauusd", "Gold", "market_asset", 60, 680, formula="-usd_carry * {{coef_carry_gold}} - rate_expectation * {{coef_re_gold}}", unit="$/oz", instrument="XAUUSD"), + _n("gbpusd", "GBP/USD", "market_asset", 240, 680, formula="-usd_carry * {{coef_carry_gbpusd}}", unit="pips", instrument="GBPUSD"), + _n("eem", "EEM Emerg", "market_asset", 420, 680, formula="-usd_carry * {{coef_carry_eem}} - risk_premium * {{coef_rp_eem}}", unit="pts", instrument="EEM"), + _n("qqq", "QQQ Nasdaq", "market_asset", 600, 680, formula="-rate_expectation * {{coef_re_qqq}} - risk_premium * {{coef_rp_qqq}}", unit="pts", instrument="QQQ"), + ], + "edges": [ + _e("guidance_signal", "ois_forward", "solid", "policy_fwd_pricing", strength=3, sign="positive", label="hike→OIS up / cut→OIS dn"), + _e("guidance_signal", "risk_premium", "dashed", "risk_channel", strength=2, sign="positive", label="hike=risk-off / cut=risk-on"), + _e("ois_forward", "usd_carry", "solid", "carry_diff", strength=3, sign="positive"), + _e("ois_forward", "rate_expectation", "solid", "fwd_path", strength=3, sign="positive"), + # FX + _e("usd_carry", "eurusd", "solid", "fx_eurusd", strength=3, sign="negative", label="USD up → EUR/USD dn"), + _e("usd_carry", "usdjpy", "solid", "fx_usdjpy", strength=2, sign="positive", label="USD up → USD/JPY up"), + _e("usd_carry", "gbpusd", "solid", "fx_gbpusd", strength=2, sign="negative"), + _e("usd_carry", "eem", "solid", "usd_eem", strength=2, sign="negative"), + # Bonds + _e("rate_expectation","tlt", "solid", "bond_dur", strength=3, sign="negative", label="hike exp → TLT dn"), + # Equities + _e("risk_premium", "sp500", "solid", "equity_rp", strength=2, sign="negative"), + _e("rate_expectation","sp500", "dashed", "disc_rate", strength=2, sign="negative", label="hike → discount rate up → SP dn"), + _e("risk_premium", "qqq", "dashed", "tech_rp", strength=2, sign="negative"), + _e("rate_expectation","qqq", "solid", "tech_dur", strength=3, sign="negative", label="QQQ = long duration"), + _e("risk_premium", "eem", "dashed", "em_risk", strength=2, sign="negative"), + # Gold + _e("usd_carry", "xauusd", "solid", "usd_gold", strength=2, sign="negative"), + _e("rate_expectation","xauusd", "dashed", "real_rate", strength=2, sign="negative", label="hike exp → real rates up → gold dn"), + ], + "coefficients": { + "coef_signal_ois": _c(0.8, "bps signal → OIS shift (same sign: hike=up, cut=down)"), + "coef_ois_carry": _c(0.4, "OIS shift → USD carry %"), + "coef_ois_re": _c(0.9, "OIS shift → rate expectation index"), + "coef_signal_rp": _c(0.3, "signal bps → risk premium (hike=risk-off positive)"), + "coef_carry_eurusd": _c(60, "1% USD carry → EUR/USD pips fall"), + "coef_carry_usdjpy": _c(80, "1% USD carry → USD/JPY pips rise"), + "coef_carry_gbpusd": _c(50, "1% USD carry → GBP/USD pips fall"), + "coef_carry_eem": _c(20, "1% USD carry → EEM pts fall"), + "coef_carry_gold": _c(50, "1% USD carry → Gold $/oz fall"), + "coef_re_tlt": _c(12, "rate exp index → TLT pts fall (~12y duration)"), + "coef_rp_sp": _c(30, "risk premium → SP500 pts fall"), + "coef_re_sp": _c(20, "rate expectation → SP500 pts fall (discount rate)"), + "coef_re_qqq": _c(45, "rate exp → QQQ pts fall (high duration)"), + "coef_rp_qqq": _c(25, "risk premium → QQQ pts fall"), + "coef_rp_eem": _c(15, "risk premium → EEM pts fall"), + "coef_re_gold": _c(15, "rate exp → Gold $/oz fall (real rate)"), + }, + "instruments": ["EURUSD", "TLT", "USDJPY", "SP500", "XAUUSD", "GBPUSD", "EEM", "QQQ"], + "input_mapping": { + "guidance_signal": {"source": "guidance_bps", "unit": "bps", "range": [-200, 200]} + }, + }, + }, ] diff --git a/backend/services/guidance_sync.py b/backend/services/guidance_sync.py new file mode 100644 index 0000000..c257d30 --- /dev/null +++ b/backend/services/guidance_sync.py @@ -0,0 +1,248 @@ +""" +Guidance Sync — crée/met à jour des market_events de type MACRO_RATE_GUIDANCE +à partir des events de taux détectés dans ff_calendar. + +Lifecycle d'un event guidance : + 1. Event de taux détecté dans ff_calendar (pas encore d'actual) + → signal = 0 si pas de forecast (status quo attendu) + → signal = (forecast - previous) * 100 en bps si forecast dispo + 2. Guidance market_event créé, end_date = date de la réunion + → causal_event_analyses créés pour chaque instrument → visible dans la Frise + 3. À chaque sync, le signal est recalculé et mis à jour si nécessaire + 4. Quand actual_value apparaît dans ff_calendar → guidance fermé (supprimé ou ignoré) + +Appelé automatiquement après chaque calendar_sync dans main.py. +""" +import json +import logging +import re +from datetime import date, datetime +from typing import Optional + +logger = logging.getLogger(__name__) + +# ── Patterns de noms d'events de taux par devise ────────────────────────────── +# Matching insensible à la casse sur event_name de ff_calendar +RATE_PATTERNS: dict[str, list[str]] = { + "USD": ["fed funds rate", "federal funds rate", "fomc rate decision"], + "EUR": ["ecb interest rate", "main refinancing rate", "refinancing rate", "ecb rate decision"], + "GBP": ["official bank rate", "boe bank rate", "boe interest rate", "bank rate"], + "JPY": ["boj policy rate", "boj rate", "overnight call rate", "monetary policy"], + "AUD": ["cash rate", "rba cash rate", "rba rate decision"], + "CAD": ["overnight rate", "boc rate", "bank of canada rate"], + "NZD": ["official cash rate", "rbnz rate", "rbnz cash rate"], + "CHF": ["snb policy rate", "snb rate", "snb interest rate"], +} + +# Instruments affectés par devise (USD rate guidance → EURUSD, TLT, etc.) +CURRENCY_INSTRUMENTS: dict[str, list[str]] = { + "USD": ["EURUSD", "TLT", "USDJPY", "SP500", "XAUUSD", "GBPUSD", "EEM", "QQQ"], + "EUR": ["EURUSD", "GBPUSD", "SP500", "XAUUSD", "TLT"], + "GBP": ["GBPUSD", "EURUSD", "SP500"], + "JPY": ["USDJPY", "EURUSD", "XAUUSD"], + "AUD": ["EURUSD"], + "CAD": ["EURUSD"], + "NZD": ["EURUSD"], + "CHF": ["EURUSD", "XAUUSD"], +} + +GUIDANCE_ORIGIN = "guidance_sync" +GUIDANCE_SUBTYPE = "rate_guidance" # sub_type dans market_events + + +# ── Helpers ──────────────────────────────────────────────────────────────────── + +def _parse_rate(s: Optional[str]) -> Optional[float]: + """'4.25%' ou '4.25' → 4.25, None si vide ou non parseable.""" + if not s: + return None + cleaned = re.sub(r"[^\d.\-]", "", str(s).strip()) + try: + return float(cleaned) + except (ValueError, TypeError): + return None + + +def _signal_bps(forecast: Optional[str], previous: Optional[str]) -> float: + """ + Signal en bps = (forecast - previous) * 100. + Positif = hike attendu, négatif = cut attendu, 0 = status quo ou pas de forecast. + """ + f = _parse_rate(forecast) + p = _parse_rate(previous) + if f is None or p is None: + return 0.0 + return round((f - p) * 100, 1) + + +def _get_guidance_template_id(conn) -> Optional[int]: + row = conn.execute( + "SELECT id FROM causal_graph_templates WHERE name = 'Macro Rate Guidance'" + ).fetchone() + return row["id"] if row else None + + +def _find_existing(conn, currency: str, meeting_date: str) -> Optional[dict]: + """Trouve le guidance market_event existant pour cette devise+date de réunion.""" + row = conn.execute( + """SELECT * FROM market_events + WHERE origin=? AND end_date=? AND sub_type=? + LIMIT 1""", + (GUIDANCE_ORIGIN, meeting_date, GUIDANCE_SUBTYPE + "_" + currency) + ).fetchone() + return dict(row) if row else None + + +def _upsert_analyses(conn, event_id: int, template_id: int, instruments: list[str], signal: float): + """ + Crée/recrée les causal_event_analyses pour que l'event apparaisse + dans la Frise de chaque instrument concerné. + """ + conn.execute( + "DELETE FROM causal_event_analyses WHERE market_event_id=?", (event_id,) + ) + inputs = json.dumps({"guidance_signal": signal}) + for inst in instruments: + conn.execute(""" + INSERT INTO causal_event_analyses + (market_event_id, template_id, instrument, inputs_json, analyzed_at) + VALUES (?, ?, ?, ?, datetime('now')) + """, (event_id, template_id, inst, inputs)) + + +def _build_title(currency: str, signal: float, meeting_date: str) -> str: + try: + month = datetime.strptime(meeting_date, "%Y-%m-%d").strftime("%b %Y") + except ValueError: + month = meeting_date + if signal == 0.0: + action = "Status Quo" + elif signal < 0: + action = f"Cut {abs(signal):.0f}bps" + else: + action = f"Hike {signal:.0f}bps" + return f"{currency} Rate Guidance: {action} ({month})" + + +def _build_description(currency: str, signal: float, event_name: str) -> str: + if signal == 0.0: + sentiment = "Pas de changement attendu (status quo). Aucun forecast disponible ou consensus = pas de mouvement." + elif signal < 0: + sentiment = f"Marché anticipe un cut de {abs(signal):.0f}bps. Pression baissière sur {currency} (carry), haussière sur obligations." + else: + sentiment = f"Marché anticipe un hike de {signal:.0f}bps. Pression haussière sur {currency} (carry), baissière sur obligations." + return f"[{event_name}] {sentiment}" + + +# ── Main ─────────────────────────────────────────────────────────────────────── + +def sync_guidance() -> dict: + """ + Scanne ff_calendar pour les events de taux à venir, crée/met à jour + les guidance market_events correspondants. + Retourne un dict de stats. + """ + from services.database import get_conn + from services.causal_graphs import init_tables + + conn = get_conn() + init_tables(conn) + + today = date.today().isoformat() + + template_id = _get_guidance_template_id(conn) + if template_id is None: + logger.warning("[guidance_sync] Template 'Macro Rate Guidance' introuvable — lance seed_templates d'abord") + conn.close() + return {"error": "template_not_found"} + + # Tous les events de taux futurs (sans actual) dans ff_calendar + rows = conn.execute(""" + SELECT currency, event_date, event_name, forecast_value, previous_value, actual_value + FROM ff_calendar + WHERE event_date >= ? + AND impact = 'high' + AND (actual_value IS NULL OR actual_value = '') + ORDER BY event_date ASC + LIMIT 500 + """, (today,)).fetchall() + + created = updated = closed = skipped = 0 + seen: set[tuple] = set() # (currency, meeting_date) + + for row in rows: + currency = (row["currency"] or "").strip() + event_name = (row["event_name"] or "").lower() + evt_date = row["event_date"] + forecast = row["forecast_value"] + previous = row["previous_value"] + actual = row["actual_value"] + + # Filtre : correspond à un pattern de taux ? + patterns = RATE_PATTERNS.get(currency, []) + if not any(p in event_name for p in patterns): + skipped += 1 + continue + + # Évite les doublons (même devise × même date) + key = (currency, evt_date) + if key in seen: + continue + seen.add(key) + + # Si actual est là, l'event est résolu → on ne crée pas de guidance + if actual and str(actual).strip(): + # Supprimer l'ancien guidance si présent (l'event est passé) + existing = _find_existing(conn, currency, evt_date) + if existing: + conn.execute("DELETE FROM market_events WHERE id=?", (existing["id"],)) + conn.execute( + "DELETE FROM causal_event_analyses WHERE market_event_id=?", (existing["id"],) + ) + closed += 1 + continue + + signal = _signal_bps(forecast, previous) + instruments = CURRENCY_INSTRUMENTS.get(currency, ["EURUSD"]) + title = _build_title(currency, signal, evt_date) + description = _build_description(currency, signal, row["event_name"] or "") + impact = round(min(1.0, 0.6 + abs(signal) / 300), 2) + affected = json.dumps(instruments) + sub_type = GUIDANCE_SUBTYPE + "_" + currency + + existing = _find_existing(conn, currency, evt_date) + + if existing: + ev_id = existing["id"] + # Met à jour titre / description / signal si changé + conn.execute(""" + UPDATE market_events + SET name=?, description=?, expected_value=?, impact_score=?, affected_assets=? + WHERE id=? + """, (title, description, str(signal), impact, affected, ev_id)) + _upsert_analyses(conn, ev_id, template_id, instruments, signal) + updated += 1 + else: + cur = conn.execute(""" + INSERT INTO market_events + (name, start_date, end_date, level, category, sub_type, + description, affected_assets, impact_score, + origin, expected_value, unit, source_refs) + VALUES (?, ?, ?, 'high', 'central_bank', ?, + ?, ?, ?, + ?, ?, 'bps', '[]') + """, ( + title, today, evt_date, sub_type, + description, affected, impact, + GUIDANCE_ORIGIN, str(signal), + )) + ev_id = cur.lastrowid + _upsert_analyses(conn, ev_id, template_id, instruments, signal) + created += 1 + + conn.commit() + conn.close() + + stats = {"created": created, "updated": updated, "closed": closed, "skipped": skipped} + logger.info(f"[guidance_sync] {stats}") + return stats