Implement A5 fundamentals cutover activation

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(

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",

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
+    )

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)
+    )

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

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

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.

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"]

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):