From bb614936c35c78dfe5d8eb60eeaca6312abf54b0 Mon Sep 17 00:00:00 2001 From: OpenSquared Date: Tue, 30 Jun 2026 19:08:33 +0200 Subject: [PATCH] feat: macro series --- backend/routers/eco.py | 60 ++++++ backend/services/calendar_sync.py | 11 ++ backend/services/database.py | 23 +++ backend/services/macro_series_log.py | 159 +++++++++++++++ frontend/src/pages/MacroSeriesPage.tsx | 255 ++++++++++++++++--------- 5 files changed, 417 insertions(+), 91 deletions(-) create mode 100644 backend/services/macro_series_log.py diff --git a/backend/routers/eco.py b/backend/routers/eco.py index 95db791..1fb160d 100644 --- a/backend/routers/eco.py +++ b/backend/routers/eco.py @@ -740,3 +740,63 @@ def eco_status() -> Dict[str, Any]: } finally: conn.close() + + +# ── Macro Series Log ────────────────────────────────────────────────────────── + +@router.get("/series-log") +def get_series_log( + series_id: str = Query(...), + from_date: str = Query("2020-01-01"), + limit: int = Query(500, le=2000), +) -> List[Dict[str, Any]]: + """Return logged actual/forecast snapshots for a given series (change events only).""" + from services.database import get_conn + conn = get_conn() + try: + rows = conn.execute( + """SELECT series_id, event_name, event_date, actual_value, forecast_value, + previous_value, currency, impact, logged_at, source, changed + FROM macro_series_log + WHERE series_id = ? AND event_date >= ? + ORDER BY event_date ASC, logged_at ASC + LIMIT ?""", + (series_id, from_date, limit) + ).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + +@router.get("/series-log/distinct-series") +def list_logged_series() -> List[Dict[str, Any]]: + """Return all series_id that have entries in macro_series_log.""" + from services.database import get_conn + conn = get_conn() + try: + rows = conn.execute( + """SELECT series_id, event_name, + COUNT(*) as log_count, + MIN(event_date) as first_date, + MAX(event_date) as last_date, + MAX(logged_at) as last_logged + FROM macro_series_log + GROUP BY series_id + ORDER BY last_date DESC""" + ).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + +@router.post("/series-log/backfill") +def backfill_series_log() -> Dict[str, Any]: + """Seed macro_series_log from existing ff_calendar rows (one-time historical import).""" + from services.database import get_conn + from services.macro_series_log import backfill_from_ff_calendar + conn = get_conn() + try: + result = backfill_from_ff_calendar(conn) + return result + finally: + conn.close() diff --git a/backend/services/calendar_sync.py b/backend/services/calendar_sync.py index ba02d51..d78bae4 100644 --- a/backend/services/calendar_sync.py +++ b/backend/services/calendar_sync.py @@ -47,6 +47,17 @@ def calendar_sync(weeks_ahead: int = 8) -> Dict[str, Any]: result = {"error": str(e)} source = "error" + # After calendar upsert, log any forecast/actual changes to macro_series_log + try: + from services.database import get_conn + from services.macro_series_log import scan_and_log_all + _conn = get_conn() + log_result = scan_and_log_all(_conn, source=source) + _conn.close() + result["_log"] = log_result + except Exception as e: + logger.warning(f"[calendar_sync] macro_series_log scan failed: {e}") + _sync_status["running"] = False _sync_status["last_result"] = result _sync_status["source"] = source diff --git a/backend/services/database.py b/backend/services/database.py index ad509af..9656ace 100644 --- a/backend/services/database.py +++ b/backend/services/database.py @@ -715,6 +715,29 @@ def init_db(): except Exception: pass # Column already exists + # ── Macro Series Log ─────────────────────────────────────────────────────── + # One row per change: actual or forecast changed vs previous snapshot. + # Allows tracking forecast revisions over time + actual vs consensus history. + c.execute("""CREATE TABLE IF NOT EXISTS macro_series_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + series_id TEXT NOT NULL, + event_name TEXT NOT NULL DEFAULT '', + event_date TEXT NOT NULL, + actual_value REAL, + forecast_value REAL, + previous_value REAL, + currency TEXT DEFAULT '', + impact TEXT DEFAULT '', + logged_at TEXT NOT NULL DEFAULT (datetime('now')), + source TEXT DEFAULT 'calendar_sync', + changed TEXT DEFAULT 'new' + )""") + try: + c.execute("CREATE INDEX IF NOT EXISTS idx_msl_series ON macro_series_log(series_id, event_date)") + c.execute("CREATE INDEX IF NOT EXISTS idx_msl_logged ON macro_series_log(logged_at DESC)") + except Exception: + pass + # ── Specialist Desks ─────────────────────────────────────────────────────── c.execute("""CREATE TABLE IF NOT EXISTS specialist_reports ( id TEXT PRIMARY KEY, diff --git a/backend/services/macro_series_log.py b/backend/services/macro_series_log.py new file mode 100644 index 0000000..10d2f82 --- /dev/null +++ b/backend/services/macro_series_log.py @@ -0,0 +1,159 @@ +""" +Macro series log — tracks changes to actual_value and forecast_value over time. +Called after each calendar sync to detect and persist revisions. +""" +import logging +from datetime import datetime, timezone +from typing import Optional + +logger = logging.getLogger(__name__) + + +def _parse_num(v) -> Optional[float]: + """Convert FF text value (e.g. '178K', '0.3%', '-92K') to float or None.""" + if v is None: + return None + s = str(v).strip().replace('%', '').replace('K', 'e3').replace('M', 'e6').replace('B', 'e9') + try: + return float(s) + except Exception: + return None + + +def log_if_changed(conn, series_id: str, event_name: str, event_date: str, + actual_value, forecast_value, previous_value, + currency: str = '', impact: str = '', + source: str = 'calendar_sync') -> bool: + """ + Compare (actual, forecast) against the last log entry for this series+date. + Insert a new row only if something changed or it's the first time we see it. + Returns True if a row was inserted. + """ + actual_num = _parse_num(actual_value) + forecast_num = _parse_num(forecast_value) + previous_num = _parse_num(previous_value) + + # Nothing to log if both actual and forecast are None + if actual_num is None and forecast_num is None: + return False + + last = conn.execute( + """SELECT actual_value, forecast_value FROM macro_series_log + WHERE series_id=? AND event_date=? + ORDER BY logged_at DESC LIMIT 1""", + (series_id, event_date) + ).fetchone() + + if last is None: + changed = 'new' + else: + prev_actual = last['actual_value'] + prev_forecast = last['forecast_value'] + actual_changed = actual_num != prev_actual and actual_num is not None + forecast_changed = forecast_num != prev_forecast and forecast_num is not None + if not actual_changed and not forecast_changed: + return False + if actual_changed and forecast_changed: + changed = 'both' + elif actual_changed: + changed = 'actual' + else: + changed = 'forecast' + + conn.execute( + """INSERT INTO macro_series_log + (series_id, event_name, event_date, actual_value, forecast_value, + previous_value, currency, impact, logged_at, source, changed) + VALUES (?,?,?,?,?,?,?,?,?,?,?)""", + (series_id, event_name or '', event_date, + actual_num, forecast_num, previous_num, + currency or '', impact or '', + datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S'), + source, changed) + ) + conn.commit() + return True + + +def backfill_from_ff_calendar(conn) -> dict: + """ + Seed macro_series_log from existing ff_calendar rows (historical). + Only inserts rows that don't already exist (idempotent). + Skips events without series_id. + """ + rows = conn.execute( + """SELECT series_id, event_name, event_date, actual_value, forecast_value, + previous_value, currency, impact, fetched_at + FROM ff_calendar + WHERE series_id IS NOT NULL AND series_id != '' + AND (actual_value IS NOT NULL OR forecast_value IS NOT NULL) + ORDER BY event_date ASC""" + ).fetchall() + + inserted = 0 + for r in rows: + already = conn.execute( + "SELECT 1 FROM macro_series_log WHERE series_id=? AND event_date=? AND source='backfill' LIMIT 1", + (r['series_id'], r['event_date']) + ).fetchone() + if already: + continue + + actual_num = _parse_num(r['actual_value']) + forecast_num = _parse_num(r['forecast_value']) + previous_num = _parse_num(r['previous_value']) + if actual_num is None and forecast_num is None: + continue + + logged_at = r['fetched_at'] or r['event_date'] + changed = 'actual' if actual_num is not None else 'forecast' + + conn.execute( + """INSERT OR IGNORE INTO macro_series_log + (series_id, event_name, event_date, actual_value, forecast_value, + previous_value, currency, impact, logged_at, source, changed) + VALUES (?,?,?,?,?,?,?,?,?,?,?)""", + (r['series_id'], r['event_name'] or '', r['event_date'], + actual_num, forecast_num, previous_num, + r['currency'] or '', r['impact'] or '', + logged_at, 'backfill', changed) + ) + inserted += 1 + + conn.commit() + logger.info(f"[macro_series_log] backfill: {inserted} rows inserted from ff_calendar") + return {"inserted": inserted, "scanned": len(rows)} + + +def scan_and_log_all(conn, source: str = 'calendar_sync') -> dict: + """ + After a calendar sync, scan all ff_calendar rows with series_id and log any changes. + Typically called at end of each 6h sync cycle. + """ + rows = conn.execute( + """SELECT series_id, event_name, event_date, actual_value, forecast_value, + previous_value, currency, impact + FROM ff_calendar + WHERE series_id IS NOT NULL AND series_id != '' + AND (actual_value IS NOT NULL OR forecast_value IS NOT NULL)""" + ).fetchall() + + logged = 0 + for r in rows: + did_log = log_if_changed( + conn, + series_id=r['series_id'], + event_name=r['event_name'] or '', + event_date=r['event_date'], + actual_value=r['actual_value'], + forecast_value=r['forecast_value'], + previous_value=r['previous_value'], + currency=r['currency'] or '', + impact=r['impact'] or '', + source=source, + ) + if did_log: + logged += 1 + + logger.info(f"[macro_series_log] scan_and_log_all: {logged} changes logged out of {len(rows)} rows") + return {"logged": logged, "scanned": len(rows)} diff --git a/frontend/src/pages/MacroSeriesPage.tsx b/frontend/src/pages/MacroSeriesPage.tsx index 8b0a38a..fe97571 100644 --- a/frontend/src/pages/MacroSeriesPage.tsx +++ b/frontend/src/pages/MacroSeriesPage.tsx @@ -1,9 +1,9 @@ import { useState, useEffect, useCallback } from 'react' import clsx from 'clsx' -import { RefreshCw, TrendingUp, TrendingDown, Minus } from 'lucide-react' +import { RefreshCw, TrendingUp, TrendingDown, Minus, Database } from 'lucide-react' import { - ComposedChart, Line, XAxis, YAxis, CartesianGrid, Tooltip, - ResponsiveContainer, ReferenceLine, Area, + ComposedChart, Line, Bar, XAxis, YAxis, CartesianGrid, Tooltip, + ResponsiveContainer, ReferenceLine, Area, Legend, } from 'recharts' const API = '' @@ -46,6 +46,17 @@ interface SeriesHistory { events: FFEvent[] } +interface LogEntry { + series_id: string + event_name: string + event_date: string + actual_value: number | null + forecast_value: number | null + previous_value: number | null + logged_at: string + changed: string // 'new' | 'actual' | 'forecast' | 'both' | 'backfill' +} + // ── Constants ───────────────────────────────────────────────────────────────── const CATEGORY_COLORS: Record = { @@ -254,9 +265,11 @@ export default function MacroSeriesPage() { const [selectedId, setSelectedId] = useState('PAYEMS') const [range, setRange] = useState('5y') const [history, setHistory] = useState(null) + const [logEntries, setLogEntries] = useState([]) const [loading, setLoading] = useState(false) + const [backfilling, setBackfilling] = useState(false) + const [tab, setTab] = useState<'chart' | 'log'>('chart') - // Load series catalog useEffect(() => { fetch(`${API}/api/eco/series`).then(r => r.json()).then(setSeriesList).catch(() => {}) }, []) @@ -267,35 +280,57 @@ export default function MacroSeriesPage() { try { const months = RANGE_OPTIONS.find(r => r.key === range)?.months ?? 60 const from = fromDateForRange(months) - const r: SeriesHistory = await fetch( - `${API}/api/eco/series/${selectedId}/history?from_date=${from}` - ).then(x => x.json()) - setHistory(r) + const [hist, log] = await Promise.all([ + fetch(`${API}/api/eco/series/${selectedId}/history?from_date=${from}`).then(x => x.json()), + fetch(`${API}/api/eco/series-log?series_id=${selectedId}&from_date=${from}&limit=1000`).then(x => x.json()), + ]) + setHistory(hist) + setLogEntries(Array.isArray(log) ? log : []) } catch {} finally { setLoading(false) } }, [selectedId, range]) useEffect(() => { loadHistory() }, [loadHistory]) - // Build chart data — merge FRED time series + FF events by date + const handleBackfill = async () => { + setBackfilling(true) + try { + await fetch(`${API}/api/eco/series-log/backfill`, { method: 'POST' }) + await loadHistory() + } catch {} + finally { setBackfilling(false) } + } + + // Build chart data — merge FRED time series + log entries by event_date + // For each release date: actual from log (last entry with actual) + last forecast before actual const chartData = (() => { if (!history) return [] const ffByDate = new Map() for (const ev of history.events) ffByDate.set(ev.event_date, ev) + // Group log entries by event_date, pick last forecast before actual and first actual + const logByDate = new Map() + for (const e of logEntries) { + const existing = logByDate.get(e.event_date) ?? { actual: null, lastForecast: null } + if (e.actual_value != null) existing.actual = e.actual_value + if (e.forecast_value != null && e.actual_value == null) existing.lastForecast = e.forecast_value + logByDate.set(e.event_date, existing) + } + return history.timeseries.map(pt => { const ff = ffByDate.get(pt.event_date) + const logged = logByDate.get(pt.event_date) return { - event_date: pt.event_date, - label: fmtDate(pt.event_date), - value: pt.actual_value, - zscore: pt.surprise_zscore, - direction: pt.surprise_direction, - ff_event: ff?.event_name ?? null, - ff_actual: ff?.actual_value ?? null, - ff_forecast:ff?.forecast_value ?? null, - // For reference lines — mark surprise events - is_surprise: pt.surprise_zscore !== null && Math.abs(pt.surprise_zscore) >= 1.5, + event_date: pt.event_date, + label: fmtDate(pt.event_date), + value: pt.actual_value, + forecast: logged?.lastForecast ?? (ff?.forecast_value ? parseFloat(ff.forecast_value) || null : null), + zscore: pt.surprise_zscore, + direction: pt.surprise_direction, + ff_event: ff?.event_name ?? null, + ff_actual: ff?.actual_value ?? null, + ff_forecast: ff?.forecast_value ?? null, + is_surprise: pt.surprise_zscore !== null && Math.abs(pt.surprise_zscore) >= 1.5, } }) })() @@ -346,26 +381,33 @@ export default function MacroSeriesPage() {
+ {/* Tab selector */} +
+ {(['chart', 'log'] as const).map(t => ( + + ))} +
{/* Range selector */}
{RANGE_OPTIONS.map(r => ( - ))}
- +
@@ -397,107 +439,138 @@ export default function MacroSeriesPage() { )} - {/* Chart */} + {tab === 'chart' && (
- Time Series - {history && ({chartData.length} data points)} + Actual vs Consensus + {history && ({chartData.length} releases)} {bigSurprises.length > 0 && ( - - {bigSurprises.length} big surprises (|z| ≥ 1.5) - + {bigSurprises.length} surprises (|z|≥1.5) )}
- {loading && ( -
- Loading… -
- )} - + {loading &&
Chargement…
} {!loading && chartData.length === 0 && (
- No data — run FRED bootstrap first in the Calendar page. + Pas de données — lancez le bootstrap FRED depuis Calendar.
)} {!loading && chartData.length > 0 && ( - + - + - - `${v}${history?.unit === '%' || history?.unit === 'pp' ? '' : ''}`} - /> + + } /> + - {/* Zero line if series crosses zero */} {minVal !== null && maxVal !== null && minVal < 0 && maxVal > 0 && ( )} - - {/* Big surprise markers */} {bigSurprises.map(s => ( - + strokeWidth={1} strokeDasharray="3 3" opacity={0.5} /> ))} - + {/* Actual — area + line */} + + + {/* Forecast — dashed line */} + )} +
+ )} - {/* Legend for reference lines */} - {bigSurprises.length > 0 && ( -
- - - Bullish surprise (|z|≥1.5) - - - - Bearish surprise - + {tab === 'log' && ( +
+
+ + Journal des changements — {selectedId} + ({logEntries.length} entrées) + + {logEntries.length === 0 && ( + + )} +
+ {logEntries.length === 0 ? ( +
+ Aucune entrée — cliquez "Importer historique" ou attendez le prochain sync (6h). +
+ ) : ( +
+ + + + + + + + + + + + + {[...logEntries].reverse().map((e, i) => { + const isActual = e.actual_value != null + const isForecastChange = e.changed === 'forecast' || e.changed === 'both' + return ( + + + + + + + + + ) + })} + +
ReleaseCapturé àActualForecastPreviousChangement
{e.event_date}{e.logged_at.slice(0, 16)} + {e.actual_value != null ? e.actual_value.toFixed(2) : '—'} + + {e.forecast_value != null ? e.forecast_value.toFixed(2) : '—'} + + {e.previous_value != null ? e.previous_value.toFixed(2) : '—'} + + + {e.changed} + +
)}
+ )} {/* Events table */}
FF Calendar Events - {history && ({history.events.length} linked)} + {history && ({history.events.length} liés)}
{history && }