admin/OpenFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat:: causal lab

Browse Source
This commit is contained in:
OpenSquared
2026-06-28 15:38:19 +02:00
parent 863ba67610
commit 1e44557551
7 changed files with 321 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
backend/routers/causal_lab.py
View File

@@ -321,6 +321,11 @@ 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
@@ -635,6 +640,97 @@ def recommend_template(body: RecommendRequest):
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
- Utilise les plages "range" si fournies (ex: [-3, 3] pour z-score)
- Magnitude proportionnelle à l'ampleur de la surprise
- Si surprise_pct disponible, utilise-la pour calibrer (ex: -29.1% → nœud_surprise ≈ -2.5)
Retourne UNIQUEMENT un JSON : {{ "node_id": valeur_numerique, ... }}
Inclure uniquement les nœuds listés ci-dessus."""
client = _openai_client()
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):
"""

13
backend/routers/cycle_actions.py
View File

@@ -51,12 +51,13 @@ router = APIRouter(prefix="/api/actions", tags=["Cycle Actions"])
class CheckEventsRequest(BaseModel):
sources: Optional[List[str]] = None # ['news','eco','technical','reports']
news_impact_min: float = 0.55
news_lookback_hours: int = 48
eco_z_threshold: float = 1.5
eco_days: int = 7
technical_lookback_days: int = 7
report_days: int = 7
report_min_importance: int = 3
# Unified date window for news + eco (replaces eco_days + news_lookback_hours)
date_from: Optional[str] = None # ISO date, e.g. "2026-06-21"
date_to: Optional[str] = None # ISO date, e.g. "2026-06-28"
class ActionResult(BaseModel):
@@ -151,12 +152,12 @@ def check_market_events(req: CheckEventsRequest):
result = check_new_market_events(
sources=req.sources,
news_impact_min=req.news_impact_min,
news_lookback_hours=req.news_lookback_hours,
eco_z_threshold=req.eco_z_threshold,
eco_days=req.eco_days,
technical_lookback_days=req.technical_lookback_days,
report_days=req.report_days,
report_min_importance=req.report_min_importance,
date_from=req.date_from,
date_to=req.date_to,
)
except Exception as e:
_release("check-market-events")
@@ -204,12 +205,12 @@ def check_market_events_background(
check_new_market_events(
sources=req.sources,
news_impact_min=req.news_impact_min,
news_lookback_hours=req.news_lookback_hours,
eco_z_threshold=req.eco_z_threshold,
eco_days=req.eco_days,
technical_lookback_days=req.technical_lookback_days,
report_days=req.report_days,
report_min_importance=req.report_min_importance,
date_from=req.date_from,
date_to=req.date_to,
)
except Exception as e:
logger.error(f"[action/check-market-events/bg] {e}")

6
backend/services/instrument_service.py
View File

@@ -528,9 +528,14 @@ def _get_relevant_events(
"""
try:
from services.database import get_conn
from services.causal_graphs import init_tables
conn = get_conn()
init_tables(conn) # ensure causal_event_analyses exists
rows = conn.execute("""
SELECT e.*,
(SELECT GROUP_CONCAT(DISTINCT a.instrument)
FROM causal_event_analyses a
WHERE a.market_event_id = e.id) AS analyzed_instruments,
(SELECT a.template_id FROM causal_event_analyses a WHERE a.market_event_id = e.id ORDER BY a.id DESC LIMIT 1) AS template_id,
(SELECT a.id FROM causal_event_analyses a WHERE a.market_event_id = e.id ORDER BY a.id DESC LIMIT 1) AS analysis_id
FROM market_events e
@@ -581,6 +586,7 @@ def _get_relevant_events(
filtered.append({
"id": ev.get("id"),
"template_id": ev.get("template_id"),
"analyzed_instruments": ev.get("analyzed_instruments"), # comma-sep, e.g. "EURUSD,SP500"
"date": ev_start,
"end_date": ev.get("end_date") or None,
"title": ev.get("name") or ev.get("event_name") or "",

109
backend/services/market_event_detector.py
View File

@@ -187,12 +187,13 @@ Réponds JSON: {{"is_duplicate": true/false, "reason": "courte phrase"}}"""
def _check_news(desk_cfg: Dict[str, Any]) -> List[Dict[str, Any]]:
from services.data_fetcher import fetch_geo_news
min_impact = float(desk_cfg.get("min_impact", 0.55))
lookback_hours = int(desk_cfg.get("lookback_hours", 48))
max_evaluate = int(desk_cfg.get("max_evaluate", 15))
dedup_enabled = bool(desk_cfg.get("dedup_enabled", True))
dedup_days = int(desk_cfg.get("dedup_lookback_days", 2))
system_prompt = desk_cfg.get("_system_prompt", "")
min_impact = float(desk_cfg.get("min_impact", 0.55))
max_evaluate = int(desk_cfg.get("max_evaluate", 15))
dedup_enabled = bool(desk_cfg.get("dedup_enabled", True))
dedup_days = int(desk_cfg.get("dedup_lookback_days", 2))
system_prompt = desk_cfg.get("_system_prompt", "")
date_from = desk_cfg.get("date_from")
date_to = desk_cfg.get("date_to")
api_key = _get_api_key()
if not api_key:
@@ -205,14 +206,22 @@ def _check_news(desk_cfg: Dict[str, Any]) -> List[Dict[str, Any]]:
logger.warning(f"[check_events/news] fetch failed: {e}")
return []
cutoff_dt = datetime.utcnow() - timedelta(hours=lookback_hours)
# Build cutoff from date_from; date_to used as upper bound
try:
cutoff_from = datetime.fromisoformat(date_from) if date_from else datetime.utcnow() - timedelta(days=7)
cutoff_to = datetime.fromisoformat(date_to) if date_to else datetime.utcnow()
except Exception:
cutoff_from = datetime.utcnow() - timedelta(days=7)
cutoff_to = datetime.utcnow()
candidates = []
for n in all_news:
if (n.get("impact_score") or 0) < min_impact:
continue
pub_date = _parse_date(n.get("date", ""))
try:
if datetime.fromisoformat(pub_date) < cutoff_dt:
pub_dt = datetime.fromisoformat(pub_date)
if pub_dt < cutoff_from or pub_dt > cutoff_to:
continue
except Exception:
pass
@@ -329,7 +338,8 @@ FORMAT JSON STRICT:
def _check_ff_calendar_surprises(
currencies: List[str],
min_impact: str,
days: int,
date_from: str,
date_to: str,
min_surprise_pct: float,
lookback_releases: int,
create_evt: bool,
@@ -347,7 +357,6 @@ def _check_ff_calendar_surprises(
"low": ("high", "medium", "low"),
}
allowed_impacts = impact_map.get(min_impact, ("high", "medium"))
cutoff = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%d")
try:
conn = get_conn()
@@ -360,11 +369,12 @@ def _check_ff_calendar_surprises(
WHERE currency IN ({ccy_ph})
AND impact IN ({imp_ph})
AND event_date >= ?
AND event_date <= ?
AND actual_value IS NOT NULL
AND forecast_value IS NOT NULL
ORDER BY event_date DESC
LIMIT 200""",
(*currencies, *allowed_impacts, cutoff),
(*currencies, *allowed_impacts, date_from, date_to),
).fetchall()
conn.close()
except Exception as e:
@@ -457,28 +467,45 @@ def _check_ff_calendar_surprises(
def _check_eco(desk_cfg: Dict[str, Any]) -> List[Dict[str, Any]]:
from services.database import get_recent_economic_surprises, get_conn
from services.database import get_conn
z_threshold = float(desk_cfg.get("z_threshold", 1.5))
days = int(desk_cfg.get("days", 7))
date_from = desk_cfg.get("date_from") or (datetime.utcnow() - timedelta(days=7)).strftime("%Y-%m-%d")
date_to = desk_cfg.get("date_to") or datetime.utcnow().strftime("%Y-%m-%d")
currencies = list(desk_cfg.get("currencies") or ["USD", "EUR", "GBP", "JPY"])
min_impact = str(desk_cfg.get("min_impact", "medium")).lower()
create_evt = bool(desk_cfg.get("create_market_event", True))
lookback_releases = int(desk_cfg.get("lookback_releases", 3))
min_rank = _IMPACT_RANKS.get(min_impact, 2)
# ff_calendar surprise threshold: z_threshold used as rough proxy (×10 → % equivalent)
ff_surprise_min = max(10.0, z_threshold * 10)
min_rank = _IMPACT_RANKS.get(min_impact, 2)
ff_surprise_min = max(10.0, z_threshold * 10)
existing = _existing_event_keys()
created: List[Dict] = []
# ── FRED path (USD only) ──────────────────────────────────────────────────
# ── Local economic_events table (replaces FRED API call) ─────────────────
if "USD" in currencies:
try:
releases = get_recent_economic_surprises(days=days, min_zscore=z_threshold)
conn = get_conn()
rows = conn.execute(
"""SELECT * FROM economic_events
WHERE event_date >= ? AND event_date <= ?
AND ABS(COALESCE(surprise_zscore, 0)) >= ?
ORDER BY event_date DESC, ABS(COALESCE(surprise_zscore, 0)) DESC
LIMIT 50""",
(date_from, date_to, z_threshold),
).fetchall()
conn.close()
releases = []
for r in rows:
d = dict(r)
try:
d["assets_impacted"] = json.loads(d.get("assets_impacted") or "[]")
except Exception:
d["assets_impacted"] = []
releases.append(d)
except Exception as e:
logger.warning(f"[check_events/eco] FRED query failed: {e}")
logger.warning(f"[check_events/eco] local query failed: {e}")
releases = []
for rel in releases:
@@ -572,7 +599,8 @@ def _check_eco(desk_cfg: Dict[str, Any]) -> List[Dict[str, Any]]:
created += _check_ff_calendar_surprises(
currencies=non_usd,
min_impact=min_impact,
days=days,
date_from=date_from,
date_to=date_to,
min_surprise_pct=ff_surprise_min,
lookback_releases=lookback_releases,
create_evt=create_evt,
@@ -860,13 +888,14 @@ def _check_fundamental(desk_cfg: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Corporate fundamental events: layoffs, M&A, earnings, credit, regulatory."""
from services.data_fetcher import fetch_geo_news
min_impact = float(desk_cfg.get("min_impact", 0.45))
lookback_hours = int(desk_cfg.get("lookback_hours", 72))
max_evaluate = int(desk_cfg.get("max_evaluate", 20))
dedup_enabled = bool(desk_cfg.get("dedup_enabled", True))
dedup_days = int(desk_cfg.get("dedup_lookback_days", 3))
system_prompt = desk_cfg.get("_system_prompt", "")
focus_types = desk_cfg.get("focus_types", ["layoffs","earnings","ma","credit","regulatory","guidance"])
min_impact = float(desk_cfg.get("min_impact", 0.45))
max_evaluate = int(desk_cfg.get("max_evaluate", 20))
dedup_enabled = bool(desk_cfg.get("dedup_enabled", True))
dedup_days = int(desk_cfg.get("dedup_lookback_days", 3))
system_prompt = desk_cfg.get("_system_prompt", "")
focus_types = desk_cfg.get("focus_types", ["layoffs","earnings","ma","credit","regulatory","guidance"])
date_from = desk_cfg.get("date_from") or (datetime.utcnow() - timedelta(days=7)).strftime("%Y-%m-%d")
date_to = desk_cfg.get("date_to") or datetime.utcnow().strftime("%Y-%m-%d")
api_key = _get_api_key()
if not api_key:
@@ -878,11 +907,10 @@ def _check_fundamental(desk_cfg: Dict[str, Any]) -> List[Dict[str, Any]]:
logger.warning(f"[check_events/fundamental] fetch failed: {e}")
return []
cutoff_dt = datetime.utcnow() - timedelta(hours=lookback_hours)
candidates = [
n for n in all_news
if (n.get("impact_score") or 0) >= min_impact
and _parse_date(n.get("date", "")) >= cutoff_dt.strftime("%Y-%m-%d")
and date_from <= _parse_date(n.get("date", "")) <= date_to
][:max_evaluate]
if not candidates:
@@ -1617,11 +1645,12 @@ def _check_macro_gauges(desk_cfg: Dict[str, Any]) -> List[Dict[str, Any]]:
def check_new_market_events(
sources: Optional[List[str]] = None,
# Unified date window — overrides eco_days/news_lookback_hours when provided
date_from: Optional[str] = None,
date_to: Optional[str] = None,
# Legacy overrides — used when called without an active desk
news_impact_min: float = 0.55,
news_lookback_hours: int = 48,
eco_z_threshold: float = 1.5,
eco_days: int = 7,
technical_lookback_days: int = 7,
report_days: int = 7,
report_min_importance: int = 3,
@@ -1655,17 +1684,27 @@ def check_new_market_events(
cfg["_system_prompt"] = desk.get("system_prompt") or ""
return cfg
# Compute default date_from fallback from desk config or 7-day default
_now = datetime.utcnow()
_date_to = date_to or _now.strftime("%Y-%m-%d")
_date_from = date_from or (_now - timedelta(days=7)).strftime("%Y-%m-%d")
news_cfg = _desk_cfg(news_desk, {
"min_impact": news_impact_min, "lookback_hours": news_lookback_hours,
"min_impact": news_impact_min,
"max_evaluate": 15, "dedup_enabled": False, "dedup_lookback_days": 2,
})
news_cfg["date_from"] = _date_from
news_cfg["date_to"] = _date_to
fundamental_cfg = _desk_cfg(fundamental_desk, {
"min_impact": 0.45, "lookback_hours": 72, "max_evaluate": 20,
"min_impact": 0.45, "max_evaluate": 20,
"dedup_enabled": True, "dedup_lookback_days": 3,
})
fundamental_cfg["date_from"] = _date_from
fundamental_cfg["date_to"] = _date_to
eco_cfg = _desk_cfg(eco_desk, {
"z_threshold": eco_z_threshold,
"days": eco_days,
"gauge_signals": {
"regime_transition": {"enabled": True},
"yield_curve_inversion": {"enabled": True, "threshold": 0.0},
@@ -1674,6 +1713,8 @@ def check_new_market_events(
"gold_copper_ratio": {"enabled": True, "fear_threshold": 700, "growth_threshold": 500},
},
})
eco_cfg["date_from"] = _date_from
eco_cfg["date_to"] = _date_to
tech_cfg = _desk_cfg(tech_desk, {
"lookback_days": technical_lookback_days,
"signals": {