475 lines
17 KiB
Python
475 lines
17 KiB
Python
"""
|
|
Instrument Models Router — graphes causaux exhaustifs par instrument.
|
|
"""
|
|
from typing import Any, Dict, List, Optional
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from pydantic import BaseModel
|
|
|
|
router = APIRouter(prefix="/api/instrument-models", tags=["instrument-models"])
|
|
|
|
|
|
class OverrideBody(BaseModel):
|
|
value: float
|
|
note: Optional[str] = ""
|
|
|
|
|
|
class BulkOverrideItem(BaseModel):
|
|
node_id: str
|
|
value: float
|
|
note: Optional[str] = ""
|
|
|
|
|
|
class BulkOverrideBody(BaseModel):
|
|
overrides: List[BulkOverrideItem]
|
|
|
|
|
|
class VirtualEvent(BaseModel):
|
|
date: str
|
|
category: str = "unclassified"
|
|
pips: float = 0.0
|
|
label: str = "Event virtuel"
|
|
absorption_days: int = 14
|
|
rise_days: int = 1
|
|
plateau_days: int = 0
|
|
decay_type: str = "exp"
|
|
|
|
|
|
class WhatIfBody(BaseModel):
|
|
period: str = "1y"
|
|
virtual_events: List[VirtualEvent] = []
|
|
|
|
|
|
class CalibrateBody(BaseModel):
|
|
ref_date: Optional[str] = None
|
|
|
|
|
|
class CalibrationOverrideBody(BaseModel):
|
|
absorption_days: int = 7
|
|
rise_days: int = 0
|
|
plateau_days: int = 0
|
|
decay_type: str = "exp"
|
|
|
|
|
|
@router.get("", response_model=List[Dict[str, Any]])
|
|
def list_instrument_models():
|
|
from services.database import get_conn
|
|
from services.instrument_models import INSTRUMENT_MODELS
|
|
conn = get_conn()
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT instrument, updated_at FROM instrument_models ORDER BY instrument"
|
|
).fetchall()
|
|
result = []
|
|
for r in rows:
|
|
inst = r["instrument"]
|
|
meta = INSTRUMENT_MODELS.get(inst, {})
|
|
counts: dict[str, int] = {}
|
|
for n in meta.get("nodes", []):
|
|
t = n.get("node_type", "unknown")
|
|
counts[t] = counts.get(t, 0) + 1
|
|
result.append({
|
|
"instrument": inst,
|
|
"name": meta.get("name", inst),
|
|
"description": meta.get("description", ""),
|
|
"n_input_event": counts.get("input_event", 0),
|
|
"n_input_manual": counts.get("input_manual", 0),
|
|
"n_intermediate": counts.get("intermediate", 0),
|
|
"updated_at": r["updated_at"],
|
|
})
|
|
return result
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.get("/{instrument}/gauge-suggestions")
|
|
def get_gauge_suggestions(
|
|
instrument: str,
|
|
at_date: Optional[str] = Query(None, description="YYYY-MM-DD — snapshot le plus proche de cette date"),
|
|
) -> Dict[str, Any]:
|
|
"""Suggestions de valeurs depuis les gauges de marché (optionnellement à une date historique)."""
|
|
from services.database import get_conn
|
|
from services.gauge_sync import suggest_from_gauges, get_latest_gauges
|
|
from services.instrument_models import INSTRUMENT_MODELS
|
|
conn = get_conn()
|
|
try:
|
|
inst = instrument.upper()
|
|
if inst not in INSTRUMENT_MODELS:
|
|
raise HTTPException(status_code=404, detail=f"Instrument {inst} non supporté")
|
|
gauges, snap_date = get_latest_gauges(conn, at_date=at_date)
|
|
if not gauges:
|
|
raise HTTPException(status_code=404, detail="Aucun snapshot de gauges disponible")
|
|
suggestions = suggest_from_gauges(inst, gauges)
|
|
return {
|
|
"instrument": inst,
|
|
"gauge_date": snap_date,
|
|
"requested_date": at_date,
|
|
"n_suggestions": len(suggestions),
|
|
"suggestions": suggestions,
|
|
"gauges_available": list(gauges.keys()),
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.post("/{instrument}/apply-suggestions")
|
|
def apply_gauge_suggestions(
|
|
instrument: str,
|
|
body: BulkOverrideBody,
|
|
) -> Dict[str, Any]:
|
|
"""Applique une liste d'overrides en bulk (depuis gauge sync ou saisie rapide)."""
|
|
from services.database import get_conn
|
|
from services.instrument_models import set_node_override
|
|
conn = get_conn()
|
|
try:
|
|
inst = instrument.upper()
|
|
saved = []
|
|
for item in body.overrides:
|
|
set_node_override(conn, inst, item.node_id, item.value, item.note or "")
|
|
saved.append(item.node_id)
|
|
return {"ok": True, "instrument": inst, "saved": saved, "count": len(saved)}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.delete("/{instrument}/overrides")
|
|
def clear_all_overrides(instrument: str) -> Dict[str, Any]:
|
|
"""Supprime TOUTES les overrides d'un instrument (reset à zéro)."""
|
|
from services.database import get_conn
|
|
conn = get_conn()
|
|
try:
|
|
inst = instrument.upper()
|
|
conn.execute("DELETE FROM instrument_node_overrides WHERE instrument=?", (inst,))
|
|
conn.commit()
|
|
return {"ok": True, "instrument": inst}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.get("/{instrument}/regime")
|
|
def get_instrument_regime(
|
|
instrument: str,
|
|
at_date: Optional[str] = Query(None),
|
|
) -> Dict[str, Any]:
|
|
"""Régime de marché courant pour cet instrument (détecté depuis events actifs)."""
|
|
from services.database import get_conn
|
|
from services.instrument_models import _compute_event_by_category, detect_regime
|
|
from datetime import datetime, date as date_type
|
|
conn = get_conn()
|
|
try:
|
|
try:
|
|
ref_date = date_type.fromisoformat(at_date) if at_date else datetime.utcnow().date()
|
|
except ValueError:
|
|
ref_date = datetime.utcnow().date()
|
|
ev_by_cat = _compute_event_by_category(conn, instrument.upper(), ref_date)
|
|
return detect_regime(ev_by_cat)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.get("/{instrument}/price-history")
|
|
def get_price_history(
|
|
instrument: str,
|
|
period: str = Query("1y", description="5d|1mo|3mo|6mo|1y|2y"),
|
|
refresh: bool = Query(False),
|
|
) -> Dict[str, Any]:
|
|
"""Cours historiques réels depuis yfinance (cache SQLite 6h)."""
|
|
from services.database import get_conn
|
|
from services.price_history import get_price_history as fetch_prices
|
|
from services.instrument_models import INSTRUMENT_MODELS
|
|
conn = get_conn()
|
|
try:
|
|
inst = instrument.upper()
|
|
prices = fetch_prices(conn, inst, period, force_refresh=refresh)
|
|
meta = INSTRUMENT_MODELS.get(inst, {})
|
|
return {
|
|
"instrument": inst,
|
|
"ticker": meta.get("yf_ticker", ""),
|
|
"period": period,
|
|
"n_points": len(prices),
|
|
"prices": prices,
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.get("/{instrument}/timeline")
|
|
def get_instrument_timeline(
|
|
instrument: str,
|
|
period: str = Query("1y", description="5d|1mo|3mo|6mo|1y|2y"),
|
|
) -> List[Dict[str, Any]]:
|
|
"""Simulation jour par jour de tous les nœuds du modèle sur la période."""
|
|
from services.database import get_conn
|
|
from services.instrument_models import simulate_timeline
|
|
conn = get_conn()
|
|
try:
|
|
data = simulate_timeline(conn, instrument.upper(), period)
|
|
if not data:
|
|
raise HTTPException(status_code=404, detail=f"Modèle introuvable pour {instrument.upper()}")
|
|
return data
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.post("/{instrument}/timeline-whatif")
|
|
def timeline_whatif(
|
|
instrument: str,
|
|
body: WhatIfBody,
|
|
) -> List[Dict[str, Any]]:
|
|
"""Simulation what-if avec events virtuels injectés dans la timeline."""
|
|
from services.database import get_conn
|
|
from services.instrument_models import simulate_timeline
|
|
conn = get_conn()
|
|
try:
|
|
ve_list = [ve.dict() for ve in body.virtual_events]
|
|
data = simulate_timeline(conn, instrument.upper(), body.period, virtual_events=ve_list)
|
|
if not data:
|
|
raise HTTPException(status_code=404, detail=f"Modèle introuvable pour {instrument.upper()}")
|
|
return data
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
_INSTRUMENT_CURRENCIES: Dict[str, List[str]] = {
|
|
"EURUSD": ["EUR", "USD"],
|
|
"USDJPY": ["USD", "JPY"],
|
|
"XAUUSD": ["USD"],
|
|
"SP500": ["USD"],
|
|
"TLT": ["USD"],
|
|
"GBPUSD": ["GBP", "USD"],
|
|
"EEM": ["USD", "CNY"],
|
|
"QQQ": ["USD"],
|
|
}
|
|
|
|
|
|
def _ff_event_to_category(event_name: str) -> str:
|
|
n = event_name.lower()
|
|
if any(k in n for k in ["fomc", "fed ", "ecb", "boe", "boj", "rba", "interest rate", "rate decision", "monetary policy", "central bank"]):
|
|
return "central_bank"
|
|
if any(k in n for k in ["cpi", "pce", "ppi", "inflation", "core price"]):
|
|
return "monetary_shock"
|
|
if any(k in n for k in ["nfp", "non-farm", "payroll", "employment change", "unemployment"]):
|
|
return "monetary_shock"
|
|
if any(k in n for k in ["gdp", "growth", "retail sales", "manufacturing", "pmi", "ism"]):
|
|
return "growth_shock"
|
|
if any(k in n for k in ["oil", "opec", "crude", "energy", "natural gas"]):
|
|
return "commodity"
|
|
if any(k in n for k in ["tarif", "trade war", "sanction", "geopolit"]):
|
|
return "geopolitical"
|
|
return "unclassified"
|
|
|
|
|
|
@router.get("/{instrument}/calendar-events")
|
|
def get_calendar_events(
|
|
instrument: str,
|
|
days_back: int = Query(90, description="Jours en arriere depuis at_date"),
|
|
days_forward: int = Query(180, description="Jours en avant depuis at_date"),
|
|
impacts: Optional[str] = Query("high,medium", description="Filtre impact FF (high,medium,low)"),
|
|
at_date: Optional[str] = Query(None),
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Events economiques du calendrier Forex Factory pour cet instrument.
|
|
Source : ff_calendar (Forex Factory). Decorrele des market_events / CausalLab.
|
|
Auto-sync live si ff_calendar vide pour la periode demandee.
|
|
"""
|
|
from services.database import get_conn
|
|
from datetime import datetime, date as date_type, timedelta
|
|
conn = get_conn()
|
|
try:
|
|
inst_upper = instrument.upper()
|
|
try:
|
|
ref = date_type.fromisoformat(at_date) if at_date else datetime.utcnow().date()
|
|
except ValueError:
|
|
ref = datetime.utcnow().date()
|
|
|
|
date_from = str(ref - timedelta(days=days_back))
|
|
date_to = str(ref + timedelta(days=days_forward))
|
|
|
|
currencies = _INSTRUMENT_CURRENCIES.get(inst_upper, ["USD"])
|
|
impact_filter = [i.strip() for i in impacts.split(",")] if impacts else ["high", "medium"]
|
|
|
|
# Auto-sync live si ff_calendar vide pour cette periode
|
|
ph = ",".join("?" * len(currencies))
|
|
n_existing = conn.execute(
|
|
f"SELECT COUNT(*) FROM ff_calendar WHERE event_date>=? AND event_date<=? AND currency IN ({ph})",
|
|
[date_from, date_to, *currencies]
|
|
).fetchone()[0]
|
|
|
|
if n_existing == 0:
|
|
try:
|
|
from services.ff_calendar import sync_live
|
|
sync_live()
|
|
except Exception:
|
|
pass
|
|
|
|
pi = ",".join("?" * len(impact_filter))
|
|
rows = conn.execute(
|
|
f"""SELECT event_date, event_time, currency, impact, event_name,
|
|
actual_value, forecast_value, previous_value
|
|
FROM ff_calendar
|
|
WHERE event_date >= ? AND event_date <= ?
|
|
AND currency IN ({ph})
|
|
AND impact IN ({pi})
|
|
ORDER BY event_date ASC, event_time ASC""",
|
|
[date_from, date_to, *currencies, *impact_filter]
|
|
).fetchall()
|
|
|
|
result = []
|
|
for row in rows:
|
|
r = dict(row)
|
|
ev_date = r["event_date"]
|
|
is_future = ev_date > str(ref)
|
|
category = _ff_event_to_category(r["event_name"])
|
|
result.append({
|
|
"date": ev_date,
|
|
"event_time": r.get("event_time", ""),
|
|
"currency": r["currency"],
|
|
"impact": r["impact"],
|
|
"event_name": r["event_name"],
|
|
"label": f"[{r['currency']}] {r['event_name']}",
|
|
"actual_value": r.get("actual_value"),
|
|
"forecast_value": r.get("forecast_value"),
|
|
"previous_value": r.get("previous_value"),
|
|
"category": category,
|
|
"is_future": is_future,
|
|
})
|
|
return result
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.put("/{instrument}/events/{analysis_id}/calibration")
|
|
def update_event_calibration(
|
|
instrument: str,
|
|
analysis_id: int,
|
|
body: CalibrationOverrideBody,
|
|
) -> Dict[str, Any]:
|
|
"""Écrase les paramètres de lifecycle d'une analyse event (override par-event)."""
|
|
from services.database import get_conn
|
|
import json as _json
|
|
conn = get_conn()
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT id FROM causal_event_analyses WHERE id=? AND instrument=?",
|
|
(analysis_id, instrument.upper())
|
|
).fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Analysis not found")
|
|
calib = {
|
|
"absorption_days": body.absorption_days,
|
|
"rise_days": body.rise_days,
|
|
"plateau_days": body.plateau_days,
|
|
"decay_type": body.decay_type,
|
|
}
|
|
conn.execute("""
|
|
INSERT INTO event_calibration_overrides (analysis_id, calibration_json)
|
|
VALUES (?, ?)
|
|
ON CONFLICT(analysis_id) DO UPDATE SET calibration_json=excluded.calibration_json, updated_at=datetime('now')
|
|
""", (analysis_id, _json.dumps(calib)))
|
|
conn.commit()
|
|
return {"ok": True, "analysis_id": analysis_id, "calibration": calib}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.delete("/{instrument}/events/{analysis_id}/calibration")
|
|
def reset_event_calibration(instrument: str, analysis_id: int) -> Dict[str, Any]:
|
|
"""Supprime l'override de calibration — retour aux paramètres du template."""
|
|
from services.database import get_conn
|
|
conn = get_conn()
|
|
try:
|
|
conn.execute("DELETE FROM event_calibration_overrides WHERE analysis_id=?", (analysis_id,))
|
|
conn.commit()
|
|
return {"ok": True, "analysis_id": analysis_id}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.post("/{instrument}/calibrate")
|
|
def calibrate_intercept(
|
|
instrument: str,
|
|
body: CalibrateBody,
|
|
) -> Dict[str, Any]:
|
|
"""Auto-calcule l'intercept depuis le cours réel à une date de référence."""
|
|
from services.database import get_conn
|
|
from services.instrument_models import get_model_state, INSTRUMENT_MODELS
|
|
from services.price_history import calibrate_intercept as do_calibrate, get_price_history
|
|
conn = get_conn()
|
|
try:
|
|
inst = instrument.upper()
|
|
state = get_model_state(conn, inst, body.ref_date)
|
|
if not state:
|
|
raise HTTPException(status_code=404, detail=f"Modèle introuvable")
|
|
# Make sure prices are cached
|
|
get_price_history(conn, inst, "3mo")
|
|
intercept = do_calibrate(conn, inst, state["structural_pips"], body.ref_date)
|
|
meta = INSTRUMENT_MODELS.get(inst, {})
|
|
return {
|
|
"instrument": inst,
|
|
"ref_date": body.ref_date,
|
|
"structural_pips": state["structural_pips"],
|
|
"pip_to_price": meta.get("pip_to_price", 0.0001),
|
|
"calibrated_intercept": intercept,
|
|
"current_intercept": meta.get("price_intercept", 0.0),
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.get("/{instrument}")
|
|
def get_instrument_model(
|
|
instrument: str,
|
|
at_date: Optional[str] = Query(None, description="YYYY-MM-DD (défaut: aujourd'hui)"),
|
|
) -> Dict[str, Any]:
|
|
"""Graphe complet avec valeurs courantes des nœuds (toutes couches)."""
|
|
from services.database import get_conn
|
|
from services.instrument_models import get_model_state
|
|
conn = get_conn()
|
|
try:
|
|
state = get_model_state(conn, instrument.upper(), at_date)
|
|
if not state:
|
|
raise HTTPException(status_code=404, detail=f"Modèle introuvable pour {instrument.upper()}")
|
|
return state
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.put("/{instrument}/nodes/{node_id}/override")
|
|
def set_override(instrument: str, node_id: str, body: OverrideBody) -> Dict[str, Any]:
|
|
"""Définit ou met à jour la valeur manuelle d'un nœud."""
|
|
from services.database import get_conn
|
|
from services.instrument_models import set_node_override
|
|
conn = get_conn()
|
|
try:
|
|
set_node_override(conn, instrument.upper(), node_id, body.value, body.note or "")
|
|
return {"ok": True, "instrument": instrument.upper(), "node_id": node_id, "value": body.value}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.delete("/{instrument}/nodes/{node_id}/override")
|
|
def clear_override(instrument: str, node_id: str) -> Dict[str, Any]:
|
|
"""Supprime l'override manuel d'un nœud (retour neutre/events)."""
|
|
from services.database import get_conn
|
|
from services.instrument_models import clear_node_override
|
|
conn = get_conn()
|
|
try:
|
|
clear_node_override(conn, instrument.upper(), node_id)
|
|
return {"ok": True, "instrument": instrument.upper(), "node_id": node_id}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.get("/{instrument}/nodes/overrides")
|
|
def get_all_overrides(instrument: str) -> List[Dict[str, Any]]:
|
|
"""Toutes les overrides manuelles pour un instrument."""
|
|
from services.database import get_conn
|
|
conn = get_conn()
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT node_id, value, note, set_at FROM instrument_node_overrides WHERE instrument=? ORDER BY set_at DESC",
|
|
(instrument.upper(),)
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
finally:
|
|
conn.close()
|