research: Task 2 closed — SUE dead, earnings gap informational
Earnings backfill sourced from the public DoltHub earnings repo at a pinned commit rather than the FMP API: reproducible for anyone re-running the study, and it burns no request quota. 12,414 events, 98.6% of symbols with >=8 announcements, 99.2% paired actual/estimate, no keyed duplicates. 2a earnings-gap diagnostic: INFORMATIONAL, no filter shipped. The pre-earnings cohort's right tail was better, so the registered avoid-earnings condition failed. Note the raw 23/266 vs 115/574 incidence gap is largely a duration confound -- severe losses stop out fast and have less time to span an announcement -- so it is not evidence that holding through earnings is safe. 2b SUE: FAIL against the pre-registered +0.03 bar (unconditional IC +0.0151 over 56 reliable windows, momentum-conditional +0.0213). Signs stable across eras, so this is a clean null rather than an ambiguous one, consistent with post-earnings drift having decayed in large caps. Closes the Tier-1 arc: Task 1 dead on deep evidence, Task 2 dead here, Task 3 complete as diagnostic. No in-sample research thread remains open. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Claude Fable 5
+388
-405
@@ -1,15 +1,13 @@
|
||||
"""Backfill historical earnings into a snapshot ``earnings_events`` table.
|
||||
"""Bulk-only historical earnings backfill for a local SQLite snapshot.
|
||||
|
||||
Prefers FMP bulk date-range ``earnings-calendar`` (one request per window).
|
||||
On free-tier 402/403, falls back to per-symbol ``/stable/earnings`` with
|
||||
resume support and request counting (≈250 req/day free tier).
|
||||
The job uses FMP's date-range earnings-calendar endpoint. One request covers all
|
||||
symbols in a date window; per-symbol endpoints are intentionally not available
|
||||
in this task runner. Successful windows are committed independently so a later
|
||||
run resumes after a daily quota boundary without repeating completed windows.
|
||||
|
||||
Research only — writes to the local snapshot SQLite, never production Postgres.
|
||||
|
||||
Example
|
||||
-------
|
||||
python scripts/backfill_earnings_events.py \\
|
||||
--snapshot backtest_snapshots/prod.sqlite --limit 250
|
||||
Example:
|
||||
python scripts/backfill_earnings_events.py --snapshot backtest_snapshots/prod.sqlite \
|
||||
--from-date 2012-01-01 --window-days 30 --limit 250
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -17,10 +15,11 @@ from __future__ import annotations
|
||||
import argparse
|
||||
import asyncio
|
||||
import json
|
||||
import math
|
||||
import sys
|
||||
import time
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from sqlalchemy import create_engine, text
|
||||
@@ -34,7 +33,7 @@ from app.ssl_bootstrap import bootstrap_ssl # noqa: E402
|
||||
bootstrap_ssl()
|
||||
|
||||
FMP_STABLE = "https://financialmodelingprep.com/stable"
|
||||
DDL = """
|
||||
EVENTS_DDL = """
|
||||
CREATE TABLE IF NOT EXISTS earnings_events (
|
||||
id INTEGER PRIMARY KEY,
|
||||
symbol TEXT NOT NULL,
|
||||
@@ -49,7 +48,6 @@ CREATE TABLE IF NOT EXISTS earnings_events (
|
||||
UNIQUE(symbol, announce_date)
|
||||
)
|
||||
"""
|
||||
# Side table tracks which symbols have been fully pulled (resume).
|
||||
META_DDL = """
|
||||
CREATE TABLE IF NOT EXISTS earnings_backfill_meta (
|
||||
symbol TEXT PRIMARY KEY,
|
||||
@@ -59,239 +57,238 @@ CREATE TABLE IF NOT EXISTS earnings_backfill_meta (
|
||||
note TEXT
|
||||
)
|
||||
"""
|
||||
WINDOW_DDL = """
|
||||
CREATE TABLE IF NOT EXISTS earnings_backfill_windows (
|
||||
from_date TEXT NOT NULL,
|
||||
to_date TEXT NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
requests INTEGER NOT NULL DEFAULT 0,
|
||||
rows_raw INTEGER NOT NULL DEFAULT 0,
|
||||
rows_universe INTEGER NOT NULL DEFAULT 0,
|
||||
duplicate_rows INTEGER NOT NULL DEFAULT 0,
|
||||
restated_rows INTEGER NOT NULL DEFAULT 0,
|
||||
updated_at TEXT NOT NULL,
|
||||
note TEXT,
|
||||
PRIMARY KEY(from_date, to_date)
|
||||
)
|
||||
"""
|
||||
|
||||
|
||||
def _parse_args() -> argparse.Namespace:
|
||||
p = argparse.ArgumentParser(description=__doc__)
|
||||
p.add_argument("--snapshot", default="backtest_snapshots/prod.sqlite")
|
||||
p.add_argument(
|
||||
"--from-date",
|
||||
default="2020-01-01",
|
||||
help="Bulk calendar window start (also filters per-symbol rows).",
|
||||
)
|
||||
p.add_argument(
|
||||
"--to-date",
|
||||
default=None,
|
||||
help="Bulk calendar window end (default: today).",
|
||||
)
|
||||
p.add_argument(
|
||||
"--limit",
|
||||
type=int,
|
||||
default=250,
|
||||
help="Max FMP requests this run (free-tier cushion).",
|
||||
)
|
||||
p.add_argument("--sleep", type=float, default=0.35)
|
||||
p.add_argument(
|
||||
"--force-symbol",
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--snapshot", default="backtest_snapshots/prod.sqlite")
|
||||
parser.add_argument("--from-date", default="2012-01-01")
|
||||
parser.add_argument("--to-date", default=None)
|
||||
parser.add_argument("--window-days", type=int, default=30)
|
||||
parser.add_argument("--limit", type=int, default=250)
|
||||
parser.add_argument("--sleep", type=float, default=0.35)
|
||||
parser.add_argument(
|
||||
"--refetch-windows",
|
||||
action="store_true",
|
||||
help="Skip bulk attempt; go straight to per-symbol.",
|
||||
help="Re-fetch date windows already logged as done.",
|
||||
)
|
||||
p.add_argument(
|
||||
"--refetch-done",
|
||||
action="store_true",
|
||||
help="Re-fetch symbols already marked done.",
|
||||
)
|
||||
p.add_argument(
|
||||
"--provider",
|
||||
choices=("fmp", "alpha_vantage", "auto"),
|
||||
default="auto",
|
||||
help="Earnings provider. auto tries FMP bulk then FMP/AV per-symbol.",
|
||||
)
|
||||
return p.parse_args()
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def _ensure_tables(engine) -> None:
|
||||
with engine.begin() as conn:
|
||||
conn.execute(text(DDL))
|
||||
conn.execute(text(EVENTS_DDL))
|
||||
conn.execute(text(META_DDL))
|
||||
conn.execute(text(WINDOW_DDL))
|
||||
|
||||
|
||||
def _upsert_events(conn, rows: list[dict], source: str) -> int:
|
||||
if not rows:
|
||||
return 0
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
written = 0
|
||||
for r in rows:
|
||||
conn.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO earnings_events (
|
||||
symbol, announce_date, announce_time,
|
||||
eps_estimate, eps_actual, revenue_estimate, revenue_actual,
|
||||
source, fetched_at
|
||||
) VALUES (
|
||||
:symbol, :announce_date, :announce_time,
|
||||
:eps_estimate, :eps_actual, :revenue_estimate, :revenue_actual,
|
||||
:source, :fetched_at
|
||||
)
|
||||
ON CONFLICT(symbol, announce_date) DO UPDATE SET
|
||||
announce_time=excluded.announce_time,
|
||||
eps_estimate=excluded.eps_estimate,
|
||||
eps_actual=excluded.eps_actual,
|
||||
revenue_estimate=excluded.revenue_estimate,
|
||||
revenue_actual=excluded.revenue_actual,
|
||||
source=excluded.source,
|
||||
fetched_at=excluded.fetched_at
|
||||
"""
|
||||
),
|
||||
{
|
||||
"symbol": r["symbol"],
|
||||
"announce_date": r["announce_date"],
|
||||
"announce_time": r.get("announce_time"),
|
||||
"eps_estimate": r.get("eps_estimate"),
|
||||
"eps_actual": r.get("eps_actual"),
|
||||
"revenue_estimate": r.get("revenue_estimate"),
|
||||
"revenue_actual": r.get("revenue_actual"),
|
||||
"source": source,
|
||||
"fetched_at": now,
|
||||
},
|
||||
)
|
||||
written += 1
|
||||
return written
|
||||
def _number(value: Any) -> float | None:
|
||||
if value is None or value == "":
|
||||
return None
|
||||
try:
|
||||
result = float(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return result if math.isfinite(result) else None
|
||||
|
||||
|
||||
def _normalise_session(value: Any) -> str | None:
|
||||
if value is None:
|
||||
return None
|
||||
cleaned = str(value).strip().lower().replace("_", " ").replace("-", " ")
|
||||
aliases = {
|
||||
"bmo": "bmo",
|
||||
"before market open": "bmo",
|
||||
"before open": "bmo",
|
||||
"amc": "amc",
|
||||
"after market close": "amc",
|
||||
"after close": "amc",
|
||||
"during market hours": "during",
|
||||
"dmh": "during",
|
||||
}
|
||||
return aliases.get(cleaned, cleaned or None)
|
||||
|
||||
|
||||
def _parse_bulk_item(item: dict) -> dict | None:
|
||||
sym = (item.get("symbol") or "").strip().upper()
|
||||
d = item.get("date") or item.get("earningsDate")
|
||||
if not sym or not d:
|
||||
symbol = str(item.get("symbol") or "").strip().upper().replace(".", "-")
|
||||
raw_date = item.get("date") or item.get("earningsDate")
|
||||
if not symbol or not raw_date:
|
||||
return None
|
||||
return {
|
||||
"symbol": sym.replace(".", "-"),
|
||||
"announce_date": str(d)[:10],
|
||||
"announce_time": item.get("time") or item.get("announceTime"),
|
||||
"eps_estimate": _f(item.get("epsEstimated") or item.get("estimatedEarning")),
|
||||
"eps_actual": _f(item.get("epsActual") or item.get("eps")),
|
||||
"revenue_estimate": _f(item.get("revenueEstimated")),
|
||||
"revenue_actual": _f(item.get("revenueActual")),
|
||||
"symbol": symbol,
|
||||
"announce_date": str(raw_date)[:10],
|
||||
"announce_time": _normalise_session(
|
||||
item.get("time") or item.get("announceTime")
|
||||
),
|
||||
"eps_estimate": _number(
|
||||
item.get("epsEstimated")
|
||||
if item.get("epsEstimated") is not None
|
||||
else item.get("estimatedEarning")
|
||||
),
|
||||
"eps_actual": _number(
|
||||
item.get("epsActual")
|
||||
if item.get("epsActual") is not None
|
||||
else item.get("eps")
|
||||
),
|
||||
"revenue_estimate": _number(item.get("revenueEstimated")),
|
||||
"revenue_actual": _number(item.get("revenueActual")),
|
||||
}
|
||||
|
||||
|
||||
def _parse_symbol_item(item: dict, symbol: str) -> dict | None:
|
||||
d = item.get("date")
|
||||
if not d:
|
||||
return None
|
||||
return {
|
||||
"symbol": symbol.replace(".", "-").upper(),
|
||||
"announce_date": str(d)[:10],
|
||||
"announce_time": item.get("time"),
|
||||
"eps_estimate": _f(item.get("epsEstimated")),
|
||||
"eps_actual": _f(item.get("epsActual")),
|
||||
"revenue_estimate": _f(item.get("revenueEstimated")),
|
||||
"revenue_actual": _f(item.get("revenueActual")),
|
||||
}
|
||||
def _windows(start: date, end: date, window_days: int) -> list[tuple[date, date]]:
|
||||
if window_days < 1:
|
||||
raise ValueError("window_days must be positive")
|
||||
result: list[tuple[date, date]] = []
|
||||
cursor = start
|
||||
while cursor <= end:
|
||||
window_end = min(end, cursor + timedelta(days=window_days - 1))
|
||||
result.append((cursor, window_end))
|
||||
cursor = window_end + timedelta(days=1)
|
||||
return result
|
||||
|
||||
|
||||
def _f(v) -> float | None:
|
||||
if v is None or v == "":
|
||||
return None
|
||||
try:
|
||||
return float(v)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
def _dedupe_bulk_rows(rows: list[dict]) -> tuple[list[dict], int, int]:
|
||||
"""Prefer the most complete duplicate; use the later row as the tie-break."""
|
||||
fields = (
|
||||
"announce_time",
|
||||
"eps_estimate",
|
||||
"eps_actual",
|
||||
"revenue_estimate",
|
||||
"revenue_actual",
|
||||
)
|
||||
chosen: dict[tuple[str, str], dict] = {}
|
||||
duplicate_extras = 0
|
||||
restated = 0
|
||||
for row in rows:
|
||||
key = (str(row["symbol"]), str(row["announce_date"]))
|
||||
previous = chosen.get(key)
|
||||
if previous is None:
|
||||
chosen[key] = row
|
||||
continue
|
||||
duplicate_extras += 1
|
||||
if any(
|
||||
previous.get(field) is not None
|
||||
and row.get(field) is not None
|
||||
and previous.get(field) != row.get(field)
|
||||
for field in fields
|
||||
):
|
||||
restated += 1
|
||||
previous_score = sum(previous.get(field) is not None for field in fields)
|
||||
new_score = sum(row.get(field) is not None for field in fields)
|
||||
if new_score >= previous_score:
|
||||
chosen[key] = row
|
||||
return list(chosen.values()), duplicate_extras, restated
|
||||
|
||||
|
||||
async def _try_bulk(
|
||||
client: httpx.AsyncClient,
|
||||
api_key: str,
|
||||
def _upsert_events(conn, rows: list[dict]) -> int:
|
||||
if not rows:
|
||||
return 0
|
||||
fetched_at = datetime.now(timezone.utc).isoformat()
|
||||
statement = text(
|
||||
"""
|
||||
INSERT INTO earnings_events (
|
||||
symbol, announce_date, announce_time, eps_estimate, eps_actual,
|
||||
revenue_estimate, revenue_actual, source, fetched_at
|
||||
) VALUES (
|
||||
:symbol, :announce_date, :announce_time, :eps_estimate, :eps_actual,
|
||||
:revenue_estimate, :revenue_actual, 'fmp_earnings_calendar', :fetched_at
|
||||
)
|
||||
ON CONFLICT(symbol, announce_date) DO UPDATE SET
|
||||
announce_time=COALESCE(excluded.announce_time, earnings_events.announce_time),
|
||||
eps_estimate=COALESCE(excluded.eps_estimate, earnings_events.eps_estimate),
|
||||
eps_actual=COALESCE(excluded.eps_actual, earnings_events.eps_actual),
|
||||
revenue_estimate=COALESCE(excluded.revenue_estimate, earnings_events.revenue_estimate),
|
||||
revenue_actual=COALESCE(excluded.revenue_actual, earnings_events.revenue_actual),
|
||||
source=excluded.source,
|
||||
fetched_at=excluded.fetched_at
|
||||
"""
|
||||
)
|
||||
conn.execute(statement, [{**row, "fetched_at": fetched_at} for row in rows])
|
||||
return len(rows)
|
||||
|
||||
|
||||
async def _fetch_bulk_window(
|
||||
client: httpx.AsyncClient, api_key: str, start: date, end: date
|
||||
) -> tuple[list[dict], int, str | None]:
|
||||
response = await client.get(
|
||||
f"{FMP_STABLE}/earnings-calendar",
|
||||
params={"from": start.isoformat(), "to": end.isoformat(), "apikey": api_key},
|
||||
)
|
||||
if response.status_code in (402, 403):
|
||||
return [], response.status_code, "bulk_endpoint_unavailable"
|
||||
if response.status_code == 429:
|
||||
return [], response.status_code, "daily_limit_reached"
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
if not isinstance(payload, list):
|
||||
return [], response.status_code, f"unexpected_payload:{type(payload).__name__}"
|
||||
rows = []
|
||||
for item in payload:
|
||||
if isinstance(item, dict):
|
||||
parsed = _parse_bulk_item(item)
|
||||
if parsed:
|
||||
rows.append(parsed)
|
||||
return rows, response.status_code, None
|
||||
|
||||
|
||||
def _write_window_status(
|
||||
engine,
|
||||
*,
|
||||
start: date,
|
||||
end: date,
|
||||
*,
|
||||
window_days: int = 30,
|
||||
) -> tuple[list[dict], int, str | None]:
|
||||
"""Return (rows, requests_used, error_note)."""
|
||||
rows: list[dict] = []
|
||||
reqs = 0
|
||||
cur = start
|
||||
while cur <= end:
|
||||
win_end = min(end, cur + timedelta(days=window_days - 1))
|
||||
resp = await client.get(
|
||||
f"{FMP_STABLE}/earnings-calendar",
|
||||
params={
|
||||
"from": cur.isoformat(),
|
||||
"to": win_end.isoformat(),
|
||||
"apikey": api_key,
|
||||
status: str,
|
||||
raw_n: int = 0,
|
||||
universe_n: int = 0,
|
||||
duplicate_n: int = 0,
|
||||
restated_n: int = 0,
|
||||
note: str | None = None,
|
||||
) -> None:
|
||||
with engine.begin() as conn:
|
||||
conn.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO earnings_backfill_windows(
|
||||
from_date, to_date, status, requests, rows_raw, rows_universe,
|
||||
duplicate_rows, restated_rows, updated_at, note
|
||||
) VALUES (:a, :b, :status, 1, :raw, :uni, :dup, :rest, :now, :note)
|
||||
ON CONFLICT(from_date, to_date) DO UPDATE SET
|
||||
status=excluded.status,
|
||||
requests=earnings_backfill_windows.requests + 1,
|
||||
rows_raw=excluded.rows_raw,
|
||||
rows_universe=excluded.rows_universe,
|
||||
duplicate_rows=excluded.duplicate_rows,
|
||||
restated_rows=excluded.restated_rows,
|
||||
updated_at=excluded.updated_at,
|
||||
note=excluded.note
|
||||
"""
|
||||
),
|
||||
{
|
||||
"a": start.isoformat(),
|
||||
"b": end.isoformat(),
|
||||
"status": status,
|
||||
"raw": raw_n,
|
||||
"uni": universe_n,
|
||||
"dup": duplicate_n,
|
||||
"rest": restated_n,
|
||||
"now": datetime.now(timezone.utc).isoformat(),
|
||||
"note": note,
|
||||
},
|
||||
)
|
||||
reqs += 1
|
||||
if resp.status_code in (402, 403):
|
||||
return [], reqs, f"bulk_unavailable status={resp.status_code}"
|
||||
if resp.status_code == 429:
|
||||
return rows, reqs, "rate_limited"
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
if not isinstance(data, list):
|
||||
return [], reqs, f"unexpected bulk payload type={type(data)}"
|
||||
for item in data:
|
||||
if isinstance(item, dict):
|
||||
parsed = _parse_bulk_item(item)
|
||||
if parsed:
|
||||
rows.append(parsed)
|
||||
cur = win_end + timedelta(days=1)
|
||||
return rows, reqs, None
|
||||
|
||||
|
||||
async def _fetch_symbol(
|
||||
client: httpx.AsyncClient, api_key: str, symbol: str
|
||||
) -> list[dict]:
|
||||
resp = await client.get(
|
||||
f"{FMP_STABLE}/earnings",
|
||||
params={"symbol": symbol, "apikey": api_key},
|
||||
)
|
||||
if resp.status_code == 429:
|
||||
raise RuntimeError("rate_limited")
|
||||
if resp.status_code == 402:
|
||||
return []
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
if not isinstance(data, list):
|
||||
return []
|
||||
out: list[dict] = []
|
||||
for item in data:
|
||||
if isinstance(item, dict):
|
||||
parsed = _parse_symbol_item(item, symbol)
|
||||
if parsed:
|
||||
out.append(parsed)
|
||||
return out
|
||||
|
||||
|
||||
async def _fetch_symbol_alpha_vantage(
|
||||
client: httpx.AsyncClient, api_key: str, symbol: str
|
||||
) -> list[dict]:
|
||||
"""Alpha Vantage EARNINGS — includes reportedDate (announce) + estimate/actual."""
|
||||
resp = await client.get(
|
||||
"https://www.alphavantage.co/query",
|
||||
params={"function": "EARNINGS", "symbol": symbol, "apikey": api_key},
|
||||
)
|
||||
if resp.status_code == 429:
|
||||
raise RuntimeError("rate_limited")
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
if not isinstance(data, dict):
|
||||
return []
|
||||
note = str(data.get("Note") or data.get("Information") or "")
|
||||
if "rate limit" in note.lower() or "Thank you for using Alpha Vantage" in note:
|
||||
raise RuntimeError("rate_limited")
|
||||
if data.get("Error Message"):
|
||||
return []
|
||||
quarterly = data.get("quarterlyEarnings") or []
|
||||
out: list[dict] = []
|
||||
for item in quarterly:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
# Prefer announce (reportedDate); fall back to fiscal end (worse PIT).
|
||||
ad = item.get("reportedDate") or item.get("fiscalDateEnding")
|
||||
if not ad:
|
||||
continue
|
||||
out.append({
|
||||
"symbol": symbol.replace(".", "-").upper(),
|
||||
"announce_date": str(ad)[:10],
|
||||
"announce_time": item.get("reportTime"),
|
||||
"eps_estimate": _f(item.get("estimatedEPS")),
|
||||
"eps_actual": _f(item.get("reportedEPS")),
|
||||
"revenue_estimate": None,
|
||||
"revenue_actual": None,
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
async def _main() -> None:
|
||||
@@ -304,228 +301,214 @@ async def _main() -> None:
|
||||
|
||||
if not settings.fmp_api_key:
|
||||
raise SystemExit("FMP_API_KEY required")
|
||||
|
||||
start = date.fromisoformat(args.from_date)
|
||||
end = date.fromisoformat(args.to_date) if args.to_date else date.today()
|
||||
engine = create_engine(
|
||||
f"sqlite:///{snapshot.resolve().as_posix()}",
|
||||
future=True,
|
||||
)
|
||||
_ensure_tables(engine)
|
||||
if start > end:
|
||||
raise SystemExit("--from-date must not be after --to-date")
|
||||
|
||||
engine = create_engine(f"sqlite:///{snapshot.resolve().as_posix()}", future=True)
|
||||
_ensure_tables(engine)
|
||||
all_windows = _windows(start, end, int(args.window_days))
|
||||
with engine.connect() as conn:
|
||||
symbols = [
|
||||
str(r[0]).upper().replace(".", "-")
|
||||
for r in conn.execute(text("SELECT symbol FROM tickers ORDER BY symbol"))
|
||||
str(row[0]).upper().replace(".", "-")
|
||||
for row in conn.execute(text("SELECT symbol FROM tickers ORDER BY symbol"))
|
||||
]
|
||||
done = set()
|
||||
if not args.refetch_done:
|
||||
done = {
|
||||
str(r[0])
|
||||
for r in conn.execute(
|
||||
text(
|
||||
"SELECT symbol FROM earnings_backfill_meta "
|
||||
"WHERE status='done' AND n_events > 0"
|
||||
)
|
||||
completed = {
|
||||
(str(row[0]), str(row[1]))
|
||||
for row in conn.execute(
|
||||
text(
|
||||
"SELECT from_date, to_date FROM earnings_backfill_windows "
|
||||
"WHERE status='done'"
|
||||
)
|
||||
}
|
||||
|
||||
pending = [s for s in symbols if s not in done]
|
||||
print(f"Snapshot: {snapshot}")
|
||||
print(f"Universe: {len(symbols)}; pending: {len(pending)}; done: {len(done)}")
|
||||
print(f"Window filter: {start} → {end}")
|
||||
print(f"Provider: {args.provider}")
|
||||
|
||||
req_budget = int(args.limit)
|
||||
reqs_used = 0
|
||||
events_written = 0
|
||||
mode = "per_symbol"
|
||||
use_av = args.provider in ("alpha_vantage", "auto") and bool(
|
||||
getattr(settings, "alpha_vantage_api_key", "")
|
||||
)
|
||||
use_fmp = args.provider in ("fmp", "auto") and bool(settings.fmp_api_key)
|
||||
|
||||
async with httpx.AsyncClient(timeout=60.0) as client:
|
||||
if (
|
||||
not args.force_symbol
|
||||
and req_budget > 0
|
||||
and use_fmp
|
||||
and args.provider != "alpha_vantage"
|
||||
):
|
||||
print("Attempting bulk earnings-calendar…")
|
||||
bulk_rows, bulk_reqs, err = await _try_bulk(
|
||||
client, settings.fmp_api_key, start, end
|
||||
)
|
||||
reqs_used += bulk_reqs
|
||||
if err:
|
||||
print(f" Bulk unavailable: {err} (requests={bulk_reqs})")
|
||||
else:
|
||||
# Filter to universe.
|
||||
uni = set(symbols)
|
||||
bulk_rows = [r for r in bulk_rows if r["symbol"] in uni]
|
||||
with engine.begin() as conn:
|
||||
events_written += _upsert_events(conn, bulk_rows, "fmp_earnings_calendar")
|
||||
for sym in symbols:
|
||||
n = conn.execute(
|
||||
text(
|
||||
"SELECT COUNT(*) FROM earnings_events WHERE symbol=:s"
|
||||
),
|
||||
{"s": sym},
|
||||
).scalar_one()
|
||||
conn.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO earnings_backfill_meta(symbol, status, n_events, updated_at, note)
|
||||
VALUES (:s, 'done', :n, :t, 'bulk')
|
||||
ON CONFLICT(symbol) DO UPDATE SET
|
||||
status='done', n_events=excluded.n_events,
|
||||
updated_at=excluded.updated_at, note=excluded.note
|
||||
"""
|
||||
),
|
||||
{
|
||||
"s": sym,
|
||||
"n": int(n),
|
||||
"t": datetime.now(timezone.utc).isoformat(),
|
||||
},
|
||||
)
|
||||
mode = "bulk"
|
||||
print(f" Bulk wrote {events_written} events; requests={bulk_reqs}")
|
||||
pending = []
|
||||
}
|
||||
pending = [
|
||||
window
|
||||
for window in all_windows
|
||||
if args.refetch_windows
|
||||
or (window[0].isoformat(), window[1].isoformat()) not in completed
|
||||
]
|
||||
universe = set(symbols)
|
||||
print(f"Snapshot: {snapshot}")
|
||||
print(f"Universe: {len(symbols)} symbols")
|
||||
print(f"Window: {start} -> {end}")
|
||||
print(
|
||||
f"Bulk windows: {len(all_windows)} total; "
|
||||
f"{len(all_windows) - len(pending)} done; {len(pending)} pending"
|
||||
)
|
||||
print("Provider: FMP bulk earnings-calendar only")
|
||||
|
||||
# Per-symbol fallback / completion.
|
||||
fmp_limited = False
|
||||
for sym in pending:
|
||||
if reqs_used >= req_budget:
|
||||
print(f"Request budget exhausted ({req_budget}). Resume later.")
|
||||
requests_this_run = 0
|
||||
rows_upserted = 0
|
||||
duplicate_rows = 0
|
||||
restated_rows = 0
|
||||
stop_note: str | None = None
|
||||
async with httpx.AsyncClient(timeout=60.0) as client:
|
||||
for index, (window_start, window_end) in enumerate(pending, 1):
|
||||
if requests_this_run >= int(args.limit):
|
||||
stop_note = "request_budget_exhausted"
|
||||
break
|
||||
items: list[dict] = []
|
||||
source = "fmp_earnings"
|
||||
note = "per_symbol"
|
||||
try:
|
||||
if use_fmp and not fmp_limited and args.provider != "alpha_vantage":
|
||||
items = await _fetch_symbol(client, settings.fmp_api_key, sym)
|
||||
source = "fmp_earnings"
|
||||
note = "fmp_per_symbol"
|
||||
# Empty list may mean soft-limit or no data — try AV if available.
|
||||
if not items and use_av:
|
||||
items = await _fetch_symbol_alpha_vantage(
|
||||
client, settings.alpha_vantage_api_key, sym
|
||||
)
|
||||
source = "alpha_vantage_earnings"
|
||||
note = "av_after_fmp_empty"
|
||||
reqs_used += 1 # count AV call separately below too
|
||||
elif use_av:
|
||||
items = await _fetch_symbol_alpha_vantage(
|
||||
client, settings.alpha_vantage_api_key, sym
|
||||
)
|
||||
source = "alpha_vantage_earnings"
|
||||
note = "av_per_symbol"
|
||||
else:
|
||||
raise RuntimeError("no provider available")
|
||||
raw_rows, status_code, error = await _fetch_bulk_window(
|
||||
client, settings.fmp_api_key, window_start, window_end
|
||||
)
|
||||
except Exception as exc:
|
||||
msg = str(exc)
|
||||
print(f" FAIL {sym}: {msg}")
|
||||
reqs_used += 1
|
||||
if "rate_limited" in msg and note.startswith("fmp"):
|
||||
fmp_limited = True
|
||||
with engine.begin() as conn:
|
||||
raw_rows, status_code = [], 0
|
||||
error = f"request_error:{type(exc).__name__}:{exc}"
|
||||
requests_this_run += 1
|
||||
if error:
|
||||
_write_window_status(
|
||||
engine,
|
||||
start=window_start,
|
||||
end=window_end,
|
||||
status="error",
|
||||
note=f"http={status_code} {error}"[:300],
|
||||
)
|
||||
stop_note = error
|
||||
print(
|
||||
f"STOP {window_start}..{window_end}: {error} "
|
||||
f"(http={status_code}, request={requests_this_run})"
|
||||
)
|
||||
break
|
||||
|
||||
in_universe = [row for row in raw_rows if row["symbol"] in universe]
|
||||
deduped, duplicate_n, restated_n = _dedupe_bulk_rows(in_universe)
|
||||
with engine.begin() as conn:
|
||||
rows_upserted += _upsert_events(conn, deduped)
|
||||
_write_window_status(
|
||||
engine,
|
||||
start=window_start,
|
||||
end=window_end,
|
||||
status="done",
|
||||
raw_n=len(raw_rows),
|
||||
universe_n=len(deduped),
|
||||
duplicate_n=duplicate_n,
|
||||
restated_n=restated_n,
|
||||
note="bulk",
|
||||
)
|
||||
duplicate_rows += duplicate_n
|
||||
restated_rows += restated_n
|
||||
if index == 1 or index % 10 == 0 or index == len(pending):
|
||||
print(
|
||||
f"progress windows={index}/{len(pending)} "
|
||||
f"requests={requests_this_run}/{args.limit} "
|
||||
f"last={window_start}..{window_end} rows={len(deduped)}"
|
||||
)
|
||||
if args.sleep > 0:
|
||||
await asyncio.sleep(float(args.sleep))
|
||||
|
||||
with engine.begin() as conn:
|
||||
windows_done = int(
|
||||
conn.execute(
|
||||
text(
|
||||
"SELECT COUNT(*) FROM earnings_backfill_windows "
|
||||
"WHERE status='done' AND from_date >= :a AND to_date <= :b"
|
||||
),
|
||||
{"a": start.isoformat(), "b": end.isoformat()},
|
||||
).scalar_one()
|
||||
)
|
||||
complete = windows_done >= len(all_windows)
|
||||
if complete:
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
for symbol in symbols:
|
||||
count = int(
|
||||
conn.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO earnings_backfill_meta(symbol, status, n_events, updated_at, note)
|
||||
VALUES (:s, 'error', 0, :t, :n)
|
||||
ON CONFLICT(symbol) DO UPDATE SET
|
||||
status='error', updated_at=excluded.updated_at, note=excluded.note
|
||||
"""
|
||||
"SELECT COUNT(*) FROM earnings_events "
|
||||
"WHERE symbol=:symbol AND announce_date BETWEEN :a AND :b"
|
||||
),
|
||||
{
|
||||
"s": sym,
|
||||
"t": datetime.now(timezone.utc).isoformat(),
|
||||
"n": msg[:200],
|
||||
},
|
||||
)
|
||||
if args.sleep > 0:
|
||||
await asyncio.sleep(args.sleep)
|
||||
continue
|
||||
|
||||
reqs_used += 1
|
||||
# Keep all rows with dates on/before end — SUE needs trailing history.
|
||||
filtered = [
|
||||
r for r in items if r["announce_date"] <= end.isoformat()
|
||||
]
|
||||
# Do NOT mark empty as done — leave pending for another provider/day.
|
||||
status = "done" if filtered else "empty"
|
||||
with engine.begin() as conn:
|
||||
n_w = _upsert_events(conn, filtered, source) if filtered else 0
|
||||
events_written += n_w
|
||||
{"symbol": symbol, "a": start.isoformat(), "b": end.isoformat()},
|
||||
).scalar_one()
|
||||
)
|
||||
conn.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO earnings_backfill_meta(symbol, status, n_events, updated_at, note)
|
||||
VALUES (:s, :st, :n, :t, :note)
|
||||
VALUES (:symbol, 'done', :count, :now, 'bulk_complete')
|
||||
ON CONFLICT(symbol) DO UPDATE SET
|
||||
status=excluded.status, n_events=excluded.n_events,
|
||||
updated_at=excluded.updated_at, note=excluded.note
|
||||
status='done', n_events=excluded.n_events,
|
||||
updated_at=excluded.updated_at, note=excluded.note
|
||||
"""
|
||||
),
|
||||
{
|
||||
"s": sym,
|
||||
"st": status,
|
||||
"n": len(filtered),
|
||||
"t": datetime.now(timezone.utc).isoformat(),
|
||||
"note": note,
|
||||
},
|
||||
{"symbol": symbol, "count": count, "now": now},
|
||||
)
|
||||
if reqs_used % 10 == 0 or reqs_used == 1:
|
||||
print(
|
||||
f" progress reqs={reqs_used}/{req_budget} last={sym} "
|
||||
f"events_batch={len(filtered)} src={source}"
|
||||
)
|
||||
# AV free tier is ~5/min or 25/day — be polite when using it.
|
||||
sleep_s = float(args.sleep)
|
||||
if source.startswith("alpha_vantage"):
|
||||
sleep_s = max(sleep_s, 12.0)
|
||||
if sleep_s > 0:
|
||||
await asyncio.sleep(sleep_s)
|
||||
|
||||
with engine.connect() as conn:
|
||||
params = {"a": start.isoformat(), "b": end.isoformat()}
|
||||
total_events = int(
|
||||
conn.execute(text("SELECT COUNT(*) FROM earnings_events")).scalar_one()
|
||||
conn.execute(
|
||||
text(
|
||||
"SELECT COUNT(*) FROM earnings_events "
|
||||
"WHERE symbol IN (SELECT symbol FROM tickers) "
|
||||
"AND announce_date BETWEEN :a AND :b"
|
||||
),
|
||||
params,
|
||||
).scalar_one()
|
||||
)
|
||||
done_n = int(
|
||||
paired_events = int(
|
||||
conn.execute(
|
||||
text(
|
||||
"SELECT COUNT(*) FROM earnings_events "
|
||||
"WHERE symbol IN (SELECT symbol FROM tickers) "
|
||||
"AND announce_date BETWEEN :a AND :b "
|
||||
"AND eps_actual IS NOT NULL AND eps_estimate IS NOT NULL"
|
||||
),
|
||||
params,
|
||||
).scalar_one()
|
||||
)
|
||||
date_range = conn.execute(
|
||||
text(
|
||||
"SELECT MIN(announce_date), MAX(announce_date) FROM earnings_events "
|
||||
"WHERE symbol IN (SELECT symbol FROM tickers) "
|
||||
"AND announce_date BETWEEN :a AND :b"
|
||||
),
|
||||
params,
|
||||
).fetchone()
|
||||
done_symbols = int(
|
||||
conn.execute(
|
||||
text("SELECT COUNT(*) FROM earnings_backfill_meta WHERE status='done'")
|
||||
).scalar_one()
|
||||
)
|
||||
d_range = conn.execute(
|
||||
text("SELECT MIN(announce_date), MAX(announce_date) FROM earnings_events")
|
||||
totals = conn.execute(
|
||||
text(
|
||||
"SELECT COALESCE(SUM(requests),0), COALESCE(SUM(duplicate_rows),0), "
|
||||
"COALESCE(SUM(restated_rows),0) FROM earnings_backfill_windows "
|
||||
"WHERE from_date >= :a AND to_date <= :b"
|
||||
),
|
||||
params,
|
||||
).fetchone()
|
||||
with_actual = int(
|
||||
conn.execute(
|
||||
text(
|
||||
"SELECT COUNT(*) FROM earnings_events "
|
||||
"WHERE eps_actual IS NOT NULL AND eps_estimate IS NOT NULL"
|
||||
)
|
||||
).scalar_one()
|
||||
)
|
||||
|
||||
summary = {
|
||||
"mode": mode,
|
||||
"fmp_requests": reqs_used,
|
||||
"events_written_this_run": events_written,
|
||||
"total_events": total_events,
|
||||
"symbols_done": done_n,
|
||||
"mode": "fmp_bulk_date_range_only",
|
||||
"window": {"from": start.isoformat(), "to": end.isoformat()},
|
||||
"window_days": int(args.window_days),
|
||||
"bulk_windows_total": len(all_windows),
|
||||
"bulk_windows_done": windows_done,
|
||||
"bulk_requests_this_run": requests_this_run,
|
||||
"bulk_requests_logged_total": int(totals[0]),
|
||||
"rows_upserted_this_run": rows_upserted,
|
||||
"duplicate_rows_this_run": duplicate_rows,
|
||||
"restated_rows_this_run": restated_rows,
|
||||
"duplicate_rows_logged_total": int(totals[1]),
|
||||
"restated_rows_logged_total": int(totals[2]),
|
||||
"dedupe_policy": (
|
||||
"UNIQUE(symbol, announce_date); prefer more non-null fields, then "
|
||||
"the provider's later occurrence; non-null bulk fields replace prior "
|
||||
"values while null bulk fields retain existing values"
|
||||
),
|
||||
"events_in_window": total_events,
|
||||
"events_with_actual_and_estimate": paired_events,
|
||||
"symbols_done": done_symbols,
|
||||
"symbols_universe": len(symbols),
|
||||
"announce_date_range": {"min": d_range[0], "max": d_range[1]},
|
||||
"events_with_actual_and_estimate": with_actual,
|
||||
"budget": req_budget,
|
||||
"complete": done_n >= len(symbols),
|
||||
"announce_date_range": {"min": date_range[0], "max": date_range[1]},
|
||||
"request_budget": int(args.limit),
|
||||
"stop_note": stop_note,
|
||||
"complete": complete,
|
||||
}
|
||||
output = Path("reports/earnings-backfill-status.json")
|
||||
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))
|
||||
out = Path("reports") / "earnings-backfill-status.json"
|
||||
out.parent.mkdir(parents=True, exist_ok=True)
|
||||
out.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
|
||||
print(f"Wrote {out}")
|
||||
print(f"Wrote {output}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user