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

feat: Market Events admin page — source_refs + per-instrument impacts

Browse Source 
- DB: colonne source_refs sur market_events (migration idempotente)
- save/update_market_event: persist source_refs (JSON array de {url,title,source})
- market_event_detector: stocke source_refs + évalue impacts instruments après chaque création
- Nouveau router /api/market-events: CRUD + evaluate + impacts CRUD
- Page MarketEvents.tsx: liste filtrée/triée + panneau détail (sources cliquables,
  tableau impacts par instrument avec score/direction/rationale inline-éditables,
  ajout manuel, bulk-evaluate)
- Sidebar: entrée Market Events (Radio icon)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OpenSquared
2026-06-25 18:29:09 +02:00
parent e66d3ddb46
commit d5da4737ef
7 changed files with 1283 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
backend/services/database.py
View File

@@ -942,6 +942,15 @@ def init_db():
except Exception:
pass
# Idempotent column additions
for _col_sql in [
"ALTER TABLE market_events ADD COLUMN source_refs TEXT DEFAULT '[]'",
]:
try:
c.execute(_col_sql)
except Exception:
pass
conn.commit()
conn.close()
@@ -4545,18 +4554,22 @@ def save_market_event(ev: Dict[str, Any]) -> int:
import json
conn = get_conn()
try:
src_refs = ev.get("source_refs", [])
if isinstance(src_refs, list):
src_refs = json.dumps(src_refs)
cur = conn.execute("""INSERT INTO market_events
(name, start_date, end_date, level, category, description, market_impact,
affected_assets, impact_score, parent_event_id,
expected_value, actual_value, surprise_pct, unit, sub_type, absorption_pct)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
expected_value, actual_value, surprise_pct, unit, sub_type, absorption_pct,
source_refs)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(ev["name"], ev["start_date"], ev.get("end_date"),
ev["level"], ev.get("category", "macro"), ev.get("description", ""),
ev.get("market_impact", ""), json.dumps(ev.get("affected_assets", [])),
ev.get("impact_score", 0.5), ev.get("parent_event_id"),
ev.get("expected_value"), ev.get("actual_value"),
ev.get("surprise_pct"), ev.get("unit"), ev.get("sub_type"),
ev.get("absorption_pct")))
ev.get("absorption_pct"), src_refs))
conn.commit()
return cur.lastrowid
finally:
@@ -4567,17 +4580,20 @@ def update_market_event(event_id: int, ev: Dict[str, Any]) -> bool:
import json
conn = get_conn()
try:
src_refs = ev.get("source_refs", [])
if isinstance(src_refs, list):
src_refs = json.dumps(src_refs)
conn.execute("""UPDATE market_events SET
name=?, start_date=?, end_date=?, level=?, category=?, description=?,
market_impact=?, affected_assets=?, impact_score=?,
absorption_pct=?, relevant_indicators=?
absorption_pct=?, relevant_indicators=?, source_refs=?
WHERE id=?""",
(ev["name"], ev["start_date"], ev.get("end_date"),
ev["level"], ev.get("category", "macro"), ev.get("description", ""),
ev.get("market_impact", ""), json.dumps(ev.get("affected_assets", [])),
ev.get("impact_score", 0.5),
ev.get("absorption_pct"), json.dumps(ev.get("relevant_indicators", [])),
event_id))
src_refs, event_id))
conn.commit()
return True
finally:

183
backend/services/market_event_detector.py
View File

@@ -6,6 +6,9 @@ Scans 4 sources and creates market_events for significant findings:
- eco : FRED economic releases with high surprise z-score
- technical: MA50/MA100/MA200 crossovers on key instruments
- reports : institutional reports (COT, EIA) with high importance
After each event is created, instrument impacts are evaluated immediately
via the AI (impact_service.evaluate_event_impacts).
"""
import json
import logging
@@ -14,7 +17,6 @@ from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
# Instruments monitored for technical crossovers
WATCH_INSTRUMENTS = [
"SPY", "QQQ", "IWM", "EEM", "EFA",
"GLD", "SLV", "USO", "TLT", "HYG",
@@ -43,7 +45,6 @@ def _get_api_key() -> str:
def _existing_event_keys() -> set:
"""Lowercase 50-char prefix of existing market_event names for fast dedup."""
from services.database import get_all_market_events
return {ev["name"].lower()[:50] for ev in get_all_market_events()}
@@ -53,15 +54,12 @@ def _is_dup(name: str, existing: set) -> bool:
def _parse_date(raw: str) -> str:
"""Return YYYY-MM-DD from any date string; fallback to today."""
if not raw:
return datetime.utcnow().strftime("%Y-%m-%d")
# ISO-like
try:
return datetime.fromisoformat(raw[:19]).strftime("%Y-%m-%d")
except Exception:
pass
# RFC 2822 (RSS)
try:
from email.utils import parsedate_to_datetime
return parsedate_to_datetime(raw).strftime("%Y-%m-%d")
@@ -70,6 +68,27 @@ def _parse_date(raw: str) -> str:
return raw[:10] if len(raw) >= 10 else datetime.utcnow().strftime("%Y-%m-%d")
def _save_and_evaluate(ev: Dict, existing: set) -> Optional[Dict]:
"""Save a market_event and immediately evaluate instrument impacts. Returns created dict or None."""
from services.database import save_market_event
try:
event_id = save_market_event(ev)
existing.add(ev["name"].lower()[:50])
logger.info(f"[check_events] ✓ saved event #{event_id}: {ev['name']}")
except Exception as e:
logger.error(f"[check_events] save failed for '{ev['name']}': {e}")
return None
# Evaluate instrument impacts immediately
try:
from services.impact_service import evaluate_event_impacts
evaluate_event_impacts(event_id, force=False)
except Exception as e:
logger.warning(f"[check_events] impact eval failed for #{event_id}: {e}")
return {"name": ev["name"], "category": ev.get("category", ""), "date": ev.get("start_date", ""), "event_id": event_id}
# ── Source 1: Geopolitical / macro news ──────────────────────────────────────
def _check_news(
@@ -77,12 +96,7 @@ def _check_news(
lookback_hours: int = 48,
max_to_evaluate: int = 15,
) -> List[Dict[str, Any]]:
"""
Fetch recent RSS news, keep those with impact_score >= threshold,
ask GPT-4o-mini which ones deserve a permanent market_event record.
"""
from services.data_fetcher import fetch_geo_news
from services.database import save_market_event
api_key = _get_api_key()
if not api_key:
@@ -170,6 +184,14 @@ FORMAT JSON STRICT:
if _is_dup(ev_name, existing):
continue
source_ref = {
"title": title,
"source": n.get("source", ""),
"url": n.get("url") or n.get("link", ""),
"date": _parse_date(n.get("date", "")),
"original_score": round(float(n.get("impact_score", 0)), 3),
}
ev = {
"name": ev_name,
"start_date": _parse_date(n.get("date", "")),
@@ -180,14 +202,12 @@ FORMAT JSON STRICT:
"market_impact": "",
"affected_assets": parsed.get("affected_assets", []),
"impact_score": float(parsed.get("impact_score", 0.6)),
"source_refs": [source_ref],
}
try:
save_market_event(ev)
existing.add(ev_name.lower()[:50])
created.append({"name": ev_name, "category": ev["category"], "date": ev["start_date"], "source": "news"})
logger.info(f"[check_events/news] ✓ {ev_name}")
except Exception as e:
logger.error(f"[check_events/news] save failed: {e}")
result = _save_and_evaluate(ev, existing)
if result:
result["source"] = "news"
created.append(result)
return created
@@ -195,11 +215,7 @@ FORMAT JSON STRICT:
# ── Source 2: Eco calendar — FRED surprises ───────────────────────────────────
def _check_eco(z_threshold: float = 1.5, days: int = 7) -> List[Dict[str, Any]]:
"""
Reads economic_events table (FRED releases already stored by fred_fetcher).
Creates market_events for releases with |z-score| >= threshold.
"""
from services.database import get_recent_economic_surprises, save_market_event
from services.database import get_recent_economic_surprises
try:
releases = get_recent_economic_surprises(days=days, min_zscore=z_threshold)
@@ -233,6 +249,14 @@ def _check_eco(z_threshold: float = 1.5, days: int = 7) -> List[Dict[str, Any]]:
except Exception:
assets = []
source_ref = {
"title": f"FRED release: {ev_name_base} ({ev_date})",
"source": "FRED",
"url": f"https://fred.stlouisfed.org/series/{s_id}" if s_id else "",
"date": ev_date,
"original_score": round(min(0.95, 0.35 + z * 0.15), 3),
}
ev = {
"name": ev_name,
"start_date": ev_date,
@@ -251,14 +275,12 @@ def _check_eco(z_threshold: float = 1.5, days: int = 7) -> List[Dict[str, Any]]:
"actual_value": str(rel.get("actual_value", "")),
"expected_value": str(rel.get("forecast_value", "")),
"surprise_pct": float(s_pct),
"source_refs": [source_ref],
}
try:
save_market_event(ev)
existing.add(ev_name.lower()[:50])
created.append({"name": ev_name, "category": "event_calendar", "date": ev_date, "source": "eco"})
logger.info(f"[check_events/eco] ✓ {ev_name}")
except Exception as e:
logger.error(f"[check_events/eco] save failed: {e}")
result = _save_and_evaluate(ev, existing)
if result:
result["source"] = "eco"
created.append(result)
return created
@@ -266,10 +288,6 @@ def _check_eco(z_threshold: float = 1.5, days: int = 7) -> List[Dict[str, Any]]:
# ── Source 3: MA crossovers (technical) ──────────────────────────────────────
def _check_technical(instruments: List[str] = None, lookback_days: int = 7) -> List[Dict[str, Any]]:
"""
Downloads recent OHLCV for each instrument, detects MA50/MA100/MA200 crossovers
in the last `lookback_days` days. Creates technical market_events.
"""
try:
import yfinance as yf
import pandas as pd
@@ -277,8 +295,6 @@ def _check_technical(instruments: List[str] = None, lookback_days: int = 7) -> L
logger.warning("[check_events/technical] yfinance/pandas not available")
return []
from services.database import save_market_event
if instruments is None:
instruments = WATCH_INSTRUMENTS
@@ -299,7 +315,6 @@ def _check_technical(instruments: List[str] = None, lookback_days: int = 7) -> L
recent = df.tail(lookback_days + 2)
# Check consecutive row pairs for crossovers
for i in range(1, len(recent)):
date_str = str(recent.index[i])[:10]
if date_str < cutoff:
@@ -308,12 +323,12 @@ def _check_technical(instruments: List[str] = None, lookback_days: int = 7) -> L
prev = recent.iloc[i - 1]
curr = recent.iloc[i]
def cross(fast_prev, fast_curr, slow_prev, slow_curr):
if any(pd.isna(v) for v in [fast_prev, fast_curr, slow_prev, slow_curr]):
def cross(fp, fc, sp, sc):
if any(pd.isna(v) for v in [fp, fc, sp, sc]):
return None
if fast_prev < slow_prev and fast_curr >= slow_curr:
if fp < sp and fc >= sc:
return "golden"
if fast_prev > slow_prev and fast_curr <= slow_curr:
if fp > sp and fc <= sc:
return "death"
return None
@@ -336,6 +351,14 @@ def _check_technical(instruments: List[str] = None, lookback_days: int = 7) -> L
direction = "bullish" if kind == "golden" else "bearish"
level = "medium" if slow_lbl == "MA200" else "short"
source_ref = {
"title": f"Technical signal: {ev_name}",
"source": "yfinance/computed",
"url": f"https://finance.yahoo.com/quote/{ticker}",
"date": date_str,
"original_score": 0.65 if slow_lbl == "MA200" else 0.45,
}
ev = {
"name": ev_name,
"start_date": date_str,
@@ -343,21 +366,21 @@ def _check_technical(instruments: List[str] = None, lookback_days: int = 7) -> L
"category": "technical",
"sub_type": f"{fast_lbl}/{slow_lbl} Cross",
"description": (
f"{cross_label} : {fast_lbl} passe {'au-dessus' if kind == 'golden' else 'en-dessous'} "
f"{cross_label} : {fast_lbl} passe "
f"{'au-dessus' if kind == 'golden' else 'en-dessous'} "
f"de la {slow_lbl} sur {ticker}. "
f"Signal {direction} de tendance {'long terme' if slow_lbl == 'MA200' else 'moyen terme'}."
f"Signal {direction} de tendance "
f"{'long terme' if slow_lbl == 'MA200' else 'moyen terme'}."
),
"market_impact": f"Signal {direction} sur {ticker}",
"affected_assets": [ticker],
"impact_score": 0.65 if slow_lbl == "MA200" else 0.45,
"source_refs": [source_ref],
}
try:
save_market_event(ev)
existing.add(ev_name.lower()[:50])
created.append({"name": ev_name, "category": "technical", "date": date_str, "source": "technical"})
logger.info(f"[check_events/technical] ✓ {ev_name}")
except Exception as e:
logger.error(f"[check_events/technical] save failed: {e}")
result = _save_and_evaluate(ev, existing)
if result:
result["source"] = "technical"
created.append(result)
except Exception as e:
logger.debug(f"[check_events/technical] {ticker} failed: {e}")
@@ -368,11 +391,7 @@ def _check_technical(instruments: List[str] = None, lookback_days: int = 7) -> L
# ── Source 4: Institutional reports ──────────────────────────────────────────
def _check_reports(days: int = 7, min_importance: int = 3) -> List[Dict[str, Any]]:
"""
Reads institutional_reports table for recent high-importance entries.
Creates report market_events.
"""
from services.database import get_conn, save_market_event
from services.database import get_conn
try:
cutoff = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%d")
@@ -410,7 +429,6 @@ def _check_reports(days: int = 7, min_importance: int = 3) -> List[Dict[str, Any
except Exception:
pass
# Derive affected assets from signal columns
assets: List[str] = []
for sig_col, asset_list in [
("signal_energy", ["USO", "XOM"]),
@@ -421,6 +439,14 @@ def _check_reports(days: int = 7, min_importance: int = 3) -> List[Dict[str, Any
if rpt.get(sig_col, "neutral") not in ("neutral", "", None):
assets.extend(asset_list)
source_ref = {
"title": title,
"source": rpt.get("source", rpt_type),
"url": "",
"date": rpt_date,
"original_score": round(min(0.9, 0.3 + rpt.get("importance", 2) * 0.12), 3),
}
ev = {
"name": ev_name,
"start_date": rpt_date,
@@ -431,14 +457,12 @@ def _check_reports(days: int = 7, min_importance: int = 3) -> List[Dict[str, Any
"market_impact": rpt.get("trading_implications", ""),
"affected_assets": list(set(assets)),
"impact_score": min(0.9, 0.3 + rpt.get("importance", 2) * 0.12),
"source_refs": [source_ref],
}
try:
save_market_event(ev)
existing.add(ev_name.lower()[:50])
created.append({"name": ev_name, "category": "report", "date": rpt_date, "source": "reports"})
logger.info(f"[check_events/reports] ✓ {ev_name}")
except Exception as e:
logger.error(f"[check_events/reports] save failed: {e}")
result = _save_and_evaluate(ev, existing)
if result:
result["source"] = "reports"
created.append(result)
return created
@@ -456,62 +480,43 @@ def check_new_market_events(
report_min_importance: int = 3,
) -> Dict[str, Any]:
"""
Isolated cycle action — scans all (or selected) sources and creates
market_events for significant findings.
sources: subset of ['news', 'eco', 'technical', 'reports']
default = all four
Isolated cycle action — scans all (or selected) sources, creates
market_events with source_refs, and immediately evaluates instrument impacts.
"""
if sources is None:
sources = ["news", "eco", "technical", "reports"]
results: Dict[str, Any] = {
"news": [],
"eco": [],
"technical": [],
"reports": [],
"news": [], "eco": [], "technical": [], "reports": [],
"total_created": 0,
"ran_at": datetime.utcnow().isoformat(),
}
if "news" in sources:
try:
results["news"] = _check_news(
min_impact=news_impact_min,
lookback_hours=news_lookback_hours,
)
results["news"] = _check_news(min_impact=news_impact_min, lookback_hours=news_lookback_hours)
except Exception as e:
logger.error(f"[check_events] news source error: {e}")
if "eco" in sources:
try:
results["eco"] = _check_eco(
z_threshold=eco_z_threshold,
days=eco_days,
)
results["eco"] = _check_eco(z_threshold=eco_z_threshold, days=eco_days)
except Exception as e:
logger.error(f"[check_events] eco source error: {e}")
if "technical" in sources:
try:
results["technical"] = _check_technical(
lookback_days=technical_lookback_days,
)
results["technical"] = _check_technical(lookback_days=technical_lookback_days)
except Exception as e:
logger.error(f"[check_events] technical source error: {e}")
if "reports" in sources:
try:
results["reports"] = _check_reports(
days=report_days,
min_importance=report_min_importance,
)
results["reports"] = _check_reports(days=report_days, min_importance=report_min_importance)
except Exception as e:
logger.error(f"[check_events] reports source error: {e}")
results["total_created"] = sum(
len(results[s]) for s in ["news", "eco", "technical", "reports"]
)
results["total_created"] = sum(len(results[s]) for s in ["news", "eco", "technical", "reports"])
logger.info(
f"[check_events] Done — {results['total_created']} new events: "
f"news={len(results['news'])} eco={len(results['eco'])} "