feat: add fundamentals parity reporting

2026-07-23 21:17:58 +02:00
parent 361cfd7883
commit ce8c60d957
26 changed files with 1331 additions and 8 deletions
@@ -52,6 +52,10 @@ SEC_REQUEST_SPACING_SECONDS=0.2
 SEC_MAX_RETRIES=4
 SEC_REQUEST_TIMEOUT_SECONDS=30.0

+# A5 read-only parity report archive. In production keep this outside the
+# rsync deployment tree, e.g. /var/lib/signal-platform/reports/fundamentals-parity.
+FUNDAMENTALS_PARITY_REPORT_DIR=reports/fundamentals-parity
+
 # Regime Monitor — FRED (VIX + HY credit spreads). Free key: https://fred.stlouisfed.org/docs/api/api_key.html
 # Optional: without it the volatility (V1) and credit (C1) pillars show as n/a.
 FRED_API_KEY=
@@ -51,3 +51,6 @@ backtest_snapshots/
 reports/*.pkl
 reports/*.pk1
 reports/.cache/
+# Runtime A5 parity bundles are generated on the production server. Research
+# conclusions belong in docs/research, not as an ever-growing artifact archive.
+reports/fundamentals-parity/
@@ -61,6 +61,10 @@ class Settings(BaseSettings):
    sec_max_retries: int = 4
    sec_request_timeout_seconds: float = 30.0

+    # A5 read-only comparison artifacts. Production must keep this outside the
+    # rsync deployment tree so the 5-7 day review window survives deploys.
+    fundamentals_parity_report_dir: str = "reports/fundamentals-parity"
+
    # Regime Monitor — FRED (VIX level + HY credit spreads). Optional: without it
    # the volatility (P5) and credit-spread (F2) signals are reported as n/a.
    fred_api_key: str = ""
@@ -453,6 +453,36 @@ async def toggle_job(
    )


+@router.get("/admin/fundamentals-parity", response_model=APIEnvelope)
+async def get_fundamentals_parity_report(
+    _admin: User = Depends(require_admin),
+):
+    """Latest read-only A5 source/score comparison, or null before first run."""
+    return APIEnvelope(
+        status="success", data=admin_service.get_fundamentals_parity_report()
+    )
+
+
+@router.get("/admin/fundamentals-parity/csv", response_model=APIEnvelope)
+async def get_fundamentals_parity_csv(
+    _admin: User = Depends(require_admin),
+):
+    """Latest flattened A5 report for an authenticated browser download."""
+    artifact = admin_service.get_fundamentals_parity_csv()
+    data = None if artifact is None else {"filename": artifact[0], "content": artifact[1]}
+    return APIEnvelope(status="success", data=data)
+
+
+@router.get("/admin/fundamentals-parity/json", response_model=APIEnvelope)
+async def get_fundamentals_parity_json(
+    _admin: User = Depends(require_admin),
+):
+    """Canonical A5 JSON artifact for an authenticated browser download."""
+    artifact = admin_service.get_fundamentals_parity_json()
+    data = None if artifact is None else {"filename": artifact[0], "content": artifact[1]}
+    return APIEnvelope(status="success", data=data)
+
+
 # ---------------------------------------------------------------------------
 # System events (operational warnings / errors)
 # ---------------------------------------------------------------------------
@@ -40,6 +40,7 @@ from app.services import (
    sentiment_service,
    settings_store,
    shadow_book_service,
+    fundamentals_parity_service,
 )
 from app.services.data_import import STATUS_FAILED, SourceImporter, run_import
 from app.services.dolt_earnings_importer import DoltEarningsImporter
@@ -98,6 +99,7 @@ _JOB_NAMES = [
    "fundamental_collector",
    "dolt_earnings_import",
    "sec_fundamentals_import",
+    "fundamentals_parity_report",
    "rr_scanner",
    "ticker_universe_sync",
    "alerts",
@@ -981,6 +983,49 @@ async def run_sec_fundamentals_import() -> None:
    await _run_shadow_import("sec_fundamentals_import", SecFundamentalsImporter())


+async def run_fundamentals_parity_report() -> None:
+    """Generate the A5 comparison bundle without mutating live fundamentals/scores."""
+    job_name = "fundamentals_parity_report"
+    _log_event(logging.INFO, "job_start", job=job_name)
+    _runtime_start(job_name, total=1)
+    try:
+        async with async_session_factory() as db:
+            if not await _is_job_enabled(db, job_name):
+                _runtime_finish(
+                    job_name, "skipped", processed=0, total=1, message="Disabled"
+                )
+                return
+            report, artifacts = await fundamentals_parity_service.generate_and_store(
+                db, settings.fundamentals_parity_report_dir
+            )
+        summary = report["summary"]
+        message = (
+            f"{summary['universe_count']} tickers · "
+            f"{summary['fundamental_score_material_changes']} material score changes"
+        )
+        _runtime_finish(job_name, "completed", processed=1, total=1, message=message)
+        _log_event(
+            logging.INFO,
+            "job_complete",
+            job=job_name,
+            generated_at=report["generated_at"],
+            json_path=artifacts["json"],
+            csv_path=artifacts["csv"],
+        )
+    except asyncio.CancelledError:
+        _runtime_finish(job_name, "error", processed=0, total=1, message="Cancelled")
+        raise
+    except Exception as exc:
+        _runtime_finish(job_name, "error", processed=0, total=1, message=str(exc))
+        _log_event(
+            logging.ERROR,
+            "job_error",
+            job=job_name,
+            error_type=type(exc).__name__,
+            message=str(exc),
+        )
+
+
 # ---------------------------------------------------------------------------
 # Job: R:R Scanner
 # ---------------------------------------------------------------------------
@@ -1524,6 +1569,7 @@ SCHEDULE_DEFAULTS: dict[str, str] = {
    # Shadow source imports. They never write legacy fundamental_data before A5.
    "schedule_dolt_earnings_cron": "30 2 * * *",
    "schedule_sec_fundamentals_cron": "0 4 * * *",
+    "schedule_fundamentals_parity_cron": "30 5 * * *",
    # Fetch in-progress bars → scan → Telegram (manual MOC window).
    "schedule_near_close_pipeline_cron": "30 15 * * mon-fri",
    # Fetch final bars → outcome eval (must not run on the partial near-close bar).
@@ -1539,6 +1585,7 @@ _CRON_JOBS: dict[str, str] = {
    "daily_pipeline": "schedule_daily_pipeline_cron",
    "dolt_earnings_import": "schedule_dolt_earnings_cron",
    "sec_fundamentals_import": "schedule_sec_fundamentals_cron",
+    "fundamentals_parity_report": "schedule_fundamentals_parity_cron",
    "near_close_pipeline": "schedule_near_close_pipeline_cron",
    "after_close_pipeline": "schedule_after_close_pipeline_cron",
    "intraday_pipeline": "schedule_intraday_pipeline_cron",
@@ -1645,6 +1692,17 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
        name="SEC Fundamentals Import (shadow)",
        replace_existing=True,
    )
+    scheduler.add_job(
+        run_fundamentals_parity_report,
+        _cron_trigger(
+            cfg["schedule_fundamentals_parity_cron"],
+            tz,
+            "schedule_fundamentals_parity_cron",
+        ),
+        id="fundamentals_parity_report",
+        name="Fundamentals Parity Report (read-only)",
+        replace_existing=True,
+    )
    scheduler.add_job(
        run_near_close_pipeline,
        _cron_trigger(
@@ -1720,6 +1778,9 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
        },
        dolt_earnings_import={"cron": cfg["schedule_dolt_earnings_cron"]},
        sec_fundamentals_import={"cron": cfg["schedule_sec_fundamentals_cron"]},
+        fundamentals_parity_report={
+            "cron": cfg["schedule_fundamentals_parity_cron"]
+        },
        near_close_pipeline={
            "cron": cfg["schedule_near_close_pipeline_cron"],
            "steps": [name for name, _ in _NEAR_CLOSE_PIPELINE_STEPS],
@@ -78,6 +78,9 @@ class ScheduleConfigUpdate(BaseModel):
    (min hour dom month dow); timezone is an IANA name (e.g. America/New_York)."""
    schedule_timezone: str | None = Field(default=None, max_length=64)
    schedule_daily_pipeline_cron: str | None = Field(default=None, max_length=120)
+    schedule_dolt_earnings_cron: str | None = Field(default=None, max_length=120)
+    schedule_sec_fundamentals_cron: str | None = Field(default=None, max_length=120)
+    schedule_fundamentals_parity_cron: str | None = Field(default=None, max_length=120)
    schedule_near_close_pipeline_cron: str | None = Field(default=None, max_length=120)
    schedule_after_close_pipeline_cron: str | None = Field(default=None, max_length=120)
    schedule_intraday_pipeline_cron: str | None = Field(default=None, max_length=120)
@@ -26,6 +26,7 @@ class MetricItem(BaseModel):
    industry: MetricIndustry | None = None
    period_end: str | None = None
    filed_date: str | None = None
+    caveat: str | None = None
    source: str = "sec"


@@ -614,6 +614,7 @@ VALID_JOB_NAMES = {
    "fundamental_collector",
    "dolt_earnings_import",
    "sec_fundamentals_import",
+    "fundamentals_parity_report",
    "rr_scanner",
    "ticker_universe_sync",
    "outcome_evaluator",
@@ -637,6 +638,7 @@ JOB_LABELS = {
    "fundamental_collector": "Fundamental Collector",
    "dolt_earnings_import": "Dolt Earnings Import (shadow)",
    "sec_fundamentals_import": "SEC Fundamentals Import (shadow)",
+    "fundamentals_parity_report": "Fundamentals Parity Report (read-only)",
    "rr_scanner": "R:R Scanner",
    "ticker_universe_sync": "Ticker Universe Sync",
    "outcome_evaluator": "Outcome Evaluator",
@@ -775,3 +777,30 @@ async def toggle_job(db: AsyncSession, job_name: str, enabled: bool) -> SystemSe

    key = f"job_{job_name}_enabled"
    return await update_setting(db, key, str(enabled).lower())
+
+
+def get_fundamentals_parity_report() -> dict | None:
+    """Return the latest compact A5 summary, if the job has run."""
+    from app.config import settings
+    from app.services.fundamentals_parity_service import load_latest
+
+    report = load_latest(settings.fundamentals_parity_report_dir)
+    if report is not None:
+        report.pop("rows", None)  # full per-ticker data is download-only
+    return report
+
+
+def get_fundamentals_parity_csv() -> tuple[str, str] | None:
+    """Return the latest A5 CSV filename and content for authenticated download."""
+    from app.config import settings
+    from app.services.fundamentals_parity_service import load_latest_csv
+
+    return load_latest_csv(settings.fundamentals_parity_report_dir)
+
+
+def get_fundamentals_parity_json() -> tuple[str, str] | None:
+    """Return the canonical A5 JSON artifact for authenticated download."""
+    from app.config import settings
+    from app.services.fundamentals_parity_service import load_latest_json
+
+    return load_latest_json(settings.fundamentals_parity_report_dir)
@@ -121,6 +121,7 @@ def _build_metrics(derived, peer_derived, two: str | None) -> list[dict[str, Any
            "industry": industry,
            "period_end": _iso(series.period_end) if series else None,
            "filed_date": _iso(series.filed_date) if series else None,
+            "caveat": series.caveat if series else None,
            "source": "sec",
        })
    return out
@@ -303,7 +304,8 @@ async def _latest_close(db, ticker_id: int) -> tuple[float, date] | None:

 def _empty_metrics() -> list[dict[str, Any]]:
    return [{"key": k, "value": None, "history": [], "industry": None,
-             "period_end": None, "filed_date": None, "source": "sec"} for k in METRIC_KEYS]
+             "period_end": None, "filed_date": None, "caveat": None,
+             "source": "sec"} for k in METRIC_KEYS]


 def _empty_earnings() -> dict[str, Any]:
@@ -27,6 +27,11 @@ _FP_TO_Q = {"Q1": 1, "Q2": 2, "Q3": 3, "FY": 4}
 _Q_TO_FP = {1: "Q1", 2: "Q2", 3: "Q3", 4: "FY"}
 _PREV_FP = {"Q2": "Q1", "Q3": "Q2", "FY": "Q3"}
 TAPE_LEN = 4  # quarter-tape length
+SPLIT_SUSPECT_SHARE_CHANGE_PCT = 25.0
+SPLIT_SENSITIVE_CAVEAT = (
+    "Not comparable: share count changed at least 25%; possible split or "
+    "corporate action."
+)

 # Duration (flow) fields differenced from YTD into discrete quarters + summed to TTM.
 _FLOW_FIELDS = (
@@ -47,6 +52,7 @@ class MetricSeries:
    history: list[MetricPoint] = field(default_factory=list)  # oldest -> newest, <= TAPE_LEN
    period_end: date | None = None
    filed_date: date | None = None
+    caveat: str | None = None


@dataclass
@@ -96,6 +102,7 @@ def derive(snapshots: Iterable[Any]) -> DerivedFundamentals:
        "net_debt_to_ebitda": _leverage_series(selected, discrete, tape),
        "share_count_change_yoy": _share_change_series(selected, tape),
    }
+    _guard_split_sensitive_metrics(result.metrics)
    for series in result.metrics.values():
        series.period_end = latest_row.period_end
        series.filed_date = latest_row.filed_date
@@ -248,6 +255,40 @@ def _share_change_series(selected, tape) -> MetricSeries:
    return _series(pts)


+def _guard_split_sensitive_metrics(metrics: dict[str, MetricSeries]) -> None:
+    """Suppress historical comparisons likely distorted by a corporate action.
+
+    Company Facts has no point-in-time split factors. A large YoY share-count
+    move can therefore make both the point-in-time share comparison and
+    per-share EPS growth non-comparable. Keep the raw facts in snapshots, but
+    expose nulls plus an explicit caveat in the user-facing derived series.
+    """
+    shares = metrics.get("share_count_change_yoy")
+    eps = metrics.get("eps_growth_yoy")
+    if shares is None or eps is None:
+        return
+
+    suspect_periods = {
+        point.period_end
+        for point in shares.history
+        if point.value is not None
+        and abs(point.value) >= SPLIT_SUSPECT_SHARE_CHANGE_PCT
+    }
+    if not suspect_periods:
+        return
+
+    for series in (shares, eps):
+        latest_guarded = bool(
+            series.history and series.history[-1].period_end in suspect_periods
+        )
+        for point in series.history:
+            if point.period_end in suspect_periods:
+                point.value = None
+        series.value = series.history[-1].value if series.history else None
+        if latest_guarded:
+            series.caveat = SPLIT_SENSITIVE_CAVEAT
+
+
 def _net_debt(row: Any) -> float | None:
    if row is None:
        return None
@@ -0,0 +1,598 @@
+"""Read-only A5 comparison of legacy and SEC/Dolt fundamental inputs.
+
+The report deliberately does not write ``fundamental_data`` or score tables.
+It reconstructs the current legacy and candidate fundamental scores, projects
+their composite-score/rank effect with the active weights, and archives a
+timestamped JSON + CSV bundle for explicit human approval.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+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.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
+
+REPORT_VERSION = 1
+APPROVAL_STATUS = "pending_explicit_approval"
+FIELD_KEYS = ("pe_ratio", "revenue_growth", "earnings_surprise")
+MIN_SCORE_METRICS = 2
+
+# Materiality is a review aid, never an automatic cutover verdict. Definition
+# changes remain visible even when a delta falls inside these bands.
+FIELD_TOLERANCES = {
+    "pe_ratio": {"absolute": 1.0, "relative_pct": 10.0},
+    "revenue_growth": {"absolute": 2.0, "relative_pct": None},
+    "earnings_surprise": {"absolute": 2.0, "relative_pct": None},
+}
+DEFINITION_NOTES = {
+    "pe_ratio": (
+        "Legacy provider P/E convention versus latest close divided by "
+        "SEC-derived TTM diluted EPS."
+    ),
+    "revenue_growth": (
+        "Legacy provider growth convention versus SEC-derived TTM revenue YoY."
+    ),
+    "earnings_surprise": (
+        "Legacy provider latest surprise versus latest completed Dolt earnings "
+        "event with actual and estimate."
+    ),
+}
+
+
+def fundamental_score(
+    pe_ratio: float | None,
+    revenue_growth: float | None,
+    earnings_surprise: float | None,
+) -> float | None:
+    """Match the production fundamental-dimension formula without persistence."""
+    scores: list[float] = []
+    if _finite(pe_ratio) and pe_ratio > 0:
+        scores.append(max(0.0, min(100.0, 100.0 - (pe_ratio - 15.0) * (100.0 / 30.0))))
+    if _finite(revenue_growth):
+        scores.append(max(0.0, min(100.0, 50.0 + revenue_growth * 2.5)))
+    if _finite(earnings_surprise):
+        scores.append(max(0.0, min(100.0, 50.0 + earnings_surprise * 5.0)))
+    return sum(scores) / len(scores) if len(scores) >= MIN_SCORE_METRICS else None
+
+
+async def build_report(
+    db: AsyncSession,
+    *,
+    generated_at: datetime | None = None,
+    today: date | None = None,
+) -> dict[str, Any]:
+    """Build a point-in-time parity report from one database session."""
+    generated_at = generated_at or datetime.now(timezone.utc)
+    today = today or datetime.now(ZoneInfo("America/New_York")).date()
+
+    # A report must not mix rows from before and after a concurrent import
+    # promotion. The scheduled job provides a fresh session, so establish the
+    # production snapshot before its first query and have Postgres enforce the
+    # no-write contract as well. SQLite tests retain their normal transaction.
+    if db.get_bind().dialect.name == "postgresql":
+        connection = await db.connection(
+            execution_options={"isolation_level": "REPEATABLE READ"}
+        )
+        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})
+
+    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),
+        }
+        legacy_values = {
+            "pe_ratio": legacy.pe_ratio if legacy else None,
+            "revenue_growth": legacy.revenue_growth if legacy else None,
+            "earnings_surprise": legacy.earnings_surprise if legacy else None,
+        }
+        fields = {
+            key: _field_comparison(key, legacy_values[key], candidate[key])
+            for key in FIELD_KEYS
+        }
+        legacy_score = fundamental_score(**legacy_values)
+        candidate_score = fundamental_score(**candidate)
+        rows.append(
+            {
+                "symbol": ticker.symbol,
+                "cik": ticker.cik,
+                "legacy_fetched_at": _iso(legacy.fetched_at) if legacy else None,
+                "price_date": _iso(close[1]) if close else None,
+                "fields": fields,
+                "scores": {
+                    "legacy_fundamental": _round(legacy_score),
+                    "candidate_fundamental": _round(candidate_score),
+                    "fundamental_delta": _delta(legacy_score, candidate_score),
+                    "legacy_fundamental_rank": None,
+                    "candidate_fundamental_rank": None,
+                    "fundamental_rank_change": None,
+                },
+            }
+        )
+
+    _attach_ranks(rows, "legacy_fundamental", "legacy_fundamental_rank")
+    _attach_ranks(rows, "candidate_fundamental", "candidate_fundamental_rank")
+    for row in rows:
+        scores = row["scores"]
+        scores["fundamental_rank_change"] = _rank_change(
+            scores["legacy_fundamental_rank"], scores["candidate_fundamental_rank"]
+        )
+
+    return {
+        "report_version": REPORT_VERSION,
+        "generated_at": generated_at.isoformat(),
+        "as_of_date": today.isoformat(),
+        "approval_status": APPROVAL_STATUS,
+        "read_only": True,
+        "fundamental_score_formula": (
+            "Equal-weighted mean of 2+ available sub-scores: P/E = "
+            "clamp(100-(pe-15)*(100/30)); revenue growth = "
+            "clamp(50+growth*2.5); earnings surprise = "
+            "clamp(50+surprise*5)."
+        ),
+        "source_runs": source_runs,
+        "definition_notes": DEFINITION_NOTES,
+        "materiality_notes": {
+            "fields": FIELD_TOLERANCES,
+            "fundamental_score_absolute": 5.0,
+            "automatic_cutover": False,
+        },
+        "summary": _summary(rows),
+        "rows": rows,
+    }
+
+
+def store_report(report: dict[str, Any], report_dir: str | Path) -> dict[str, str]:
+    """Atomically archive JSON/CSV artifacts and update the latest manifest."""
+    directory = Path(report_dir).expanduser().resolve()
+    directory.mkdir(parents=True, exist_ok=True)
+    stamp = _artifact_stamp(report["generated_at"])
+    json_name = f"fundamentals-parity-{stamp}.json"
+    csv_name = f"fundamentals-parity-{stamp}.csv"
+    json_path = directory / json_name
+    csv_path = directory / csv_name
+
+    _atomic_write(json_path, json.dumps(report, indent=2, sort_keys=True) + "\n")
+    _atomic_write(csv_path, report_csv(report))
+    manifest = {
+        "generated_at": report["generated_at"],
+        "json_file": json_name,
+        "csv_file": csv_name,
+    }
+    _atomic_write(
+        directory / "latest.json",
+        json.dumps(manifest, indent=2, sort_keys=True) + "\n",
+    )
+    return {
+        "json": str(json_path),
+        "csv": str(csv_path),
+        "manifest": str(directory / "latest.json"),
+    }
+
+
+async def generate_and_store(
+    db: AsyncSession,
+    report_dir: str | Path,
+    *,
+    generated_at: datetime | None = None,
+    today: date | None = None,
+) -> tuple[dict[str, Any], dict[str, str]]:
+    report = await build_report(db, generated_at=generated_at, today=today)
+    return report, store_report(report, report_dir)
+
+
+def load_latest(report_dir: str | Path) -> dict[str, Any] | None:
+    manifest = _load_manifest(report_dir)
+    if manifest is None:
+        return None
+    try:
+        path = _manifest_artifact(report_dir, manifest, "json_file")
+        loaded = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError, TypeError, ValueError):
+        return None
+    return loaded if isinstance(loaded, dict) else None
+
+
+def load_latest_csv(report_dir: str | Path) -> tuple[str, str] | None:
+    return _load_latest_text_artifact(report_dir, "csv_file")
+
+
+def load_latest_json(report_dir: str | Path) -> tuple[str, str] | None:
+    return _load_latest_text_artifact(report_dir, "json_file")
+
+
+def _load_latest_text_artifact(
+    report_dir: str | Path, manifest_key: str
+) -> tuple[str, str] | None:
+    manifest = _load_manifest(report_dir)
+    if manifest is None:
+        return None
+    try:
+        path = _manifest_artifact(report_dir, manifest, manifest_key)
+        return path.name, path.read_text(encoding="utf-8")
+    except (OSError, TypeError, ValueError):
+        return None
+
+
+def report_csv(report: dict[str, Any]) -> str:
+    output = io.StringIO(newline="")
+    columns = [
+        "symbol",
+        "cik",
+        "legacy_fetched_at",
+        "price_date",
+        *(
+            f"{field}_{suffix}"
+            for field in FIELD_KEYS
+            for suffix in ("legacy", "candidate", "absolute_delta", "relative_delta_pct", "material")
+        ),
+        "legacy_fundamental",
+        "candidate_fundamental",
+        "fundamental_delta",
+        "legacy_fundamental_rank",
+        "candidate_fundamental_rank",
+        "fundamental_rank_change",
+    ]
+    writer = csv.DictWriter(output, fieldnames=columns)
+    writer.writeheader()
+    for row in report.get("rows", []):
+        flat = {
+            "symbol": row["symbol"],
+            "cik": row.get("cik"),
+            "legacy_fetched_at": row.get("legacy_fetched_at"),
+            "price_date": row.get("price_date"),
+            **row["scores"],
+        }
+        for field in FIELD_KEYS:
+            comparison = row["fields"][field]
+            for suffix in (
+                "legacy",
+                "candidate",
+                "absolute_delta",
+                "relative_delta_pct",
+                "material",
+            ):
+                flat[f"{field}_{suffix}"] = comparison.get(suffix)
+        writer.writerow(flat)
+    return output.getvalue()
+
+
+async def _legacy_values(
+    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 _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 = (
+        await db.execute(
+            select(DataImportRun)
+            .where(
+                DataImportRun.source.in_(sources),
+                DataImportRun.status.in_(("promoted", "no_op")),
+            )
+            .order_by(DataImportRun.id.desc())
+        )
+    ).scalars()
+    latest: dict[str, dict[str, Any] | None] = {source: None for source in sources}
+    for row in rows:
+        if latest[row.source] is None:
+            latest[row.source] = {
+                "run_id": row.id,
+                "status": row.status,
+                "revision": row.revision,
+                "source_max_date": _iso(row.source_max_date),
+                "completed_at": _iso(row.completed_at),
+            }
+    return latest
+
+
+def _field_comparison(
+    key: str, legacy: float | None, candidate: float | None
+) -> dict[str, Any]:
+    legacy = float(legacy) if _finite(legacy) else None
+    candidate = float(candidate) if _finite(candidate) else None
+    absolute = _delta(legacy, candidate)
+    relative = (
+        None
+        if absolute is None or legacy in (None, 0)
+        else round(absolute / abs(legacy) * 100.0, 4)
+    )
+    tolerance = FIELD_TOLERANCES[key]
+    material = False
+    if absolute is not None:
+        material = abs(absolute) > tolerance["absolute"]
+        relative_limit = tolerance["relative_pct"]
+        if relative_limit is not None:
+            material = material and relative is not None and abs(relative) > relative_limit
+    return {
+        "legacy": _round(legacy),
+        "candidate": _round(candidate),
+        "absolute_delta": absolute,
+        "relative_delta_pct": relative,
+        "material": material,
+        "definition_changed": True,
+    }
+
+
+def _attach_ranks(rows: list[dict[str, Any]], value_key: str, rank_key: str) -> None:
+    values = [
+        row["scores"][value_key]
+        for row in rows
+        if _finite(row["scores"][value_key])
+    ]
+    for row in rows:
+        value = row["scores"][value_key]
+        row["scores"][rank_key] = (
+            1 + sum(other > value for other in values) if _finite(value) else None
+        )
+
+
+def _summary(rows: list[dict[str, Any]]) -> dict[str, Any]:
+    field_stats = {}
+    for key in FIELD_KEYS:
+        comparisons = [row["fields"][key] for row in rows]
+        deltas = [
+            abs(item["absolute_delta"])
+            for item in comparisons
+            if item["absolute_delta"] is not None
+        ]
+        field_stats[key] = {
+            "legacy_available": sum(item["legacy"] is not None for item in comparisons),
+            "candidate_available": sum(
+                item["candidate"] is not None for item in comparisons
+            ),
+            "both_available": len(deltas),
+            "material_differences": sum(item["material"] for item in comparisons),
+            "median_absolute_delta": _round(statistics.median(deltas) if deltas else None),
+            "p95_absolute_delta": _round(_percentile(deltas, 0.95)),
+            "max_absolute_delta": _round(max(deltas) if deltas else None),
+        }
+
+    fundamental_deltas = _score_deltas(rows, "fundamental_delta")
+    changed_rows = sorted(
+        (
+            {
+                "symbol": row["symbol"],
+                "fundamental_delta": row["scores"]["fundamental_delta"],
+                "fundamental_rank_change": row["scores"]["fundamental_rank_change"],
+            }
+            for row in rows
+            if row["scores"]["fundamental_delta"] is not None
+        ),
+        key=lambda item: (
+            abs(item["fundamental_delta"] or 0),
+        ),
+        reverse=True,
+    )[:20]
+    return {
+        "universe_count": len(rows),
+        "legacy_fundamental_score_available": _count_score(
+            rows, "legacy_fundamental"
+        ),
+        "candidate_fundamental_score_available": _count_score(
+            rows, "candidate_fundamental"
+        ),
+        "fundamental_scores_compared": len(fundamental_deltas),
+        "fundamental_score_material_changes": sum(
+            abs(delta) > 5.0 for delta in fundamental_deltas
+        ),
+        "fundamental_rank_changes": _rank_change_count(
+            rows, "fundamental_rank_change"
+        ),
+        "field_stats": field_stats,
+        "largest_changes": changed_rows,
+    }
+
+
+def _score_deltas(rows: Iterable[dict[str, Any]], key: str) -> list[float]:
+    return [
+        row["scores"][key]
+        for row in rows
+        if row["scores"][key] is not None
+    ]
+
+
+def _count_score(rows: Iterable[dict[str, Any]], key: str) -> int:
+    return sum(row["scores"][key] is not None for row in rows)
+
+
+def _rank_change_count(rows: Iterable[dict[str, Any]], key: str) -> int:
+    return sum(
+        row["scores"][key] not in (None, 0)
+        for row in rows
+    )
+
+
+def _rank_change(legacy: int | None, candidate: int | None) -> int | None:
+    # Positive means the candidate improved its rank.
+    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
+    return round(candidate - legacy, 4)
+
+
+def _round(value: float | None, digits: int = 4) -> float | None:
+    return round(float(value), digits) if _finite(value) else None
+
+
+def _percentile(values: list[float], quantile: float) -> float | None:
+    if not values:
+        return None
+    ordered = sorted(values)
+    index = max(0, math.ceil(quantile * len(ordered)) - 1)
+    return ordered[index]
+
+
+def _finite(value: Any) -> bool:
+    return (
+        isinstance(value, (int, float))
+        and not isinstance(value, bool)
+        and math.isfinite(value)
+    )
+
+
+def _iso(value: Any) -> str | None:
+    return value.isoformat() if value is not None else None
+
+
+def _artifact_stamp(raw: str) -> str:
+    parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+    return parsed.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
+
+
+def _atomic_write(path: Path, content: str) -> None:
+    temp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
+    temp.write_text(content, encoding="utf-8", newline="")
+    os.replace(temp, path)
+
+
+def _load_manifest(report_dir: str | Path) -> dict[str, Any] | None:
+    path = Path(report_dir).expanduser().resolve() / "latest.json"
+    try:
+        loaded = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError, TypeError, ValueError):
+        return None
+    return loaded if isinstance(loaded, dict) else None
+
+
+def _manifest_artifact(
+    report_dir: str | Path, manifest: dict[str, Any], key: str
+) -> Path:
+    directory = Path(report_dir).expanduser().resolve()
+    name = Path(str(manifest.get(key, ""))).name
+    if not name:
+        raise ValueError(f"Latest parity manifest has no {key}")
+    return directory / name
@@ -15,6 +15,7 @@ MIN_FREE_GB="${DOLT_MIN_FREE_DISK_GB:-5}"
 EARNINGS_DIR="${DOLT_DATA_DIR}/${DOLT_EARNINGS_SUBDIR}"
 DOLT_IDENTITY_NAME="${DOLT_IDENTITY_NAME:-Signal Platform}"
 DOLT_IDENTITY_EMAIL="${DOLT_IDENTITY_EMAIL:-signal-platform@localhost}"
+FUNDAMENTALS_PARITY_REPORT_DIR="${FUNDAMENTALS_PARITY_REPORT_DIR:-/var/lib/signal-platform/reports/fundamentals-parity}"

 fail() {
  echo "ERROR: $*" >&2
@@ -79,6 +80,8 @@ check_env() {
    || fail "set DOLT_EARNINGS_SUBDIR=$DOLT_EARNINGS_SUBDIR in $ENV_FILE"
  grep -Eq '^SEC_USER_AGENT=.*@.*' "$ENV_FILE" \
    || fail "SEC_USER_AGENT in $ENV_FILE must contain a real contact email"
+  grep -Fqx "FUNDAMENTALS_PARITY_REPORT_DIR=$FUNDAMENTALS_PARITY_REPORT_DIR" "$ENV_FILE" \
+    || fail "set FUNDAMENTALS_PARITY_REPORT_DIR=$FUNDAMENTALS_PARITY_REPORT_DIR in $ENV_FILE"
 }

 check_all() {
@@ -101,6 +104,15 @@ check_all() {
  identity_email="$(repo_config_value user.email 2>/dev/null || true)"
  [[ -n "$identity_name" ]] || fail "missing Dolt user.name for $EARNINGS_DIR"
  [[ -n "$identity_email" ]] || fail "missing Dolt user.email for $EARNINGS_DIR"
+  [[ -d "$FUNDAMENTALS_PARITY_REPORT_DIR" ]] \
+    || fail "missing parity report directory: $FUNDAMENTALS_PARITY_REPORT_DIR"
+  if [[ "$(id -un)" == "$APP_USER" ]]; then
+    [[ -w "$FUNDAMENTALS_PARITY_REPORT_DIR" ]] \
+      || fail "parity report directory is not writable by $APP_USER"
+  else
+    runuser -u "$APP_USER" -- test -w "$FUNDAMENTALS_PARITY_REPORT_DIR" \
+      || fail "parity report directory is not writable by $APP_USER"
+  fi
  check_free_space
  check_env
  echo "OK: Dolt $DOLT_VERSION and earnings clone are provisioned"
@@ -127,6 +139,7 @@ fi
 version_ok || fail "Dolt $DOLT_VERSION installation failed"

 install -d -o "$APP_USER" -g "$APP_GROUP" -m 0750 "$DOLT_DATA_DIR"
+install -d -o "$APP_USER" -g "$APP_GROUP" -m 0750 "$FUNDAMENTALS_PARITY_REPORT_DIR"
 check_free_space

 if [[ ! -d "$EARNINGS_DIR/.dolt" ]]; then
@@ -426,6 +426,8 @@ workstream B — Alpaca remains the price source throughout.
  API values across the tracked universe, report per-field deltas and resulting
  fundamental-score/ranking changes, require explicit approval. Definition
  changes (e.g. TTM vs provider convention) called out, not averaged away.
+  The read-only report job and Admin summary/download are implemented; production
+  observation and explicit cutover approval remain pending.
 - A6. Remove FMP/Finnhub/Alpha Vantage; keep monitoring + manual fallback.

 **Workstream B (independent, start when wanted):**
@@ -8,6 +8,7 @@ approval. Do not add OS cron entries: the application scheduler owns both jobs.

 - `Dolt Earnings Import (shadow)` runs daily at 02:30 America/New_York.
 - `SEC Fundamentals Import (shadow)` runs daily at 04:00 America/New_York.
+- `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.
 - Every attempt is recorded in `data_import_runs`; failures also create a system
@@ -28,11 +29,14 @@ DOLT_EARNINGS_SUBDIR=earnings
 DOLT_MIN_FREE_DISK_GB=5.0
 SEC_USER_AGENT=signal-platform/1.0 (contact: real-address@example.com)
 SEC_REQUEST_SPACING_SECONDS=0.2
+FUNDAMENTALS_PARITY_REPORT_DIR=/var/lib/signal-platform/reports/fundamentals-parity
 ```

 Use a real monitored contact address. Keep at least 5 GB free at the Dolt data
 path; 8–10 GB gives comfortable growth headroom. The data directory must stay
 outside `/opt/signalplatform`, because deployments use `rsync --delete` there.
+The parity-report directory is also persistent and owned by the service user;
+its small timestamped JSON/CSV bundles form the temporary A5 review trail.

 ## One-time provisioning

@@ -79,6 +83,28 @@ In Admin → Jobs, wait until no other job is running, then:
 5. Open several ticker pages and confirm the fundamentals panel has populated
   data and still handles partial/missing issuers cleanly.

+## A5 parity observation window
+
+After both shadow imports are healthy, trigger **Fundamentals Parity Report
+(read-only)** once in Admin → Jobs. The **A5 Fundamentals Parity** card above
+the jobs shows the latest coverage/delta summary and provides authenticated JSON
+and CSV downloads. The canonical server-side bundles are archived at:
+
+```text
+/var/lib/signal-platform/reports/fundamentals-parity/
+```
+
+The scheduler then generates one report daily at 05:30 New York time, after the
+02:30 Dolt and 04:00 SEC jobs. Review 5–7 consecutive reports before making the
+cutover decision. A report never writes `fundamental_data`, dimension/composite
+scores, rankings, qualification state, or an approval flag. Materiality bands
+only highlight rows for review; A5 still requires explicit approval.
+
+Each bundle contains legacy and candidate P/E, revenue growth, and earnings
+surprise; definition notes; source revisions and price dates; recomputed legacy
+and candidate fundamental scores; and per-universe fundamental-rank changes.
+Definition changes remain explicit even when numeric deltas are small.
+
 Optional database verification:

 ```sql
@@ -141,3 +167,6 @@ locks. A second Admin trigger should independently report the job as busy.
  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.
+- 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.
@@ -233,6 +233,40 @@ export interface TriggerJobResponse {
  cadence?: BacktestCadence;
 }

+export interface ParityFieldStats {
+  legacy_available: number;
+  candidate_available: number;
+  both_available: number;
+  material_differences: number;
+  median_absolute_delta: number | null;
+  p95_absolute_delta: number | null;
+  max_absolute_delta: number | null;
+}
+
+export interface FundamentalsParityReport {
+  report_version: number;
+  generated_at: string;
+  as_of_date: string;
+  approval_status: string;
+  read_only: boolean;
+  summary: {
+    universe_count: number;
+    legacy_fundamental_score_available: number;
+    candidate_fundamental_score_available: number;
+    fundamental_scores_compared: number;
+    fundamental_score_material_changes: number;
+    fundamental_rank_changes: number;
+    field_stats: Record<string, ParityFieldStats>;
+  };
+  source_runs: Record<string, {
+    run_id: number;
+    status: string;
+    revision: string | null;
+    source_max_date: string | null;
+    completed_at: string | null;
+  } | null>;
+}
+
 export type BacktestTargetModel = 'production_gtl' | 'structural_sr';
 export type BacktestCadence = 'weekly' | 'daily';

@@ -259,6 +293,24 @@ export function triggerJob(
    .then((r) => r.data);
 }

+export function getFundamentalsParityReport() {
+  return apiClient
+    .get<FundamentalsParityReport | null>('admin/fundamentals-parity')
+    .then((r) => r.data);
+}
+
+export function getFundamentalsParityCsv() {
+  return apiClient
+    .get<{ filename: string; content: string } | null>('admin/fundamentals-parity/csv')
+    .then((r) => r.data);
+}
+
+export function getFundamentalsParityJson() {
+  return apiClient
+    .get<{ filename: string; content: string } | null>('admin/fundamentals-parity/json')
+    .then((r) => r.data);
+}
+
 // System events (operational warnings / errors)
 export interface SystemEvent {
  id: number;
@@ -0,0 +1,156 @@
+import { useState } from 'react';
+import {
+  getFundamentalsParityCsv,
+  getFundamentalsParityJson,
+} from '../../api/admin';
+import { useFundamentalsParityReport } from '../../hooks/useAdmin';
+import { SkeletonTable } from '../ui/Skeleton';
+
+const FIELD_LABELS: Record<string, string> = {
+  pe_ratio: 'P/E',
+  revenue_growth: 'Revenue growth',
+  earnings_surprise: 'Earnings surprise',
+};
+
+function downloadText(filename: string, content: string, type: string) {
+  const blob = new Blob([content], { type });
+  const url = URL.createObjectURL(blob);
+  const anchor = document.createElement('a');
+  anchor.href = url;
+  anchor.download = filename;
+  anchor.click();
+  URL.revokeObjectURL(url);
+}
+
+export function FundamentalsParityPanel() {
+  const { data: report, isLoading, isError, error } = useFundamentalsParityReport();
+  const [downloading, setDownloading] = useState(false);
+
+  if (isLoading) return <SkeletonTable rows={2} cols={4} />;
+  if (isError) {
+    return <p className="text-sm text-red-400">{(error as Error).message}</p>;
+  }
+
+  if (!report) {
+    return (
+      <div className="glass p-5">
+        <h3 className="text-sm font-semibold text-gray-200">A5 Fundamentals Parity</h3>
+        <p className="mt-1 text-xs text-gray-500">
+          No report yet. Trigger “Fundamentals Parity Report (read-only)” below.
+        </p>
+      </div>
+    );
+  }
+
+  const summary = report.summary;
+  const generated = new Date(report.generated_at).toLocaleString();
+
+  async function downloadCsv() {
+    setDownloading(true);
+    try {
+      const artifact = await getFundamentalsParityCsv();
+      if (artifact) downloadText(artifact.filename, artifact.content, 'text/csv;charset=utf-8');
+    } finally {
+      setDownloading(false);
+    }
+  }
+
+  async function downloadJson() {
+    setDownloading(true);
+    try {
+      const artifact = await getFundamentalsParityJson();
+      if (artifact) downloadText(artifact.filename, artifact.content, 'application/json;charset=utf-8');
+    } finally {
+      setDownloading(false);
+    }
+  }
+
+  return (
+    <div className="glass p-5 space-y-4">
+      <div className="flex flex-wrap items-start justify-between gap-3">
+        <div>
+          <div className="flex flex-wrap items-center gap-2">
+            <h3 className="text-sm font-semibold text-gray-200">A5 Fundamentals Parity</h3>
+            <span className="rounded-full border border-amber-400/20 bg-amber-400/10 px-2 py-0.5 text-[10px] uppercase tracking-wide text-amber-300">
+              approval pending
+            </span>
+            <span className="rounded-full border border-cyan-400/20 bg-cyan-400/10 px-2 py-0.5 text-[10px] uppercase tracking-wide text-cyan-300">
+              read-only
+            </span>
+          </div>
+          <p className="mt-1 text-xs text-gray-500">
+            Generated {generated} · as of {report.as_of_date} · {summary.universe_count} tracked tickers
+          </p>
+        </div>
+        <div className="flex gap-2">
+          <button
+            type="button"
+            className="rounded border border-white/10 px-3 py-1.5 text-xs text-gray-300 hover:text-white"
+            onClick={downloadJson}
+            disabled={downloading}
+          >
+            Download JSON
+          </button>
+          <button
+            type="button"
+            className="rounded border border-white/10 px-3 py-1.5 text-xs text-gray-300 hover:text-white disabled:opacity-50"
+            onClick={downloadCsv}
+            disabled={downloading}
+          >
+            {downloading ? 'Preparing…' : 'Download CSV'}
+          </button>
+        </div>
+      </div>
+
+      <div className="grid gap-3 sm:grid-cols-2 lg:grid-cols-4">
+        <Summary label="Candidate score coverage" value={`${summary.candidate_fundamental_score_available}/${summary.universe_count}`} />
+        <Summary label="Scores compared" value={summary.fundamental_scores_compared} />
+        <Summary label="Material score moves" value={summary.fundamental_score_material_changes} />
+        <Summary label="Fundamental rank moves" value={summary.fundamental_rank_changes} />
+      </div>
+
+      <div className="overflow-x-auto">
+        <table className="w-full text-left text-xs">
+          <thead className="text-[10px] uppercase tracking-wider text-gray-500">
+            <tr>
+              <th className="pb-2 pr-4 font-medium">Field</th>
+              <th className="pb-2 px-3 font-medium">Legacy</th>
+              <th className="pb-2 px-3 font-medium">Candidate</th>
+              <th className="pb-2 px-3 font-medium">Compared</th>
+              <th className="pb-2 px-3 font-medium">Material</th>
+              <th className="pb-2 pl-3 font-medium">Median |Δ|</th>
+            </tr>
+          </thead>
+          <tbody className="divide-y divide-white/[0.06] text-gray-300">
+            {Object.entries(summary.field_stats).map(([key, stats]) => (
+              <tr key={key}>
+                <td className="py-2.5 pr-4">{FIELD_LABELS[key] ?? key}</td>
+                <td className="py-2.5 px-3 num">{stats.legacy_available}</td>
+                <td className="py-2.5 px-3 num">{stats.candidate_available}</td>
+                <td className="py-2.5 px-3 num">{stats.both_available}</td>
+                <td className="py-2.5 px-3 num">{stats.material_differences}</td>
+                <td className="py-2.5 pl-3 num">
+                  {stats.median_absolute_delta == null ? 'n/a' : stats.median_absolute_delta.toFixed(2)}
+                </td>
+              </tr>
+            ))}
+          </tbody>
+        </table>
+      </div>
+
+      <p className="text-[11px] leading-relaxed text-gray-500">
+        Materiality bands highlight review candidates only. They do not approve a cutover or write fundamentals,
+        scores, rankings, or qualification state.
+      </p>
+    </div>
+  );
+}
+
+function Summary({ label, value }: { label: string; value: string | number }) {
+  return (
+    <div className="rounded-lg border border-white/[0.07] bg-white/[0.025] px-3 py-2.5">
+      <div className="text-[10px] uppercase tracking-wider text-gray-500">{label}</div>
+      <div className="mt-1 num text-lg text-gray-200">{value}</div>
+    </div>
+  );
+}
@@ -8,6 +8,7 @@ const DEFAULTS: ScheduleConfig = {
  schedule_daily_pipeline_cron: '0 2 * * *',
  schedule_dolt_earnings_cron: '30 2 * * *',
  schedule_sec_fundamentals_cron: '0 4 * * *',
+  schedule_fundamentals_parity_cron: '30 5 * * *',
  schedule_near_close_pipeline_cron: '30 15 * * mon-fri',
  schedule_after_close_pipeline_cron: '45 16 * * mon-fri',
  schedule_intraday_pipeline_cron: '0 10-15 * * mon-fri',
@@ -38,6 +39,12 @@ const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: b
    hint: 'Import tracked-universe SEC facts daily at 04:00 ET. Unchanged revisions become no-op runs.',
    mono: true,
  },
+  {
+    key: 'schedule_fundamentals_parity_cron',
+    label: 'Fundamentals parity report',
+    hint: 'Read-only legacy vs SEC/Dolt comparison daily at 05:30 ET, after the shadow imports.',
+    mono: true,
+  },
  {
    key: 'schedule_near_close_pipeline_cron',
    label: 'Near-close pipeline (scan + alert)',
@@ -138,7 +138,8 @@ export function FundamentalsPanel({ data }: FundamentalsPanelProps) {
              <div className="mt-2.5 space-y-3.5">
                {trendRows.map((r) => (
                  <TrendRow key={r.key} label={r.label} kind={r.kind}
-                            metric={metrics[r.key]} read={reads[r.key]} />
+                            metric={metrics[r.key]} read={reads[r.key]}
+                            caveat={metrics[r.key]?.caveat} />
                ))}
              </div>
            </div>
@@ -188,9 +189,10 @@ function Bullet({ label, value, rail, comparison }: {

 // ---- operating-trend row (delta vs reference, favorable = right) ------------

-function TrendRow({ label, kind, metric, read }: {
+function TrendRow({ label, kind, metric, read, caveat }: {
  label: string; kind: 'growth' | 'margin' | 'share';
  metric: MetricItem | undefined; read: string | null | undefined;
+  caveat: string | null | undefined;
 }) {
  const tone = readTone(read);
  const value = finiteOrNull(metric?.value);
@@ -213,7 +215,9 @@ function TrendRow({ label, kind, metric, read }: {
  }
  const delta = value != null && ref != null ? value - ref : null;

-  const comparison = value == null ? (
+  const comparison = caveat ? (
+    <span style={{ color: HZ.muted }}>{caveat}</span>
+  ) : value == null ? (
    <span style={{ color: HZ.track }}>n/a</span>
  ) : delta == null ? (
    <span style={{ color: HZ.track }}>history n/a</span>
@@ -23,11 +23,13 @@ function dateFromToday(days: number): string {
 }

 function metric(key: string, value: number | null, hist: (number | null)[],
-                industry: MetricItem['industry'] = null): MetricItem {
+                industry: MetricItem['industry'] = null,
+                caveat: string | null = null): MetricItem {
  return {
    key: key as MetricItem['key'], value,
    history: hist.map((v, i) => h(P[i], v)),
-    industry, period_end: '2026-03-28', filed_date: '2026-05-01', source: 'sec',
+    industry, period_end: '2026-03-28', filed_date: '2026-05-01', caveat,
+    source: 'sec',
  };
 }
 const ind = (median: number, favorable_percentile: number) =>
@@ -78,12 +80,24 @@ const partial: FundamentalResponse = {
  earnings: { next: { date: dateFromToday(0), session: 'unknown', days_until: 0 }, recent: [] },
  metrics: [
    metric('revenue_growth_yoy', 12, [null, 8, 10, 12], null),
-    metric('eps_growth_yoy', null, [null, null, null, null], null),
+    metric(
+      'eps_growth_yoy',
+      null,
+      [null, null, null, null],
+      null,
+      'Not comparable: share count changed at least 25%; possible split or corporate action.',
+    ),
    metric('operating_margin', 25, [24, 24, 25, 25], null),
    metric('fcf_margin', null, [null, null, null, null], null),
    metric('net_debt', null, [], null),
    metric('net_debt_to_ebitda', 1.9, [1.7, 1.8, 1.9, 1.9], null),
-    metric('share_count_change_yoy', 2.1, [1.8, 2.0, 2.0, 2.1], null),
+    metric(
+      'share_count_change_yoy',
+      null,
+      [1.8, 2.0, 2.0, null],
+      null,
+      'Not comparable: share count changed at least 25%; possible split or corporate action.',
+    ),
  ],
  valuation: {
    pe: 15.2, fcf_yield: null, market_cap_est: 5.4e8,
@@ -316,6 +316,14 @@ export function useJobs() {
  });
 }

+export function useFundamentalsParityReport() {
+  return useQuery({
+    queryKey: ['admin', 'fundamentals-parity'],
+    queryFn: () => adminApi.getFundamentalsParityReport(),
+    refetchInterval: 15_000,
+  });
+}
+
 export function usePipelineReadiness() {
  return useQuery({
    queryKey: ['admin', 'pipeline-readiness'],
@@ -193,6 +193,7 @@ export interface ScheduleConfig {
  schedule_daily_pipeline_cron: string;
  schedule_dolt_earnings_cron: string;
  schedule_sec_fundamentals_cron: string;
+  schedule_fundamentals_parity_cron: string;
  schedule_near_close_pipeline_cron: string;
  schedule_after_close_pipeline_cron: string;
  schedule_intraday_pipeline_cron: string;
@@ -733,6 +734,7 @@ export interface MetricItem {
  industry: MetricIndustry | null;
  period_end: string | null;
  filed_date: string | null;
+  caveat: string | null;
  source: string; // 'sec' | 'legacy_api'
 }

@@ -5,6 +5,7 @@ import { AlertSettings } from '../components/admin/AlertSettings';
 import { SentimentProviderSettings } from '../components/admin/SentimentProviderSettings';
 import { DataCleanup } from '../components/admin/DataCleanup';
 import { JobControls } from '../components/admin/JobControls';
+import { FundamentalsParityPanel } from '../components/admin/FundamentalsParityPanel';
 import { PerformanceSettings } from '../components/admin/PerformanceSettings';
 import { PipelineReadinessPanel } from '../components/admin/PipelineReadinessPanel';
 import { SystemEventsPanel } from '../components/admin/SystemEventsPanel';
@@ -48,6 +49,7 @@ export default function AdminPage() {
        {activeTab === 'Jobs' && (
          <div className="space-y-4">
            <ScheduleSettings />
+            <FundamentalsParityPanel />
            <JobControls />
            <PipelineReadinessPanel />
          </div>
@@ -104,6 +104,22 @@ def test_net_debt_leverage_and_share_dilution():
    assert d.metrics["share_count_change_yoy"].value == pytest.approx(-10.0, abs=1e-6)


+def test_split_suspect_share_move_suppresses_share_and_eps_comparisons():
+    rows = _two_years()
+    for row in rows:
+        if row.fiscal_year == 2026:
+            row.shares_outstanding = 2000  # +100% resembles an unadjusted 2-for-1 split
+
+    d = fd.derive(rows)
+
+    for key in ("share_count_change_yoy", "eps_growth_yoy"):
+        series = d.metrics[key]
+        assert series.value is None
+        assert series.history[-1].value is None
+        assert "possible split" in series.caveat
+    assert d.metrics["revenue_growth_yoy"].value == pytest.approx(10.0)
+
+
 def test_valuation_inputs():
    d = fd.derive(_two_years())
    # TTM diluted EPS FY2026 = 1.1+1.21+1.32+1.43 = 5.06
@@ -0,0 +1,209 @@
+"""A5 fundamentals parity report: read-only comparison + artifact archive."""
+
+from __future__ import annotations
+
+from datetime import date, datetime, timezone
+
+import pytest
+from sqlalchemy import func, select
+
+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.fundamentals_parity_service import (
+    build_report,
+    fundamental_score,
+    load_latest,
+    load_latest_csv,
+    load_latest_json,
+    store_report,
+)
+
+UTC = timezone.utc
+GENERATED = datetime(2026, 7, 23, 10, 30, tzinfo=UTC)
+
+
+def _snapshot_rows(cik: str) -> list[FundamentalSnapshot]:
+    rows = []
+    periods = ("Q1", "Q2", "Q3", "FY")
+    months = (3, 6, 9, 12)
+    for fy, multiplier in ((2025, 1.0), (2026, 1.1)):
+        revenues = [100 * multiplier, 110 * multiplier, 120 * multiplier, 130 * multiplier]
+        eps = [1.0 * multiplier, 1.1 * multiplier, 1.2 * multiplier, 1.3 * multiplier]
+        for index, period in enumerate(periods):
+            period_end = date(fy, months[index], 28)
+            rows.append(
+                FundamentalSnapshot(
+                    cik=cik,
+                    accession=f"{cik}-{fy}-{period}",
+                    form="10-K" if period == "FY" else "10-Q",
+                    filed_date=period_end,
+                    accepted_at=datetime(fy, months[index], 28, tzinfo=UTC),
+                    period_end=period_end,
+                    fiscal_year=fy,
+                    fiscal_period=period,
+                    revenue=sum(revenues[: index + 1]),
+                    operating_income=sum(revenues[: index + 1]) * 0.2,
+                    diluted_eps=sum(eps[: index + 1]),
+                    cfo=sum(revenues[: index + 1]) * 0.25,
+                    capex=sum(revenues[: index + 1]) * 0.05,
+                    depreciation_amortization=sum(revenues[: index + 1]) * 0.05,
+                    cash_and_st_investments=40,
+                    total_debt=100,
+                    shares_outstanding=1000,
+                )
+            )
+    return rows
+
+
+async def _seed(db_session):
+    first = Ticker(symbol="AAA", cik="0000000001", sic="3571")
+    second = Ticker(symbol="BBB", cik=None, sic=None)
+    db_session.add_all([first, second])
+    await db_session.flush()
+    db_session.add_all(_snapshot_rows(first.cik))
+    db_session.add_all(
+        [
+            FundamentalData(
+                ticker_id=first.id,
+                pe_ratio=25,
+                revenue_growth=5,
+                earnings_surprise=0,
+                fetched_at=GENERATED,
+            ),
+            FundamentalData(
+                ticker_id=second.id,
+                pe_ratio=12,
+                revenue_growth=3,
+                earnings_surprise=None,
+                fetched_at=GENERATED,
+            ),
+            OHLCVRecord(
+                ticker_id=first.id,
+                date=date(2026, 7, 22),
+                open=100,
+                high=100,
+                low=100,
+                close=100,
+                volume=100,
+            ),
+            EarningsEvent(
+                ticker_id=first.id,
+                announce_date=date(2026, 7, 1),
+                session="amc",
+                eps_estimate=2,
+                eps_actual=2.2,
+                source="dolt_earnings",
+            ),
+            DataImportRun(
+                source="sec_facts",
+                revision="sec-rev",
+                status="promoted",
+                source_max_date=date(2026, 7, 22),
+                started_at=GENERATED,
+                completed_at=GENERATED,
+            ),
+            DataImportRun(
+                source="dolt_earnings",
+                revision="dolt-rev",
+                status="no_op",
+                source_max_date=date(2026, 7, 22),
+                started_at=GENERATED,
+                completed_at=GENERATED,
+            ),
+        ]
+    )
+    await db_session.flush()
+
+
+def test_score_formula_matches_production_rules():
+    score = fundamental_score(pe_ratio=15, revenue_growth=0, earnings_surprise=0)
+    assert score == pytest.approx((100 + 50 + 50) / 3)
+    assert fundamental_score(pe_ratio=15, revenue_growth=None, earnings_surprise=None) is None
+
+async def test_report_compares_sources_and_leaves_database_untouched(db_session):
+    await _seed(db_session)
+    before = await db_session.scalar(select(func.count()).select_from(FundamentalData))
+
+    report = await build_report(
+        db_session,
+        generated_at=GENERATED,
+        today=date(2026, 7, 23),
+    )
+
+    after = await db_session.scalar(select(func.count()).select_from(FundamentalData))
+    assert before == after == 2
+    assert not db_session.new and not db_session.dirty and not db_session.deleted
+    assert report["read_only"] is True
+    assert report["approval_status"] == "pending_explicit_approval"
+    assert report["source_runs"]["sec_facts"]["revision"] == "sec-rev"
+    assert report["source_runs"]["dolt_earnings"]["revision"] == "dolt-rev"
+
+    first = next(row for row in report["rows"] if row["symbol"] == "AAA")
+    assert first["fields"]["pe_ratio"]["candidate"] == pytest.approx(
+        100 / 5.06, abs=1e-4
+    )
+    assert first["fields"]["revenue_growth"]["candidate"] == pytest.approx(10)
+    assert first["fields"]["earnings_surprise"]["candidate"] == pytest.approx(10)
+    assert first["scores"]["candidate_fundamental"] is not None
+    assert report["summary"]["universe_count"] == 2
+    assert report["summary"]["field_stats"]["pe_ratio"]["both_available"] == 1
+
+
+async def test_artifacts_archive_and_latest_manifest(db_session, tmp_path):
+    await _seed(db_session)
+    report = await build_report(
+        db_session,
+        generated_at=GENERATED,
+        today=date(2026, 7, 23),
+    )
+
+    paths = store_report(report, tmp_path)
+
+    assert tmp_path.joinpath("latest.json").exists()
+    assert paths["json"].endswith(".json") and paths["csv"].endswith(".csv")
+    assert load_latest(tmp_path)["generated_at"] == GENERATED.isoformat()
+    csv_artifact = load_latest_csv(tmp_path)
+    assert csv_artifact is not None
+    assert csv_artifact[0].endswith(".csv")
+    assert "legacy_fundamental,candidate_fundamental" in csv_artifact[1]
+    assert "AAA" in csv_artifact[1]
+    json_artifact = load_latest_json(tmp_path)
+    assert json_artifact is not None and '"rows"' in json_artifact[1]
+
+
+async def test_admin_endpoints_return_compact_summary_and_downloads(
+    client, db_session, tmp_path, monkeypatch
+):
+    from app.config import settings
+    from app.dependencies import require_admin
+    from app.main import app
+
+    await _seed(db_session)
+    report = await build_report(
+        db_session,
+        generated_at=GENERATED,
+        today=date(2026, 7, 23),
+    )
+    store_report(report, tmp_path)
+    monkeypatch.setattr(settings, "fundamentals_parity_report_dir", str(tmp_path))
+    app.dependency_overrides[require_admin] = lambda: None
+    try:
+        summary_response = await client.get("/api/v1/admin/fundamentals-parity")
+        assert summary_response.status_code == 200
+        summary = summary_response.json()["data"]
+        assert summary["summary"]["universe_count"] == 2
+        assert "rows" not in summary
+
+        csv_response = await client.get("/api/v1/admin/fundamentals-parity/csv")
+        assert csv_response.status_code == 200
+        assert "AAA" in csv_response.json()["data"]["content"]
+
+        json_response = await client.get("/api/v1/admin/fundamentals-parity/json")
+        assert json_response.status_code == 200
+        assert '"rows"' in json_response.json()["data"]["content"]
+    finally:
+        app.dependency_overrides.pop(require_admin, None)
@@ -85,6 +85,7 @@ class TestTradingDayCrons:
        (
            ("schedule_dolt_earnings_cron", 2, 30),
            ("schedule_sec_fundamentals_cron", 4, 0),
+            ("schedule_fundamentals_parity_cron", 5, 30),
        ),
    )
    def test_shadow_imports_run_daily_at_expected_et_time(
@@ -11,6 +11,7 @@ from app.scheduler import (
    _resume_tickers,
    _last_successful,
    _run_shadow_import,
+    run_fundamentals_parity_report,
    configure_scheduler,
    get_job_runtime_snapshot,
    queue_backtest_options,
@@ -112,6 +113,7 @@ class TestConfigureScheduler:
            "fundamental_collector",
            "dolt_earnings_import",
            "sec_fundamentals_import",
+            "fundamentals_parity_report",
            "rr_scanner",
            "shadow_book",
            "ticker_universe_sync",
@@ -145,6 +147,7 @@ class TestConfigureScheduler:
            "fundamental_collector",
            "dolt_earnings_import",
            "sec_fundamentals_import",
+            "fundamentals_parity_report",
            "market_regime",
            "near_close_pipeline",
            "regime_monitor",
@@ -243,3 +246,32 @@ class TestShadowImportJobs:
        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):
+        return True
+
+    async def generated(db, report_dir):
+        return (
+            {
+                "generated_at": "2026-07-23T10:30:00+00:00",
+                "summary": {
+                    "universe_count": 511,
+                    "fundamental_score_material_changes": 12,
+                },
+            },
+            {"json": "report.json", "csv": "report.csv"},
+        )
+
+    monkeypatch.setattr("app.scheduler.async_session_factory", TestShadowImportJobs._session_factory)
+    monkeypatch.setattr("app.scheduler._is_job_enabled", enabled)
+    monkeypatch.setattr(
+        "app.scheduler.fundamentals_parity_service.generate_and_store", generated
+    )
+
+    await run_fundamentals_parity_report()
+
+    runtime = get_job_runtime_snapshot("fundamentals_parity_report")
+    assert runtime["status"] == "completed"
+    assert runtime["message"] == "511 tickers · 12 material score changes"