Spaces:
Sleeping
Sleeping
File size: 3,507 Bytes
232b6f9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | """Backfill per-product / per-region results for filings already ingested.
``ingest.py`` now pulls the segment detail as it goes, but every filing
ingested before that lives in ``metrics.db`` with only its consolidated
figures. This walks those rows, fetches each filing's XBRL instance once and
fills the ``segments`` table.
One HTTP request per filing, ~1 MB each, so it is rate-limited to stay inside
the SEC's fair-access policy.
Usage::
python -m scripts.backfill_segments AAPL
python -m scripts.backfill_segments --all
python -m scripts.backfill_segments AAPL --dry-run
"""
from __future__ import annotations
import argparse
import sys
import time
from ingestion.segments import PRODUCT_AXIS, SEGMENT_AXIS, fetch_segments
from storage import metrics_db, segments_db
# The SEC asks for no more than 10 requests a second; one filing per second is
# far inside that and keeps a full backfill polite.
_PAUSE_SECONDS = 1.0
def _cik_and_doc(source_url: str) -> tuple[str, str]:
"""Pull the CIK and primary document out of a stored filing URL."""
try:
cik = source_url.split("/data/")[1].split("/")[0]
except (IndexError, AttributeError):
return "", ""
return cik, source_url.rsplit("/", 1)[-1]
def backfill(ticker: str, *, dry_run: bool = False) -> int:
rows = metrics_db.get_all_metrics(ticker.upper())
filings = [
r for r in rows
if r.get("accession") and r.get("source_url") and r.get("period_basis") != "annual"
]
if not filings:
print(f"{ticker}: no ingested filings carrying an accession")
return 1
total = failures = 0
for row in filings:
cik, primary_doc = _cik_and_doc(row["source_url"])
if not cik:
print(f" {row['period']}: unparseable source_url, skipped")
failures += 1
continue
try:
facts, url = fetch_segments(cik, row["accession"], primary_doc)
except Exception as exc: # noqa: BLE001 - one bad filing must not stop the run
print(f" {row['period']}: fetch failed ({exc})")
failures += 1
continue
products = sum(1 for f in facts if f.axis == PRODUCT_AXIS)
regions = sum(1 for f in facts if f.axis == SEGMENT_AXIS)
print(f" {row['period']}: {len(facts)} line(s) — {products} product, {regions} region")
if not dry_run:
total += segments_db.upsert_segments(
ticker, facts, row["accession"], url
)
time.sleep(_PAUSE_SECONDS)
if dry_run:
print(f"{ticker}: dry run, nothing written")
else:
print(f"{ticker}: {total} row(s) written")
return 1 if failures and not total else 0
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("ticker", nargs="?", help="Ticker to backfill.")
parser.add_argument("--all", action="store_true", help="Every ingested ticker.")
parser.add_argument("--dry-run", action="store_true", help="Fetch and report without writing.")
args = parser.parse_args()
if args.all:
tickers = metrics_db.list_tickers()
elif args.ticker:
tickers = [args.ticker.upper()]
else:
parser.error("give a ticker or --all")
failures = 0
for ticker in tickers:
print(f"\n== {ticker} ==")
failures += backfill(ticker, dry_run=args.dry_run)
return 1 if failures else 0
if __name__ == "__main__":
raise SystemExit(main())
|