OpenFin/backend/services/macro_series_log.py

"""
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)}