From b0537ebe9afd975a510fcd3f0d84597424473787 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Fri, 24 Jul 2026 14:19:22 +0200 Subject: [PATCH] Implement A5 fundamentals cutover activation --- app/scheduler.py | 86 ++++- app/services/admin_service.py | 2 +- .../fundamental_data_refresh_service.py | 179 +++++++++++ .../fundamentals_candidate_service.py | 288 +++++++++++++++++ app/services/fundamentals_parity_service.py | 130 +------- docs/dolt-integration-plan.md | 19 +- docs/fundamentals-deployment.md | 77 ++++- tests/unit/test_fundamental_data_refresh.py | 293 ++++++++++++++++++ tests/unit/test_scheduler.py | 90 ++++++ 9 files changed, 1023 insertions(+), 141 deletions(-) create mode 100644 app/services/fundamental_data_refresh_service.py create mode 100644 app/services/fundamentals_candidate_service.py create mode 100644 tests/unit/test_fundamental_data_refresh.py diff --git a/app/scheduler.py b/app/scheduler.py index 24db4d2..e35ad07 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -41,6 +41,7 @@ from app.services import ( settings_store, shadow_book_service, fundamentals_parity_service, + fundamental_data_refresh_service, ) from app.services.data_import import STATUS_FAILED, SourceImporter, run_import from app.services.dolt_earnings_importer import DoltEarningsImporter @@ -652,7 +653,7 @@ async def run_shadow_book() -> None: if not await _is_job_enabled(db, job_name): _log_event(logging.INFO, "job_skipped", job=job_name, reason="disabled") _runtime_finish(job_name, "skipped", processed=0, total=1, message="Disabled") - return + return False if not await shadow_book_service.is_enabled(db): _log_event(logging.INFO, "job_skipped", job=job_name, reason="not enabled in settings") _runtime_finish(job_name, "skipped", processed=0, total=1, message="Not enabled") @@ -924,8 +925,13 @@ async def collect_fundamentals() -> None: # --------------------------------------------------------------------------- -async def _run_shadow_import(job_name: str, importer: SourceImporter) -> None: - """Run one source importer and surface its audit result in Admin → Jobs.""" +async def _run_shadow_import(job_name: str, importer: SourceImporter) -> bool: + """Run an importer and return whether its scheduled job was enabled. + + The SEC wrapper uses the return value to run its activated local cache step + after failed, no-op, promoted, or source-locked attempts while still honoring + the job-level disable switch. + """ _log_event(logging.INFO, "job_start", job=job_name) _runtime_start(job_name, total=1) @@ -941,7 +947,7 @@ async def _run_shadow_import(job_name: str, importer: SourceImporter) -> None: message = "Another import for this source is already running" _log_event(logging.INFO, "job_skipped", job=job_name, reason="source_locked") _runtime_finish(job_name, "skipped", processed=0, total=1, message=message) - return + return True revision = f" · {run.revision[:12]}" if run.revision else "" message = f"{run.status}{revision}" @@ -949,7 +955,7 @@ async def _run_shadow_import(job_name: str, importer: SourceImporter) -> None: message = run.error_details or message _log_event(logging.ERROR, "job_error", job=job_name, message=message) _runtime_finish(job_name, "error", processed=0, total=1, message=message) - return + return True _log_event( logging.INFO, @@ -959,6 +965,7 @@ async def _run_shadow_import(job_name: str, importer: SourceImporter) -> None: revision=run.revision, ) _runtime_finish(job_name, "completed", processed=1, total=1, message=message) + return True except asyncio.CancelledError: _runtime_finish(job_name, "error", processed=0, total=1, message="Cancelled") raise @@ -971,6 +978,7 @@ async def _run_shadow_import(job_name: str, importer: SourceImporter) -> None: message=str(exc), ) _runtime_finish(job_name, "error", processed=0, total=1, message=str(exc)) + return True async def run_dolt_earnings_import() -> None: @@ -979,8 +987,67 @@ async def run_dolt_earnings_import() -> None: async def run_sec_fundamentals_import() -> None: - """Import tracked-universe SEC facts in shadow.""" - await _run_shadow_import("sec_fundamentals_import", SecFundamentalsImporter()) + """Import SEC facts, then run the activated local compat-cache refresh. + + The refresh is deliberately separate from the network import result. Once + activated it therefore still runs from stored snapshots/earnings/prices when + SEC is unavailable, unchanged, or another SEC import owns the source lock. + """ + job_name = "sec_fundamentals_import" + job_enabled = await _run_shadow_import(job_name, SecFundamentalsImporter()) + if not job_enabled: + return + + try: + async with async_session_factory() as db: + summary = await fundamental_data_refresh_service.refresh_if_enabled(db) + except asyncio.CancelledError: + _runtime_finish( + job_name, "error", processed=0, total=1, message="Cancelled" + ) + raise + except Exception as exc: + message = f"Local fundamental_data refresh failed: {exc}" + _log_event( + logging.ERROR, + "fundamental_data_refresh_error", + job=job_name, + error_type=type(exc).__name__, + message=str(exc), + ) + _runtime_finish(job_name, "error", processed=0, total=1, message=message) + return + + if not summary["enabled"]: + _log_event( + logging.INFO, + "fundamental_data_refresh_skipped", + job=job_name, + reason="cutover_disabled", + setting=fundamental_data_refresh_service.ACTIVATION_KEY, + ) + return + + _log_event( + logging.INFO, + "fundamental_data_refresh_complete", + job=job_name, + **summary, + ) + runtime = get_job_runtime_snapshot(job_name) + if runtime.get("status") == "completed": + import_message = runtime.get("message") or "import completed" + cache_message = ( + f"cache {summary['refreshed']} · " + f"{summary['score_inputs_changed']} score inputs changed" + ) + _runtime_finish( + job_name, + "completed", + processed=1, + total=1, + message=f"{import_message} · {cache_message}", + ) async def run_fundamentals_parity_report() -> None: @@ -1566,7 +1633,8 @@ SCHEDULE_DEFAULTS: dict[str, str] = { "schedule_timezone": "America/New_York", # Morning data/display refresh (no qualifying R:R scan). "schedule_daily_pipeline_cron": "0 2 * * *", - # Shadow source imports. They never write legacy fundamental_data before A5. + # Bulk source imports. The SEC job writes the legacy compat cache only after + # the explicit, default-off A5 cutover setting is enabled. "schedule_dolt_earnings_cron": "30 2 * * *", "schedule_sec_fundamentals_cron": "0 4 * * *", "schedule_fundamentals_parity_cron": "30 5 * * *", @@ -1689,7 +1757,7 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None: "schedule_sec_fundamentals_cron", ), id="sec_fundamentals_import", - name="SEC Fundamentals Import (shadow)", + name="SEC Fundamentals Import", replace_existing=True, ) scheduler.add_job( diff --git a/app/services/admin_service.py b/app/services/admin_service.py index 4cf2424..f7f850b 100644 --- a/app/services/admin_service.py +++ b/app/services/admin_service.py @@ -637,7 +637,7 @@ JOB_LABELS = { "sentiment_collector": "Sentiment Collector", "fundamental_collector": "Fundamental Collector", "dolt_earnings_import": "Dolt Earnings Import (shadow)", - "sec_fundamentals_import": "SEC Fundamentals Import (shadow)", + "sec_fundamentals_import": "SEC Fundamentals Import", "fundamentals_parity_report": "Fundamentals Parity Report (read-only)", "rr_scanner": "R:R Scanner", "ticker_universe_sync": "Ticker Universe Sync", diff --git a/app/services/fundamental_data_refresh_service.py b/app/services/fundamental_data_refresh_service.py new file mode 100644 index 0000000..a171e3a --- /dev/null +++ b/app/services/fundamental_data_refresh_service.py @@ -0,0 +1,179 @@ +"""A5 activation: refresh the legacy fundamentals cache from local bulk data.""" + +from __future__ import annotations + +import json +from datetime import date, datetime, timezone +from typing import Any + +from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import insert_for_session +from app.models.fundamental import FundamentalData +from app.models.score import CompositeScore, DimensionScore +from app.services import fundamentals_candidate_service, settings_store + + +# Absence is deliberately false. Production activation therefore requires one +# explicit, durable SystemSetting change after the A5 evidence is approved. +ACTIVATION_KEY = "fundamental_data_sec_dolt_cutover_enabled" +_SCORE_FIELDS = ("pe_ratio", "revenue_growth", "earnings_surprise") + + +async def is_enabled(db: AsyncSession) -> bool: + raw = await settings_store.get_value(db, ACTIVATION_KEY, "false") + return str(raw).strip().lower() == "true" + + +async def refresh_if_enabled( + db: AsyncSession, + *, + now: datetime | None = None, + today: date | None = None, +) -> dict[str, Any]: + """Refresh atomically when activated; otherwise perform no writes.""" + if not await is_enabled(db): + return { + "enabled": False, + "refreshed": 0, + "score_inputs_changed": 0, + "dimension_scores_staled": 0, + "composite_scores_staled": 0, + } + return await refresh(db, now=now, today=today) + + +async def refresh( + db: AsyncSession, + *, + now: datetime | None = None, + today: date | None = None, +) -> dict[str, Any]: + """Replace every ticker's compat-cache row in one database transaction. + + Candidate values are assembled before the first write and use only local + PostgreSQL tables. A failure rolls the whole refresh back. Only changes to + the three scoring inputs invalidate cached scores; market cap and the next + earnings date are display-only. + """ + refreshed_at = now or datetime.now(timezone.utc) + candidates = await fundamentals_candidate_service.build_candidates( + db, today=today + ) + ticker_ids = [candidate.ticker_id for candidate in candidates] + existing = await _existing_by_ticker(db, ticker_ids) + changed_ids = { + candidate.ticker_id + for candidate in candidates + if _score_inputs_changed(existing.get(candidate.ticker_id), candidate) + } + + for candidate in candidates: + unavailable_json = json.dumps( + candidate.unavailable_fields, sort_keys=True + ) + stmt = insert_for_session(db, FundamentalData).values( + ticker_id=candidate.ticker_id, + pe_ratio=candidate.pe_ratio, + revenue_growth=candidate.revenue_growth, + earnings_surprise=candidate.earnings_surprise, + market_cap=candidate.market_cap, + next_earnings_date=candidate.next_earnings_date, + fetched_at=refreshed_at, + unavailable_fields_json=unavailable_json, + ) + await db.execute( + stmt.on_conflict_do_update( + index_elements=["ticker_id"], + set_={ + "pe_ratio": stmt.excluded.pe_ratio, + "revenue_growth": stmt.excluded.revenue_growth, + "earnings_surprise": stmt.excluded.earnings_surprise, + "market_cap": stmt.excluded.market_cap, + "next_earnings_date": stmt.excluded.next_earnings_date, + "fetched_at": stmt.excluded.fetched_at, + "unavailable_fields_json": ( + stmt.excluded.unavailable_fields_json + ), + }, + ) + ) + + dimension_ids = await _fundamental_dimension_ids(db, changed_ids) + composite_ids = await _composite_ids(db, changed_ids) + if dimension_ids: + await db.execute( + update(DimensionScore) + .where(DimensionScore.ticker_id.in_(dimension_ids)) + .values(is_stale=True) + ) + if composite_ids: + await db.execute( + update(CompositeScore) + .where(CompositeScore.ticker_id.in_(composite_ids)) + .values(is_stale=True) + ) + + await db.commit() + return { + "enabled": True, + "refreshed": len(candidates), + "score_inputs_changed": len(changed_ids), + "dimension_scores_staled": len(dimension_ids), + "composite_scores_staled": len(composite_ids), + } + + +async def _existing_by_ticker( + db: AsyncSession, ticker_ids: list[int] +) -> dict[int, FundamentalData]: + if not ticker_ids: + return {} + rows = ( + await db.execute( + select(FundamentalData).where( + FundamentalData.ticker_id.in_(ticker_ids) + ) + ) + ).scalars() + return {row.ticker_id: row for row in rows} + + +async def _fundamental_dimension_ids( + db: AsyncSession, ticker_ids: set[int] +) -> set[int]: + if not ticker_ids: + return set() + rows = await db.execute( + select(DimensionScore.ticker_id).where( + DimensionScore.ticker_id.in_(ticker_ids), + DimensionScore.dimension == "fundamental", + ) + ) + return set(rows.scalars()) + + +async def _composite_ids( + db: AsyncSession, ticker_ids: set[int] +) -> set[int]: + if not ticker_ids: + return set() + rows = await db.execute( + select(CompositeScore.ticker_id).where( + CompositeScore.ticker_id.in_(ticker_ids) + ) + ) + return set(rows.scalars()) + + +def _score_inputs_changed( + existing: FundamentalData | None, + candidate: fundamentals_candidate_service.CandidateFundamentals, +) -> bool: + if existing is None: + return True + return any( + getattr(existing, field) != getattr(candidate, field) + for field in _SCORE_FIELDS + ) diff --git a/app/services/fundamentals_candidate_service.py b/app/services/fundamentals_candidate_service.py new file mode 100644 index 0000000..03d61f1 --- /dev/null +++ b/app/services/fundamentals_candidate_service.py @@ -0,0 +1,288 @@ +"""Local SEC/Dolt candidate values for the legacy fundamentals cache. + +This is the single read path shared by the A5 parity report and the activated +``fundamental_data`` refresh. It never contacts SEC or Dolt: every input comes +from PostgreSQL, so price- and earnings-driven values can still refresh when an +upstream import is unchanged or unavailable. +""" + +from __future__ import annotations + +import math +from collections import defaultdict +from dataclasses import dataclass, field +from datetime import date, datetime +from typing import Any +from zoneinfo import ZoneInfo + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.earnings_event import EarningsEvent +from app.models.fundamental_snapshot import FundamentalSnapshot +from app.models.ohlcv import OHLCVRecord +from app.models.ticker import Ticker +from app.services import fundamentals_derivation as deriv + + +@dataclass(frozen=True) +class CandidateFundamentals: + ticker_id: int + symbol: str + cik: str | None + pe_ratio: float | None + revenue_growth: float | None + earnings_surprise: float | None + market_cap: float | None + next_earnings_date: date | None + price_date: date | None + unavailable_fields: dict[str, str] = field(default_factory=dict) + + +async def build_candidates( + db: AsyncSession, + *, + today: date | None = None, +) -> list[CandidateFundamentals]: + """Derive current cache candidates using only already-stored data.""" + today = today or datetime.now(ZoneInfo("America/New_York")).date() + tickers = list( + (await db.execute(select(Ticker).order_by(Ticker.symbol))).scalars() + ) + if not tickers: + return [] + + ticker_ids = [ticker.id for ticker in tickers] + ciks = sorted({ticker.cik for ticker in tickers if ticker.cik}) + derived_by_cik = await _derived_by_cik(db, ciks) + closes_by_ticker = await _latest_closes(db, ticker_ids) + surprise_by_ticker, next_by_ticker = await _earnings_values( + db, ticker_ids, today + ) + + out: list[CandidateFundamentals] = [] + for ticker in tickers: + derived = derived_by_cik.get(ticker.cik) if ticker.cik else None + close = closes_by_ticker.get(ticker.id) + price = close[0] if close is not None else None + price_date = close[1] if close is not None else None + growth_series = ( + derived.metrics.get("revenue_growth_yoy") + if derived is not None + else None + ) + + pe_ratio = ( + _pe(price, derived.ttm_diluted_eps) + if derived is not None + else None + ) + revenue_growth = ( + float(growth_series.value) + if growth_series is not None and _finite(growth_series.value) + else None + ) + earnings_surprise = surprise_by_ticker.get(ticker.id) + market_cap = ( + _market_cap(price, derived.shares_outstanding) + if derived is not None + else None + ) + next_earnings_date = next_by_ticker.get(ticker.id) + + out.append( + CandidateFundamentals( + ticker_id=ticker.id, + symbol=ticker.symbol, + cik=ticker.cik, + pe_ratio=pe_ratio, + revenue_growth=revenue_growth, + earnings_surprise=earnings_surprise, + market_cap=market_cap, + next_earnings_date=next_earnings_date, + price_date=price_date, + unavailable_fields=_availability_metadata( + derived=derived, + price=price, + pe_ratio=pe_ratio, + revenue_growth=revenue_growth, + earnings_surprise=earnings_surprise, + market_cap=market_cap, + next_earnings_date=next_earnings_date, + ), + ) + ) + return out + + +async def _derived_by_cik( + db: AsyncSession, ciks: list[str] +) -> dict[str, deriv.DerivedFundamentals]: + if not ciks: + return {} + grouped: dict[str, list[FundamentalSnapshot]] = defaultdict(list) + rows = ( + await db.execute( + select(FundamentalSnapshot).where(FundamentalSnapshot.cik.in_(ciks)) + ) + ).scalars() + for row in rows: + grouped[row.cik].append(row) + return {cik: deriv.derive(grouped.get(cik, [])) for cik in ciks} + + +async def _latest_closes( + db: AsyncSession, ticker_ids: list[int] +) -> dict[int, tuple[float, date]]: + latest = ( + select( + OHLCVRecord.ticker_id, + func.max(OHLCVRecord.date).label("max_date"), + ) + .where(OHLCVRecord.ticker_id.in_(ticker_ids)) + .group_by(OHLCVRecord.ticker_id) + .subquery() + ) + rows = ( + await db.execute( + select( + OHLCVRecord.ticker_id, + OHLCVRecord.close, + OHLCVRecord.date, + ).join( + latest, + (OHLCVRecord.ticker_id == latest.c.ticker_id) + & (OHLCVRecord.date == latest.c.max_date), + ) + ) + ).all() + return { + ticker_id: (float(close), close_date) + for ticker_id, close, close_date in rows + if _finite(close) + } + + +async def _earnings_values( + db: AsyncSession, + ticker_ids: list[int], + today: date, +) -> tuple[dict[int, float], dict[int, date]]: + rows = ( + await db.execute( + select(EarningsEvent) + .where(EarningsEvent.ticker_id.in_(ticker_ids)) + .order_by(EarningsEvent.ticker_id, EarningsEvent.announce_date.desc()) + ) + ).scalars() + surprises: dict[int, float] = {} + upcoming: dict[int, date] = {} + for row in rows: + if row.announce_date >= today: + current = upcoming.get(row.ticker_id) + if current is None or row.announce_date < current: + upcoming[row.ticker_id] = row.announce_date + continue + if row.ticker_id in surprises: + continue + surprise = _surprise(row.eps_estimate, row.eps_actual) + if surprise is not None: + surprises[row.ticker_id] = surprise + return surprises, upcoming + + +def _availability_metadata( + *, + derived: deriv.DerivedFundamentals | None, + price: float | None, + pe_ratio: float | None, + revenue_growth: float | None, + earnings_surprise: float | None, + market_cap: float | None, + next_earnings_date: date | None, +) -> dict[str, str]: + metadata: dict[str, str] = {} + + if pe_ratio is not None: + metadata["source_pe_ratio"] = "sec_facts+ohlcv_records" + elif derived is None or derived.latest_period_end is None: + metadata["pe_ratio"] = "no SEC fundamental snapshots" + elif not _finite(price) or price <= 0: + metadata["pe_ratio"] = "no usable PostgreSQL close" + elif derived.ttm_diluted_eps_caveat: + metadata["pe_ratio"] = derived.ttm_diluted_eps_caveat + else: + metadata["pe_ratio"] = "no positive SEC-derived TTM diluted EPS" + + if revenue_growth is not None: + metadata["source_revenue_growth"] = "sec_facts" + else: + metadata["revenue_growth"] = "SEC-derived TTM revenue growth unavailable" + + if earnings_surprise is not None: + metadata["source_earnings_surprise"] = "dolt_earnings" + else: + metadata["earnings_surprise"] = ( + "no completed earnings event with actual and nonzero estimate" + ) + + if market_cap is not None: + metadata["source_market_cap"] = "sec_facts+ohlcv_records" + if derived is not None and derived.shares_outstanding_estimated: + metadata["market_cap_estimated"] = ( + "shares use the SEC weighted-average diluted fallback" + ) + elif derived is None or derived.latest_period_end is None: + metadata["market_cap"] = "no SEC fundamental snapshots" + elif not _finite(price) or price <= 0: + metadata["market_cap"] = "no usable PostgreSQL close" + else: + metadata["market_cap"] = "SEC-derived shares outstanding unavailable" + + if next_earnings_date is not None: + metadata["source_next_earnings_date"] = "dolt_earnings" + else: + metadata["next_earnings_date"] = "no upcoming earnings event" + return metadata + + +def _surprise( + estimate: float | None, + actual: float | None, +) -> float | None: + if not _finite(estimate) or not _finite(actual) or estimate == 0: + return None + return (float(actual) - float(estimate)) / abs(float(estimate)) * 100.0 + + +def _pe(price: float | None, ttm_eps: float | None) -> float | None: + if ( + not _finite(price) + or price <= 0 + or not _finite(ttm_eps) + or ttm_eps <= 0 + ): + return None + return float(price) / float(ttm_eps) + + +def _market_cap( + price: float | None, + shares_outstanding: float | None, +) -> float | None: + if ( + not _finite(price) + or price <= 0 + or not _finite(shares_outstanding) + or shares_outstanding <= 0 + ): + return None + return float(price) * float(shares_outstanding) + + +def _finite(value: Any) -> bool: + return ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(value) + ) diff --git a/app/services/fundamentals_parity_service.py b/app/services/fundamentals_parity_service.py index 8b99f42..de43060 100644 --- a/app/services/fundamentals_parity_service.py +++ b/app/services/fundamentals_parity_service.py @@ -14,22 +14,17 @@ import json import math import os import statistics -from collections import defaultdict from datetime import date, datetime, timezone from pathlib import Path from typing import Any, Iterable from zoneinfo import ZoneInfo -from sqlalchemy import func, select, text +from sqlalchemy import select, text from sqlalchemy.ext.asyncio import AsyncSession from app.models.data_import_run import DataImportRun -from app.models.earnings_event import EarningsEvent from app.models.fundamental import FundamentalData -from app.models.fundamental_snapshot import FundamentalSnapshot -from app.models.ohlcv import OHLCVRecord -from app.models.ticker import Ticker -from app.services import fundamentals_derivation as deriv +from app.services import fundamentals_candidate_service as candidate_service REPORT_VERSION = 1 APPROVAL_STATUS = "pending_explicit_approval" @@ -94,33 +89,18 @@ async def build_report( ) await connection.execute(text("SET TRANSACTION READ ONLY")) - tickers = list((await db.execute(select(Ticker).order_by(Ticker.symbol))).scalars()) - ticker_ids = [ticker.id for ticker in tickers] - ciks = sorted({ticker.cik for ticker in tickers if ticker.cik}) - + candidates = await candidate_service.build_candidates(db, today=today) + ticker_ids = [candidate.ticker_id for candidate in candidates] legacy_by_ticker = await _legacy_values(db, ticker_ids) - derived_by_cik = await _derived_by_cik(db, ciks) - closes_by_ticker = await _latest_closes(db, ticker_ids) - surprise_by_ticker = await _latest_surprises(db, ticker_ids, today) source_runs = await _source_runs(db) rows: list[dict[str, Any]] = [] - for ticker in tickers: - legacy = legacy_by_ticker.get(ticker.id) - derived = derived_by_cik.get(ticker.cik) if ticker.cik else None - close = closes_by_ticker.get(ticker.id) - candidate_pe = ( - _pe(close[0], derived.ttm_diluted_eps) - if close is not None and derived is not None - else None - ) - growth_series = ( - derived.metrics.get("revenue_growth_yoy") if derived is not None else None - ) - candidate = { - "pe_ratio": candidate_pe, - "revenue_growth": growth_series.value if growth_series else None, - "earnings_surprise": surprise_by_ticker.get(ticker.id), + for candidate in candidates: + legacy = legacy_by_ticker.get(candidate.ticker_id) + candidate_values = { + "pe_ratio": candidate.pe_ratio, + "revenue_growth": candidate.revenue_growth, + "earnings_surprise": candidate.earnings_surprise, } legacy_values = { "pe_ratio": legacy.pe_ratio if legacy else None, @@ -128,17 +108,17 @@ async def build_report( "earnings_surprise": legacy.earnings_surprise if legacy else None, } fields = { - key: _field_comparison(key, legacy_values[key], candidate[key]) + key: _field_comparison(key, legacy_values[key], candidate_values[key]) for key in FIELD_KEYS } legacy_score = fundamental_score(**legacy_values) - candidate_score = fundamental_score(**candidate) + candidate_score = fundamental_score(**candidate_values) rows.append( { - "symbol": ticker.symbol, - "cik": ticker.cik, + "symbol": candidate.symbol, + "cik": candidate.cik, "legacy_fetched_at": _iso(legacy.fetched_at) if legacy else None, - "price_date": _iso(close[1]) if close else None, + "price_date": _iso(candidate.price_date), "fields": fields, "scores": { "legacy_fundamental": _round(legacy_score), @@ -311,74 +291,6 @@ async def _legacy_values( return {row.ticker_id: row for row in rows} -async def _derived_by_cik( - db: AsyncSession, ciks: list[str] -) -> dict[str, deriv.DerivedFundamentals]: - if not ciks: - return {} - grouped: dict[str, list[FundamentalSnapshot]] = defaultdict(list) - rows = ( - await db.execute( - select(FundamentalSnapshot).where(FundamentalSnapshot.cik.in_(ciks)) - ) - ).scalars() - for row in rows: - grouped[row.cik].append(row) - return {cik: deriv.derive(grouped.get(cik, [])) for cik in ciks} - - -async def _latest_closes( - db: AsyncSession, ticker_ids: list[int] -) -> dict[int, tuple[float, date]]: - if not ticker_ids: - return {} - latest = ( - select(OHLCVRecord.ticker_id, func.max(OHLCVRecord.date).label("max_date")) - .where(OHLCVRecord.ticker_id.in_(ticker_ids)) - .group_by(OHLCVRecord.ticker_id) - .subquery() - ) - rows = ( - await db.execute( - select(OHLCVRecord.ticker_id, OHLCVRecord.close, OHLCVRecord.date).join( - latest, - (OHLCVRecord.ticker_id == latest.c.ticker_id) - & (OHLCVRecord.date == latest.c.max_date), - ) - ) - ).all() - return { - ticker_id: (float(close), close_date) - for ticker_id, close, close_date in rows - if _finite(close) - } - - -async def _latest_surprises( - db: AsyncSession, ticker_ids: list[int], today: date -) -> dict[int, float]: - if not ticker_ids: - return {} - rows = ( - await db.execute( - select(EarningsEvent) - .where( - EarningsEvent.ticker_id.in_(ticker_ids), - EarningsEvent.announce_date < today, - ) - .order_by(EarningsEvent.ticker_id, EarningsEvent.announce_date.desc()) - ) - ).scalars() - out: dict[int, float] = {} - for row in rows: - if row.ticker_id in out: - continue - surprise = _surprise(row.eps_estimate, row.eps_actual) - if surprise is not None: - out[row.ticker_id] = surprise - return out - - async def _source_runs(db: AsyncSession) -> dict[str, dict[str, Any] | None]: sources = ("sec_facts", "dolt_earnings") rows = ( @@ -526,18 +438,6 @@ def _rank_change(legacy: int | None, candidate: int | None) -> int | None: return legacy - candidate if legacy is not None and candidate is not None else None -def _surprise(estimate: float | None, actual: float | None) -> float | None: - if not _finite(estimate) or not _finite(actual) or estimate == 0: - return None - return (actual - estimate) / abs(estimate) * 100.0 - - -def _pe(price: float | None, ttm_eps: float | None) -> float | None: - if not _finite(price) or price <= 0 or not _finite(ttm_eps) or ttm_eps <= 0: - return None - return price / ttm_eps - - def _delta(legacy: float | None, candidate: float | None) -> float | None: if not _finite(legacy) or not _finite(candidate): return None diff --git a/docs/dolt-integration-plan.md b/docs/dolt-integration-plan.md index 80cea90..8758e67 100644 --- a/docs/dolt-integration-plan.md +++ b/docs/dolt-integration-plan.md @@ -427,8 +427,9 @@ workstream B — Alpaca remains the price source throughout. fundamental-score/ranking changes, require explicit approval. Definition changes (e.g. TTM vs provider convention) called out, not averaged away. **Status 2026-07-24: the gate has been exercised and the evidence supports - approval** — see the handoff section below. What remains of A5 is the - activation itself: implementing step (c) and flipping it on. + approval** — see the handoff section below. Step (c) is implemented behind the + default-off `fundamental_data_sec_dolt_cutover_enabled` SystemSetting; the + remaining production action is flipping that switch on and observing it. - A6. Remove FMP/Finnhub/Alpha Vantage; keep monitoring + manual fallback. **Workstream B (independent, start when wanted):** @@ -491,16 +492,18 @@ Post-fix: candidate scores 504 of 511 vs legacy's 507 (gap = PSKY/Q new registra FITB, all explained); revenue-growth agreement 0.0038 median abs delta where both exist. Dennis reviewed the evidence 2026-07-24 and directed proceeding to cutover. -**Task 1 — A5 activation (implement step (c) above, ~line 207).** The post-activation -local refresh of `fundamental_data` does not exist yet. Per the spec: `pe_ratio` and +**Task 1 — A5 activation (IMPLEMENTED 2026-07-24; production switch remains).** The +post-activation local refresh of `fundamental_data` derives `pe_ratio` and `market_cap` from newest valid snapshots × latest PostgreSQL close, `revenue_growth` from snapshots, `earnings_surprise`/`next_earnings_date` from `earnings_events`; mark affected cached fundamental scores stale; must run identically when SEC is unreachable. -Implementation notes from the parity work: consume `fundamentals_derivation.derive()` -outputs, NOT raw snapshot fields — that path carries the split guard (`ttm_diluted_eps` +It consumes `fundamentals_derivation.derive()` outputs, NOT raw snapshot fields — +that path carries the split guard (`ttm_diluted_eps` nulls when contaminated, with `ttm_diluted_eps_caveat`) and the multi-class share -fallback (`shares_outstanding` + `shares_outstanding_estimated`). Activation should be -an explicit switch (SystemSetting, like `sec_cik_overrides`), default off. +fallback (`shares_outstanding` + `shares_outstanding_estimated`). Parity and activation +share the same candidate builder. Activation is the explicit +`fundamental_data_sec_dolt_cutover_enabled` SystemSetting and defaults off; see +`docs/fundamentals-deployment.md` for the production flip and rollback procedure. **Task 2 — A6 decommissioning.** After a short observation window: remove FMP/Finnhub/Alpha Vantage providers, config and env keys; keep monitoring + manual diff --git a/docs/fundamentals-deployment.md b/docs/fundamentals-deployment.md index 5f25ca7..f6bbb39 100644 --- a/docs/fundamentals-deployment.md +++ b/docs/fundamentals-deployment.md @@ -1,13 +1,15 @@ # Fundamentals production deployment This is the one-time production setup for the Dolt earnings and SEC fundamentals -imports. Both imports remain shadow inputs until the separate A5 scoring-cutover -approval. Do not add OS cron entries: the application scheduler owns both jobs. +imports. The A5 scoring cutover was approved on 2026-07-24; the compat-cache write +path is still default-off until the explicit production switch below is set. Do +not add OS cron entries: the application scheduler owns both jobs. ## What the deployment adds - `Dolt Earnings Import (shadow)` runs daily at 02:30 America/New_York. -- `SEC Fundamentals Import (shadow)` runs daily at 04:00 America/New_York. +- `SEC Fundamentals Import` runs daily at 04:00 America/New_York. Its local + `fundamental_data` refresh runs only when the A5 switch is enabled. - `Fundamentals Parity Report (read-only)` runs daily at 05:30 America/New_York. - Both jobs are visible, toggleable, and manually triggerable in Admin → Jobs. - Cron expressions are editable in Admin → Schedule. @@ -75,7 +77,7 @@ In Admin → Jobs, wait until no other job is running, then: 1. Trigger **Dolt Earnings Import (shadow)**. Expect `completed` with import status `promoted`; a repeat without an upstream change should report `no_op`. -2. Trigger **SEC Fundamentals Import (shadow)**. The first run performs the +2. Trigger **SEC Fundamentals Import**. The first run performs the tracked-universe history backfill and can take materially longer than a daily incremental run. Expect `completed` with import status `promoted`. 3. Check Admin → System Events. There should be no new import error. @@ -154,10 +156,69 @@ Expect `OK: source lock is busy`. This is the remaining live-PostgreSQL mutual-exclusion check; SQLite unit tests cannot exercise PostgreSQL advisory locks. A second Admin trigger should independently report the job as busy. +## A5 production activation (approved 2026-07-24) + +The write path is controlled by the SystemSetting +`fundamental_data_sec_dolt_cutover_enabled`. An absent value, `false`, or any +value other than `true` leaves `fundamental_data` untouched. Before enabling it, +confirm the normal PostgreSQL backup containing `fundamental_data` is current. + +Enable the cutover in PostgreSQL: + +```sql +INSERT INTO system_settings (key, value, updated_at) +VALUES ('fundamental_data_sec_dolt_cutover_enabled', 'true', now()) +ON CONFLICT (key) DO UPDATE +SET value = EXCLUDED.value, updated_at = now(); +``` + +Then trigger **SEC Fundamentals Import** once in Admin → Jobs. The import may be +`promoted` or `no_op`; either result runs the local refresh. Once enabled, the +same refresh also runs after an SEC network/validation failure or a source-lock +skip, because it reads only PostgreSQL snapshots, earnings events, and closes. +The job message appends the cache row count and changed score-input count when +the import itself completed successfully. + +Verify the switch and refreshed rows: + +```sql +SELECT key, value, updated_at +FROM system_settings +WHERE key = 'fundamental_data_sec_dolt_cutover_enabled'; + +SELECT count(*) AS rows, + max(fetched_at) AS refreshed_at, + count(pe_ratio) AS pe_available, + count(revenue_growth) AS growth_available, + count(earnings_surprise) AS surprise_available, + count(next_earnings_date) AS next_date_available +FROM fundamental_data; + +SELECT dimension, is_stale, count(*) +FROM dimension_scores +WHERE dimension = 'fundamental' +GROUP BY dimension, is_stale; + +SELECT is_stale, count(*) +FROM composite_scores +GROUP BY is_stale; +``` + +The first refresh intentionally marks affected fundamental and composite score +caches stale. The normal 15:30 near-close scanner recomputes them before using +the rankings; until then, reads truthfully expose the stale state. Observe at +least several scheduled cycles before A6 removes the legacy providers. + ## Failure and rollback -- Disable the failing shadow job in Admin → Jobs. This stops scheduled imports - without changing existing data or the legacy scoring path. +- To stop the A5 cache writes without stopping SEC snapshot ingestion, set + `fundamental_data_sec_dolt_cutover_enabled` back to `false` with the SQL above + (changing only the value). This prevents the next local refresh but does not + restore rows already replaced. Restore `fundamental_data` from the pre-cutover + database backup, or—before A6—manually run the legacy Fundamental Collector if + its provider keys and quota are still available. +- Disable a failing source-import job in Admin → Jobs only when ingestion itself + must stop. Existing promoted snapshots/events remain available. - Inspect the job runtime, latest `data_import_runs.validation_json`, service logs, and Admin → System Events before retrying. - Re-run `sudo -u deploy bash ./deploy/provision_fundamentals.sh --check` for @@ -165,8 +226,8 @@ locks. A second Admin trigger should independently report the job as busy. - The Dolt clone is a reproducible cache and does not need a bespoke backup. PostgreSQL (including `earnings_events`, `fundamental_snapshots`, and import audit rows) must remain covered by the normal production database backup. -- Do not proceed to A5 while either shadow feed is unhealthy or the parity gate - has not received explicit approval. +- Do not proceed to A6 until the activated cache has completed the observation + window and the forward earnings calendar remains timely. - If report generation fails, inspect Admin → System Events and verify `FUNDAMENTALS_PARITY_REPORT_DIR` exists and is writable by `deploy`. Existing reports and all live data remain untouched. diff --git a/tests/unit/test_fundamental_data_refresh.py b/tests/unit/test_fundamental_data_refresh.py new file mode 100644 index 0000000..2508fd7 --- /dev/null +++ b/tests/unit/test_fundamental_data_refresh.py @@ -0,0 +1,293 @@ +"""A5 activation: local candidate derivation and compat-cache refresh.""" + +from __future__ import annotations + +import json +from datetime import date, datetime, timedelta, timezone + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app.database import Base +from app.models.earnings_event import EarningsEvent +from app.models.fundamental import FundamentalData +from app.models.fundamental_snapshot import FundamentalSnapshot +from app.models.ohlcv import OHLCVRecord +from app.models.score import CompositeScore, DimensionScore +from app.models.settings import SystemSetting +from app.models.ticker import Ticker +from app.services import fundamentals_candidate_service as candidates +from app.services import fundamentals_derivation as deriv +from app.services import fundamental_data_refresh_service as refresh_service + + +UTC = timezone.utc +NOW = datetime(2026, 7, 24, 10, 0, tzinfo=UTC) +TODAY = date(2026, 7, 24) + +_engine = create_async_engine("sqlite+aiosqlite://", echo=False) +_session_factory = async_sessionmaker( + _engine, class_=AsyncSession, expire_on_commit=False +) + + +@pytest.fixture(autouse=True) +async def _setup_tables(): + async with _engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + yield + async with _engine.begin() as connection: + await connection.run_sync(Base.metadata.drop_all) + + +@pytest.fixture +async def session() -> AsyncSession: + async with _session_factory() as db: + yield db + + +def _snapshot_rows(cik: str) -> list[FundamentalSnapshot]: + rows: list[FundamentalSnapshot] = [] + periods = ("Q1", "Q2", "Q3", "FY") + months = (3, 6, 9, 12) + for fiscal_year, multiplier in ((2025, 1.0), (2026, 1.1)): + revenue = [100 * multiplier, 110 * multiplier, 120 * multiplier, 130 * multiplier] + eps = [1.0 * multiplier, 1.1 * multiplier, 1.2 * multiplier, 1.3 * multiplier] + for index, fiscal_period in enumerate(periods): + period_end = date(fiscal_year, months[index], 28) + rows.append( + FundamentalSnapshot( + cik=cik, + accession=f"{cik}-{fiscal_year}-{fiscal_period}", + form="10-K" if fiscal_period == "FY" else "10-Q", + filed_date=period_end, + accepted_at=datetime( + fiscal_year, months[index], 28, tzinfo=UTC + ), + period_end=period_end, + fiscal_year=fiscal_year, + fiscal_period=fiscal_period, + revenue=sum(revenue[: index + 1]), + diluted_eps=sum(eps[: index + 1]), + shares_outstanding=1_000, + ) + ) + return rows + + +async def test_default_off_performs_no_candidate_read_or_write( + session: AsyncSession, monkeypatch +): + ticker = Ticker(symbol="AAA") + session.add(ticker) + await session.flush() + session.add( + FundamentalData( + ticker_id=ticker.id, + pe_ratio=12, + revenue_growth=3, + earnings_surprise=1, + market_cap=100, + fetched_at=NOW, + ) + ) + await session.commit() + + async def should_not_read(*args, **kwargs): + raise AssertionError("default-off refresh derived candidates") + + monkeypatch.setattr(candidates, "build_candidates", should_not_read) + summary = await refresh_service.refresh_if_enabled(session, today=TODAY) + + stored = await session.scalar( + select(FundamentalData).where(FundamentalData.ticker_id == ticker.id) + ) + assert summary == { + "enabled": False, + "refreshed": 0, + "score_inputs_changed": 0, + "dimension_scores_staled": 0, + "composite_scores_staled": 0, + } + assert stored.pe_ratio == 12 + + +async def test_activated_refresh_updates_all_fields_and_invalidates_scores( + session: AsyncSession, +): + session.add( + SystemSetting(key=refresh_service.ACTIVATION_KEY, value="true") + ) + first = Ticker(symbol="AAA", cik="0000000001") + second = Ticker(symbol="AAB", cik="0000000001") + session.add_all([first, second]) + await session.flush() + session.add_all(_snapshot_rows(first.cik)) + session.add_all( + [ + OHLCVRecord( + ticker_id=first.id, + date=TODAY - timedelta(days=1), + open=100, + high=100, + low=100, + close=100, + volume=100, + ), + OHLCVRecord( + ticker_id=second.id, + date=TODAY - timedelta(days=1), + open=200, + high=200, + low=200, + close=200, + volume=100, + ), + EarningsEvent( + ticker_id=first.id, + announce_date=TODAY - timedelta(days=10), + session="amc", + eps_estimate=2, + eps_actual=2.2, + source="dolt_earnings", + ), + EarningsEvent( + ticker_id=first.id, + announce_date=TODAY, + session="amc", + source="dolt_earnings", + ), + ] + ) + for ticker in (first, second): + session.add( + FundamentalData( + ticker_id=ticker.id, + pe_ratio=1, + revenue_growth=1, + earnings_surprise=1, + market_cap=1, + fetched_at=NOW - timedelta(days=1), + ) + ) + session.add( + DimensionScore( + ticker_id=ticker.id, + dimension="fundamental", + score=50, + is_stale=False, + computed_at=NOW, + ) + ) + session.add( + CompositeScore( + ticker_id=ticker.id, + score=50, + is_stale=False, + weights_json="{}", + computed_at=NOW, + ) + ) + await session.commit() + + summary = await refresh_service.refresh_if_enabled( + session, now=NOW, today=TODAY + ) + + stored = { + row.ticker_id: row + for row in ( + await session.execute(select(FundamentalData)) + ).scalars() + } + assert summary["refreshed"] == 2 + assert summary["score_inputs_changed"] == 2 + assert stored[first.id].pe_ratio == pytest.approx(100 / 5.06) + assert stored[second.id].pe_ratio == pytest.approx(200 / 5.06) + assert stored[first.id].revenue_growth == pytest.approx(10) + assert stored[first.id].earnings_surprise == pytest.approx(10) + assert stored[first.id].market_cap == pytest.approx(100_000) + assert stored[first.id].next_earnings_date == TODAY + metadata = json.loads(stored[first.id].unavailable_fields_json) + assert metadata["source_pe_ratio"] == "sec_facts+ohlcv_records" + assert metadata["source_next_earnings_date"] == "dolt_earnings" + + dimensions = ( + await session.execute(select(DimensionScore)) + ).scalars().all() + composites = ( + await session.execute(select(CompositeScore)) + ).scalars().all() + assert all(row.is_stale for row in dimensions) + assert all(row.is_stale for row in composites) + + for row in (*dimensions, *composites): + row.is_stale = False + await session.commit() + unchanged = await refresh_service.refresh_if_enabled( + session, now=NOW + timedelta(hours=1), today=TODAY + ) + assert unchanged["score_inputs_changed"] == 0 + assert not any( + (await session.execute(select(DimensionScore.is_stale))).scalars() + ) + assert not any( + (await session.execute(select(CompositeScore.is_stale))).scalars() + ) + + +async def test_candidate_uses_guarded_derive_outputs_and_share_fallback( + session: AsyncSession, monkeypatch +): + ticker = Ticker(symbol="GUARD", cik="0000000002") + session.add(ticker) + await session.flush() + session.add( + FundamentalSnapshot( + cik=ticker.cik, + accession="raw-accession", + form="10-Q", + filed_date=TODAY, + accepted_at=NOW, + period_end=TODAY, + fiscal_year=2026, + fiscal_period="Q2", + diluted_eps=99, + shares_outstanding=999, + ) + ) + session.add( + OHLCVRecord( + ticker_id=ticker.id, + date=TODAY, + open=50, + high=50, + low=50, + close=50, + volume=100, + ) + ) + await session.commit() + + def guarded(_rows): + return deriv.DerivedFundamentals( + metrics={ + "revenue_growth_yoy": deriv.MetricSeries(value=7) + }, + ttm_diluted_eps=None, + ttm_diluted_eps_caveat="split guard applied", + shares_outstanding=123, + shares_outstanding_estimated=True, + latest_period_end=TODAY, + latest_filed_date=TODAY, + ) + + monkeypatch.setattr(candidates.deriv, "derive", guarded) + candidate = (await candidates.build_candidates(session, today=TODAY))[0] + + assert candidate.pe_ratio is None + assert candidate.market_cap == 50 * 123 + assert candidate.revenue_growth == 7 + assert candidate.unavailable_fields["pe_ratio"] == "split guard applied" + assert "weighted-average" in candidate.unavailable_fields["market_cap_estimated"] diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index 07fb8dd..8a9a23b 100644 --- a/tests/unit/test_scheduler.py +++ b/tests/unit/test_scheduler.py @@ -12,6 +12,7 @@ from app.scheduler import ( _last_successful, _run_shadow_import, run_fundamentals_parity_report, + run_sec_fundamentals_import, configure_scheduler, get_job_runtime_snapshot, queue_backtest_options, @@ -247,6 +248,95 @@ class TestShadowImportJobs: assert runtime["status"] == "skipped" assert runtime["message"] == "Disabled" + async def test_sec_failure_still_runs_activated_local_refresh(self, monkeypatch): + calls = [] + + async def enabled(db, job_name): + return True + + async def unavailable(importer): + raise RuntimeError("SEC unavailable") + + async def refreshed(db): + calls.append(db) + return { + "enabled": True, + "refreshed": 511, + "score_inputs_changed": 2, + "dimension_scores_staled": 2, + "composite_scores_staled": 2, + } + + monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) + monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) + monkeypatch.setattr("app.scheduler.run_import", unavailable) + monkeypatch.setattr( + "app.scheduler.fundamental_data_refresh_service.refresh_if_enabled", + refreshed, + ) + + await run_sec_fundamentals_import() + + assert len(calls) == 1 + runtime = get_job_runtime_snapshot("sec_fundamentals_import") + assert runtime["status"] == "error" + assert runtime["message"] == "SEC unavailable" + + async def test_sec_success_surfaces_activated_refresh_summary(self, monkeypatch): + async def enabled(db, job_name): + return True + + async def imported(importer): + return SimpleNamespace( + status="no_op", revision="abcdef1234567890", error_details=None + ) + + async def refreshed(db): + return { + "enabled": True, + "refreshed": 511, + "score_inputs_changed": 2, + "dimension_scores_staled": 2, + "composite_scores_staled": 2, + } + + monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) + monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) + monkeypatch.setattr("app.scheduler.run_import", imported) + monkeypatch.setattr( + "app.scheduler.fundamental_data_refresh_service.refresh_if_enabled", + refreshed, + ) + + await run_sec_fundamentals_import() + + runtime = get_job_runtime_snapshot("sec_fundamentals_import") + assert runtime["status"] == "completed" + assert runtime["message"] == ( + "no_op · abcdef123456 · cache 511 · 2 score inputs changed" + ) + + async def test_disabled_sec_job_does_not_run_local_refresh(self, monkeypatch): + async def disabled(db, job_name): + return False + + async def should_not_run(*args, **kwargs): + raise AssertionError("disabled SEC job ran work") + + monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) + monkeypatch.setattr("app.scheduler._is_job_enabled", disabled) + monkeypatch.setattr("app.scheduler.run_import", should_not_run) + monkeypatch.setattr( + "app.scheduler.fundamental_data_refresh_service.refresh_if_enabled", + should_not_run, + ) + + await run_sec_fundamentals_import() + + runtime = get_job_runtime_snapshot("sec_fundamentals_import") + assert runtime["status"] == "skipped" + assert runtime["message"] == "Disabled" + async def test_fundamentals_parity_job_surfaces_report_summary(monkeypatch): async def enabled(db, job_name):