935 lines
35 KiB
Python
935 lines
35 KiB
Python
"""
|
|
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]) -> dict:
|
|
"""
|
|
Télécharge les prix autour de l'événement pour chaque instrument demandé.
|
|
Retourne dict { instrument: [{"t": iso, "c": float}, ...] }
|
|
"""
|
|
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) pour les FX / actifs principaux
|
|
use_intraday = days_ago < 55
|
|
|
|
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
|
|
start = (event_dt - timedelta(days=5)).strftime("%Y-%m-%d")
|
|
end = (event_dt + timedelta(days=5)).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) -> 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}
|
|
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:
|
|
pre = [b for b in series if b["t"] < edate]
|
|
same = [b for b in series if b["t"] == edate]
|
|
if not pre or not same:
|
|
return empty
|
|
mult = 10000 if inst in ("EURUSD",) else 10
|
|
pre_pips = None
|
|
post_pips = round((same[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."
|
|
)
|
|
|
|
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": <int or null>,
|
|
"template_name": "<name>",
|
|
"confidence": <0.0-1.0>,
|
|
"rationale": "<2-3 sentences explaining why this template fits>",
|
|
"coefficient_suggestions": {{
|
|
"<coef_key>": <suggested_value>
|
|
}},
|
|
"instrument_focus": ["<primary instrument>"],
|
|
"alternative_template_id": <int or null>,
|
|
"notes": "<any additional context>"
|
|
}}"""
|
|
|
|
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 AnalyzeRequest(BaseModel):
|
|
market_event_id: int
|
|
template_id: int
|
|
instrument: str = "EURUSD"
|
|
inputs: dict = {}
|
|
coef_overrides: dict = {}
|
|
|
|
|
|
# ── 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)
|
|
count = conn.execute("SELECT COUNT(*) FROM causal_graph_templates").fetchone()[0]
|
|
conn.close()
|
|
return {
|
|
"built_in_templates": len(BUILT_IN_TEMPLATES),
|
|
"db_templates": count,
|
|
"status": "ok",
|
|
}
|
|
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 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/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" and event.get("surprise_pct") is not None:
|
|
inputs[input_id] = float(event["surprise_pct"])
|
|
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 des lag_min sur toutes les arêtes menant à un market_asset
|
|
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")}
|
|
effective_lag = max(
|
|
(e.get("lag_min", 0) or 0 for e in edges if e.get("to") in output_ids),
|
|
default=0,
|
|
)
|
|
|
|
# Prix réels
|
|
instruments = list({body.instrument} | set(tmpl.get("instruments", [body.instrument])))
|
|
prices = _fetch_prices(event["start_date"], instruments)
|
|
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)
|
|
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": body.instrument,
|
|
"inputs": inputs,
|
|
"node_values": node_values,
|
|
"actual_moves": actual_moves,
|
|
"yields": yields,
|
|
"activation": activation,
|
|
"drift": drift_by_inst.get(body.instrument, {}),
|
|
"prices_mode": prices.get("mode", "none"),
|
|
"effective_lag_min": effective_lag,
|
|
"analyzed_at": analyzed_at,
|
|
}
|
|
|
|
# Persistance
|
|
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,
|
|
body.instrument,
|
|
json.dumps(inputs),
|
|
json.dumps(body.coef_overrides),
|
|
json.dumps(node_values),
|
|
json.dumps(actual_moves),
|
|
activation.get("score"),
|
|
json.dumps(drift_by_inst.get(body.instrument, {})),
|
|
analyzed_at,
|
|
))
|
|
conn.commit()
|
|
|
|
# Calibration
|
|
update_calibration(conn, body.template_id, {
|
|
"activation_score": activation.get("score"),
|
|
"instrument": body.instrument,
|
|
"pred_pips": node_values.get(body.instrument.lower(), 0) or
|
|
next((v for k, v in node_values.items()
|
|
if body.instrument.lower() in k.lower()), 0),
|
|
"actual_pips": actual_moves.get(body.instrument),
|
|
"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))
|
|
|
|
|
|
@router.get("/api/causal-lab/analyses")
|
|
def list_analyses(
|
|
template_id: int = Query(0),
|
|
instrument: str = Query(""),
|
|
limit: int = Query(100, le=500),
|
|
):
|
|
try:
|
|
from services.database import get_conn
|
|
conn = get_conn()
|
|
_init(conn)
|
|
|
|
conditions = ["1=1"]
|
|
if template_id:
|
|
conditions.append(f"a.template_id = {template_id}")
|
|
if instrument:
|
|
conditions.append(f"a.instrument = '{instrument}'")
|
|
where = " AND ".join(conditions)
|
|
|
|
rows = conn.execute(f"""
|
|
SELECT a.*, e.name as event_name, e.category, e.start_date,
|
|
t.name as template_name
|
|
FROM causal_event_analyses a
|
|
LEFT JOIN market_events e ON e.id = a.market_event_id
|
|
LEFT JOIN causal_graph_templates t ON t.id = a.template_id
|
|
WHERE {where}
|
|
ORDER BY a.analyzed_at DESC
|
|
LIMIT {limit}
|
|
""").fetchall()
|
|
conn.close()
|
|
|
|
result = []
|
|
for r in rows:
|
|
d = dict(r)
|
|
for f in ("inputs_json", "override_params", "prediction_json", "actual_json", "drift_json"):
|
|
try:
|
|
d[f] = json.loads(d[f] or "{}")
|
|
except Exception:
|
|
d[f] = {}
|
|
result.append(d)
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"[causal_lab] list_analyses: {e}")
|
|
raise HTTPException(500, str(e))
|
|
|
|
|
|
@router.get("/api/causal-lab/calibration")
|
|
def get_calibration():
|
|
"""Statistiques de calibration par template (avg_activation, avg_pred vs avg_actual)."""
|
|
try:
|
|
from services.database import get_conn
|
|
conn = get_conn()
|
|
_init(conn)
|
|
|
|
rows = conn.execute("""
|
|
SELECT t.id, t.name, t.category, t.sub_type,
|
|
t.calibration_json, t.heuristic_ver,
|
|
COUNT(a.id) as n_analyses,
|
|
AVG(a.activation_score) as avg_activation
|
|
FROM causal_graph_templates t
|
|
LEFT JOIN causal_event_analyses a ON a.template_id = t.id
|
|
GROUP BY t.id
|
|
ORDER BY t.category, t.name
|
|
""").fetchall()
|
|
conn.close()
|
|
|
|
result = []
|
|
for r in rows:
|
|
d = dict(r)
|
|
try:
|
|
d["calibration"] = json.loads(d.pop("calibration_json") or "{}")
|
|
except Exception:
|
|
d["calibration"] = {}
|
|
if d.get("avg_activation") is not None:
|
|
d["avg_activation"] = round(d["avg_activation"], 3)
|
|
result.append(d)
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"[causal_lab] calibration: {e}")
|
|
raise HTTPException(500, str(e))
|