#!/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())