From e1607ddbff27de97ba891e66da3409e1061b93d7 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Fri, 7 Aug 2026 11:46:34 +0200 Subject: [PATCH] fix: don't double-report a failed SEC run, and correct the rollback doc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- app/scheduler.py | 22 +++++++++++++++++----- docs/dolt-integration-plan.md | 24 +++++++++++------------- docs/fundamentals-deployment.md | 26 +++++++++++++++++--------- tests/unit/test_scheduler.py | 27 ++++++++++++++++++--------- 4 files changed, 63 insertions(+), 36 deletions(-) diff --git a/app/scheduler.py b/app/scheduler.py index da5b11c..942fc5a 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -254,7 +254,14 @@ def _runtime_finish( processed: int, total: int | None, message: str | None = None, + emit_event: bool = True, ) -> 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.update({ "running": False, @@ -268,7 +275,7 @@ def _runtime_finish( }) _job_runtime[job_name] = runtime # 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" try: 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. 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: """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: @@ -873,7 +880,7 @@ async def run_sec_fundamentals_import() -> None: no filing does, and `fundamental_data` feeds scoring. """ job_name = "sec_fundamentals_import" - import_ran = await _run_shadow_import(job_name, SecFundamentalsImporter()) + import_ran = await _run_source_import(job_name, SecFundamentalsImporter()) try: 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 # source-locked ones. The import status is what varies; the refresh always # 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) if import_ran: status = str(runtime.get("status") or "completed") @@ -921,6 +932,7 @@ async def run_sec_fundamentals_import() -> None: processed=processed, total=1, message=f"{import_message} · {cache_message}", + emit_event=False, ) diff --git a/docs/dolt-integration-plan.md b/docs/dolt-integration-plan.md index 60180af..0fed5a6 100644 --- a/docs/dolt-integration-plan.md +++ b/docs/dolt-integration-plan.md @@ -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. Dennis reviewed the evidence 2026-07-24 and directed proceeding to cutover. -**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. +**Task 1 — A5 activation: DONE.** Implemented 2026-07-24, switched on and observed +in production, and made unconditional by A6 (2026-08-07) — there is no longer a +switch, an Admin card, or a weekly legacy collector to skip. The local refresh of +`fundamental_data` derives `pe_ratio` and `market_cap` from newest valid snapshots × +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 — -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`). Parity and activation -share the same candidate builder. Activation is the explicit -`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. +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`). See `docs/fundamentals-deployment.md` for current +operations and rollback. **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 diff --git a/docs/fundamentals-deployment.md b/docs/fundamentals-deployment.md index 06df284..e20ae29 100644 --- a/docs/fundamentals-deployment.md +++ b/docs/fundamentals-deployment.md @@ -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: -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`. 2. Trigger **SEC Fundamentals Import**. The first run performs the 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 -- **There is no provider fallback any more.** To recover bad `fundamental_data` - values, restore the table from the PostgreSQL backup; the next scheduled run - will rebuild it from the current snapshots. Confirm a recent backup exists - before any change that could corrupt the snapshots. -- Disable a failing source-import job in Admin → Jobs only when SEC network - access itself must stop — the local cache refresh keeps running, and existing - promoted snapshots/events remain available. To freeze the cache as well, - disable the job *and* accept that P/E, market cap and earnings dates go stale. +- **There is no provider fallback any more, and no Admin switch that freezes the + cache.** Disabling **SEC Fundamentals Import** stops SEC network access only; + the 04:00 job still rebuilds `fundamental_data` from the stored snapshots, + earnings events and closes. +- Restoring `fundamental_data` from the PostgreSQL backup is therefore a + *temporary* fix on its own: if the bad values come from the snapshots or from + the derivation code, the next scheduled run reproduces them. Fix the cause — + 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 logs, and Admin → System Events before retrying. - `unresolved_filing` is emitted once when a filing enters automatic retry. It diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index c31715a..1fb6dda 100644 --- a/tests/unit/test_scheduler.py +++ b/tests/unit/test_scheduler.py @@ -1,5 +1,6 @@ """Unit tests for app.scheduler module.""" +import asyncio from types import SimpleNamespace import pytest @@ -12,7 +13,7 @@ from app.scheduler import ( _parse_frequency, _resume_tickers, _last_successful, - _run_shadow_import, + _run_source_import, run_sec_fundamentals_import, configure_scheduler, get_job_runtime_snapshot, @@ -175,7 +176,7 @@ class _SessionContext: return None -class TestShadowImportJobs: +class TestSourceImportJobs: @staticmethod def _session_factory(): return _SessionContext() @@ -193,7 +194,7 @@ class TestShadowImportJobs: monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) 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") assert runtime["status"] == "completed" @@ -213,7 +214,7 @@ class TestShadowImportJobs: monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) 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") assert runtime["status"] == "error" @@ -235,7 +236,7 @@ class TestShadowImportJobs: monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) 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") assert runtime["status"] == STATUS_DEFERRED @@ -253,7 +254,7 @@ class TestShadowImportJobs: monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) 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") assert runtime["status"] == "skipped" @@ -270,7 +271,7 @@ class TestShadowImportJobs: monkeypatch.setattr("app.scheduler._is_job_enabled", disabled) 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") assert runtime["status"] == "skipped" @@ -278,6 +279,7 @@ class TestShadowImportJobs: async def test_sec_failure_still_runs_local_cache_refresh(self, monkeypatch): calls = [] + events = [] async def enabled(db, job_name): return True @@ -294,15 +296,20 @@ class TestShadowImportJobs: "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._is_job_enabled", enabled) monkeypatch.setattr("app.scheduler.run_import", unavailable) + monkeypatch.setattr("app.scheduler._record_system_event", record) monkeypatch.setattr( "app.scheduler.fundamental_data_refresh_service.refresh", refreshed, ) await run_sec_fundamentals_import() + await asyncio.sleep(0) # let the fire-and-forget event task run assert len(calls) == 1 runtime = get_job_runtime_snapshot("sec_fundamentals_import") @@ -311,6 +318,10 @@ class TestShadowImportJobs: assert runtime["message"] == ( "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 enabled(db, job_name): @@ -421,5 +432,3 @@ class TestShadowImportJobs: assert runtime["message"] == ( "Import disabled · cache 511 · 2 score inputs changed" ) - -