2041 lines
83 KiB
Python
2041 lines
83 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], 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",
|
|
"GBPUSD": "GBPUSD=X",
|
|
"USDJPY": "USDJPY=X",
|
|
"USDCHF": "USDCHF=X",
|
|
"AUDUSD": "AUDUSD=X",
|
|
"NZDUSD": "NZDUSD=X",
|
|
"USDCAD": "USDCAD=X",
|
|
"XAUUSD": "GC=F",
|
|
"XAGUSD": "SI=F",
|
|
"SP500": "^GSPC",
|
|
"NASDAQ": "^IXIC",
|
|
"DAX": "^GDAXI",
|
|
"FTSE": "^FTSE",
|
|
"BRENT": "BZ=F",
|
|
"WTI": "CL=F",
|
|
"US2Y": "US2YT=RR",
|
|
"US10Y": "^TNX",
|
|
"EU10Y": "GE10YT=RR",
|
|
"US30Y": "^TYX",
|
|
"DXY": "DX-Y.NYB",
|
|
}
|
|
|
|
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": <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 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])
|
|
inputs[input_id] = round(max(lo, min(hi, raw)), 4)
|
|
else:
|
|
inputs[input_id] = round(raw, 4)
|
|
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",
|
|
"lag_days": 3})
|
|
|
|
# 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",
|
|
"lag_days": 3})
|
|
|
|
instruments = list({out.get("instrument", "EURUSD") for out in outputs})
|
|
return {
|
|
"nodes": nodes, "edges": edges,
|
|
"coefficients": coefficients, "input_mapping": input_mapping,
|
|
"instruments": instruments,
|
|
}
|
|
|
|
|
|
def _run_auto_analysis(event: dict, template_id: int) -> dict:
|
|
"""
|
|
Causal analysis for the auto_template pipeline.
|
|
Returns {"ok": bool, "inputs": dict, "node_values": dict, "actual_moves": dict, "error": str}.
|
|
"""
|
|
try:
|
|
from services.database import get_conn
|
|
from services.causal_graphs import get_template, evaluate_graph
|
|
|
|
conn = get_conn()
|
|
tmpl = get_template(conn, template_id)
|
|
if not tmpl:
|
|
conn.close()
|
|
logger.warning(f"[auto_analysis] template #{template_id} introuvable")
|
|
return {"ok": False, "error": f"template #{template_id} introuvable"}
|
|
|
|
graph = tmpl["graph_json"]
|
|
inputs = {}
|
|
mapping = graph.get("input_mapping", {})
|
|
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 == "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
|
|
|
|
elif src in ("surprise_bps", "surprise"):
|
|
# Step 1: use stored surprise_pct
|
|
sp = event.get("surprise_pct")
|
|
# Step 2: compute from actual_value / expected_value
|
|
if sp is None:
|
|
try:
|
|
act_s = str(event.get("actual_value") or "").replace(",", ".").strip()
|
|
exp_s = str(event.get("expected_value") or "").replace(",", ".").strip()
|
|
if act_s and exp_s:
|
|
act_f = float(act_s)
|
|
exp_f = float(exp_s)
|
|
sp = (act_f - exp_f) / abs(exp_f) * 100 if abs(exp_f) > 1e-9 else (act_f - exp_f)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
# Step 3: look up ff_calendar by currency + date
|
|
if sp is None:
|
|
try:
|
|
currency = (event.get("sub_type") or "").upper()
|
|
if currency:
|
|
row_ff = conn.execute("""
|
|
SELECT actual_value, forecast_value FROM ff_calendar
|
|
WHERE currency = ? AND event_date <= ? AND actual_value IS NOT NULL
|
|
AND forecast_value IS NOT NULL
|
|
ORDER BY event_date DESC LIMIT 1
|
|
""", (currency, edate_str)).fetchone()
|
|
if row_ff:
|
|
a_f = float(row_ff["actual_value"])
|
|
e_f = float(row_ff["forecast_value"])
|
|
sp = (a_f - e_f) / abs(e_f) * 100 if abs(e_f) > 1e-9 else (a_f - e_f)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
if sp is not None:
|
|
raw = float(sp)
|
|
node_range = cfg.get("range")
|
|
if src == "surprise_bps":
|
|
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)
|
|
else:
|
|
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 == "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"])
|
|
|
|
node_values = evaluate_graph(graph, inputs, {})
|
|
|
|
instruments = list(set(tmpl.get("instruments", []))) or ["EURUSD"]
|
|
all_instruments_csv = ",".join(sorted(instruments)) or "EURUSD"
|
|
analyzed_at = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
event_id = event["id"]
|
|
|
|
# Effective lag from template edges (for price window and chip width)
|
|
edges = graph.get("edges", [])
|
|
effective_lag_days = max((e.get("lag_days", 0) or 0 for e in edges), default=0)
|
|
if effective_lag_days == 0:
|
|
effective_lag_days = 5 # Default: measure close-to-close impact over 5 trading days
|
|
|
|
# Fetch actual prices to compute actual pips for scoring
|
|
actual_moves: dict = {}
|
|
drift_by_inst: dict = {}
|
|
try:
|
|
prices = _fetch_prices(event["start_date"], instruments, lag_days=effective_lag_days)
|
|
for inst in instruments:
|
|
drift = _drift_metrics(prices, event["start_date"], inst,
|
|
lag_min=0, lag_days=effective_lag_days)
|
|
drift_by_inst[inst] = drift
|
|
if drift.get("post_pips") is not None:
|
|
actual_moves[inst] = drift["post_pips"]
|
|
except Exception as _pe:
|
|
logger.warning(f"[auto_analysis] price fetch failed for event #{event_id}: {_pe}")
|
|
|
|
logger.info(
|
|
f"[auto_analysis] event #{event_id} → tmpl #{template_id} | "
|
|
f"surprise_pct={event.get('surprise_pct')} inputs={inputs} | "
|
|
f"nodes={len(node_values)} | instruments={all_instruments_csv} | actual_moves={actual_moves}"
|
|
)
|
|
|
|
existing = conn.execute(
|
|
"SELECT id FROM causal_event_analyses WHERE market_event_id=? AND template_id=?",
|
|
(event_id, template_id),
|
|
).fetchone()
|
|
|
|
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({}),
|
|
json.dumps(node_values), json.dumps(actual_moves), None,
|
|
json.dumps(drift_by_inst), analyzed_at,
|
|
event_id, 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 (?,?,?,?,?,?,?,?,?,?)
|
|
""", (
|
|
event_id, template_id, all_instruments_csv,
|
|
json.dumps(inputs), json.dumps({}),
|
|
json.dumps(node_values), json.dumps(actual_moves),
|
|
None, json.dumps(drift_by_inst), analyzed_at,
|
|
))
|
|
conn.commit()
|
|
conn.close()
|
|
logger.info(f"[auto_analysis] upserted event #{event_id}, tmpl #{template_id}, actual_moves={actual_moves}")
|
|
return {
|
|
"ok": True,
|
|
"inputs": inputs,
|
|
"node_values": node_values,
|
|
"actual_moves": actual_moves,
|
|
"instruments": instruments,
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"[auto_analysis] event #{event.get('id')}: {e}", exc_info=True)
|
|
return {"ok": False, "error": str(e)}
|
|
|
|
|
|
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 auto_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()
|
|
_run_auto_analysis(event, tmpl_id)
|
|
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": "<nom court, ex: 'CPI US — Taux/USD'>",
|
|
"category": "macro_us",
|
|
"description": "<1 phrase décrivant le mécanisme>",
|
|
"input_node": {{"id": "<snake_case>", "label": "<label court>", "unit": "<unité>"}},
|
|
"intermediate_nodes": [{{"id": "<snake_case>", "label": "<label>", "coef_name": "<coef_snake_case>", "coef_value": 0.5, "coef_desc": "<description>"}}],
|
|
"output_nodes": [{{"id": "<instr_lower>_pip", "label": "<INSTRUMENT>", "instrument": "<EURUSD|XAUUSD|SP500|BRENT|US10Y>", "source_intermediate": "<id>", "coef_name": "<coef_snake_case>", "coef_value": 100, "coef_desc": "<description>"}}]
|
|
}}
|
|
Règles : coef intermédiaire ∈ [-5,5] ; coef output en pips, |val| ∈ [30,200]."""
|
|
|
|
resp = client.chat.completions.create(
|
|
model="gpt-4o",
|
|
messages=[{"role": "user", "content": prompt}],
|
|
response_format={"type": "json_object"},
|
|
temperature=0.3,
|
|
max_tokens=900,
|
|
)
|
|
spec = json.loads(resp.choices[0].message.content or "{}")
|
|
|
|
graph_json = _build_graph_json_from_spec(spec)
|
|
category = spec.get("category", "macro_us")
|
|
instruments = graph_json.get("instruments", ["EURUSD"])
|
|
|
|
conn2 = get_conn()
|
|
cur = conn2.execute("""
|
|
INSERT INTO causal_graph_templates
|
|
(name, category, sub_type, description, instruments, graph_json, heuristic_ver)
|
|
VALUES (?, ?, ?, ?, ?, ?, 1)
|
|
""", (
|
|
spec.get("name", event.get("name", "Template IA")),
|
|
category,
|
|
event.get("sub_type", ""),
|
|
spec.get("description", ""),
|
|
json.dumps(instruments),
|
|
json.dumps(graph_json),
|
|
))
|
|
new_id = cur.lastrowid
|
|
conn2.execute("UPDATE market_events SET auto_template_id = ? WHERE id = ?", (new_id, event_id))
|
|
conn2.commit()
|
|
conn2.close()
|
|
|
|
logger.info(f"[auto_template] Created template #{new_id} '{spec.get('name')}' for event #{event_id}")
|
|
_run_auto_analysis(event, new_id)
|
|
return {"template_id": new_id, "action": "created", "name": spec.get("name", "Template IA"), "confidence": confidence}
|
|
|
|
except Exception as e:
|
|
logger.error(f"[auto_template] event #{event_id}: {e}")
|
|
return {"error": str(e)}
|
|
|
|
|
|
@router.post("/api/causal-lab/auto-analyze/refresh")
|
|
def refresh_auto_analyses():
|
|
"""
|
|
Re-run _run_auto_analysis for all causal_event_analyses rows with empty actual_json.
|
|
Returns detailed per-event debug info.
|
|
"""
|
|
try:
|
|
from services.database import get_conn
|
|
conn = get_conn()
|
|
|
|
# Also return a DB state summary for all analyses (not just empty ones)
|
|
all_rows = conn.execute("""
|
|
SELECT a.id AS cea_id, a.market_event_id, a.template_id,
|
|
a.actual_json, a.prediction_json,
|
|
e.name AS event_name, e.start_date
|
|
FROM causal_event_analyses a
|
|
LEFT JOIN market_events e ON e.id = a.market_event_id
|
|
ORDER BY a.id DESC
|
|
""").fetchall()
|
|
|
|
db_state = []
|
|
for r in all_rows:
|
|
actual = r["actual_json"] or ""
|
|
pred = r["prediction_json"] or ""
|
|
db_state.append({
|
|
"cea_id": r["cea_id"],
|
|
"event_id": r["market_event_id"],
|
|
"event_name": r["event_name"],
|
|
"start_date": r["start_date"],
|
|
"template_id": r["template_id"],
|
|
"actual_empty": actual in (None, "", "{}", "null"),
|
|
"pred_empty": pred in (None, "", "{}", "null"),
|
|
"actual_keys": list(json.loads(actual).keys()) if actual not in (None, "", "{}", "null") else [],
|
|
"pred_keys": list(json.loads(pred).keys()) if pred not in (None, "", "{}", "null") else [],
|
|
})
|
|
|
|
_empty = (None, "", "{}", "null", "[]")
|
|
rows = [r for r in all_rows if
|
|
(r["actual_json"] or "") in _empty or
|
|
(r["prediction_json"] or "") in _empty]
|
|
print(f"[refresh] {len(rows)} analyses with empty actual_json out of {len(all_rows)} total", flush=True)
|
|
conn.close()
|
|
|
|
details = []
|
|
refreshed, failed = 0, 0
|
|
for row in rows:
|
|
detail: dict = {"cea_id": row["cea_id"], "event_id": row["market_event_id"],
|
|
"event_name": row["event_name"], "template_id": row["template_id"]}
|
|
try:
|
|
conn2 = get_conn()
|
|
ev_row = conn2.execute(
|
|
"SELECT * FROM market_events WHERE id = ?", (row["market_event_id"],)
|
|
).fetchone()
|
|
conn2.close()
|
|
if not ev_row:
|
|
detail["result"] = "event_not_found"
|
|
details.append(detail)
|
|
continue
|
|
ev = dict(ev_row)
|
|
detail["surprise_pct"] = ev.get("surprise_pct")
|
|
detail["actual_value"] = ev.get("actual_value")
|
|
detail["expected_value"] = ev.get("expected_value")
|
|
detail["start_date"] = ev.get("start_date")
|
|
res = _run_auto_analysis(ev, row["template_id"])
|
|
detail["result"] = "ok" if res["ok"] else "failed"
|
|
detail["inputs"] = res.get("inputs", {})
|
|
detail["node_keys"] = list(res.get("node_values", {}).keys())
|
|
detail["actual_moves"] = res.get("actual_moves", {})
|
|
detail["run_error"] = res.get("error")
|
|
if res["ok"]:
|
|
refreshed += 1
|
|
else:
|
|
failed += 1
|
|
except Exception as _e:
|
|
logger.warning(f"[refresh_auto_analyses] cea#{row['cea_id']}: {_e}")
|
|
detail["result"] = f"exception: {_e}"
|
|
failed += 1
|
|
details.append(detail)
|
|
|
|
print(f"[refresh] done: {refreshed} refreshed, {failed} failed", flush=True)
|
|
return {
|
|
"refreshed": refreshed,
|
|
"failed": failed,
|
|
"total": len(rows),
|
|
"db_state": db_state,
|
|
"details": details,
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"[refresh_auto_analyses] {e}")
|
|
raise HTTPException(500, str(e))
|
|
|
|
|
|
@router.post("/api/causal-lab/create-from-event")
|
|
def create_template_from_event(body: CreateFromEventRequest):
|
|
"""GPT-4o génère et enregistre un template causal adapté à l'événement."""
|
|
try:
|
|
from services.database import get_conn, get_config
|
|
from services.causal_graphs import init_tables
|
|
import openai
|
|
|
|
key = get_config("openai_api_key") or ""
|
|
if not key:
|
|
raise HTTPException(400, "Clé OpenAI manquante dans la configuration")
|
|
|
|
conn = get_conn()
|
|
init_tables(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, "Événement introuvable")
|
|
event = dict(ev_row)
|
|
|
|
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": "<nom court, ex: 'CPI US — Taux/USD'>",
|
|
"category": "macro_us",
|
|
"description": "<1 phrase décrivant le mécanisme>",
|
|
"input_node": {{
|
|
"id": "<snake_case>",
|
|
"label": "<label court>",
|
|
"unit": "<unité>"
|
|
}},
|
|
"intermediate_nodes": [
|
|
{{
|
|
"id": "<snake_case>",
|
|
"label": "<label>",
|
|
"coef_name": "<coef_snake_case>",
|
|
"coef_value": <float [-5,5]>,
|
|
"coef_desc": "<description>"
|
|
}}
|
|
],
|
|
"output_nodes": [
|
|
{{
|
|
"id": "<instr_lower>_pip",
|
|
"label": "<INSTRUMENT>",
|
|
"instrument": "<EURUSD|XAUUSD|SP500|BRENT|US10Y>",
|
|
"source_intermediate": "<id_nœud_intermédiaire>",
|
|
"coef_name": "<coef_snake_case>",
|
|
"coef_value": <pips/unité, valeur absolue 30-200, signe selon impact>,
|
|
"coef_desc": "<description>"
|
|
}}
|
|
]
|
|
}}
|
|
|
|
Règles : coef intermédiaire ∈ [-5,5] ; coef output en pips, |val| ∈ [30,200].
|
|
Surprise CPI US supérieur → taux ↑ → USD ↑ → EURUSD ↓ (coef négatif), XAU ↓, SP500 ↓."""
|
|
|
|
client = openai.OpenAI(api_key=key)
|
|
resp = client.chat.completions.create(
|
|
model="gpt-4o",
|
|
messages=[{"role": "user", "content": prompt}],
|
|
response_format={"type": "json_object"},
|
|
temperature=0.3,
|
|
max_tokens=900,
|
|
)
|
|
spec = json.loads(resp.choices[0].message.content or "{}")
|
|
|
|
graph_json = _build_graph_json_from_spec(spec)
|
|
category = spec.get("category", "macro_us")
|
|
instruments = graph_json.get("instruments", ["EURUSD"])
|
|
|
|
cur = conn.execute("""
|
|
INSERT INTO causal_graph_templates
|
|
(name, category, sub_type, description, instruments, graph_json, heuristic_ver)
|
|
VALUES (?, ?, ?, ?, ?, ?, 1)
|
|
""", (
|
|
spec.get("name", event.get("name", "Template IA")),
|
|
category,
|
|
event.get("sub_type", ""),
|
|
spec.get("description", ""),
|
|
json.dumps(instruments),
|
|
json.dumps(graph_json),
|
|
))
|
|
new_id = cur.lastrowid
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
return {
|
|
"id": new_id,
|
|
"name": spec.get("name", "Template IA"),
|
|
"category": category,
|
|
"instruments": instruments,
|
|
"graph_json": graph_json,
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[causal_lab] create-from-event: {e}")
|
|
raise HTTPException(500, str(e))
|
|
|
|
|
|
@router.get("/api/causal-lab/analyses")
|
|
def list_analyses(
|
|
template_id: int = Query(0),
|
|
market_event_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 market_event_id:
|
|
conditions.append(f"a.market_event_id = {market_event_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, t.graph_json as template_graph_json
|
|
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] = {}
|
|
try:
|
|
d["graph_json"] = json.loads(d.get("template_graph_json") or "{}")
|
|
except Exception:
|
|
d["graph_json"] = {}
|
|
result.append(d)
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"[causal_lab] list_analyses: {e}")
|
|
raise HTTPException(500, str(e))
|
|
|
|
|
|
@router.delete("/api/causal-lab/analyses/{analysis_id}")
|
|
def delete_analysis(analysis_id: int):
|
|
try:
|
|
from services.database import get_conn
|
|
conn = get_conn()
|
|
_init(conn)
|
|
row = conn.execute("SELECT id FROM causal_event_analyses WHERE id=?", (analysis_id,)).fetchone()
|
|
if not row:
|
|
conn.close(); raise HTTPException(404, "Analyse introuvable")
|
|
conn.execute("DELETE FROM causal_event_analyses WHERE id=?", (analysis_id,))
|
|
conn.commit(); conn.close()
|
|
return {"ok": True}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[causal_lab] delete_analysis: {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))
|
|
|
|
|
|
@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))
|
|
|
|
|
|
# ── AI graph modifier & grammar compiler ─────────────────────────────────────
|
|
|
|
class AiModifyRequest(BaseModel):
|
|
prompt: str
|
|
current_graph: dict = {} # client can pass current in-memory state
|
|
|
|
|
|
GRAMMAR_SPEC = """\
|
|
Grammar for causal graph patch operations (one operation per line, # = comment):
|
|
|
|
NODE OPERATIONS
|
|
+node id=<id> label="<text>" type=<macro_event|observable|latent|market_asset> [x=<int>] [y=<int>] [instrument=<str>] [formula=<expr>] [unit=<str>]
|
|
~node id=<id> [label="<text>"] [type=...] [x=<int>] [y=<int>] [instrument=<str>]
|
|
-node id=<id>
|
|
|
|
EDGE OPERATIONS (sign: + positive - negative = neutral)
|
|
+edge from=<id> to=<id> [sign=+|-|=] [s=1|2|3] [style=solid|dashed] [label="<text>"] [lag=<int>] [decay=<float|null>]
|
|
~edge from=<id> to=<id> [sign=...] [s=...] [label="<text>"] [lag=<int>] [decay=<float|null>]
|
|
-edge from=<id> to=<id>
|
|
|
|
COEFFICIENT OPERATIONS
|
|
+coef key=<str> value=<float> [desc="<text>"]
|
|
~coef key=<str> [value=<float>] [desc="<text>"]
|
|
-coef key=<str>
|
|
|
|
INPUT MAPPING
|
|
~input node=<id> source=<surprise|actual_value|user_input|impact_score_scaled> [key=<str>]
|
|
-input node=<id>
|
|
|
|
GRAPH METADATA
|
|
~instruments EURUSD,SP500,...
|
|
|
|
Vocabulary:
|
|
- macro_event: source event node (NFP surprise, etc.)
|
|
- observable: directly measurable market variable (OIS rate, US 2Y yield, spread)
|
|
- latent: hidden variable (risk aversion, growth expectations, sentiment)
|
|
- market_asset: traded instrument (EUR/USD, S&P 500 — final output nodes)
|
|
- s=1 weak / s=2 medium (default) / s=3 strong causal link
|
|
- style=dashed for indirect/uncertain channels
|
|
- lag: trading days before effect onset; decay: half-life in days (null = permanent shift)
|
|
"""
|
|
|
|
|
|
class AiToGrammarRequest(BaseModel):
|
|
prompt: str
|
|
current_graph: dict = {}
|
|
|
|
|
|
@router.post("/api/causal-lab/template/{template_id}/ai-to-grammar")
|
|
def ai_to_grammar(template_id: int, body: AiToGrammarRequest):
|
|
"""
|
|
Convert a natural-language modification request into graph patch grammar.
|
|
Returns only the grammar lines — user reviews before applying.
|
|
"""
|
|
try:
|
|
from services.database import get_conn, get_config
|
|
from services.causal_graphs import get_template
|
|
import openai
|
|
|
|
key = get_config("openai_api_key") or ""
|
|
if not key:
|
|
raise HTTPException(400, "Clé OpenAI manquante dans la configuration")
|
|
|
|
conn = get_conn()
|
|
t = get_template(conn, template_id)
|
|
conn.close()
|
|
if not t:
|
|
raise HTTPException(404, f"Template {template_id} non trouvé")
|
|
|
|
gj = body.current_graph if body.current_graph.get("nodes") else t["graph_json"]
|
|
|
|
# Compact node/edge summary for context
|
|
node_summary = "\n".join(
|
|
f" {n['id']} ({n['type']}) — {n['label']}"
|
|
for n in gj.get("nodes", [])
|
|
)
|
|
edge_summary = "\n".join(
|
|
f" {e['from']} → {e['to']} sign={e.get('sign','?')} lag={e.get('lag_days','?')} [{e.get('label','')}]"
|
|
for e in gj.get("edges", [])
|
|
)
|
|
|
|
system_prompt = (
|
|
"You are a causal graph patch generator for the GeoOptions macro-finance platform.\n"
|
|
"Convert the user's natural language modification request into graph patch operations.\n"
|
|
"Return ONLY the grammar lines — no explanation, no markdown, no fences.\n\n"
|
|
+ GRAMMAR_SPEC
|
|
)
|
|
|
|
user_prompt = (
|
|
f'Current graph nodes:\n{node_summary}\n\n'
|
|
f'Current graph edges:\n{edge_summary}\n\n'
|
|
f'Modification request:\n{body.prompt}\n\n'
|
|
"Return ONLY the grammar patch lines, one per line."
|
|
)
|
|
|
|
client = openai.OpenAI(api_key=key)
|
|
resp = client.chat.completions.create(
|
|
model="gpt-4o",
|
|
messages=[
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_prompt},
|
|
],
|
|
temperature=0.15,
|
|
max_tokens=800,
|
|
)
|
|
grammar = (resp.choices[0].message.content or "").strip()
|
|
# Strip any accidental markdown fences
|
|
grammar = "\n".join(
|
|
l for l in grammar.splitlines()
|
|
if not l.strip().startswith("```")
|
|
)
|
|
return {"grammar": grammar}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[causal_lab] ai_to_grammar {template_id}: {e}")
|
|
raise HTTPException(500, str(e))
|
|
|
|
|
|
@router.post("/api/causal-lab/template/{template_id}/ai-modify")
|
|
def ai_modify_template(template_id: int, body: AiModifyRequest):
|
|
"""
|
|
GPT-4o modifie un graph_json selon un prompt utilisateur.
|
|
Retourne le graph_json modifié sans sauvegarder — l'utilisateur valide puis sauvegarde.
|
|
"""
|
|
try:
|
|
from services.database import get_conn, get_config
|
|
from services.causal_graphs import get_template
|
|
import openai
|
|
|
|
key = get_config("openai_api_key") or ""
|
|
if not key:
|
|
raise HTTPException(400, "Clé OpenAI manquante dans la configuration")
|
|
|
|
conn = get_conn()
|
|
t = get_template(conn, template_id)
|
|
conn.close()
|
|
if not t:
|
|
raise HTTPException(404, f"Template {template_id} non trouvé")
|
|
|
|
# Use the graph sent by client (latest in-memory edits) or fall back to DB
|
|
gj = body.current_graph if body.current_graph.get("nodes") else t["graph_json"]
|
|
|
|
system_prompt = (
|
|
"You are a causal graph editor for the GeoOptions macro-finance platform.\n"
|
|
"You receive a causal graph in JSON and a modification request in natural language.\n"
|
|
"Return ONLY the complete modified graph_json as a valid JSON object — no explanation, no markdown.\n\n"
|
|
"Graph format:\n"
|
|
"{\n"
|
|
' "nodes": [{"id": str, "label": str, "type": "macro_event|observable|latent|market_asset",\n'
|
|
' "x": int, "y": int, "formula": str_opt, "unit": str_opt, "instrument": str_opt}],\n'
|
|
' "edges": [{"from": str, "to": str, "style": "solid|dashed", "strength": 1|2|3,\n'
|
|
' "sign": "positive|negative|neutral", "label": str_opt,\n'
|
|
' "lag_days": int_opt, "decay_days": float_opt}],\n'
|
|
' "coefficients": {"key": {"value": float, "calibrated": null, "description": str}},\n'
|
|
' "instruments": [str],\n'
|
|
' "input_mapping": {"node_id": {"source": "user_input|surprise|actual_value|impact_score_scaled",\n'
|
|
' "key": str_opt, "field": str_opt}}\n'
|
|
"}\n\n"
|
|
"Rules:\n"
|
|
"- Preserve existing node x/y positions unless layout is clearly broken\n"
|
|
"- All edge from/to must reference valid node ids\n"
|
|
"- instruments list must include instrument field of all market_asset nodes\n"
|
|
"- Return the COMPLETE graph_json (not just the changed parts)"
|
|
)
|
|
|
|
user_prompt = (
|
|
f'Current graph — template "{t["name"]}" ({t["category"]}, sub_type: {t.get("sub_type","")}):\n'
|
|
f"{json.dumps(gj, ensure_ascii=False, indent=2)}\n\n"
|
|
f"Modification request:\n{body.prompt}\n\n"
|
|
"Return the complete modified graph_json as JSON only."
|
|
)
|
|
|
|
client = openai.OpenAI(api_key=key)
|
|
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.25,
|
|
max_tokens=3000,
|
|
)
|
|
raw = resp.choices[0].message.content or "{}"
|
|
modified = json.loads(raw)
|
|
|
|
if "nodes" not in modified or "edges" not in modified:
|
|
raise ValueError("L'IA n'a pas retourné un graph_json valide (nodes/edges manquants)")
|
|
|
|
return {"graph_json": modified, "template_name": t["name"]}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[causal_lab] ai_modify_template {template_id}: {e}")
|
|
raise HTTPException(500, str(e))
|