signal-platform/scripts/import_dolthub_earnings.py

"""Import the public post-no-preference/earnings DoltHub database.

The earnings calendar and EPS history are separate tables in the source. This
importer aligns them monotonically per symbol, keeps every calendar event for
the defensive gap study, and stores the longer EPS history separately for SUE
scaling. EPS history without an announcement date is never exposed as a live
signal event.
"""

from __future__ import annotations

import argparse
import csv
import json
import math
import sqlite3
from collections import defaultdict
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any


EVENTS_DDL = """
CREATE TABLE IF NOT EXISTS earnings_events (
    id INTEGER PRIMARY KEY,
    symbol TEXT NOT NULL,
    announce_date TEXT NOT NULL,
    announce_time TEXT,
    eps_estimate REAL,
    eps_actual REAL,
    revenue_estimate REAL,
    revenue_actual REAL,
    source TEXT NOT NULL,
    fetched_at TEXT NOT NULL,
    period_end_date TEXT,
    UNIQUE(symbol, announce_date)
)
"""
META_DDL = """
CREATE TABLE IF NOT EXISTS earnings_backfill_meta (
    symbol TEXT PRIMARY KEY,
    status TEXT NOT NULL,
    n_events INTEGER NOT NULL DEFAULT 0,
    updated_at TEXT NOT NULL,
    note TEXT
)
"""
SURPRISE_HISTORY_DDL = """
CREATE TABLE IF NOT EXISTS earnings_surprise_history (
    symbol TEXT NOT NULL,
    period_end_date TEXT NOT NULL,
    eps_estimate REAL,
    eps_actual REAL,
    source TEXT NOT NULL,
    fetched_at TEXT NOT NULL,
    PRIMARY KEY(symbol, period_end_date)
)
"""

SKIP_EVENT_COST = 45.0
SKIP_PERIOD_COST = 45.0


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--snapshot", default="backtest_snapshots/prod.sqlite")
    parser.add_argument("--calendar-csv", required=True)
    parser.add_argument("--history-csv", required=True)
    parser.add_argument("--from-date", default="2020-01-22")
    parser.add_argument("--to-date", required=True)
    parser.add_argument("--source-commit", required=True)
    parser.add_argument(
        "--source-url",
        default="https://www.dolthub.com/repositories/post-no-preference/earnings",
    )
    parser.add_argument("--max-period-lag-days", type=int, default=90)
    parser.add_argument("--max-period-lead-days", type=int, default=14)
    parser.add_argument(
        "--status-output", default="reports/earnings-backfill-status.json"
    )
    return parser.parse_args()


def _normalise_symbol(value: Any) -> str:
    return str(value or "").strip().upper().replace(".", "-")


def _normalise_session(value: Any) -> str | None:
    cleaned = str(value or "").strip().lower().replace("_", " ").replace("-", " ")
    aliases = {
        "before market open": "bmo",
        "before open": "bmo",
        "bmo": "bmo",
        "after market close": "amc",
        "after close": "amc",
        "amc": "amc",
        "during market hours": "during",
        "dmh": "during",
    }
    return aliases.get(cleaned, cleaned or None)


def _number(value: Any) -> float | None:
    if value is None or str(value).strip() == "":
        return None
    try:
        result = float(value)
    except (TypeError, ValueError):
        return None
    return result if math.isfinite(result) else None


def _read_calendar(
    path: Path,
    universe: set[str],
    start: date,
    end: date,
) -> tuple[dict[str, list[dict[str, Any]]], dict[str, int]]:
    by_key: dict[tuple[str, date], dict[str, Any]] = {}
    raw_rows = 0
    universe_rows = 0
    duplicate_rows = 0
    restated_rows = 0
    with path.open(newline="", encoding="utf-8-sig") as handle:
        for raw in csv.DictReader(handle):
            raw_rows += 1
            symbol = _normalise_symbol(raw.get("act_symbol"))
            raw_date = str(raw.get("date") or "")[:10]
            if symbol not in universe or not raw_date:
                continue
            event_date = date.fromisoformat(raw_date)
            if not start <= event_date <= end:
                continue
            universe_rows += 1
            row = {
                "symbol": symbol,
                "announce_date": event_date,
                "announce_time": _normalise_session(raw.get("when")),
            }
            key = (symbol, event_date)
            previous = by_key.get(key)
            if previous is not None:
                duplicate_rows += 1
                if (
                    previous.get("announce_time") is not None
                    and row.get("announce_time") is not None
                    and previous["announce_time"] != row["announce_time"]
                ):
                    restated_rows += 1
                if row.get("announce_time") is not None:
                    by_key[key] = row
            else:
                by_key[key] = row
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in by_key.values():
        grouped[row["symbol"]].append(row)
    for rows in grouped.values():
        rows.sort(key=lambda item: item["announce_date"])
    return grouped, {
        "raw_rows": raw_rows,
        "universe_rows_in_window": universe_rows,
        "deduped_rows_in_window": len(by_key),
        "duplicate_rows": duplicate_rows,
        "restated_rows": restated_rows,
    }


def _read_history(
    path: Path, universe: set[str]
) -> tuple[dict[str, list[dict[str, Any]]], dict[str, int]]:
    by_key: dict[tuple[str, date], dict[str, Any]] = {}
    raw_rows = 0
    universe_rows = 0
    duplicate_rows = 0
    restated_rows = 0
    fields = ("eps_actual", "eps_estimate")
    with path.open(newline="", encoding="utf-8-sig") as handle:
        for raw in csv.DictReader(handle):
            raw_rows += 1
            symbol = _normalise_symbol(raw.get("act_symbol"))
            raw_date = str(raw.get("period_end_date") or "")[:10]
            if symbol not in universe or not raw_date:
                continue
            universe_rows += 1
            period_end = date.fromisoformat(raw_date)
            row = {
                "symbol": symbol,
                "period_end_date": period_end,
                "eps_actual": _number(raw.get("reported")),
                "eps_estimate": _number(raw.get("estimate")),
            }
            key = (symbol, period_end)
            previous = by_key.get(key)
            if previous is not None:
                duplicate_rows += 1
                if any(
                    previous.get(field) is not None
                    and row.get(field) is not None
                    and previous[field] != row[field]
                    for field in fields
                ):
                    restated_rows += 1
                previous_score = sum(previous.get(field) is not None for field in fields)
                row_score = sum(row.get(field) is not None for field in fields)
                if row_score >= previous_score:
                    by_key[key] = row
            else:
                by_key[key] = row
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in by_key.values():
        grouped[row["symbol"]].append(row)
    for rows in grouped.values():
        rows.sort(key=lambda item: item["period_end_date"])
    return grouped, {
        "raw_rows": raw_rows,
        "universe_rows": universe_rows,
        "deduped_rows": len(by_key),
        "duplicate_rows": duplicate_rows,
        "restated_rows": restated_rows,
    }


def _match_cost(event: dict[str, Any], period: dict[str, Any]) -> float:
    delta = (event["announce_date"] - period["period_end_date"]).days
    missing_session_penalty = 3.0 if event.get("announce_time") is None else 0.0
    return float(abs(delta - 30)) + missing_session_penalty


def _align_symbol(
    events: list[dict[str, Any]],
    periods: list[dict[str, Any]],
    *,
    max_lag_days: int,
    max_lead_days: int,
) -> tuple[list[tuple[int, int]], list[int], list[int]]:
    """Return a minimum-cost monotonic calendar-to-period alignment."""
    n_events = len(events)
    n_periods = len(periods)
    scores = [[0.0] * (n_periods + 1) for _ in range(n_events + 1)]
    choices = [[""] * (n_periods + 1) for _ in range(n_events + 1)]
    for event_index in range(n_events - 1, -1, -1):
        scores[event_index][n_periods] = (
            scores[event_index + 1][n_periods] + SKIP_EVENT_COST
        )
        choices[event_index][n_periods] = "event"
    for period_index in range(n_periods - 1, -1, -1):
        scores[n_events][period_index] = (
            scores[n_events][period_index + 1] + SKIP_PERIOD_COST
        )
        choices[n_events][period_index] = "period"

    for event_index in range(n_events - 1, -1, -1):
        for period_index in range(n_periods - 1, -1, -1):
            options = [
                (
                    scores[event_index + 1][period_index] + SKIP_EVENT_COST,
                    2,
                    "event",
                ),
                (
                    scores[event_index][period_index + 1] + SKIP_PERIOD_COST,
                    1,
                    "period",
                ),
            ]
            delta = (
                events[event_index]["announce_date"]
                - periods[period_index]["period_end_date"]
            ).days
            if -max_lead_days <= delta <= max_lag_days:
                options.append(
                    (
                        scores[event_index + 1][period_index + 1]
                        + _match_cost(events[event_index], periods[period_index]),
                        0,
                        "match",
                    )
                )
            score, _, choice = min(options)
            scores[event_index][period_index] = score
            choices[event_index][period_index] = choice

    matches: list[tuple[int, int]] = []
    unmatched_events: list[int] = []
    unmatched_periods: list[int] = []
    event_index = 0
    period_index = 0
    while event_index < n_events or period_index < n_periods:
        if event_index >= n_events:
            unmatched_periods.extend(range(period_index, n_periods))
            break
        if period_index >= n_periods:
            unmatched_events.extend(range(event_index, n_events))
            break
        choice = choices[event_index][period_index]
        if choice == "match":
            matches.append((event_index, period_index))
            event_index += 1
            period_index += 1
        elif choice == "period":
            unmatched_periods.append(period_index)
            period_index += 1
        else:
            unmatched_events.append(event_index)
            event_index += 1
    return matches, unmatched_events, unmatched_periods


def _ensure_schema(connection: sqlite3.Connection) -> None:
    connection.execute(EVENTS_DDL)
    columns = {
        str(row[1])
        for row in connection.execute("PRAGMA table_info(earnings_events)")
    }
    if "period_end_date" not in columns:
        connection.execute("ALTER TABLE earnings_events ADD COLUMN period_end_date TEXT")
    connection.execute(META_DDL)
    connection.execute(SURPRISE_HISTORY_DDL)


def _main() -> None:
    args = _parse_args()
    snapshot = Path(args.snapshot)
    calendar_csv = Path(args.calendar_csv)
    history_csv = Path(args.history_csv)
    for path in (snapshot, calendar_csv, history_csv):
        if not path.exists():
            raise SystemExit(f"Missing input: {path}")
    start = date.fromisoformat(args.from_date)
    end = date.fromisoformat(args.to_date)
    if start > end:
        raise SystemExit("--from-date must not be after --to-date")

    connection = sqlite3.connect(snapshot)
    try:
        universe = {
            _normalise_symbol(row[0])
            for row in connection.execute("SELECT symbol FROM tickers")
        }
    finally:
        connection.close()
    calendar, calendar_stats = _read_calendar(calendar_csv, universe, start, end)
    history, history_stats = _read_history(history_csv, universe)

    aligned_events: list[dict[str, Any]] = []
    pairing_deltas: list[int] = []
    unmatched_calendar = 0
    unmatched_periods_in_pairing_window = 0
    matched = 0
    for symbol in sorted(universe):
        events = calendar.get(symbol, [])
        lower = start - timedelta(days=int(args.max_period_lag_days))
        upper = end + timedelta(days=int(args.max_period_lead_days))
        periods = [
            row
            for row in history.get(symbol, [])
            if lower <= row["period_end_date"] <= upper
        ]
        matches, unmatched_events, unmatched_periods = _align_symbol(
            events,
            periods,
            max_lag_days=int(args.max_period_lag_days),
            max_lead_days=int(args.max_period_lead_days),
        )
        matched_by_event = {event_index: period_index for event_index, period_index in matches}
        matched += len(matches)
        unmatched_calendar += len(unmatched_events)
        unmatched_periods_in_pairing_window += len(unmatched_periods)
        for event_index, event in enumerate(events):
            row = dict(event)
            period_index = matched_by_event.get(event_index)
            if period_index is None:
                row.update(
                    {
                        "period_end_date": None,
                        "eps_actual": None,
                        "eps_estimate": None,
                    }
                )
            else:
                period = periods[period_index]
                row.update(
                    {
                        "period_end_date": period["period_end_date"],
                        "eps_actual": period["eps_actual"],
                        "eps_estimate": period["eps_estimate"],
                    }
                )
                pairing_deltas.append(
                    (event["announce_date"] - period["period_end_date"]).days
                )
            aligned_events.append(row)

    now = datetime.now(timezone.utc).isoformat()
    source = f"dolthub_post_no_preference@{args.source_commit}"
    conflicting_existing_rows = 0
    conflicting_existing_fields = 0
    preserved_existing_fields = 0
    incoming_keys = {
        (row["symbol"], row["announce_date"].isoformat()) for row in aligned_events
    }
    connection = sqlite3.connect(snapshot)
    try:
        _ensure_schema(connection)
        existing = {
            (str(row[0]), str(row[1])): row
            for row in connection.execute(
                """
                SELECT symbol, announce_date, announce_time, eps_estimate,
                       eps_actual, period_end_date, source
                FROM earnings_events
                WHERE announce_date BETWEEN ? AND ?
                """,
                (start.isoformat(), end.isoformat()),
            )
        }
        upsert = """
            INSERT INTO earnings_events(
                symbol, announce_date, announce_time, eps_estimate, eps_actual,
                revenue_estimate, revenue_actual, source, fetched_at, period_end_date
            ) VALUES (?, ?, ?, ?, ?, NULL, NULL, ?, ?, ?)
            ON CONFLICT(symbol, announce_date) DO UPDATE SET
                announce_time=COALESCE(earnings_events.announce_time, excluded.announce_time),
                eps_estimate=COALESCE(earnings_events.eps_estimate, excluded.eps_estimate),
                eps_actual=COALESCE(earnings_events.eps_actual, excluded.eps_actual),
                period_end_date=COALESCE(excluded.period_end_date, earnings_events.period_end_date),
                source=excluded.source,
                fetched_at=excluded.fetched_at
        """
        for row in aligned_events:
            key = (row["symbol"], row["announce_date"].isoformat())
            old = existing.get(key)
            retained = 0
            conflicts = 0
            if old is not None:
                old_values = {
                    "announce_time": old[2],
                    "eps_estimate": old[3],
                    "eps_actual": old[4],
                    "period_end_date": old[5],
                }
                new_values = {
                    "announce_time": row.get("announce_time"),
                    "eps_estimate": row.get("eps_estimate"),
                    "eps_actual": row.get("eps_actual"),
                    "period_end_date": (
                        row["period_end_date"].isoformat()
                        if row.get("period_end_date")
                        else None
                    ),
                }
                for field, new_value in new_values.items():
                    old_value = old_values[field]
                    if field != "period_end_date" and old_value is not None:
                        retained += 1
                    if new_value is not None and old_value is not None:
                        if field in {"eps_estimate", "eps_actual"}:
                            differs = not math.isclose(
                                float(new_value), float(old_value), rel_tol=0.0, abs_tol=1e-9
                            )
                        else:
                            differs = str(new_value) != str(old_value)
                        conflicts += int(differs)
                conflicting_existing_rows += int(conflicts > 0)
                conflicting_existing_fields += conflicts
                preserved_existing_fields += retained
            row_source = source
            if retained and old is not None:
                row_source = f"{old[6]}+calendar:{source}"
            connection.execute(
                upsert,
                (
                    row["symbol"],
                    row["announce_date"].isoformat(),
                    row.get("announce_time"),
                    row.get("eps_estimate"),
                    row.get("eps_actual"),
                    row_source,
                    now,
                    (
                        row["period_end_date"].isoformat()
                        if row.get("period_end_date")
                        else None
                    ),
                ),
            )

        history_upsert = """
            INSERT INTO earnings_surprise_history(
                symbol, period_end_date, eps_estimate, eps_actual, source, fetched_at
            ) VALUES (?, ?, ?, ?, ?, ?)
            ON CONFLICT(symbol, period_end_date) DO UPDATE SET
                eps_estimate=COALESCE(excluded.eps_estimate, earnings_surprise_history.eps_estimate),
                eps_actual=COALESCE(excluded.eps_actual, earnings_surprise_history.eps_actual),
                source=excluded.source,
                fetched_at=excluded.fetched_at
        """
        for symbol, rows in history.items():
            connection.executemany(
                history_upsert,
                [
                    (
                        symbol,
                        row["period_end_date"].isoformat(),
                        row.get("eps_estimate"),
                        row.get("eps_actual"),
                        source,
                        now,
                    )
                    for row in rows
                ],
            )

        for symbol in sorted(universe):
            count = int(
                connection.execute(
                    """
                    SELECT COUNT(*) FROM earnings_events
                    WHERE symbol=? AND announce_date BETWEEN ? AND ?
                    """,
                    (symbol, start.isoformat(), end.isoformat()),
                ).fetchone()[0]
            )
            connection.execute(
                """
                INSERT INTO earnings_backfill_meta(symbol, status, n_events, updated_at, note)
                VALUES (?, 'done', ?, ?, 'dolthub_bulk_complete')
                ON CONFLICT(symbol) DO UPDATE SET
                    status='done', n_events=excluded.n_events,
                    updated_at=excluded.updated_at, note=excluded.note
                """,
                (symbol, count, now),
            )
        connection.commit()

        params = (start.isoformat(), end.isoformat())
        total_events = int(
            connection.execute(
                """
                SELECT COUNT(*) FROM earnings_events
                WHERE symbol IN (SELECT symbol FROM tickers)
                  AND announce_date BETWEEN ? AND ?
                """,
                params,
            ).fetchone()[0]
        )
        paired_events = int(
            connection.execute(
                """
                SELECT COUNT(*) FROM earnings_events
                WHERE symbol IN (SELECT symbol FROM tickers)
                  AND announce_date BETWEEN ? AND ?
                  AND eps_actual IS NOT NULL AND eps_estimate IS NOT NULL
                """,
                params,
            ).fetchone()[0]
        )
        date_range = connection.execute(
            """
            SELECT MIN(announce_date), MAX(announce_date) FROM earnings_events
            WHERE symbol IN (SELECT symbol FROM tickers)
              AND announce_date BETWEEN ? AND ?
            """,
            params,
        ).fetchone()
        source_symbols = set(calendar)
        history_complete = int(
            connection.execute(
                """
                SELECT COUNT(*) FROM earnings_surprise_history
                WHERE symbol IN (SELECT symbol FROM tickers)
                  AND eps_actual IS NOT NULL AND eps_estimate IS NOT NULL
                """
            ).fetchone()[0]
        )
    finally:
        connection.close()

    deltas = sorted(pairing_deltas)
    summary = {
        "mode": "dolthub_public_bulk_clone",
        "window": {"from": start.isoformat(), "to": end.isoformat()},
        "coverage_amendment": {
            "approved_by_user": True,
            "reason": "FMP free tier blocks historical bulk earnings",
            "original_start": "2016-01-04",
            "amended_announcement_start": start.isoformat(),
        },
        "source": {
            "repository": args.source_url,
            "commit": args.source_commit,
            "license": "CC-BY-SA-4.0",
            "upstream_provider_documented": False,
        },
        "bulk_windows_total": 1,
        "bulk_windows_done": 1,
        "bulk_requests_logged_total": 1,
        "bulk_exports": 2,
        "calendar": calendar_stats,
        "eps_history": {**history_stats, "complete_actual_and_estimate": history_complete},
        "pairing": {
            "method": "minimum-cost monotonic alignment per symbol",
            "allowed_announce_minus_period_end_days": [
                -int(args.max_period_lead_days),
                int(args.max_period_lag_days),
            ],
            "matched_calendar_events": matched,
            "unmatched_calendar_events": unmatched_calendar,
            "unmatched_periods_in_pairing_window": unmatched_periods_in_pairing_window,
            "announce_minus_period_end_days": {
                "min": min(deltas) if deltas else None,
                "median": deltas[len(deltas) // 2] if deltas else None,
                "max": max(deltas) if deltas else None,
            },
            "pre_2020_eps_history_use": (
                "trailing_surprise_stdev_only; never treated as an announcement "
                "or live signal event"
            ),
        },
        "duplicate_rows_logged_total": (
            calendar_stats["duplicate_rows"] + history_stats["duplicate_rows"]
        ),
        "restated_rows_logged_total": (
            calendar_stats["restated_rows"]
            + history_stats["restated_rows"]
            + conflicting_existing_rows
        ),
        "conflicting_existing_rows": conflicting_existing_rows,
        "conflicting_existing_fields": conflicting_existing_fields,
        "preserved_existing_fields": preserved_existing_fields,
        "existing_enrichment_events_not_in_dolthub_calendar": max(
            0, total_events - len(incoming_keys)
        ),
        "dedupe_policy": (
            "UNIQUE(symbol, announce_date); normalise dot/dash symbols; retain one "
            "calendar row per key; preserve existing non-null session/EPS values from "
            "the prior FMP/Alpha Vantage partial backfill, then fill nulls and all "
            "remaining symbols from DoltHub; attach DoltHub period-end alignment"
        ),
        "events_in_window": total_events,
        "events_with_actual_and_estimate": paired_events,
        "symbols_done": len(universe),
        "symbols_universe": len(universe),
        "symbols_with_dolthub_calendar": len(source_symbols),
        "symbols_without_dolthub_calendar": sorted(universe - source_symbols),
        "announce_date_range": {"min": date_range[0], "max": date_range[1]},
        "complete": True,
    }
    output = Path(args.status_output)
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(summary, indent=2))
    print(f"Wrote {output}")


if __name__ == "__main__":
    _main()