230 lines
8.0 KiB
Python
230 lines
8.0 KiB
Python
"""
|
|
Instrument Dashboard Router.
|
|
Exposes per-instrument snapshot (price, indicators, regime, trend, events) and AI narrative.
|
|
"""
|
|
import json
|
|
import math
|
|
from datetime import datetime, timedelta
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from pydantic import BaseModel
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
from services.instrument_service import (
|
|
get_all_instruments,
|
|
get_instrument,
|
|
get_snapshot,
|
|
get_narrative,
|
|
update_instrument_drivers,
|
|
)
|
|
|
|
class DriverUpdate(BaseModel):
|
|
drivers: List[Dict[str, Any]]
|
|
|
|
router = APIRouter(prefix="/api/instruments", tags=["instruments"])
|
|
|
|
|
|
@router.get("", response_model=List[Dict[str, Any]])
|
|
def list_instruments() -> List[Dict[str, Any]]:
|
|
"""
|
|
Return all instrument configurations (no price data).
|
|
"""
|
|
return get_all_instruments()
|
|
|
|
|
|
@router.get("/{instrument_id}/snapshot")
|
|
async def instrument_snapshot(
|
|
instrument_id: str,
|
|
period: str = Query(default="1y", description="yfinance period string (e.g. 1y, 6mo, 3mo)"),
|
|
interval: str = Query(default="1d", description="yfinance interval string (e.g. 1d, 1wk)"),
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Full snapshot for a single instrument: price data, indicators, regime, trend, events.
|
|
"""
|
|
config = get_instrument(instrument_id)
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail=f"Instrument '{instrument_id}' not found")
|
|
|
|
snapshot = await get_snapshot(instrument_id, period=period, interval=interval)
|
|
|
|
if "error" in snapshot:
|
|
raise HTTPException(status_code=500, detail=snapshot["error"])
|
|
|
|
return snapshot
|
|
|
|
|
|
@router.post("/{instrument_id}/narrative")
|
|
async def generate_narrative(
|
|
instrument_id: str,
|
|
force: bool = Query(default=False, description="Force regeneration, bypass cache"),
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Generate (or return cached) a French AI narrative for the instrument.
|
|
"""
|
|
config = get_instrument(instrument_id)
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail=f"Instrument '{instrument_id}' not found")
|
|
|
|
narrative = await get_narrative(instrument_id, force=force)
|
|
|
|
return {
|
|
"instrument_id": instrument_id.upper(),
|
|
"instrument_name": config.get("name", instrument_id),
|
|
"narrative": narrative,
|
|
}
|
|
|
|
|
|
@router.put("/{instrument_id}/drivers")
|
|
def update_drivers(instrument_id: str, body: DriverUpdate) -> Dict[str, Any]:
|
|
"""
|
|
Persist updated drivers (label, weight, keywords) for an instrument to instruments.json.
|
|
"""
|
|
config = get_instrument(instrument_id)
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail=f"Instrument '{instrument_id}' not found")
|
|
|
|
try:
|
|
update_instrument_drivers(instrument_id, body.drivers)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
return {"ok": True, "instrument_id": instrument_id.upper(), "drivers_count": len(body.drivers)}
|
|
|
|
|
|
# ── Instrument mult (pips → price conversion) ─────────────────────────────────
|
|
|
|
_INST_MULT: Dict[str, int] = {"EURUSD": 10000, "GBPUSD": 10000, "USDJPY": 100, "AUDUSD": 10000}
|
|
|
|
def _get_mult(inst: str) -> int:
|
|
return _INST_MULT.get(inst.upper(), 10)
|
|
|
|
|
|
def _decay(days_after: int, absorption_days: int, decay_type: str) -> float:
|
|
"""Decay factor ∈ [0,1] for a given number of days after the event."""
|
|
if days_after < 0:
|
|
return 0.0
|
|
if decay_type == "step":
|
|
return 1.0 if days_after <= absorption_days else 0.0
|
|
elif decay_type == "linear":
|
|
return max(0.0, 1.0 - days_after / max(absorption_days, 1))
|
|
else: # exp — 3 time-constants reach ~5% at absorption_days
|
|
lam = 3.0 / max(absorption_days, 1)
|
|
return math.exp(-lam * days_after)
|
|
|
|
|
|
@router.get("/{instrument_id}/theoretical-curve")
|
|
def get_theoretical_curve(
|
|
instrument_id: str,
|
|
period: str = Query("1y"),
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Courbe théorique composite : pour chaque jour calendaire de la période,
|
|
somme des impacts décroissants des analyses causales stockées.
|
|
|
|
Retourne [{date, cumulative_pips, contributions: [{template_name, event_name, event_date, pips, decay_factor}]}]
|
|
"""
|
|
from services.database import get_conn
|
|
|
|
period_lookback: Dict[str, int] = {
|
|
"5d": 7, "1mo": 35, "3mo": 95, "6mo": 190,
|
|
"1y": 370, "2y": 740, "5y": 1830,
|
|
}
|
|
lookback = period_lookback.get(period, 370)
|
|
|
|
date_to = datetime.utcnow().date()
|
|
date_from = date_to - timedelta(days=lookback)
|
|
# Fetch events that started before date_from too — they may still be decaying into the window
|
|
extended_from = date_from - timedelta(days=90)
|
|
|
|
inst_upper = instrument_id.upper()
|
|
|
|
conn = get_conn()
|
|
try:
|
|
rows = conn.execute("""
|
|
SELECT a.id,
|
|
a.prediction_json,
|
|
a.activation_score,
|
|
e.start_date AS event_date,
|
|
e.name AS event_name,
|
|
t.name AS template_name,
|
|
t.calibration_json
|
|
FROM causal_event_analyses a
|
|
JOIN market_events e ON e.id = a.market_event_id
|
|
JOIN causal_graph_templates t ON t.id = a.template_id
|
|
WHERE a.instrument = ?
|
|
AND e.start_date >= ?
|
|
AND e.start_date <= ?
|
|
ORDER BY e.start_date
|
|
""", (inst_upper, str(extended_from), str(date_to))).fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
# ── Build calendar-day series ─────────────────────────────────────────────
|
|
all_dates: List[str] = []
|
|
cur = date_from
|
|
while cur <= date_to:
|
|
all_dates.append(str(cur))
|
|
cur += timedelta(days=1)
|
|
|
|
curve: Dict[str, Dict] = {
|
|
d: {"cumulative_pips": 0.0, "contributions": []} for d in all_dates
|
|
}
|
|
|
|
for row in rows:
|
|
r = dict(row)
|
|
try:
|
|
predictions = json.loads(r["prediction_json"] or "{}")
|
|
calib = json.loads(r["calibration_json"] or "{}")
|
|
except Exception:
|
|
continue
|
|
|
|
# Extract predicted pips for this instrument from node_values dict
|
|
inst_lower = inst_upper.lower()
|
|
predicted_pips: Optional[float] = None
|
|
if inst_lower in predictions:
|
|
predicted_pips = float(predictions[inst_lower])
|
|
else:
|
|
for k, v in predictions.items():
|
|
if inst_lower in k.lower():
|
|
try:
|
|
predicted_pips = float(v)
|
|
break
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
if predicted_pips is None or predicted_pips == 0:
|
|
continue
|
|
|
|
absorption_days: int = max(1, int(calib.get("absorption_days", 7)))
|
|
dtype: str = str(calib.get("decay_type", "exp"))
|
|
event_date_str: str = r["event_date"][:10]
|
|
|
|
try:
|
|
event_date = datetime.strptime(event_date_str, "%Y-%m-%d").date()
|
|
except ValueError:
|
|
continue
|
|
|
|
for d in all_dates:
|
|
cal_date = datetime.strptime(d, "%Y-%m-%d").date()
|
|
days_after = (cal_date - event_date).days
|
|
df = _decay(days_after, absorption_days, dtype)
|
|
if df < 0.01:
|
|
continue
|
|
contribution = round(predicted_pips * df, 2)
|
|
curve[d]["cumulative_pips"] += contribution
|
|
curve[d]["contributions"].append({
|
|
"template_name": r["template_name"],
|
|
"event_name": r["event_name"],
|
|
"event_date": event_date_str,
|
|
"pips": contribution,
|
|
"decay_factor": round(df, 3),
|
|
})
|
|
|
|
# Round totals and strip empty-contribution days at the edges
|
|
result = []
|
|
for d in all_dates:
|
|
entry = curve[d]
|
|
entry["cumulative_pips"] = round(entry["cumulative_pips"], 1)
|
|
result.append({"date": d, **entry})
|
|
|
|
return result
|