Spaces:
Sleeping
Sleeping
| """Decide whether a street consensus may be compared to a reported figure. | |
| This module is ARCHITECTURE.md's alignment table expressed as code. Every | |
| ``return None, reason`` corresponds to one row of it. The default is refusal: | |
| a caller that cannot prove alignment renders nothing. | |
| The function deliberately says no far more often than yes. For EPS it will | |
| usually return ``basis_mismatch``, because yfinance publishes adjusted | |
| estimates while the reported figure comes from the GAAP XBRL concept. That is | |
| the correct answer, not a defect to engineer around. | |
| """ | |
| from __future__ import annotations | |
| import re | |
| from typing import Optional | |
| from storage.forecast_store import read_snapshots | |
| from storage.metrics_db import get_all_metrics | |
| # metrics.period labels: "Q{1,2,3,4}{YYYY}" for 10-Q rows, "FY{YYYY}" for | |
| # 10-K rows. Q4 never appears on a real row (it's absorbed into the 10-K), | |
| # but the label is still the correct thing to emit for a Q3 -> Q4 rollover. | |
| _QUARTER_PERIOD_RE = re.compile(r"^Q([1-4])(\d{4})$") | |
| # The accounting basis of each reported metric, as this repo extracts it. | |
| # ingestion/edgar.py pulls the us-gaap taxonomy only, so everything it writes | |
| # to the metrics table is GAAP. | |
| _REPORTED_BASIS: dict[str, str] = { | |
| "eps": "gaap", | |
| "revenue": "gaap", | |
| } | |
| _REPORTED_CURRENCY = "USD" | |
| def aligned_consensus( | |
| ticker: str, period: str, metric: str | |
| ) -> tuple[Optional[dict], str]: | |
| """Return ``(snapshot, "ALIGNED")`` or ``(None, reason)``. | |
| All five conditions must hold: | |
| 1. the capture is a live snapshot (our timestamp, not a provider's) | |
| 2. it was taken strictly before the report date | |
| 3. its fiscal period basis matches the reported row's | |
| 4. its accounting basis matches the reported metric's | |
| 5. its currency is USD | |
| """ | |
| reported = next( | |
| (r for r in get_all_metrics(ticker) if r.get("period") == period), None | |
| ) | |
| if reported is None: | |
| return None, "no_reported_period" | |
| report_date = reported.get("report_date") | |
| if not report_date: | |
| return None, "no_report_date" | |
| # 1. Only our own timestamped captures can prove a pre-release estimate. | |
| candidates = read_snapshots( | |
| ticker=ticker, | |
| target_period=period, | |
| metric=metric, | |
| capture_kind="live_snapshot", | |
| ) | |
| if not candidates: | |
| return None, "no_live_snapshot" | |
| # 2. Strictly before the release. Compare on the date prefix: captured_at | |
| # is a full UTC timestamp, report_date is a plain date. | |
| pre_release = [ | |
| s for s in candidates | |
| if str(s.get("captured_at", ""))[:10] < str(report_date)[:10] | |
| ] | |
| if not pre_release: | |
| return None, "no_pre_release_snapshot" | |
| # The consensus the market carried into the print is the last one struck | |
| # before it. | |
| snapshot = max(pre_release, key=lambda s: str(s.get("captured_at", ""))) | |
| # 3. Same fiscal period shape (quarterly vs TTM vs annual). | |
| if snapshot.get("period_basis") != reported.get("period_basis"): | |
| return None, "period_basis_mismatch" | |
| # 4. Same accounting basis. Adjusted estimates are not comparable to GAAP | |
| # reported figures, and this is the usual outcome for EPS. | |
| if snapshot.get("basis") != _REPORTED_BASIS.get(metric): | |
| return None, "basis_mismatch" | |
| # 5. Currency. See ingestion/edgar.py:96-98 — the reported side selects | |
| # USD units by construction; the estimate side must say so explicitly. | |
| if snapshot.get("currency") != _REPORTED_CURRENCY: | |
| return None, "currency_mismatch" | |
| return snapshot, "ALIGNED" | |
| def next_unreported_period(ticker: str) -> Optional[str]: | |
| """The fiscal period the street is currently estimating for `ticker`. | |
| Derived as the quarter after the most recently reported one. Returns None | |
| when the ticker has no parseable quarterly row, in which case the caller | |
| must not fetch: an estimate with no period label cannot be archived | |
| meaningfully. | |
| Ordering is computed on parsed (year, quarter) because the label format | |
| sorts incorrectly as a string across a year boundary ("Q12027" < "Q42026"). | |
| """ | |
| parsed: list[tuple[int, int]] = [] | |
| for row in get_all_metrics(ticker): | |
| match = _QUARTER_PERIOD_RE.match(str(row.get("period") or "")) | |
| if match: | |
| quarter, year = int(match.group(1)), int(match.group(2)) | |
| parsed.append((year, quarter)) | |
| if not parsed: | |
| return None | |
| year, quarter = max(parsed) | |
| if quarter == 4: | |
| year, quarter = year + 1, 1 | |
| else: | |
| quarter += 1 | |
| return f"Q{quarter}{year}" | |