""" 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 in display unit. Strips unit suffix without multiplying — '114K' → 114.0, '0.3%' → 0.3. Matches JavaScript parseFloat behaviour so stored values match chart display.""" if v is None: return None s = str(v).strip().replace('%', '').replace(',', '') if s and s[-1].upper() in ('K', 'M', 'B'): s = s[:-1].strip() 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). Filters by ff_currency to avoid multi-country contamination. """ currency_map = _series_currency_map() 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: expected = currency_map.get(r['series_id']) if expected and r['currency'] != expected: continue 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 # For past events use fetched_at (when we got the data); for future/no-actual use now now_str = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S') if actual_num is not None: logged_at = r['fetched_at'] or r['event_date'] # historical: keep original timestamp changed = 'actual' else: logged_at = now_str # forecast-only: logged_at = time of backfill run changed = '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 _series_currency_map() -> dict: """Return {series_id: ff_currency} for series that have a currency restriction.""" try: from services.fred_bootstrap import FRED_SERIES return {sid: meta['ff_currency'] for sid, meta in FRED_SERIES.items() if meta.get('ff_currency')} except Exception: return {} 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. Filters by ff_currency to avoid multi-country flip-flop (e.g. US vs CA unemployment). """ currency_map = _series_currency_map() 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: expected = currency_map.get(r['series_id']) if expected and r['currency'] != expected: continue 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)}