143 lines
5.1 KiB
Python
143 lines
5.1 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] = ""
|
|
|
|
|
|
@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}/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}/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.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()
|