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

feat: instrument model

Browse Source
This commit is contained in:
OpenSquared
2026-07-03 00:31:30 +02:00
parent a8ee808937
commit d9e9d30889
4 changed files with 591 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
backend/services/gauge_sync.py
View File

@@ -247,12 +247,21 @@ _SUGGEST_FN = {
# ── Public API ─────────────────────────────────────────────────────────────────
def get_latest_gauges(conn) -> tuple[dict, str]:
"""Retourne (gauges_dict, snapshot_date) depuis macro_regime_history."""
row = conn.execute(
"SELECT gauges_summary_json, timestamp FROM macro_regime_history "
"ORDER BY timestamp DESC LIMIT 1"
).fetchone()
def get_latest_gauges(conn, at_date: Optional[str] = None) -> tuple[dict, str]:
"""Retourne (gauges_dict, snapshot_date) depuis macro_regime_history.
Si at_date fourni, retourne le snapshot le plus proche <= at_date."""
if at_date:
row = conn.execute(
"SELECT gauges_summary_json, timestamp FROM macro_regime_history "
"WHERE timestamp <= ? ORDER BY timestamp DESC LIMIT 1",
(at_date,)
).fetchone()
else:
row = conn.execute(
"SELECT gauges_summary_json, timestamp FROM macro_regime_history "
"ORDER BY timestamp DESC LIMIT 1"
).fetchone()
if row:
r = dict(row)
gauges = json.loads(r.get("gauges_summary_json") or "{}")
@@ -261,10 +270,17 @@ def get_latest_gauges(conn) -> tuple[dict, str]:
return gauges, date
# Fallback : macro_gauge_snapshots
row2 = conn.execute(
"SELECT gauges_json, snapshot_date FROM macro_gauge_snapshots "
"ORDER BY snapshot_date DESC LIMIT 1"
).fetchone()
if at_date:
row2 = conn.execute(
"SELECT gauges_json, snapshot_date FROM macro_gauge_snapshots "
"WHERE snapshot_date <= ? ORDER BY snapshot_date DESC LIMIT 1",
(at_date,)
).fetchone()
else:
row2 = conn.execute(
"SELECT gauges_json, snapshot_date FROM macro_gauge_snapshots "
"ORDER BY snapshot_date DESC LIMIT 1"
).fetchone()
if row2:
return json.loads(row2["gauges_json"] or "{}"), str(row2["snapshot_date"])

116
backend/services/instrument_models.py
View File

@@ -638,6 +638,11 @@ def init_instrument_model_tables(conn):
set_at TEXT DEFAULT (datetime('now')),
UNIQUE(instrument, node_id)
);
CREATE TABLE IF NOT EXISTS event_calibration_overrides (
analysis_id INTEGER PRIMARY KEY,
calibration_json TEXT NOT NULL,
updated_at TEXT DEFAULT (datetime('now'))
);
""")
conn.commit()
@@ -669,12 +674,15 @@ def _compute_event_by_category(conn, instrument: str, ref_date: date_type) -> di
extended_from = ref_date - timedelta(days=365)
rows = conn.execute("""
SELECT a.prediction_json,
e.start_date, e.end_date, e.sub_type,
t.category, t.calibration_json
SELECT a.id as analysis_id,
a.prediction_json,
e.start_date, e.end_date, e.sub_type, e.name as title,
t.category,
COALESCE(o.calibration_json, t.calibration_json) as 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
LEFT JOIN event_calibration_overrides o ON o.analysis_id = a.id
WHERE a.instrument = ?
AND e.start_date >= ?
AND e.start_date <= ?
@@ -787,6 +795,98 @@ def _graph_json_for_eval(graph_def: dict, regime_weights: Optional[dict] = None)
# ── Public API ─────────────────────────────────────────────────────────────────
def get_active_event_details(conn, instrument: str, ref_date: date_type) -> dict:
"""
Détail des events actifs par catégorie — utilisé pour enrichir les nœuds event du DAG.
Retourne : {category: [{analysis_id, title, start_date, days_since,
pip_prediction, lifecycle_factor, remaining_pips, calibration}]}
"""
inst_upper = instrument.upper()
inst_lower = inst_upper.lower()
extended_from = ref_date - timedelta(days=365)
rows = conn.execute("""
SELECT a.id as analysis_id,
a.prediction_json,
e.id as event_id, e.name as title, e.start_date, e.end_date, e.sub_type,
t.category,
COALESCE(o.calibration_json, t.calibration_json) as 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
LEFT JOIN event_calibration_overrides o ON o.analysis_id = a.id
WHERE a.instrument = ?
AND e.start_date >= ?
AND e.start_date <= ?
ORDER BY e.start_date DESC
""", (inst_upper, str(extended_from), str(ref_date))).fetchall()
by_cat: dict[str, list] = {}
for row in rows:
r = dict(row)
try:
preds = json.loads(r["prediction_json"] or "{}")
calib = json.loads(r["calibration_json"] or "{}")
except Exception:
continue
pips: Optional[float] = None
if inst_lower in preds:
pips = float(preds[inst_lower])
else:
for k, v in preds.items():
if inst_lower in k.lower():
try: pips = float(v); break
except (TypeError, ValueError): pass
if pips is None:
continue
absorption = max(1, int(calib.get("absorption_days", 7)))
dtype = str(calib.get("decay_type", "exp"))
rise = max(0, int(calib.get("rise_days", 0)))
plateau = max(0, int(calib.get("plateau_days", 0)))
ev_end = r.get("end_date")
if ev_end and str(r.get("sub_type", "")).startswith("rate_guidance"):
try:
meeting = date_type.fromisoformat(ev_end[:10])
ev_start = date_type.fromisoformat(r["start_date"][:10])
absorption = max(1, (meeting - ev_start).days)
dtype = "linear"; rise = 0; plateau = 0
except ValueError:
pass
try:
ev_date = date_type.fromisoformat(r["start_date"][:10])
except ValueError:
continue
days = (ref_date - ev_date).days
lf = _lifecycle(days, rise, plateau, absorption, dtype)
if lf < 0.01:
continue
cat = r["category"]
by_cat.setdefault(cat, []).append({
"analysis_id": r["analysis_id"],
"event_id": r.get("event_id"),
"title": r.get("title", ""),
"start_date": r["start_date"][:10],
"days_since": days,
"pip_prediction": round(pips, 1),
"lifecycle_factor": round(lf, 3),
"remaining_pips": round(pips * lf, 1),
"calibration": {
"absorption_days": absorption,
"rise_days": rise,
"plateau_days": plateau,
"decay_type": dtype,
},
})
return by_cat
def get_model_state(conn, instrument: str, at_date: Optional[str] = None) -> Optional[dict]:
"""
Full model state : DAG evaluation avec saturation (Phase 2) + poids de régime.
@@ -889,6 +989,8 @@ def get_model_state(conn, instrument: str, at_date: Optional[str] = None) -> Opt
direction = "bullish" if net_pips > 5 else "bearish" if net_pips < -5 else "neutral"
event_details = get_active_event_details(conn, inst_upper, ref_date)
return {
"instrument": inst_upper,
"name": graph_def["name"],
@@ -906,6 +1008,7 @@ def get_model_state(conn, instrument: str, at_date: Optional[str] = None) -> Opt
"nodes": nodes_out,
"output_node": output_id,
"regime": regime_info,
"event_details": event_details,
}
@@ -944,11 +1047,14 @@ def simulate_timeline(
inst_lower = inst_upper.lower()
extended = date_from - timedelta(days=365)
rows = conn.execute("""
SELECT a.prediction_json, e.start_date, e.end_date, e.sub_type,
t.category, t.calibration_json
SELECT a.id as analysis_id,
a.prediction_json, e.start_date, e.end_date, e.sub_type,
t.category,
COALESCE(o.calibration_json, t.calibration_json) as 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
LEFT JOIN event_calibration_overrides o ON o.analysis_id = a.id
WHERE a.instrument=? AND e.start_date>=? AND e.start_date<=?
""", (inst_upper, str(extended), str(today))).fetchall()