feat: causal lab

2026-06-28 18:08:09 +02:00
parent c286c7c000
commit 6ebbf4326e
5 changed files with 481 additions and 133 deletions
--- a/backend/routers/causal_lab.py
+++ b/backend/routers/causal_lab.py
@@ -431,6 +431,8 @@ def patch_template(template_id: int, body: dict):
            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}
@@ -1235,3 +1237,100 @@ def get_calibration():
    except Exception as e:
        logger.error(f"[causal_lab] calibration: {e}")
        raise HTTPException(500, str(e))
+
+
+@router.post("/api/causal-lab/template/{template_id}/generate-theory")
+def generate_theory(template_id: int):
+    """GPT-4o-mini génère absorption_days, decay_type et confidence pour le template parent."""
+    try:
+        from services.database import get_conn, get_config
+        from services.causal_graphs import get_template
+
+        conn = get_conn()
+        _init(conn)
+        tmpl = get_template(conn, template_id)
+        if not tmpl:
+            conn.close(); raise HTTPException(404, "Template introuvable")
+
+        stats = conn.execute("""
+            SELECT COUNT(*) as n, AVG(activation_score) as avg_act
+            FROM causal_event_analyses WHERE template_id = ?
+        """, (template_id,)).fetchone()
+        conn.close()
+
+        n_analyses = stats["n"] if stats else 0
+        avg_act    = round((stats["avg_act"] or 0.0), 2) if stats else 0.0
+
+        graph = tmpl.get("graph_json", {})
+        coefs = {k: v.get("value") for k, v in graph.get("coefficients", {}).items()}
+
+        key = get_config("openai_api_key") or ""
+        if not key:
+            raise HTTPException(400, "Clé OpenAI manquante dans la configuration")
+
+        import openai
+        prompt = (
+            f'Tu es un expert en microstructure de marché et dynamique d\'absorption des chocs de prix.\n\n'
+            f'Template causal : "{tmpl["name"]}"\n'
+            f'Catégorie : {tmpl["category"]} / {tmpl.get("sub_type", "")}\n'
+            f'Description : {tmpl.get("description", "")}\n'
+            f'Instruments : {tmpl.get("instruments", [])}\n'
+            f'Coefficients : {json.dumps(coefs, ensure_ascii=False)}\n'
+            f'Analyses historiques : {n_analyses} (activation directionnelle moy. : {avg_act:.0%})\n\n'
+            f'Propose les paramètres d\'absorption de l\'impact de marché :\n'
+            f'- absorption_days : jours calendaires avant absorption à >90% (entier 1-60)\n'
+            f'- decay_type : "step" (tout-ou-rien), "linear" (déclin linéaire), "exp" (exponentiel)\n'
+            f'- confidence : confiance 0.0-1.0\n'
+            f'- rationale : justification courte (max 120 chars)\n\n'
+            f'Références : décisions taux→3-7j/exp ; CPI/NFP→2-5j/exp ; '
+            f'géopolitique→5-21j/linear ; PMI secondaire→1-2j/step\n\n'
+            f'JSON uniquement : {{"absorption_days": N, "decay_type": "...", "confidence": 0.X, "rationale": "..."}}'
+        )
+
+        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=200,
+        )
+        raw = resp.choices[0].message.content or "{}"
+        params = json.loads(raw)
+
+        absorption_days = max(1, min(60, int(params.get("absorption_days", 7))))
+        decay_type = params.get("decay_type", "exp")
+        if decay_type not in ("step", "linear", "exp"):
+            decay_type = "exp"
+        confidence = round(max(0.0, min(1.0, float(params.get("confidence", 0.5)))), 2)
+        rationale  = str(params.get("rationale", ""))[:150]
+
+        conn2 = get_conn()
+        existing_calib = dict(tmpl.get("calibration_json") or {})
+        existing_calib.update({
+            "absorption_days": absorption_days,
+            "decay_type": decay_type,
+            "confidence": confidence,
+            "theory_rationale": rationale,
+            "theory_generated_at": datetime.utcnow().isoformat() + "Z",
+        })
+        conn2.execute(
+            "UPDATE causal_graph_templates SET calibration_json=?, updated_at=datetime('now') WHERE id=?",
+            (json.dumps(existing_calib), template_id),
+        )
+        conn2.commit()
+        conn2.close()
+
+        return {
+            "template_id": template_id,
+            "absorption_days": absorption_days,
+            "decay_type": decay_type,
+            "confidence": confidence,
+            "rationale": rationale,
+        }
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"[causal_lab] generate_theory {template_id}: {e}")
+        raise HTTPException(500, str(e))
--- a/backend/routers/instruments.py
+++ b/backend/routers/instruments.py
@@ -2,6 +2,9 @@
 Instrument Dashboard Router.
 Exposes per-instrument snapshot (price, indicators, regime, trend, events) and AI narrative.
 """
+import json
+import math
+from datetime import datetime, timedelta
 from fastapi import APIRouter, HTTPException, Query
 from pydantic import BaseModel
 from typing import List, Dict, Any, Optional
@@ -85,3 +88,142 @@ def update_drivers(instrument_id: str, body: DriverUpdate) -> Dict[str, Any]:
        raise HTTPException(status_code=500, detail=str(e))

    return {"ok": True, "instrument_id": instrument_id.upper(), "drivers_count": len(body.drivers)}
+
+
+# ── Instrument mult (pips → price conversion) ─────────────────────────────────
+
+_INST_MULT: Dict[str, int] = {"EURUSD": 10000, "GBPUSD": 10000, "USDJPY": 100, "AUDUSD": 10000}
+
+def _get_mult(inst: str) -> int:
+    return _INST_MULT.get(inst.upper(), 10)
+
+
+def _decay(days_after: int, absorption_days: int, decay_type: str) -> float:
+    """Decay factor ∈ [0,1] for a given number of days after the event."""
+    if days_after < 0:
+        return 0.0
+    if decay_type == "step":
+        return 1.0 if days_after <= absorption_days else 0.0
+    elif decay_type == "linear":
+        return max(0.0, 1.0 - days_after / max(absorption_days, 1))
+    else:  # exp — 3 time-constants reach ~5% at absorption_days
+        lam = 3.0 / max(absorption_days, 1)
+        return math.exp(-lam * days_after)
+
+
+@router.get("/{instrument_id}/theoretical-curve")
+def get_theoretical_curve(
+    instrument_id: str,
+    period: str = Query("1y"),
+) -> List[Dict[str, Any]]:
+    """
+    Courbe théorique composite : pour chaque jour calendaire de la période,
+    somme des impacts décroissants des analyses causales stockées.
+
+    Retourne [{date, cumulative_pips, contributions: [{template_name, event_name, event_date, pips, decay_factor}]}]
+    """
+    from services.database import get_conn
+
+    period_lookback: Dict[str, int] = {
+        "5d": 7, "1mo": 35, "3mo": 95, "6mo": 190,
+        "1y": 370, "2y": 740, "5y": 1830,
+    }
+    lookback = period_lookback.get(period, 370)
+
+    date_to   = datetime.utcnow().date()
+    date_from = date_to - timedelta(days=lookback)
+    # Fetch events that started before date_from too — they may still be decaying into the window
+    extended_from = date_from - timedelta(days=90)
+
+    inst_upper = instrument_id.upper()
+
+    conn = get_conn()
+    try:
+        rows = conn.execute("""
+            SELECT a.id,
+                   a.prediction_json,
+                   a.activation_score,
+                   e.start_date  AS event_date,
+                   e.name        AS event_name,
+                   t.name        AS template_name,
+                   t.calibration_json
+            FROM   causal_event_analyses a
+            JOIN   market_events          e ON e.id = a.market_event_id
+            JOIN   causal_graph_templates t ON t.id = a.template_id
+            WHERE  a.instrument = ?
+              AND  e.start_date >= ?
+              AND  e.start_date <= ?
+            ORDER  BY e.start_date
+        """, (inst_upper, str(extended_from), str(date_to))).fetchall()
+    finally:
+        conn.close()
+
+    # ── Build calendar-day series ─────────────────────────────────────────────
+    all_dates: List[str] = []
+    cur = date_from
+    while cur <= date_to:
+        all_dates.append(str(cur))
+        cur += timedelta(days=1)
+
+    curve: Dict[str, Dict] = {
+        d: {"cumulative_pips": 0.0, "contributions": []} for d in all_dates
+    }
+
+    for row in rows:
+        r = dict(row)
+        try:
+            predictions = json.loads(r["prediction_json"] or "{}")
+            calib       = json.loads(r["calibration_json"] or "{}")
+        except Exception:
+            continue
+
+        # Extract predicted pips for this instrument from node_values dict
+        inst_lower = inst_upper.lower()
+        predicted_pips: Optional[float] = None
+        if inst_lower in predictions:
+            predicted_pips = float(predictions[inst_lower])
+        else:
+            for k, v in predictions.items():
+                if inst_lower in k.lower():
+                    try:
+                        predicted_pips = float(v)
+                        break
+                    except (TypeError, ValueError):
+                        pass
+
+        if predicted_pips is None or predicted_pips == 0:
+            continue
+
+        absorption_days: int  = max(1, int(calib.get("absorption_days", 7)))
+        dtype:           str  = str(calib.get("decay_type", "exp"))
+        event_date_str:  str  = r["event_date"][:10]
+
+        try:
+            event_date = datetime.strptime(event_date_str, "%Y-%m-%d").date()
+        except ValueError:
+            continue
+
+        for d in all_dates:
+            cal_date   = datetime.strptime(d, "%Y-%m-%d").date()
+            days_after = (cal_date - event_date).days
+            df         = _decay(days_after, absorption_days, dtype)
+            if df < 0.01:
+                continue
+            contribution = round(predicted_pips * df, 2)
+            curve[d]["cumulative_pips"] += contribution
+            curve[d]["contributions"].append({
+                "template_name": r["template_name"],
+                "event_name":    r["event_name"],
+                "event_date":    event_date_str,
+                "pips":          contribution,
+                "decay_factor":  round(df, 3),
+            })
+
+    # Round totals and strip empty-contribution days at the edges
+    result = []
+    for d in all_dates:
+        entry = curve[d]
+        entry["cumulative_pips"] = round(entry["cumulative_pips"], 1)
+        result.append({"date": d, **entry})
+
+    return result