""" Causal Lab v2 — Bibliothèque de graphes causaux + analyse empirique. Sources d'événements : market_events (toutes catégories). Templates de graphes : causal_graph_templates (seed automatique au démarrage). IA : GPT-4o pour recommandation de template + coefficients. Endpoints : GET /api/causal-lab/templates — liste des templates GET /api/causal-lab/template/{id} — détail d'un template PUT /api/causal-lab/template/{id} — mettre à jour les coefficients GET /api/causal-lab/market-events — événements analysables POST /api/causal-lab/recommend — GPT-4o recommande un template POST /api/causal-lab/analyze — analyse (prédiction + réel) GET /api/causal-lab/analyses — historique des analyses GET /api/causal-lab/calibration — stats de calibration par template """ import json import logging import math from datetime import datetime, timedelta from typing import Optional from fastapi import APIRouter, HTTPException, Query from pydantic import BaseModel router = APIRouter() logger = logging.getLogger(__name__) # ── Initialisation DB (appelée depuis startup) ──────────────────────────────── def _init(conn): from services.causal_graphs import init_tables, seed_templates init_tables(conn) seed_templates(conn) # ── Récupération des prix autour d'un événement ─────────────────────────────── def _fetch_prices(event_date_str: str, instruments: list[str], lag_days: int = 0) -> dict: """ Télécharge les prix autour de l'événement pour chaque instrument demandé. Retourne dict { instrument: [{"t": iso, "c": float}, ...] } Si lag_days > 0, force le mode journalier (la fenêtre doit couvrir N jours après l'event). """ YFINANCE_MAP = { "EURUSD": "EURUSD=X", "XAUUSD": "GC=F", "SP500": "^GSPC", "BRENT": "BZ=F", "US2Y": "US2YT=RR", "US10Y": "^TNX", "EU10Y": "GE10YT=RR", } out: dict = {"mode": "none"} for inst in instruments + ["US2Y", "US10Y", "EU10Y"]: out[inst] = [] try: import yfinance as yf event_dt = datetime.strptime(event_date_str[:10], "%Y-%m-%d") days_ago = (datetime.utcnow() - event_dt).days # Intraday 5min (< 55 jours) sauf si lag_days > 0 (besoin fenêtre journalière étendue) use_intraday = days_ago < 55 and lag_days == 0 for inst in instruments: sym = YFINANCE_MAP.get(inst) if not sym: continue try: if use_intraday: start = (event_dt - timedelta(days=1)).strftime("%Y-%m-%d") end = (event_dt + timedelta(days=2)).strftime("%Y-%m-%d") df = yf.download(sym, start=start, end=end, interval="5m", progress=False, auto_adjust=True) if df is not None and len(df) > 0: if hasattr(df.columns, "levels"): df.columns = df.columns.get_level_values(0) day_df = df[df.index.date == event_dt.date()] rows = [ {"t": idx.isoformat(), "c": round(float(row["Close"]), 5)} for idx, row in day_df.iterrows() if not math.isnan(float(row["Close"])) ] if len(rows) >= 6: out[inst] = rows out["mode"] = "intraday_5m" continue # Fallback journalier (ou mode forcé si lag_days > 0) start = (event_dt - timedelta(days=5)).strftime("%Y-%m-%d") end = (event_dt + timedelta(days=max(5, lag_days + 3))).strftime("%Y-%m-%d") df = yf.download(sym, start=start, end=end, interval="1d", progress=False, auto_adjust=True) if df is not None and len(df) > 0: if hasattr(df.columns, "levels"): df.columns = df.columns.get_level_values(0) out[inst] = [ {"t": str(idx.date()), "c": round(float(row["Close"]), 5)} for idx, row in df.iterrows() if float(row["Close"]) == float(row["Close"]) ] if out["mode"] == "none": out["mode"] = "daily" except Exception as e: logger.debug(f"[causal_lab] price {sym}: {e}") # Taux (toujours en journalier) start = (event_dt - timedelta(days=4)).strftime("%Y-%m-%d") end = (event_dt + timedelta(days=4)).strftime("%Y-%m-%d") for key in ["US2Y", "US10Y", "EU10Y"]: sym = YFINANCE_MAP.get(key) if not sym: continue try: df = yf.download(sym, start=start, end=end, interval="1d", progress=False, auto_adjust=True) if df is None or len(df) == 0: continue if hasattr(df.columns, "levels"): df.columns = df.columns.get_level_values(0) out[key] = [ {"t": str(idx.date()), "c": round(float(row["Close"]), 3)} for idx, row in df.iterrows() if float(row["Close"]) == float(row["Close"]) ] except Exception as e: logger.debug(f"[causal_lab] yield {sym}: {e}") except Exception as e: logger.error(f"[causal_lab] _fetch_prices: {e}") return out def _drift_metrics(prices: dict, event_date_str: str, inst: str, lag_min: int = 0, lag_days: int = 0) -> dict: series = prices.get(inst, []) mode = prices.get("mode", "none") edate = event_date_str[:10] empty = {"pre_pips": None, "post_pips": None, "drift_ratio": None, "leak": "unknown", "lag_min": lag_min, "lag_days": lag_days} if not series: return empty mult = 10000 if inst in ("EURUSD",) else 10 if mode == "intraday_5m": n = len(series) if n < 8: return empty # Décale le mid en avant selon le lag (chaque barre = 5 min) lag_bars = max(0, round(lag_min / 5)) mid = min(n // 2 + lag_bars, n - 2) pre_pips = round((series[mid - 1]["c"] - series[0]["c"]) * mult) post_pips = round((series[-1]["c"] - series[mid]["c"]) * mult) else: # En mode journalier, lag_days décale la date de référence "post" if lag_days > 0: post_date = (datetime.strptime(edate, "%Y-%m-%d") + timedelta(days=lag_days)).strftime("%Y-%m-%d") pre = [b for b in series if b["t"] <= edate] post = [b for b in series if b["t"] >= post_date] else: pre = [b for b in series if b["t"] < edate] post = [b for b in series if b["t"] == edate] if not pre or not post: return empty pre_pips = None post_pips = round((post[0]["c"] - pre[-1]["c"]) * mult) ratio: Optional[float] = None if pre_pips is not None and post_pips and post_pips != 0: ratio = round(pre_pips / post_pips, 3) leak = "unknown" if ratio is not None: ar = abs(ratio) leak = "high" if ar > 0.5 else "medium" if ar > 0.25 else "low" return {"pre_pips": pre_pips, "post_pips": post_pips, "drift_ratio": ratio, "leak": leak} def _yield_delta(series: list, event_date: str) -> Optional[float]: if not series: return None pre = [b for b in series if b["t"] < event_date] post = [b for b in series if b["t"] >= event_date] if not pre or not post: return None return round(post[0]["c"] - pre[-1]["c"], 3) def _activation_score(node_values: dict, actual_moves: dict, template_graph: dict) -> dict: """ Compare les sorties du graphe (outputs) aux mouvements réels. Retourne {score, nodes: {node_id: {pred, act, status}}} """ output_nodes = [n for n in template_graph.get("nodes", []) if n.get("type") == "output"] nodes_detail: dict = {} correct = 0 evaluated = 0 for node in output_nodes: nid = node["id"] inst = node.get("instrument", "") pred = node_values.get(nid) act = actual_moves.get(inst) if pred is None or act is None: status = "unknown" elif abs(pred) < 2 and abs(act) < 2: status = "neutral" elif (pred > 0) == (act > 0): status = "correct" correct += 1 evaluated += 1 else: status = "wrong" evaluated += 1 nodes_detail[nid] = { "pred": round(pred, 2) if pred is not None else None, "act": round(act, 2) if act is not None else None, "status": status, "label": node.get("label", nid), } score = round(correct / evaluated, 2) if evaluated else None return {"score": score, "nodes": nodes_detail, "correct": correct, "total": evaluated} # ── Helpers GPT-4o ───────────────────────────────────────────────────────────── def _gpt4o_recommend(event: dict, templates: list) -> dict: """Appelle GPT-4o pour recommander un template + coefficients.""" try: from services.database import get_config import openai key = get_config("openai_api_key") or "" if not key: return {"error": "GPT-4o non configuré (clé OpenAI manquante)"} client = openai.OpenAI(api_key=key) template_summary = [ { "id": t["id"], "name": t["name"], "category": t["category"], "sub_type": t.get("sub_type", ""), "instruments": t.get("instruments", []), "description": t.get("description", ""), } for t in templates ] system_prompt = ( "You are a macro market analyst and causal graph specialist for the GeoOptions platform. " "Given a market event and a library of causal graph templates, recommend the most appropriate template " "and suggest any coefficient adjustments. Respond with a JSON object only.\n" "IMPORTANT: Only recommend a template if it is structurally compatible with the event type " "(same causal mechanism — e.g. do NOT recommend an ECB rate decision template for a sentiment survey). " "If no template fits well (structural mismatch, confidence < 0.6), set template_id to null " "and explain in 'notes' what kind of template would be needed." ) user_prompt = f"""Market event to analyze: - Name: {event.get('name')} - Category: {event.get('category')} - Sub-type: {event.get('sub_type', '')} - Description: {event.get('description', '')} - Market impact: {event.get('market_impact', '')} - Affected assets: {event.get('affected_assets', '')} - Impact score: {event.get('impact_score', '')} - Surprise (%): {event.get('surprise_pct', 'N/A')} - Expected: {event.get('expected_value', 'N/A')} - Actual: {event.get('actual_value', 'N/A')} Available templates: {json.dumps(template_summary, ensure_ascii=False, indent=2)} Respond with this exact JSON structure: {{ "template_id": , "template_name": "", "confidence": <0.0-1.0>, "rationale": "<2-3 sentences explaining why this template fits>", "coefficient_suggestions": {{ "": }}, "instrument_focus": [""], "alternative_template_id": , "notes": "" }}""" resp = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], response_format={"type": "json_object"}, temperature=0.3, max_tokens=600, ) raw = resp.choices[0].message.content or "{}" return json.loads(raw) except Exception as e: logger.error(f"[causal_lab] GPT-4o recommend: {e}") return {"error": str(e)} # ── Schémas Pydantic ────────────────────────────────────────────────────────── class UpdateCoefRequest(BaseModel): coefficients: dict # {coef_key: float} class CreateTemplateRequest(BaseModel): name: str category: str sub_type: str = "" description: str = "" instruments: list = ["EURUSD"] graph_json: dict ai_rationale: str = "" class RecommendRequest(BaseModel): market_event_id: int class InstantiateRequest(BaseModel): market_event_id: int template_id: int class AnalyzeRequest(BaseModel): market_event_id: int template_id: int instrument: str = "" # vide = analyser tous les instruments du template inputs: dict = {} coef_overrides: dict = {} class CreateFromEventRequest(BaseModel): market_event_id: int # ── Endpoints ───────────────────────────────────────────────────────────────── @router.get("/api/causal-lab/debug") def debug_causal(): """Endpoint de diagnostic — vérifie que causal_graphs est bien chargé.""" try: from services.database import get_conn from services.causal_graphs import BUILT_IN_TEMPLATES, init_tables, seed_templates conn = get_conn() init_tables(conn) seed_templates(conn) rows = conn.execute("SELECT id, name, graph_json FROM causal_graph_templates ORDER BY id").fetchall() templates_info = [] for r in rows: g = json.loads(r["graph_json"] or "{}") edges = g.get("edges", []) templates_info.append({ "id": r["id"], "name": r["name"], "edges_count": len(edges), "edges_with_lag_days": sum(1 for e in edges if (e.get("lag_days") or 0) > 0), "edges_with_lag_min": sum(1 for e in edges if (e.get("lag_min") or 0) > 0), "output_node_ids": [n["id"] for n in g.get("nodes", []) if n.get("type") in ("market_asset", "output")], "edges": [{"from": e.get("from"), "to": e.get("to"), "lag_min": e.get("lag_min"), "lag_days": e.get("lag_days")} for e in edges], }) conn.close() return { "built_in_templates": len(BUILT_IN_TEMPLATES), "db_templates": len(rows), "status": "ok", "templates": templates_info, } except Exception as e: import traceback return {"status": "error", "error": str(e), "trace": traceback.format_exc()} @router.get("/api/causal-lab/templates") def list_templates(category: str = Query("")): try: from services.database import get_conn from services.causal_graphs import get_templates conn = get_conn() _init(conn) data = get_templates(conn, category) conn.close() print(f"[causal_lab] list_templates → {len(data)} templates (category={category!r})", flush=True) return data except Exception as e: import traceback print(f"[causal_lab] list_templates ERROR: {e}\n{traceback.format_exc()}", flush=True) logger.error(f"[causal_lab] list_templates: {e}") raise HTTPException(500, str(e)) @router.post("/api/causal-lab/template") def create_template(body: CreateTemplateRequest): """Crée un nouveau template utilisateur.""" try: from services.database import get_conn from services.causal_graphs import init_tables, seed_templates conn = get_conn() init_tables(conn); seed_templates(conn) existing = conn.execute( "SELECT id FROM causal_graph_templates WHERE name = ?", (body.name,) ).fetchone() if existing: raise HTTPException(409, f"Un template nommé '{body.name}' existe déjà") cur = conn.execute(""" INSERT INTO causal_graph_templates (name, category, sub_type, instruments, description, graph_json, ai_rationale, created_by) VALUES (?, ?, ?, ?, ?, ?, ?, 'user') """, ( body.name, body.category, body.sub_type, json.dumps(body.instruments), body.description, json.dumps(body.graph_json), body.ai_rationale, )) conn.commit() new_id = cur.lastrowid conn.close() return {"id": new_id, "name": body.name} except HTTPException: raise except Exception as e: logger.error(f"[causal_lab] create_template: {e}") raise HTTPException(500, str(e)) @router.patch("/api/causal-lab/template/{template_id}") def patch_template(template_id: int, body: dict): """Met à jour un template existant en entier (graph_json, métadonnées, coefficients).""" try: from services.database import get_conn from services.causal_graphs import get_template conn = get_conn() tmpl = get_template(conn, template_id) if not tmpl: conn.close(); raise HTTPException(404, "Template introuvable") sets, params = [], [] for field in ("name", "category", "sub_type", "description", "ai_rationale"): if field in body: sets.append(f"{field}=?"); params.append(body[field]) if "instruments" in body: sets.append("instruments=?"); params.append(json.dumps(body["instruments"])) if "graph_json" in body: sets.append("graph_json=?"); params.append(json.dumps(body["graph_json"])) if "calibration_json" in body: sets.append("calibration_json=?"); params.append(json.dumps(body["calibration_json"])) if not sets: conn.close(); return {"ok": True} sets.append("updated_at=datetime('now')") params.append(template_id) conn.execute(f"UPDATE causal_graph_templates SET {', '.join(sets)} WHERE id=?", params) conn.commit(); conn.close() return {"ok": True} except HTTPException: raise except Exception as e: logger.error(f"[causal_lab] patch_template {template_id}: {e}") raise HTTPException(500, str(e)) @router.delete("/api/causal-lab/template/{template_id}") def delete_template(template_id: int): """Supprime un template (tous les templates peuvent être supprimés).""" try: from services.database import get_conn conn = get_conn() row = conn.execute( "SELECT id FROM causal_graph_templates WHERE id = ?", (template_id,) ).fetchone() if not row: conn.close(); raise HTTPException(404, "Template introuvable") conn.execute("DELETE FROM causal_graph_templates WHERE id = ?", (template_id,)) conn.commit(); conn.close() return {"ok": True} except HTTPException: raise except Exception as e: logger.error(f"[causal_lab] delete_template {template_id}: {e}") raise HTTPException(500, str(e)) @router.get("/api/causal-lab/template/{template_id}") def get_template_detail(template_id: int): try: from services.database import get_conn from services.causal_graphs import get_template, init_tables, seed_templates conn = get_conn() init_tables(conn) seed_templates(conn) tmpl = get_template(conn, template_id) conn.close() if not tmpl: raise HTTPException(404, "Template introuvable") return tmpl except HTTPException: raise except Exception as e: logger.error(f"[causal_lab] get_template {template_id}: {e}") raise HTTPException(500, str(e)) @router.put("/api/causal-lab/template/{template_id}") def update_template(template_id: int, body: UpdateCoefRequest): try: from services.database import get_conn from services.causal_graphs import update_coefficients, init_tables, seed_templates conn = get_conn() init_tables(conn) seed_templates(conn) ok = update_coefficients(conn, template_id, body.coefficients) conn.close() if not ok: raise HTTPException(404, "Template introuvable") return {"ok": True} except HTTPException: raise except Exception as e: logger.error(f"[causal_lab] update_template {template_id}: {e}") raise HTTPException(500, str(e)) @router.get("/api/causal-lab/data-sources") def get_data_sources(): """ Retourne toutes les sources de données disponibles pour le mapping des nœuds observables : - prices : tickers yfinance (market_watchlist + defaults) - macro_series : series_id distincts depuis economic_events - ff_events : événements récurrents depuis ff_calendar """ PRICE_DEFAULTS = [ {"key": "EURUSD", "label": "EUR/USD"}, {"key": "XAUUSD", "label": "Or (Gold)"}, {"key": "SP500", "label": "S&P 500"}, {"key": "BRENT", "label": "Brent Crude"}, {"key": "US2Y", "label": "US 2Y Yield"}, {"key": "US10Y", "label": "US 10Y Yield"}, {"key": "EU10Y", "label": "EU 10Y Yield"}, ] try: from services.database import get_conn conn = get_conn() wl = conn.execute( "SELECT ticker, name FROM market_watchlist ORDER BY ticker" ).fetchall() known = {x["key"] for x in PRICE_DEFAULTS} extra = [ {"key": r["ticker"], "label": r["name"] or r["ticker"]} for r in wl if r["ticker"] not in known ] macro = conn.execute( """SELECT DISTINCT series_id, event_name FROM economic_events WHERE series_id IS NOT NULL AND series_id != '' ORDER BY series_id LIMIT 200""" ).fetchall() ff = conn.execute( """SELECT event_name, currency, COUNT(*) as cnt FROM ff_calendar WHERE event_name IS NOT NULL AND event_name != '' GROUP BY event_name, currency HAVING cnt >= 2 ORDER BY cnt DESC LIMIT 100""" ).fetchall() conn.close() return { "prices": PRICE_DEFAULTS + extra, "macro_series": [ {"key": r["series_id"], "label": r["event_name"] or r["series_id"]} for r in macro ], "ff_events": [ {"key": r["event_name"], "label": f"{r['event_name']} ({r['currency']})"} for r in ff ], } except Exception as e: logger.error(f"[causal_lab] data_sources: {e}") raise HTTPException(500, str(e)) @router.get("/api/causal-lab/market-events") def list_market_events( limit: int = Query(200, le=500), category: str = Query(""), q: str = Query(""), ): """Tous les market_events analysables (avec ou sans analyse existante).""" try: from services.database import get_conn conn = get_conn() _init(conn) conditions = ["1=1"] if category: conditions.append(f"e.category = '{category}'") if q: safe_q = q.replace("'", "''") conditions.append(f"(e.name LIKE '%{safe_q}%' OR e.description LIKE '%{safe_q}%')") where = " AND ".join(conditions) rows = conn.execute(f""" SELECT e.id, e.name, e.category, e.sub_type, e.start_date, e.end_date, e.level, e.impact_score, e.market_impact, e.affected_assets, e.actual_value, e.expected_value, e.surprise_pct, a.id as analysis_id, a.template_id, a.activation_score, a.instrument, a.analyzed_at FROM market_events e LEFT JOIN causal_event_analyses a ON a.market_event_id = e.id WHERE {where} ORDER BY e.start_date DESC LIMIT {limit} """).fetchall() conn.close() return [dict(r) for r in rows] except Exception as e: logger.error(f"[causal_lab] list_market_events: {e}") raise HTTPException(500, str(e)) @router.post("/api/causal-lab/recommend") def recommend_template(body: RecommendRequest): """GPT-4o recommande le template le plus adapté à l'événement.""" try: from services.database import get_conn from services.causal_graphs import get_templates, init_tables, seed_templates conn = get_conn() init_tables(conn) seed_templates(conn) ev_row = conn.execute( "SELECT * FROM market_events WHERE id = ?", (body.market_event_id,) ).fetchone() if not ev_row: conn.close() raise HTTPException(404, "market_event introuvable") event = dict(ev_row) templates = get_templates(conn) conn.close() recommendation = _gpt4o_recommend(event, templates) return {"event": event, "recommendation": recommendation} except HTTPException: raise except Exception as e: logger.error(f"[causal_lab] recommend: {e}") raise HTTPException(500, str(e)) @router.post("/api/causal-lab/instantiate") def instantiate_template(body: InstantiateRequest): """ GPT-4o-mini suggère les valeurs des nœuds user_input du template en fonction des spécificités de l'événement (surprise +/-, magnitude, catégorie…). Retourne { inputs: { node_id: value }, rationale: str } """ try: from services.database import get_conn conn = get_conn() _init(conn) ev_row = conn.execute("SELECT * FROM market_events WHERE id = ?", (body.market_event_id,)).fetchone() tmpl_row = conn.execute("SELECT * FROM causal_graph_templates WHERE id = ?", (body.template_id,)).fetchone() conn.close() if not ev_row or not tmpl_row: raise HTTPException(404, "event ou template introuvable") event = dict(ev_row) template = dict(tmpl_row) graph_json = json.loads(template.get("graph_json") or "{}") input_mapping = graph_json.get("input_mapping", {}) # Only user_input nodes need to be suggested (auto-sourced nodes are pulled from DB) user_nodes = { k: v for k, v in input_mapping.items() if not v.get("source") or v["source"] == "user_input" } if not user_nodes: return {"inputs": {}, "rationale": "Aucun nœud manuel — tout est auto-récupéré depuis la BD"} prompt = f"""Tu es un analyste financier. Tu dois configurer un graphe causal pour un événement de marché précis. Événement : {event.get('name')} Catégorie : {event.get('category')} / {event.get('sub_type', '')} Date : {event.get('start_date')} Description : {(event.get('description') or '')[:400]} Score impact : {event.get('impact_score', 0.5)} Valeur réelle : {event.get('actual_value')} | Valeur attendue : {event.get('expected_value')} | Surprise % : {event.get('surprise_pct')} Template : {template.get('name')} (catégorie : {template.get('category')}) Nœuds à configurer (entrées manuelles) : {json.dumps(user_nodes, indent=2, ensure_ascii=False)} Règles : - Surprise NÉGATIVE → valeurs négatives pour les nœuds de surprise - Surprise POSITIVE → valeurs positives - Si le nœud a un champ "range" (ex: [-3, 3]) : normalise OBLIGATOIREMENT la valeur dans ce range. Formule : valeur = clamp(surprise_pct / 100 * abs(range_max), range_min, range_max) Exemples : surprise_pct=-33.3%, range=[-3,3] → -0.333*3 = -1.0 | surprise_pct=-29.1%, range=[-3,3] → -0.873 NE JAMAIS retourner la valeur brute surprise_pct (-33.3, -29.1…) — c'est une erreur de calibration. - Si pas de "range" défini : retourne surprise_pct / 100 (forme fractionnelle, ex: -33.3% → -0.333) - Magnitude proportionnelle à l'ampleur de la surprise dans la plage autorisée Retourne UNIQUEMENT un JSON : {{ "node_id": valeur_numerique, ... }} Inclure uniquement les nœuds listés ci-dessus.""" from services.database import get_config as _get_cfg import openai as _openai _key = _get_cfg("openai_api_key") or "" if not _key: raise HTTPException(400, "Clé OpenAI manquante dans la configuration") client = _openai.OpenAI(api_key=_key) resp = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, temperature=0.2, max_tokens=400, ) raw = resp.choices[0].message.content or "{}" suggested = json.loads(raw) # Validate: only keep known nodes with numeric values validated: dict = {} for k, v in suggested.items(): if k in user_nodes: try: validated[k] = float(v) except (TypeError, ValueError): pass surprise = event.get("surprise_pct") rationale = ( f"Surprise {'+' if (surprise or 0) >= 0 else ''}{surprise:.1f}%" if surprise is not None else "Calibration basée sur la description de l'événement" ) return {"inputs": validated, "rationale": rationale} except HTTPException: raise except Exception as e: logger.error(f"[causal_lab] instantiate: {e}") raise HTTPException(500, str(e)) @router.post("/api/causal-lab/analyze") def analyze_event(body: AnalyzeRequest): """ Analyse complète : évalue le template sur l'événement, compare aux mouvements réels, stocke le résultat. """ try: from services.database import get_conn from services.causal_graphs import ( get_template, evaluate_graph, update_calibration, init_tables, seed_templates, ) conn = get_conn() init_tables(conn) seed_templates(conn) ev_row = conn.execute( "SELECT * FROM market_events WHERE id = ?", (body.market_event_id,) ).fetchone() if not ev_row: conn.close() raise HTTPException(404, "market_event introuvable") event = dict(ev_row) tmpl = get_template(conn, body.template_id) if not tmpl: conn.close() raise HTTPException(404, "Template introuvable") graph = tmpl["graph_json"] # Inputs de base depuis l'événement + mappings DB/yfinance inputs = {} mapping = graph.get("input_mapping", {}) YFINANCE_MAP = { "EURUSD": "EURUSD=X", "XAUUSD": "GC=F", "SP500": "^GSPC", "BRENT": "BZ=F", "US2Y": "US2YT=RR", "US10Y": "^TNX", "EU10Y": "GE10YT=RR", } edate_str = event["start_date"][:10] for input_id, cfg in mapping.items(): src = cfg.get("source", "") key = cfg.get("key", "") field = cfg.get("field", "close") if src == "market_watchlist" and key: try: import yfinance as yf from datetime import datetime as _dt, timedelta as _td event_dt = _dt.strptime(edate_str, "%Y-%m-%d") start = (event_dt - _td(days=5)).strftime("%Y-%m-%d") end = (event_dt + _td(days=2)).strftime("%Y-%m-%d") sym = YFINANCE_MAP.get(key, key) df = yf.download(sym, start=start, end=end, interval="1d", progress=False, auto_adjust=True) if df is not None and len(df) > 0: if hasattr(df.columns, "levels"): df.columns = df.columns.get_level_values(0) rows_yf = [ (str(idx.date()), float(row["Close"])) for idx, row in df.iterrows() if float(row["Close"]) == float(row["Close"]) ] if field == "change_pct" and len(rows_yf) >= 2: pre = next((c for d, c in rows_yf if d < edate_str), None) cur = next((c for d, c in rows_yf if d == edate_str), None) if pre and cur and pre != 0: inputs[input_id] = round((cur - pre) / pre * 100, 4) else: val = next((c for d, c in rows_yf if d == edate_str), None) if val is None: val = next((c for d, c in reversed(rows_yf) if d <= edate_str), None) if val is not None: inputs[input_id] = round(val, 5) except Exception as _e: logger.debug(f"[causal_lab] market_watchlist fetch {key}: {_e}") elif src == "economic_events" and key: row_e = conn.execute( """SELECT actual_value, forecast_value FROM economic_events WHERE series_id = ? AND event_date <= ? AND actual_value IS NOT NULL ORDER BY event_date DESC LIMIT 1""", (key, edate_str) ).fetchone() if row_e: if field == "surprise" and row_e["forecast_value"] is not None: inputs[input_id] = round( float(row_e["actual_value"]) - float(row_e["forecast_value"]), 4 ) else: inputs[input_id] = float(row_e["actual_value"]) elif src == "ff_calendar" and key: row_f = conn.execute( """SELECT actual_value, forecast_value FROM ff_calendar WHERE event_name = ? AND event_date <= ? AND actual_value IS NOT NULL ORDER BY event_date DESC LIMIT 1""", (key, edate_str) ).fetchone() if row_f and row_f["actual_value"]: try: actual = float(row_f["actual_value"]) if field == "surprise" and row_f["forecast_value"]: inputs[input_id] = round(actual - float(row_f["forecast_value"]), 4) else: inputs[input_id] = actual except (ValueError, TypeError): pass # Sources legacy elif src == "surprise_bps" and event.get("surprise_pct") is not None: # Valeur brute en bps (ex: écart taux BCE en points de base) raw = float(event["surprise_pct"]) node_range = cfg.get("range") if node_range and len(node_range) == 2: lo, hi = float(node_range[0]), float(node_range[1]) inputs[input_id] = round(max(lo, min(hi, raw)), 4) else: inputs[input_id] = round(raw, 4) elif src == "surprise" and event.get("surprise_pct") is not None: raw = float(event["surprise_pct"]) node_range = cfg.get("range") if node_range and len(node_range) == 2: lo, hi = float(node_range[0]), float(node_range[1]) bound = max(abs(lo), abs(hi)) if bound <= 10: # Plage type z-score : normalise surprise_pct → [-bound, +bound] inputs[input_id] = round(max(lo, min(hi, raw / 100 * bound)), 4) else: inputs[input_id] = round(max(lo, min(hi, raw)), 4) else: inputs[input_id] = round(raw / 100, 4) # fractional par défaut elif src == "impact_score_scaled" and event.get("impact_score") is not None: inputs[input_id] = float(event["impact_score"]) / 10.0 elif src == "impact_score_pct" and event.get("impact_score") is not None: inputs[input_id] = float(event["impact_score"]) elif src == "actual_value" and event.get("actual_value") is not None: inputs[input_id] = float(event["actual_value"]) # Inputs manuels (saisie frontend — priorité sur auto-fetch) inputs.update(body.inputs) # Évaluation du graphe node_values = evaluate_graph(graph, inputs, body.coef_overrides or {}) # Lag effectif : max sur toutes les arêtes du graphe # (filtre output_ids optionnel mais peu fiable si types de nœuds non standards) edges = graph.get("edges", []) nodes_map = {n["id"]: n for n in graph.get("nodes", [])} output_ids = {n["id"] for n in graph.get("nodes", []) if n.get("type") in ("market_asset", "output")} edges_to_output = [e for e in edges if e.get("to") in output_ids] edges_for_lag = edges_to_output if edges_to_output else edges # fallback : toutes arêtes effective_lag = max( (e.get("lag_min", 0) or 0 for e in edges_for_lag), default=0, ) effective_lag_days = max( (e.get("lag_days", 0) or 0 for e in edges_for_lag), default=0, ) # Diagnostic : combien d'arêtes portent un lag_days > 0 edges_with_lag_days = sum(1 for e in edges if (e.get("lag_days") or 0) > 0) lag_debug = { "output_ids": list(output_ids), "edges_to_output": len(edges_to_output), "edges_with_lag_days": edges_with_lag_days, "used_all_edges": not edges_to_output, } # Instruments : utiliser tous ceux du template si aucun spécifié if body.instrument: instruments = list({body.instrument} | set(tmpl.get("instruments", [body.instrument]))) else: instruments = list(set(tmpl.get("instruments", []))) or ["EURUSD"] primary_inst = body.instrument or (instruments[0] if instruments else "EURUSD") prices = _fetch_prices(event["start_date"], instruments, lag_days=effective_lag_days) edate = event["start_date"][:10] actual_moves: dict = {} drift_by_inst: dict = {} for inst in instruments: drift = _drift_metrics(prices, event["start_date"], inst, lag_min=effective_lag, lag_days=effective_lag_days) drift_by_inst[inst] = drift if drift.get("post_pips") is not None: actual_moves[inst] = drift["post_pips"] # Activation activation = _activation_score(node_values, actual_moves, graph) # Taux yields = { "us10y": _yield_delta(prices.get("US10Y", []), edate), "eu10y": _yield_delta(prices.get("EU10Y", []), edate), "us2y": _yield_delta(prices.get("US2Y", []), edate), } analyzed_at = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") result = { "event": event, "template_id": body.template_id, "template_name": tmpl["name"], "instrument": primary_inst, "instruments": instruments, "inputs": inputs, "node_values": node_values, "actual_moves": actual_moves, "yields": yields, "activation": activation, "drift": drift_by_inst.get(primary_inst, {}), "drift_by_inst": drift_by_inst, "prices_mode": prices.get("mode", "none"), "effective_lag_min": effective_lag, "effective_lag_days": effective_lag_days, "lag_debug": lag_debug, "analyzed_at": analyzed_at, "graph_json": graph, # structure complète pour visualisation frontend } # Persistance — UPSERT : one analysis per (market_event_id, template_id) existing = conn.execute( "SELECT id FROM causal_event_analyses WHERE market_event_id=? AND template_id=?", (body.market_event_id, body.template_id), ).fetchone() # Stocker tous les instruments analysés pour que la frise Instrument Analysis les retrouve all_instruments_csv = ",".join(sorted(drift_by_inst.keys())) or primary_inst if existing: conn.execute(""" UPDATE causal_event_analyses SET instrument=?, inputs_json=?, override_params=?, prediction_json=?, actual_json=?, activation_score=?, drift_json=?, analyzed_at=? WHERE market_event_id=? AND template_id=? """, ( all_instruments_csv, json.dumps(inputs), json.dumps(body.coef_overrides), json.dumps(node_values), json.dumps(actual_moves), activation.get("score"), json.dumps(drift_by_inst), analyzed_at, body.market_event_id, body.template_id, )) else: conn.execute(""" INSERT INTO causal_event_analyses (market_event_id, template_id, instrument, inputs_json, override_params, prediction_json, actual_json, activation_score, drift_json, analyzed_at) VALUES (?,?,?,?,?,?,?,?,?,?) """, ( body.market_event_id, body.template_id, all_instruments_csv, json.dumps(inputs), json.dumps(body.coef_overrides), json.dumps(node_values), json.dumps(actual_moves), activation.get("score"), json.dumps(drift_by_inst), analyzed_at, )) conn.commit() # Calibration update_calibration(conn, body.template_id, { "activation_score": activation.get("score"), "instrument": primary_inst, "pred_pips": node_values.get(primary_inst.lower(), 0) or next((v for k, v in node_values.items() if primary_inst.lower() in k.lower()), 0), "actual_pips": actual_moves.get(primary_inst), "analyzed_at": analyzed_at, }) conn.close() return result except HTTPException: raise except Exception as e: logger.error(f"[causal_lab] analyze: {e}") raise HTTPException(500, str(e)) def _build_graph_json_from_spec(spec: dict) -> dict: """Construit un graph_json complet depuis le spec simplifié généré par GPT-4o.""" input_node = spec.get("input_node", {}) intermediates = spec.get("intermediate_nodes", []) outputs = spec.get("output_nodes", []) nodes: list = [] edges: list = [] coefficients: dict = {} input_mapping: dict = {} # Nœud d'entrée — centré en haut inp_id = input_node.get("id", "surprise") nodes.append({ "id": inp_id, "label": input_node.get("label", "Surprise"), "type": "input", "x": 200, "y": 50, "unit": input_node.get("unit", "% vs consensus"), }) input_mapping[inp_id] = { "source": "surprise", "unit": input_node.get("unit", "% vs consensus"), "description": input_node.get("label", "Surprise"), } # Nœuds intermédiaires — étalés horizontalement au milieu n_inter = max(len(intermediates), 1) x_step_i = max(160, 340 // n_inter) for i, inter in enumerate(intermediates): x = 200 - (len(intermediates) - 1) * x_step_i // 2 + i * x_step_i inter_id = inter["id"] coef_name = inter.get("coef_name", f"coef_{inter_id}") coef_val = float(inter.get("coef_value", 2.0)) coefficients[coef_name] = { "value": coef_val, "calibrated": False, "description": inter.get("coef_desc", coef_name), } nodes.append({ "id": inter_id, "label": inter.get("label", inter_id), "type": "intermediate", "x": x, "y": 175, "formula": inp_id + " * {{" + coef_name + "}}", }) edges.append({"from": inp_id, "to": inter_id, "sign": "positive" if coef_val >= 0 else "negative", "style": "solid"}) # Nœuds de sortie — étalés horizontalement en bas inter_ids = [n["id"] for n in intermediates] n_out = max(len(outputs), 1) x_step_o = max(140, 320 // n_out) for i, out in enumerate(outputs): x = 200 - (len(outputs) - 1) * x_step_o // 2 + i * x_step_o out_id = out["id"] coef_name = out.get("coef_name", f"coef_{out_id}") coef_val = float(out.get("coef_value", -100.0)) coefficients[coef_name] = { "value": coef_val, "calibrated": False, "description": out.get("coef_desc", coef_name), } src = out.get("source_intermediate", "") from_id = src if src in inter_ids else (intermediates[0]["id"] if intermediates else inp_id) nodes.append({ "id": out_id, "label": out.get("label", out_id), "type": "market_asset", "x": x, "y": 310, "unit": "pips", "instrument": out.get("instrument", "EURUSD"), "formula": from_id + " * {{" + coef_name + "}}", }) edges.append({"from": from_id, "to": out_id, "sign": "positive" if coef_val >= 0 else "negative", "style": "solid"}) instruments = list({out.get("instrument", "EURUSD") for out in outputs}) return { "nodes": nodes, "edges": edges, "coefficients": coefficients, "input_mapping": input_mapping, "instruments": instruments, } def auto_assign_template(event_id: int) -> dict: """ Auto-assign or auto-create a causal template for a market event. Called by the eco detector when auto_template=True in desk config. Returns {"template_id": int, "action": "assigned"|"created", "name": str} or {"error": ...} """ try: from services.database import get_conn, get_config from services.causal_graphs import get_templates, init_tables, seed_templates import openai key = get_config("openai_api_key") or "" if not key: return {"error": "Clé OpenAI manquante"} conn = get_conn() init_tables(conn) seed_templates(conn) ev_row = conn.execute("SELECT * FROM market_events WHERE id = ?", (event_id,)).fetchone() if not ev_row: conn.close() return {"error": f"Event {event_id} introuvable"} event = dict(ev_row) templates = get_templates(conn) # Step 1 — recommend an existing template recommendation = _gpt4o_recommend(event, templates) tmpl_id = recommendation.get("template_id") confidence = float(recommendation.get("confidence") or 0.0) if tmpl_id and confidence >= 0.6: conn.execute("UPDATE market_events SET template_id = ? WHERE id = ?", (tmpl_id, event_id)) conn.commit() row = conn.execute("SELECT name FROM causal_graph_templates WHERE id = ?", (tmpl_id,)).fetchone() conn.close() return {"template_id": tmpl_id, "action": "assigned", "name": row["name"] if row else "", "confidence": confidence} # Step 2 — no good match → generate a new template conn.close() client = openai.OpenAI(api_key=key) prompt = f"""Tu es un analyste financier spécialisé en graphes causaux macro. Crée un graphe causal pour l'événement suivant. Événement : - Nom : {event.get('name')} - Catégorie : {event.get('category')} / {event.get('sub_type', '')} - Description : {(event.get('description') or '')[:400]} - Réel : {event.get('actual_value')} | Attendu : {event.get('expected_value')} | Surprise % : {event.get('surprise_pct')} Retourne UNIQUEMENT ce JSON (sans commentaire ni markdown) : {{ "name": "", "category": "macro_us", "description": "<1 phrase décrivant le mécanisme>", "input_node": {{"id": "", "label": "