feat: Phase 2 — economic event surprise tracker (FRED actuals + z-score)

- economic_events table in DB (series_id, actual, forecast_baseline, surprise_pct, surprise_zscore, direction)
- DB helpers: save_economic_event(), get_recent_economic_surprises(), get_economic_events_for_calendar()
- fred_fetcher.py: _compute_zscore_surprise() computes 12-period MA as implied consensus + z-score deviation; save_fred_releases_to_db() persists releases per cycle; build_economic_surprise_block() formats significant surprises for AI prompt
- auto_cycle.py: saves FRED releases to economic_events each cycle, appends surprise block to fred_block for injection into both suggestion and scoring prompts
- data_fetcher.py: get_economic_calendar() now merges static upcoming events with past FRED actuals from DB (Prev/Fcst/Actual/z-score fields populated)
- CalendarPage.tsx: past events show colored z-score badge ( for |z|≥1.5, bullish/bearish colors)
- EconomicEvent type: added surprise_zscore, surprise_direction, source fields

Activates automatically once fred_api_key is set in Configuration.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OpenSquared
2026-06-22 14:00:35 +02:00
parent 3edbd6b0b7
commit d178615c74
6 changed files with 346 additions and 29 deletions

View File

@@ -157,20 +157,123 @@ def get_fred_recent_releases() -> List[Dict[str, Any]]:
return results
def _compute_zscore_surprise(series_id: str, api_key: str, latest_value: float) -> tuple:
"""Compute z-score of latest value vs 12-period MA. Returns (forecast, surprise_pct, zscore)."""
import math
obs = _fetch_series_observations(series_id, api_key, count=14)
values = [_parse_value(o["value"]) for o in obs if _parse_value(o["value"]) is not None]
if len(values) < 3:
return None, None, None
# Use obs[1:] as history (exclude latest), up to 12 periods
history = values[1:13]
if not history:
return None, None, None
mean = sum(history) / len(history)
variance = sum((x - mean) ** 2 for x in history) / len(history)
std = math.sqrt(variance) if variance > 0 else 0.0
surprise_pct = round((latest_value - mean) / abs(mean) * 100, 2) if mean != 0 else 0.0
zscore = round((latest_value - mean) / std, 2) if std > 0 else 0.0
return round(mean, 4), surprise_pct, zscore
def save_fred_releases_to_db(releases: List[Dict], api_key: str = "") -> int:
"""Persist FRED releases to economic_events table with surprise scores. Returns count saved."""
from services.database import save_economic_event
import json as _json
# Map series_id → assets_impacted from FRED_SERIES config
_assets_map = {sid: assets for sid, _, _, assets, _ in FRED_SERIES}
saved = 0
for r in releases:
sid = r.get("series_id", "")
if not sid or r.get("latest_value") is None:
continue
# Compute z-score surprise if api_key available
forecast_val, surprise_pct, zscore = (None, None, None)
if api_key:
try:
forecast_val, surprise_pct, zscore = _compute_zscore_surprise(sid, api_key, r["latest_value"])
except Exception:
pass
try:
save_economic_event(
event_name=r.get("label", sid),
series_id=sid,
event_date=r.get("latest_date", ""),
actual_value=r.get("latest_value"),
actual_unit=r.get("unit", ""),
forecast_value=forecast_val,
previous_value=r.get("previous_value"),
surprise_pct=surprise_pct,
surprise_zscore=zscore,
surprise_direction=r.get("direction", "neutral"),
assets_impacted=_assets_map.get(sid, []),
source="FRED",
)
saved += 1
except Exception as e:
logger.warning(f"[FRED] Failed to save {sid} to DB: {e}")
return saved
def build_fred_context_block(releases: List[Dict]) -> str:
"""Format FRED releases as a prompt-ready string."""
"""Format FRED releases as a prompt-ready string with surprise scoring."""
if not releases:
return ""
lines = ["## 📈 DONNÉES MACRO RÉCENTES (FRED — dernières releases)"]
lines = ["## RECENT MACRO DATA (FRED — latest releases)"]
for r in releases:
val_str = f"{r['latest_value']:.2f}{r['unit']}" if isinstance(r.get("latest_value"), float) else str(r.get("latest_value", "N/A"))
val_str = f"{r['latest_value']:.2f} {r['unit']}" if isinstance(r.get("latest_value"), float) else str(r.get("latest_value", "N/A"))
chg_str = ""
if r.get("change") is not None:
arrow = "" if r["change"] > 0 else ""
chg_str = f" ({arrow}{abs(r['change']):.3f} vs release précédente)"
dir_emoji = {"bullish": "🟢", "bearish": "🔴", "neutral": ""}.get(r.get("direction", "neutral"), "")
arrow = "+" if r["change"] > 0 else ""
chg_str = f" ({arrow}{r['change']:.3f} vs prior)"
surprise_str = ""
if r.get("surprise_zscore") is not None:
z = r["surprise_zscore"]
if abs(z) >= 1.5:
surprise_str = f" ⚡ SURPRISE z={z:+.1f}"
elif abs(z) >= 0.8:
surprise_str = f" (notable z={z:+.1f})"
dir_tag = {"bullish": "[BULLISH]", "bearish": "[BEARISH]", "neutral": "[NEUTRAL]"}.get(r.get("direction", "neutral"), "")
lines.append(
f" {dir_emoji} {r['label']} [{r['latest_date']}] : {val_str}{chg_str}{r['direction'].upper()}"
f" {r['label']} [{r['latest_date']}]: {val_str}{chg_str}{surprise_str}{dir_tag}"
)
lines.append("⚠️ Compare ces chiffres au consensus attendu pour évaluer si le marché a déjà intégré la surprise.")
lines.append("Use these figures to assess whether current market pricing already reflects macro reality.")
return "\n".join(lines)
def build_economic_surprise_block(days: int = 14) -> str:
"""Build a prompt block from stored economic_events table — recent surprises highlighted."""
try:
from services.database import get_recent_economic_surprises
events = get_recent_economic_surprises(days=days, min_zscore=0.0)
if not events:
return ""
lines = [f"## ECONOMIC SURPRISE TRACKER (last {days} days — FRED actuals vs trend baseline)"]
for ev in events[:8]:
z = ev.get("surprise_zscore") or 0.0
sp = ev.get("surprise_pct") or 0.0
direction = ev.get("surprise_direction", "neutral")
actual = ev.get("actual_value")
unit = ev.get("actual_unit", "")
forecast = ev.get("forecast_value")
flag = ""
if abs(z) >= 1.5:
flag = " ⚡ SIGNIFICANT SURPRISE"
elif abs(z) >= 0.8:
flag = " (notable)"
actual_str = f"{actual:.2f}{unit}" if actual is not None else "N/A"
forecast_str = f"{forecast:.2f}{unit}" if forecast is not None else "N/A"
dir_tag = {"bullish": "BULLISH", "bearish": "BEARISH", "neutral": "NEUTRAL"}.get(direction, "")
lines.append(
f" [{ev['event_date']}] {ev['event_name']}: actual={actual_str} vs baseline={forecast_str}"
f" (z={z:+.2f}, {sp:+.1f}%) → {dir_tag}{flag}"
)
lines.append("→ High z-scores = unexpected move vs recent trend → potential mispricing in related assets")
return "\n".join(lines)
except Exception as e:
logger.warning(f"[FRED] build_economic_surprise_block failed: {e}")
return ""