scripts/backfill_consensus.py

#!/usr/bin/env python
"""One-off import of historical consensus EPS from Alpha Vantage.

These records are explicitly weaker evidence than the daily live snapshots:
Alpha Vantage supplies the estimate but no timestamp for it, so nothing here
can prove an estimate predates the release it is compared against. They are
marked capture_kind="backfill_at_report" and are refused by
analytics.consensus.aligned_consensus by design (it filters on
capture_kind == "live_snapshot" before any other test).

Labelling. Alpha Vantage's quarterly entries don't carry this repo's
"Q{n}{YYYY}" period label, and the label can't be derived from the calendar:
this repo's fiscal quarters don't line up with calendar quarters for every
issuer (AAPL and NVDA both run offset fiscal years), so calendar-quarter
arithmetic on fiscalDateEnding mislabels them. Instead each Alpha Vantage
entry's fiscalDateEnding is matched against metrics.report_date for the same
ticker.

NOTE on metrics.report_date: despite the name, this column does NOT hold an
announcement/filing date — it holds the fiscal PERIOD END date (it comes
from SEC's filings.recent.reportDate, see ingestion/edgar.py:68-80). That is
why the join below uses Alpha Vantage's fiscalDateEnding (also a period-end
date) rather than its reportedDate (an announcement date, which sits ~30
days later and essentially never lines up with metrics.report_date — an
earlier version of this script joined on reportedDate and matched 0 of 617
real records; measured drift after switching to fiscalDateEnding was 0-5
days across the six covered tickers).

The join is further restricted to metrics rows whose period looks like
"Q{1-4}{YYYY}" (a real 10-Q row). Alpha Vantage's quarterlyEarnings entries
are quarterly by definition, so an "FY{YYYY}" (10-K/annual) row is never a
valid partner even when its report_date happens to fall inside the match
window — a fiscal Q4 often shares (or nearly shares) its period end with the
fiscal year end, and without this restriction that produces a category
error: a quarterly estimate mislabelled with an annual period. A fiscal Q4
quarter has no 10-Q row in this repo's period scheme at all, so it correctly
finds no partner and keeps the date-string fallback.

  - an exact date match, or the nearest one within 7 days (5 measured
    maximum across tickers, plus margin for 52/53-week fiscal calendars),
    against a QUARTERLY row only, borrows that row's period verbatim
    (target_period_source = "matched_metrics_report_date" — kept as-is even
    though the column is a period end, not a report date, to stay
    consistent with records already written under that source string);
  - otherwise the record falls back to Alpha Vantage's own fiscalDateEnding
    string as the identifier (target_period_source =
    "alphavantage_fiscal_date_ending"). metrics.db holds far fewer rows per
    ticker than Alpha Vantage returns quarters, so this is the majority case.
    A date string can never collide with a "Q{n}{YYYY}" label, so
    aligned_consensus finds no reported row for it and refuses on
    "no_reported_period" — but the (estimate, actual, report date) triple is
    still preserved in the archive for the calibration backtest, which needs
    it more than it needs the label to resolve.

Idempotent: snapshot_id is derived from the OBSERVED identity of the
estimate -- ticker, Alpha Vantage's fiscalDateEnding, metric, provider,
captured_at -- and never from target_period, which is derived and moves.
target_period starts as the fiscalDateEnding string and becomes a
"Q{n}{YYYY}" label as soon as ingest.py adds the matching quarterly row,
so keying identity on it would mint a new id for the same observation at
the next quarterly ingest and double-write it under two labels. Keyed on
the period end instead, a re-run produces identical ids, which the store
skips. Note the consequence: a re-run after an ingest recognises the
record as already present and writes nothing at all, so the archived
target_period stays whatever it was when first written. That is the
correct behaviour for an append-only log -- period_end_date is on every
backfill record, so a consumer that wants the current label joins on
that rather than trusting a label frozen at write time.
"""
from __future__ import annotations

import re
import sys
from datetime import date
from pathlib import Path
from typing import Optional

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from dotenv import load_dotenv

load_dotenv()

from ingestion.alphavantage import fetch_earnings
from storage.forecast_store import append_snapshots, snapshot_id
from storage.metrics_db import get_all_metrics, list_tickers

# 7 days: 5 measured maximum drift between Alpha Vantage's fiscalDateEnding
# and metrics.report_date across the six covered tickers, plus margin for
# 52/53-week fiscal calendars.
_MATCH_WINDOW_DAYS = 7

# Only a real 10-Q row is a valid match partner for a quarterly Alpha
# Vantage estimate. "FY{YYYY}" (10-K/annual) rows are excluded even when
# their report_date falls inside the window — see the module docstring.
_QUARTERLY_PERIOD_RE = re.compile(r"^Q[1-4]\d{4}$")


def _parse_date(value: Optional[str]) -> Optional[date]:
    if not value:
        return None
    try:
        return date.fromisoformat(str(value)[:10])
    except ValueError:
        return None


def _match_period(metrics_rows: list[dict], fiscal_date_ending: Optional[str]) -> Optional[str]:
    """Return the quarterly metrics row's period whose report_date (actually
    a fiscal period-end date — see the NOTE in this module's docstring)
    exactly matches ``fiscal_date_ending``, or, failing that, the nearest
    quarterly row within _MATCH_WINDOW_DAYS. Non-quarterly rows (annual
    "FY{YYYY}" rows) are never considered. Returns None if no quarterly row
    is close enough (or fiscal_date_ending is missing)."""
    target = _parse_date(fiscal_date_ending)
    if target is None:
        return None

    best_period: Optional[str] = None
    best_diff: Optional[int] = None
    for row in metrics_rows:
        period = row.get("period")
        if not period or not _QUARTERLY_PERIOD_RE.match(period):
            continue
        row_date = _parse_date(row.get("report_date"))
        if row_date is None:
            continue
        diff = abs((row_date - target).days)
        if diff == 0:
            return period
        if diff <= _MATCH_WINDOW_DAYS and (best_diff is None or diff < best_diff):
            best_diff = diff
            best_period = period
    return best_period


def _target_period(
    metrics_rows: list[dict], fiscal_date_ending: str
) -> tuple[str, str]:
    """Return (target_period, target_period_source)."""
    matched = _match_period(metrics_rows, fiscal_date_ending)
    if matched is not None:
        return matched, "matched_metrics_report_date"
    return fiscal_date_ending, "alphavantage_fiscal_date_ending"


def backfill(ticker: str) -> tuple[int, int, int]:
    """Import historical consensus EPS for one ticker.

    Returns (written, matched_count, fallback_count)."""
    # fetch_earnings returns (data, error), not a bare dict.
    payload, error = fetch_earnings(ticker)
    if not payload:
        print(f"  {ticker}: no earnings payload — {error}", file=sys.stderr)
        return 0, 0, 0

    metrics_rows = get_all_metrics(ticker)

    records = []
    matched_count = 0
    fallback_count = 0
    for quarter in payload.get("quarterlyEarnings", []):
        estimated = quarter.get("estimatedEPS")
        ending = quarter.get("fiscalDateEnding")
        reported_date = quarter.get("reportedDate")
        if estimated in (None, "None", "") or not ending:
            continue
        try:
            value = float(estimated)
        except (TypeError, ValueError):
            continue

        period, source = _target_period(metrics_rows, ending)
        if source == "matched_metrics_report_date":
            matched_count += 1
        else:
            fallback_count += 1

        # captured_at is the report date: the latest moment this estimate could
        # have been observed. It is deliberately NOT treated as pre-release.
        captured_at = reported_date or ending
        records.append({
            # Identity comes from fiscalDateEnding (stored below as
            # period_end_date), NOT from target_period. target_period is
            # DERIVED: it is this same date string until a matching quarterly
            # row lands in metrics.db, at which point it becomes
            # "Q{n}{YYYY}". ingest.py adds such rows every quarter, so an id
            # keyed on the label would change under a re-run and append the
            # same observation a second time under the other name -- the
            # duplication Ruling 13 had to clean by hand. fiscalDateEnding is
            # an observed provider fact and never moves, which leaves
            # target_period free to improve without forking the record.
            "snapshot_id": snapshot_id(
                ticker, ending, "eps", "alphavantage", captured_at
            ),
            "ticker": ticker.upper(),
            "target_period": period,
            "target_period_source": source,
            "period_end_date": ending,
            "metric": "eps",
            "value": value,
            "basis": "adjusted",
            "currency": "USD",
            "period_basis": "quarter",
            "provider": "alphavantage",
            "provider_period_code": None,
            "captured_at": captured_at,
            "capture_kind": "backfill_at_report",
        })

    written = append_snapshots(records)
    print(
        f"  {ticker}: {written} new of {len(records)} historical quarters "
        f"({matched_count} matched, {fallback_count} fallback)"
    )
    return written, matched_count, fallback_count


def main() -> int:
    tickers = list_tickers()
    if not tickers:
        print("no covered tickers found", file=sys.stderr)
        return 1
    total = 0
    total_matched = 0
    total_fallback = 0
    for t in tickers:
        written, matched_count, fallback_count = backfill(t)
        total += written
        total_matched += matched_count
        total_fallback += fallback_count
    print(
        f"\nbackfilled {total} records "
        f"({total_matched} matched, {total_fallback} fallback)"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())