Spaces:
Sleeping
Sleeping
File size: 4,887 Bytes
a95bfa1 6bdcc4d a95bfa1 6bdcc4d 5f932b3 a95bfa1 6bdcc4d a95bfa1 6bdcc4d 5f932b3 6bdcc4d ac0fe57 a95bfa1 49203e0 a95bfa1 | 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 104 105 106 107 108 109 110 111 112 113 114 | #!/usr/bin/env python
"""Capture today's street consensus for every covered ticker.
Run daily, after the US close. Each run appends one record per (ticker,
metric) to data/consensus_snapshots.jsonl, stamped with our own UTC capture
time. That timestamp is what later lets analytics/consensus.aligned_consensus
prove an estimate predates a release.
A gap in this archive cannot be backfilled from any public source, which is
why the job logs failures loudly and exits non-zero only when nothing at all
was captured.
The street doesn't publish a period label alongside its estimate — yfinance
only exposes relative codes ("0q", "+1q", "0y") — so this script derives the
label itself, as the quarter after the ticker's most recently reported one
(``analytics.consensus.next_unreported_period``). A ticker whose latest
reported filing is old enough that a newer one has probably already been
missed (see the staleness warning below) gets a label that is probably
wrong, so it is worth watching this output, not just its exit code.
"""
from __future__ import annotations
import sys
from datetime import datetime, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from analytics.consensus import next_unreported_period
from analytics.staleness import filing_staleness
from ingestion.analyst import fetch_analyst_estimates
from storage.forecast_store import CONSENSUS_PATH, read_snapshots
from storage.metrics_db import get_all_metrics, list_tickers
def main() -> int:
tickers = list_tickers()
if not tickers:
print("no covered tickers found in data/metrics.db", file=sys.stderr)
return 1
before = len(read_snapshots())
started = datetime.now(timezone.utc).isoformat()
ok, failed = [], []
for ticker in tickers:
period = next_unreported_period(ticker)
if period is None:
failed.append((ticker, "no_parseable_quarterly_period"))
print(
f" {ticker}: SKIPPED — no parseable quarterly period in "
f"metrics_db, refusing to archive an unlabeled estimate",
file=sys.stderr,
)
continue
# 120 days matches analytics/staleness.py::filing_staleness — roughly
# one quarter plus the usual filing delay. Beyond it, a newer
# quarterly report has probably already been missed, which means the
# period this script just derived is probably wrong for this ticker.
rows = get_all_metrics(ticker)
latest_report_date = rows[0].get("report_date") if rows else None
staleness = filing_staleness(latest_report_date)
if staleness["likely_newer_filing"]:
print(
f" {ticker}: WARNING — latest reported filing is "
f"{staleness['days_old']} days old (>120d); the derived "
f"target_period {period!r} is probably wrong for this ticker",
file=sys.stderr,
)
# snapshot_period is what opts this call into archiving. It is
# separate from target_period on purpose: only a caller that has
# derived the period the estimate is actually FOR (here, the next
# unreported quarter) may label an archive record, and this script
# is the only such caller. See ingestion/analyst.py's docstring.
payload, error = fetch_analyst_estimates(
ticker, target_period=period, snapshot_period=period
)
if payload is None:
failed.append((ticker, error))
print(f" {ticker}: FAILED — {error}", file=sys.stderr)
else:
ok.append(ticker)
print(f" {ticker}: captured for {payload.get('target_period')}")
written = len(read_snapshots()) - before
print(
f"\nrun {started}: {len(ok)} ok, {len(failed)} failed, "
f"{written} new records in {CONSENSUS_PATH}"
)
# The exit code tracks records WRITTEN, not fetches that returned. Those
# are different things: the append happens inside
# ingestion.analyst.record_consensus_snapshot, which swallows every
# exception and returns 0. Counting successful fetches meant a run whose
# every append failed still printed a success line per ticker and exited
# 0 -- a green check over an empty archive, on the one log whose missing
# days cannot be reconstructed from any public source. A day with nothing
# captured is a failed day, whatever the fetches did.
if not written:
print(
"captured 0 new records - treating this run as FAILED. Either "
"every append failed (see the [consensus capture] lines above) "
"or every snapshot for today is already in the archive.",
file=sys.stderr,
)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
|