fix: don't double-report a failed SEC run, and correct the rollback doc
Deploy / lint (push) Successful in 11s
Deploy / test (push) Successful in 1m24s
Deploy / deploy (push) Successful in 41s

Three follow-ups from review of the A6 commits.

The cache-summary re-finalize raised a second durable system event on a failed
run: _runtime_finish emits for `error`/`rate_limited`, and the dedup key
includes the message, so "SEC unavailable" and "SEC unavailable · cache 511 · 2
score inputs changed" landed as two unacknowledged Admin events. Adds
`emit_event` so a re-finalize that only rewords an outcome stays silent, with a
regression test asserting exactly one event.

The rollback section still claimed disabling the SEC job freezes the cache — the
opposite of what the same page says two lines earlier, and of what the code now
does. Rewritten: there is no Admin cache-off switch, restoring `fundamental_data`
alone is temporary because the next run rebuilds it from the same snapshots and
code, and a real freeze means stopping the service.

Remaining "shadow" wording: the two import jobs have never been shadow since
activation, so `_run_shadow_import` -> `_run_source_import`, its section heading,
the deployment doc's job label, and the plan doc's "production switch remains"
handoff paragraph are all brought up to date.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-07 11:46:34 +02:00
co-authored by Claude Opus 5
parent 13f3636b6a
commit e1607ddbff
4 changed files with 63 additions and 36 deletions
+17 -5
View File
@@ -254,7 +254,14 @@ def _runtime_finish(
processed: int, processed: int,
total: int | None, total: int | None,
message: str | None = None, message: str | None = None,
emit_event: bool = True,
) -> None: ) -> None:
"""Finalize a job's runtime row, optionally raising a durable event.
``emit_event=False`` is for a *re-finalize* that only rewords an outcome an
earlier call already reported. The dedup key includes the message, so a
reworded error would otherwise land in Admin → System Events twice.
"""
runtime = _job_runtime.get(job_name, {}) runtime = _job_runtime.get(job_name, {})
runtime.update({ runtime.update({
"running": False, "running": False,
@@ -268,7 +275,7 @@ def _runtime_finish(
}) })
_job_runtime[job_name] = runtime _job_runtime[job_name] = runtime
# Durable event for error / rate-limit finishes (badge + Admin → Jobs panel). # Durable event for error / rate-limit finishes (badge + Admin → Jobs panel).
if status in ("error", "rate_limited"): if emit_event and status in ("error", "rate_limited"):
severity = "error" if status == "error" else "warning" severity = "error" if status == "error" else "warning"
try: try:
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
@@ -792,11 +799,11 @@ async def collect_sentiment() -> None:
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Jobs: shadow fundamentals sources # Jobs: bulk fundamentals source imports
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
async def _run_shadow_import(job_name: str, importer: SourceImporter) -> bool: async def _run_source_import(job_name: str, importer: SourceImporter) -> bool:
"""Run an importer and return whether its scheduled job was enabled. """Run an importer and return whether its scheduled job was enabled.
The SEC wrapper uses the return value only to word its runtime message: its The SEC wrapper uses the return value only to word its runtime message: its
@@ -859,7 +866,7 @@ async def _run_shadow_import(job_name: str, importer: SourceImporter) -> bool:
async def run_dolt_earnings_import() -> None: async def run_dolt_earnings_import() -> None:
"""Pull and import the Dolt earnings calendar/results feed.""" """Pull and import the Dolt earnings calendar/results feed."""
await _run_shadow_import("dolt_earnings_import", DoltEarningsImporter()) await _run_source_import("dolt_earnings_import", DoltEarningsImporter())
async def run_sec_fundamentals_import() -> None: async def run_sec_fundamentals_import() -> None:
@@ -873,7 +880,7 @@ async def run_sec_fundamentals_import() -> None:
no filing does, and `fundamental_data` feeds scoring. no filing does, and `fundamental_data` feeds scoring.
""" """
job_name = "sec_fundamentals_import" job_name = "sec_fundamentals_import"
import_ran = await _run_shadow_import(job_name, SecFundamentalsImporter()) import_ran = await _run_source_import(job_name, SecFundamentalsImporter())
try: try:
async with async_session_factory() as db: async with async_session_factory() as db:
@@ -908,6 +915,10 @@ async def run_sec_fundamentals_import() -> None:
# Every outcome carries the cache summary — including deferred, failed and # Every outcome carries the cache summary — including deferred, failed and
# source-locked ones. The import status is what varies; the refresh always # source-locked ones. The import status is what varies; the refresh always
# happened, and Admin → Jobs is the only place an operator sees that. # happened, and Admin → Jobs is the only place an operator sees that.
#
# This only rewords what _run_source_import already finalized, so it must not
# emit a second durable event: the dedup key includes the message, and a
# failure would otherwise show up twice in Admin → System Events.
runtime = get_job_runtime_snapshot(job_name) runtime = get_job_runtime_snapshot(job_name)
if import_ran: if import_ran:
status = str(runtime.get("status") or "completed") status = str(runtime.get("status") or "completed")
@@ -921,6 +932,7 @@ async def run_sec_fundamentals_import() -> None:
processed=processed, processed=processed,
total=1, total=1,
message=f"{import_message} · {cache_message}", message=f"{import_message} · {cache_message}",
emit_event=False,
) )
+11 -13
View File
@@ -496,20 +496,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. 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. Dennis reviewed the evidence 2026-07-24 and directed proceeding to cutover.
**Task 1 — A5 activation (IMPLEMENTED 2026-07-24; production switch remains).** The **Task 1 — A5 activation: DONE.** Implemented 2026-07-24, switched on and observed
post-activation local refresh of `fundamental_data` derives `pe_ratio` and in production, and made unconditional by A6 (2026-08-07) — there is no longer a
`market_cap` from newest valid snapshots × latest PostgreSQL close, `revenue_growth` switch, an Admin card, or a weekly legacy collector to skip. The local refresh of
from snapshots, `earnings_surprise`/`next_earnings_date` from `earnings_events`; mark `fundamental_data` derives `pe_ratio` and `market_cap` from newest valid snapshots ×
affected cached fundamental scores stale; must run identically when SEC is unreachable. latest PostgreSQL close, `revenue_growth` from snapshots, and
`earnings_surprise`/`next_earnings_date` from `earnings_events`; it marks affected
cached fundamental scores stale and runs identically when SEC is unreachable.
It consumes `fundamentals_derivation.derive()` outputs, NOT raw snapshot fields — It consumes `fundamentals_derivation.derive()` outputs, NOT raw snapshot fields —
that path carries the split guard (`ttm_diluted_eps` that path carries the split guard (`ttm_diluted_eps` nulls when contaminated, with
nulls when contaminated, with `ttm_diluted_eps_caveat`) and the multi-class share `ttm_diluted_eps_caveat`) and the multi-class share fallback (`shares_outstanding` +
fallback (`shares_outstanding` + `shares_outstanding_estimated`). Parity and activation `shares_outstanding_estimated`). See `docs/fundamentals-deployment.md` for current
share the same candidate builder. Activation is the explicit operations and rollback.
`fundamental_data_sec_dolt_cutover_enabled` SystemSetting and defaults off. It is
managed by the **Fundamentals data source** card in Admin → Settings; while active,
the weekly legacy collector skips itself so it cannot overwrite the SEC/Dolt cache.
See `docs/fundamentals-deployment.md` for the production flip and rollback procedure.
**Task 2 — A6 decommissioning: DONE 2026-08-07.** The cutover ran on and was **Task 2 — A6 decommissioning: DONE 2026-08-07.** The cutover ran on and was
observed in production, so the legacy providers, their config/env keys, the weekly observed in production, so the legacy providers, their config/env keys, the weekly
+17 -9
View File
@@ -86,7 +86,7 @@ a reviewed change to `DOLT_VERSION`, followed by the same provision/check flow.
In Admin → Jobs, wait until no other job is running, then: In Admin → Jobs, wait until no other job is running, then:
1. Trigger **Dolt Earnings Import (shadow)**. Expect `completed` with import 1. Trigger **Dolt Earnings Import**. Expect `completed` with import
status `promoted`; a repeat without an upstream change should report `no_op`. status `promoted`; a repeat without an upstream change should report `no_op`.
2. Trigger **SEC Fundamentals Import**. 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 tracked-universe history backfill and can take materially longer than a daily
@@ -197,14 +197,22 @@ rollback window has closed, delete the rows and the `MANAGED_SETTINGS` filter in
## Failure and rollback ## Failure and rollback
- **There is no provider fallback any more.** To recover bad `fundamental_data` - **There is no provider fallback any more, and no Admin switch that freezes the
values, restore the table from the PostgreSQL backup; the next scheduled run cache.** Disabling **SEC Fundamentals Import** stops SEC network access only;
will rebuild it from the current snapshots. Confirm a recent backup exists the 04:00 job still rebuilds `fundamental_data` from the stored snapshots,
before any change that could corrupt the snapshots. earnings events and closes.
- Disable a failing source-import job in Admin → Jobs only when SEC network - Restoring `fundamental_data` from the PostgreSQL backup is therefore a
access itself must stop — the local cache refresh keeps running, and existing *temporary* fix on its own: if the bad values come from the snapshots or from
promoted snapshots/events remain available. To freeze the cache as well, the derivation code, the next scheduled run reproduces them. Fix the cause —
disable the job *and* accept that P/E, market cap and earnings dates go stale. restore or repair `fundamental_snapshots` / `earnings_events`, or revert the
parser change and re-run `scripts/reparse_fundamentals.py --apply`.
- To genuinely freeze the cache while you work, stop the service
(`sudo systemctl stop signalplatform.service`) — that stops the scheduler with
it. There is no finer-grained control, by design: a silently frozen scoring
input is worse than an obvious outage.
- Disable a failing source-import job in Admin → Jobs when SEC network access
itself must stop. Existing promoted snapshots and events remain available, and
the job's runtime message still reports the cache result.
- Inspect the job runtime, latest `data_import_runs.validation_json`, service - Inspect the job runtime, latest `data_import_runs.validation_json`, service
logs, and Admin → System Events before retrying. logs, and Admin → System Events before retrying.
- `unresolved_filing` is emitted once when a filing enters automatic retry. It - `unresolved_filing` is emitted once when a filing enters automatic retry. It
+18 -9
View File
@@ -1,5 +1,6 @@
"""Unit tests for app.scheduler module.""" """Unit tests for app.scheduler module."""
import asyncio
from types import SimpleNamespace from types import SimpleNamespace
import pytest import pytest
@@ -12,7 +13,7 @@ from app.scheduler import (
_parse_frequency, _parse_frequency,
_resume_tickers, _resume_tickers,
_last_successful, _last_successful,
_run_shadow_import, _run_source_import,
run_sec_fundamentals_import, run_sec_fundamentals_import,
configure_scheduler, configure_scheduler,
get_job_runtime_snapshot, get_job_runtime_snapshot,
@@ -175,7 +176,7 @@ class _SessionContext:
return None return None
class TestShadowImportJobs: class TestSourceImportJobs:
@staticmethod @staticmethod
def _session_factory(): def _session_factory():
return _SessionContext() return _SessionContext()
@@ -193,7 +194,7 @@ class TestShadowImportJobs:
monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled)
monkeypatch.setattr("app.scheduler.run_import", imported) monkeypatch.setattr("app.scheduler.run_import", imported)
await _run_shadow_import("dolt_earnings_import", object()) await _run_source_import("dolt_earnings_import", object())
runtime = get_job_runtime_snapshot("dolt_earnings_import") runtime = get_job_runtime_snapshot("dolt_earnings_import")
assert runtime["status"] == "completed" assert runtime["status"] == "completed"
@@ -213,7 +214,7 @@ class TestShadowImportJobs:
monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled)
monkeypatch.setattr("app.scheduler.run_import", imported) monkeypatch.setattr("app.scheduler.run_import", imported)
await _run_shadow_import("sec_fundamentals_import", object()) await _run_source_import("sec_fundamentals_import", object())
runtime = get_job_runtime_snapshot("sec_fundamentals_import") runtime = get_job_runtime_snapshot("sec_fundamentals_import")
assert runtime["status"] == "error" assert runtime["status"] == "error"
@@ -235,7 +236,7 @@ class TestShadowImportJobs:
monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled)
monkeypatch.setattr("app.scheduler.run_import", imported) monkeypatch.setattr("app.scheduler.run_import", imported)
await _run_shadow_import("sec_fundamentals_import", object()) await _run_source_import("sec_fundamentals_import", object())
runtime = get_job_runtime_snapshot("sec_fundamentals_import") runtime = get_job_runtime_snapshot("sec_fundamentals_import")
assert runtime["status"] == STATUS_DEFERRED assert runtime["status"] == STATUS_DEFERRED
@@ -253,7 +254,7 @@ class TestShadowImportJobs:
monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled)
monkeypatch.setattr("app.scheduler.run_import", imported) monkeypatch.setattr("app.scheduler.run_import", imported)
await _run_shadow_import("dolt_earnings_import", object()) await _run_source_import("dolt_earnings_import", object())
runtime = get_job_runtime_snapshot("dolt_earnings_import") runtime = get_job_runtime_snapshot("dolt_earnings_import")
assert runtime["status"] == "skipped" assert runtime["status"] == "skipped"
@@ -270,7 +271,7 @@ class TestShadowImportJobs:
monkeypatch.setattr("app.scheduler._is_job_enabled", disabled) monkeypatch.setattr("app.scheduler._is_job_enabled", disabled)
monkeypatch.setattr("app.scheduler.run_import", should_not_run) monkeypatch.setattr("app.scheduler.run_import", should_not_run)
await _run_shadow_import("sec_fundamentals_import", object()) await _run_source_import("sec_fundamentals_import", object())
runtime = get_job_runtime_snapshot("sec_fundamentals_import") runtime = get_job_runtime_snapshot("sec_fundamentals_import")
assert runtime["status"] == "skipped" assert runtime["status"] == "skipped"
@@ -278,6 +279,7 @@ class TestShadowImportJobs:
async def test_sec_failure_still_runs_local_cache_refresh(self, monkeypatch): async def test_sec_failure_still_runs_local_cache_refresh(self, monkeypatch):
calls = [] calls = []
events = []
async def enabled(db, job_name): async def enabled(db, job_name):
return True return True
@@ -294,15 +296,20 @@ class TestShadowImportJobs:
"composite_scores_staled": 2, "composite_scores_staled": 2,
} }
async def record(**kwargs):
events.append(kwargs)
monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory)
monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled)
monkeypatch.setattr("app.scheduler.run_import", unavailable) monkeypatch.setattr("app.scheduler.run_import", unavailable)
monkeypatch.setattr("app.scheduler._record_system_event", record)
monkeypatch.setattr( monkeypatch.setattr(
"app.scheduler.fundamental_data_refresh_service.refresh", "app.scheduler.fundamental_data_refresh_service.refresh",
refreshed, refreshed,
) )
await run_sec_fundamentals_import() await run_sec_fundamentals_import()
await asyncio.sleep(0) # let the fire-and-forget event task run
assert len(calls) == 1 assert len(calls) == 1
runtime = get_job_runtime_snapshot("sec_fundamentals_import") runtime = get_job_runtime_snapshot("sec_fundamentals_import")
@@ -311,6 +318,10 @@ class TestShadowImportJobs:
assert runtime["message"] == ( assert runtime["message"] == (
"SEC unavailable · cache 511 · 2 score inputs changed" "SEC unavailable · cache 511 · 2 score inputs changed"
) )
# Rewording the outcome must not duplicate the durable event: the dedup
# key includes the message, so a second finish would show up twice in
# Admin → System Events.
assert len(events) == 1, events
async def test_sec_success_surfaces_cache_refresh_summary(self, monkeypatch): async def test_sec_success_surfaces_cache_refresh_summary(self, monkeypatch):
async def enabled(db, job_name): async def enabled(db, job_name):
@@ -421,5 +432,3 @@ class TestShadowImportJobs:
assert runtime["message"] == ( assert runtime["message"] == (
"Import disabled · cache 511 · 2 score inputs changed" "Import disabled · cache 511 · 2 score inputs changed"
) )