feat(deploy): schedule fundamentals shadow imports

max-w-md (448px) under-sized the panel vs its real full-width tab placement and
never triggered the desktop two-column layout. Widen to max-w-3xl; note to
resize to ~390px for the mobile (single-column) check.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.env.example

@@ -32,7 +32,8 @@ ALPHA_VANTAGE_API_KEY=
 #   e.g. Windows: C:\Program Files\Dolt\bin\dolt.exe). DOLT_DATA_DIR holds the
 #   clones; in PRODUCTION it MUST be outside the deploy tree (deploy is
 #   rsync --delete) — e.g. /var/lib/signal-platform/dolt. The earnings clone lives
-#   at <DOLT_DATA_DIR>/<DOLT_EARNINGS_SUBDIR>. Set up: dolt clone post-no-preference/earnings <dir>.
+#   at <DOLT_DATA_DIR>/<DOLT_EARNINGS_SUBDIR>. Production setup is automated by
+#   deploy/provision_fundamentals.sh; see docs/fundamentals-deployment.md.
 DOLT_BINARY=dolt
 DOLT_DATA_DIR=dolt-data
 DOLT_EARNINGS_SUBDIR=earnings

app/routers/fundamentals.py

@@ -9,6 +9,7 @@ from app.dependencies import get_db, require_access
 from app.schemas.common import APIEnvelope
 from app.schemas.fundamental import FundamentalResponse
 from app.services.fundamental_service import get_fundamental
+from app.services.fundamentals_api_service import build_fundamentals_v1
 
 router = APIRouter(tags=["fundamentals"])
 
@@ -30,14 +31,13 @@ async def read_fundamentals(
     _user=Depends(require_access),
     db: AsyncSession = Depends(get_db),
 ) -> APIEnvelope:
-    """Get latest fundamental data for a symbol."""
+    """Get latest fundamental data for a symbol (legacy fields + additive v1)."""
     record = await get_fundamental(db, symbol)
+    v1 = await build_fundamentals_v1(db, symbol)
 
-    if record is None:
-        data = FundamentalResponse(symbol=symbol.strip().upper())
-    else:
-        data = FundamentalResponse(
-            symbol=symbol.strip().upper(),
+    legacy: dict = {}
+    if record is not None:
+        legacy = dict(
             pe_ratio=record.pe_ratio,
             revenue_growth=record.revenue_growth,
             earnings_surprise=record.earnings_surprise,
@@ -47,4 +47,5 @@ async def read_fundamentals(
             unavailable_fields=_parse_unavailable_fields(record.unavailable_fields_json),
         )
 
+    data = FundamentalResponse(symbol=symbol.strip().upper(), **legacy, **v1)
     return APIEnvelope(status="success", data=data.model_dump())

app/scheduler.py

@@ -41,6 +41,9 @@ from app.services import (
     settings_store,
     shadow_book_service,
 )
+from app.services.data_import import STATUS_FAILED, SourceImporter, run_import
+from app.services.dolt_earnings_importer import DoltEarningsImporter
+from app.services.sec_fundamentals_importer import SecFundamentalsImporter
 from app.services.alert_service import dispatch_alerts
 from app.services.backtest_service import (
     BACKTEST_TARGET_MODELS,
@@ -93,6 +96,8 @@ _JOB_NAMES = [
     "data_backfill",
     "sentiment_collector",
     "fundamental_collector",
+    "dolt_earnings_import",
+    "sec_fundamentals_import",
     "rr_scanner",
     "ticker_universe_sync",
     "alerts",
@@ -912,6 +917,70 @@ async def collect_fundamentals() -> None:
         _runtime_finish(job_name, "error", processed=processed, total=total, message=str(exc))
 
 
+# ---------------------------------------------------------------------------
+# Jobs: shadow fundamentals sources
+# ---------------------------------------------------------------------------
+
+
+async def _run_shadow_import(job_name: str, importer: SourceImporter) -> None:
+    """Run one source importer and surface its audit result in Admin → Jobs."""
+    _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):
+                _log_event(logging.INFO, "job_skipped", job=job_name, reason="disabled")
+                _runtime_finish(job_name, "skipped", processed=0, total=1, message="Disabled")
+                return
+
+        run = await run_import(importer)
+        if run is 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
+
+        revision = f" · {run.revision[:12]}" if run.revision else ""
+        message = f"{run.status}{revision}"
+        if run.status == STATUS_FAILED:
+            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
+
+        _log_event(
+            logging.INFO,
+            "job_complete",
+            job=job_name,
+            import_status=run.status,
+            revision=run.revision,
+        )
+        _runtime_finish(job_name, "completed", processed=1, total=1, message=message)
+    except asyncio.CancelledError:
+        _runtime_finish(job_name, "error", processed=0, total=1, message="Cancelled")
+        raise
+    except Exception as exc:
+        _log_event(
+            logging.ERROR,
+            "job_error",
+            job=job_name,
+            error_type=type(exc).__name__,
+            message=str(exc),
+        )
+        _runtime_finish(job_name, "error", processed=0, total=1, message=str(exc))
+
+
+async def run_dolt_earnings_import() -> None:
+    """Pull and import the Dolt earnings calendar/results feed in shadow."""
+    await _run_shadow_import("dolt_earnings_import", DoltEarningsImporter())
+
+
+async def run_sec_fundamentals_import() -> None:
+    """Import tracked-universe SEC facts in shadow."""
+    await _run_shadow_import("sec_fundamentals_import", SecFundamentalsImporter())
+
+
 # ---------------------------------------------------------------------------
 # Job: R:R Scanner
 # ---------------------------------------------------------------------------
@@ -1452,6 +1521,9 @@ 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.
+    "schedule_dolt_earnings_cron": "30 2 * * *",
+    "schedule_sec_fundamentals_cron": "0 4 * * *",
     # 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).
@@ -1465,6 +1537,8 @@ SCHEDULE_DEFAULTS: dict[str, str] = {
 # job id -> schedule setting key
 _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",
     "near_close_pipeline": "schedule_near_close_pipeline_cron",
     "after_close_pipeline": "schedule_after_close_pipeline_cron",
     "intraday_pipeline": "schedule_intraday_pipeline_cron",
@@ -1549,6 +1623,28 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
         _cron_trigger(cfg["schedule_daily_pipeline_cron"], tz, "schedule_daily_pipeline_cron"),
         id="daily_pipeline", name="Morning Pipeline", replace_existing=True,
     )
+    scheduler.add_job(
+        run_dolt_earnings_import,
+        _cron_trigger(
+            cfg["schedule_dolt_earnings_cron"],
+            tz,
+            "schedule_dolt_earnings_cron",
+        ),
+        id="dolt_earnings_import",
+        name="Dolt Earnings Import (shadow)",
+        replace_existing=True,
+    )
+    scheduler.add_job(
+        run_sec_fundamentals_import,
+        _cron_trigger(
+            cfg["schedule_sec_fundamentals_cron"],
+            tz,
+            "schedule_sec_fundamentals_cron",
+        ),
+        id="sec_fundamentals_import",
+        name="SEC Fundamentals Import (shadow)",
+        replace_existing=True,
+    )
     scheduler.add_job(
         run_near_close_pipeline,
         _cron_trigger(
@@ -1622,6 +1718,8 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
             "cron": cfg["schedule_daily_pipeline_cron"],
             "steps": [name for name, _ in _DAILY_PIPELINE_STEPS],
         },
+        dolt_earnings_import={"cron": cfg["schedule_dolt_earnings_cron"]},
+        sec_fundamentals_import={"cron": cfg["schedule_sec_fundamentals_cron"]},
         near_close_pipeline={
             "cron": cfg["schedule_near_close_pipeline_cron"],
             "steps": [name for name, _ in _NEAR_CLOSE_PIPELINE_STEPS],

app/schemas/fundamental.py

@@ -7,8 +7,74 @@ from datetime import date, datetime
 from pydantic import BaseModel
 
 
+class MetricIndustry(BaseModel):
+    label: str
+    median: float
+    favorable_percentile: int  # 0-100, polarity-aware (higher = more favorable)
+    peer_count: int
+
+
+class MetricHistoryPoint(BaseModel):
+    period_end: str  # YYYY-MM-DD
+    value: float | None
+
+
+class MetricItem(BaseModel):
+    key: str
+    value: float | None = None
+    history: list[MetricHistoryPoint] = []
+    industry: MetricIndustry | None = None
+    period_end: str | None = None
+    filed_date: str | None = None
+    source: str = "sec"
+
+
+class EarningsNext(BaseModel):
+    date: str
+    session: str
+    days_until: int
+
+
+class EarningsRecent(BaseModel):
+    announce_date: str
+    period_end: str | None = None
+    eps_estimate: float | None = None
+    eps_actual: float | None = None
+    surprise_pct: float | None = None
+
+
+class EarningsObject(BaseModel):
+    next: EarningsNext | None = None
+    recent: list[EarningsRecent] = []
+
+
+class Valuation(BaseModel):
+    pe: float | None = None
+    fcf_yield: float | None = None
+    market_cap_est: float | None = None
+    pe_industry: MetricIndustry | None = None
+    fcf_yield_industry: MetricIndustry | None = None
+    price_date: str | None = None
+
+
+class FundamentalsReads(BaseModel):
+    """Deterministic text outputs, separate from the numeric metrics.
+
+    ``by_key`` is a fixed map over every metric key plus ``pe`` and ``fcf_yield``,
+    each a read string or null. ``header`` is null when there is no read at all."""
+
+    header: str | None = None
+    by_key: dict[str, str | None] = {}
+
+
 class FundamentalResponse(BaseModel):
-    """Envelope-ready fundamental data response."""
+    """Envelope-ready fundamental data response.
+
+    Legacy fields are preserved unchanged (they come from ``fundamental_data`` /
+    the legacy providers). The additive v1 objects — earnings, metrics, valuation,
+    reads — are SEC/Dolt-derived and independent; a null legacy field is never
+    mapped onto the new SEC metrics and vice-versa.
+    """
 
     symbol: str
     pe_ratio: float | None = None
@@ -18,3 +84,9 @@ class FundamentalResponse(BaseModel):
     next_earnings_date: date | None = None
     fetched_at: datetime | None = None
     unavailable_fields: dict[str, str] = {}
+
+    # --- additive v1 (always present; empty/null when unavailable) ---
+    earnings: EarningsObject | None = None
+    metrics: list[MetricItem] | None = None
+    valuation: Valuation | None = None
+    reads: FundamentalsReads | None = None

app/services/admin_service.py

@@ -612,6 +612,8 @@ VALID_JOB_NAMES = {
     "benchmark_collector",
     "sentiment_collector",
     "fundamental_collector",
+    "dolt_earnings_import",
+    "sec_fundamentals_import",
     "rr_scanner",
     "ticker_universe_sync",
     "outcome_evaluator",
@@ -633,6 +635,8 @@ JOB_LABELS = {
     "benchmark_collector": "Benchmark Collector",
     "sentiment_collector": "Sentiment Collector",
     "fundamental_collector": "Fundamental Collector",
+    "dolt_earnings_import": "Dolt Earnings Import (shadow)",
+    "sec_fundamentals_import": "SEC Fundamentals Import (shadow)",
     "rr_scanner": "R:R Scanner",
     "ticker_universe_sync": "Ticker Universe Sync",
     "outcome_evaluator": "Outcome Evaluator",

app/services/fundamentals_api_service.py

@@ -0,0 +1,331 @@
+"""Assemble the additive fundamentals API v1 objects (earnings, metrics,
+valuation, reads) from SEC snapshots + Dolt earnings + the latest price.
+
+Strictly additive: the router merges these into the existing FundamentalResponse
+without touching legacy fields. Valuation ratios are computed at REQUEST TIME from
+the stored snapshots + the latest ohlcv close (no stored valuation). Peer stats are
+batched and CIK-deduplicated; invalid valuation inputs are guarded to null.
+"""
+
+from __future__ import annotations
+
+import math
+from collections import defaultdict
+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
+from app.services import fundamentals_peers as peers
+from app.services import fundamentals_reads as reads
+
+# The fixed metric row set — every key always present, value null when unavailable.
+METRIC_KEYS = (
+    "revenue_growth_yoy", "eps_growth_yoy", "operating_margin", "fcf_margin",
+    "net_debt", "net_debt_to_ebitda", "share_count_change_yoy",
+)
+
+
+async def build_fundamentals_v1(db: AsyncSession, symbol: str, *, today: date | None = None) -> dict[str, Any]:
+    today = today or _ny_today()
+    ticker = await _ticker_by_symbol(db, symbol)
+
+    earnings = await _build_earnings(db, ticker.id, today) if ticker else _empty_earnings()
+    if ticker is None or not ticker.cik:
+        # No SEC identity: metrics present but null, valuation null, empty reads.
+        return {"earnings": earnings, "metrics": _empty_metrics(), "valuation": None,
+                "reads": _empty_reads()}
+
+    subject_cik = ticker.cik
+    derived = deriv.derive((await _snapshots_for(db, [subject_cik])).get(subject_cik, []))
+
+    two = peers.two_digit_sic(ticker.sic)
+    peer_derived: dict[str, deriv.DerivedFundamentals] = {}
+    peer_price_by_cik: dict[str, tuple[float, date] | None] = {}
+    if two:
+        # Subject's representative is the REQUESTED ticker (so its price is used for
+        # the subject in the peer set); other issuers pick a deterministic-by-symbol rep.
+        group = await _peer_group(db, two, subject_cik, ticker.id)
+        peer_snaps = await _snapshots_for(db, list(group))
+        peer_derived = {cik: deriv.derive(rows) for cik, rows in peer_snaps.items()}
+        closes = await _latest_closes(db, set(group.values()))
+        peer_price_by_cik = {cik: closes.get(tid) for cik, tid in group.items()}
+
+    subject_price = await _latest_close(db, ticker.id)
+    metrics = _build_metrics(derived, peer_derived, two)
+    valuation = _build_valuation(derived, subject_price, peer_derived, peer_price_by_cik, two)
+    reads_obj = _build_reads(metrics, valuation)
+    return {"earnings": earnings, "metrics": metrics, "valuation": valuation, "reads": reads_obj}
+
+
+# -- earnings ----------------------------------------------------------------
+
+async def _build_earnings(db, ticker_id: int, today: date) -> dict[str, Any]:
+    rows = (await db.execute(
+        select(EarningsEvent).where(EarningsEvent.ticker_id == ticker_id)
+    )).scalars().all()
+    # Same-day earnings are UPCOMING (days_until 0); recent is strictly earlier.
+    upcoming = sorted((e for e in rows if e.announce_date >= today), key=lambda e: e.announce_date)
+    past = sorted((e for e in rows if e.announce_date < today), key=lambda e: e.announce_date, reverse=True)
+
+    nxt = None
+    if upcoming:
+        e = upcoming[0]
+        nxt = {"date": e.announce_date.isoformat(), "session": e.session,
+               "days_until": (e.announce_date - today).days}
+    recent = [{
+        "announce_date": e.announce_date.isoformat(),
+        "period_end": _iso(e.period_end),
+        "eps_estimate": e.eps_estimate,
+        "eps_actual": e.eps_actual,
+        "surprise_pct": _surprise_pct(e.eps_estimate, e.eps_actual),
+    } for e in past[:4]]
+    return {"next": nxt, "recent": recent}
+
+
+def _surprise_pct(estimate, actual):
+    if estimate is None or actual is None or estimate == 0:
+        return None
+    return round((actual - estimate) / abs(estimate) * 100.0, 2)
+
+
+# -- metrics -----------------------------------------------------------------
+
+def _build_metrics(derived, peer_derived, two: str | None) -> list[dict[str, Any]]:
+    out = []
+    for key in METRIC_KEYS:
+        series = derived.metrics.get(key)
+        value = series.value if series else None
+        history = [{"period_end": _iso(p.period_end), "value": p.value} for p in (series.history if series else [])]
+        industry = None
+        if two and peer_derived and key in peers.HIGHER_IS_BETTER:
+            group_values = [
+                (pd.metrics.get(key).value if pd.metrics.get(key) else None)
+                for pd in peer_derived.values()
+            ]
+            stat = peers.peer_stat_for(key, value, group_values)
+            if stat:
+                industry = {"label": f"SIC {two} peers", "median": round(stat.median, 4),
+                            "favorable_percentile": stat.favorable_percentile, "peer_count": stat.peer_count}
+        out.append({
+            "key": key,
+            "value": value,
+            "history": history,
+            "industry": industry,
+            "period_end": _iso(series.period_end) if series else None,
+            "filed_date": _iso(series.filed_date) if series else None,
+            "source": "sec",
+        })
+    return out
+
+
+# -- valuation (request-time) ------------------------------------------------
+
+def _build_valuation(derived, subject_price, peer_derived, peer_price_by_cik, two) -> dict[str, Any] | None:
+    if derived.latest_period_end is None:
+        return None  # no snapshots yet
+    price = subject_price[0] if subject_price else None
+    price_date = subject_price[1] if subject_price else None
+    if not _finite(price) or price <= 0:
+        return None  # no usable price -> valuation null (approved contract)
+
+    pe = _pe(price, derived.ttm_diluted_eps)
+    market_cap = _market_cap(price, derived.shares_outstanding)
+    fcf_yield = _fcf_yield(derived.ttm_fcf, market_cap)
+
+    pe_industry = fcf_yield_industry = None
+    if two and peer_derived:
+        pe_values = [_pe(_p(peer_price_by_cik.get(cik)), pd.ttm_diluted_eps) for cik, pd in peer_derived.items()]
+        fy_values = [
+            _fcf_yield(pd.ttm_fcf, _market_cap(_p(peer_price_by_cik.get(cik)), pd.shares_outstanding))
+            for cik, pd in peer_derived.items()
+        ]
+        pe_industry = _industry("pe", pe, pe_values, two)
+        fcf_yield_industry = _industry("fcf_yield", fcf_yield, fy_values, two)
+
+    return {
+        "pe": _round(pe, 2),
+        "fcf_yield": _round(fcf_yield, 2),
+        "market_cap_est": _round(market_cap, 0),
+        "pe_industry": pe_industry,
+        "fcf_yield_industry": fcf_yield_industry,
+        "price_date": _iso(price_date),
+    }
+
+
+def _pe(price, ttm_eps):
+    if not _finite(price) or price <= 0 or not _finite(ttm_eps) or ttm_eps <= 0:
+        return None
+    return price / ttm_eps
+
+
+def _market_cap(price, shares):
+    if not _finite(price) or price <= 0 or not _finite(shares) or shares <= 0:
+        return None
+    return price * shares
+
+
+def _fcf_yield(ttm_fcf, market_cap):
+    if not _finite(ttm_fcf) or not _finite(market_cap) or market_cap <= 0:
+        return None
+    return ttm_fcf / market_cap * 100.0
+
+
+def _industry(key, subject, group_values, two):
+    stat = peers.peer_stat_for(key, subject, group_values)
+    if stat is None:
+        return None
+    return {"label": f"SIC {two} peers", "median": round(stat.median, 4),
+            "favorable_percentile": stat.favorable_percentile, "peer_count": stat.peer_count}
+
+
+# -- reads -------------------------------------------------------------------
+
+_READ_KEYS = METRIC_KEYS + ("pe", "fcf_yield")
+
+
+def _build_reads(metrics: list[dict], valuation: dict | None) -> dict[str, Any]:
+    by_metric = {m["key"]: m for m in metrics}
+
+    def hist(key):
+        return [_Pt(p["value"]) for p in by_metric.get(key, {}).get("history", [])]
+
+    growth = reads.growth_read(hist("revenue_growth_yoy"))
+    eps_growth = reads.growth_read(hist("eps_growth_yoy"))
+    op_margin = reads.margin_read(hist("operating_margin"))
+    fcf_margin = reads.margin_read(hist("fcf_margin"))
+    share = reads.share_count_read(by_metric.get("share_count_change_yoy", {}).get("value"))
+    leverage = reads.peer_read("net_debt_to_ebitda", _pct(by_metric.get("net_debt_to_ebitda", {}).get("industry")))
+    pe_read = reads.peer_read("pe", _pct(valuation.get("pe_industry"))) if valuation else None
+    fcf_yield_read = reads.peer_read("fcf_yield", _pct(valuation.get("fcf_yield_industry"))) if valuation else None
+
+    # Fixed by_key map over every metric + pe + fcf_yield (null where unavailable).
+    by_key: dict[str, str | None] = {k: None for k in _READ_KEYS}
+    by_key.update({
+        "revenue_growth_yoy": growth,
+        "eps_growth_yoy": eps_growth,
+        "operating_margin": op_margin,
+        "fcf_margin": fcf_margin,
+        "share_count_change_yoy": share,
+        "net_debt_to_ebitda": leverage,
+        "pe": pe_read,
+        "fcf_yield": fcf_yield_read,
+    })
+    header = reads.header_sentence(growth, op_margin, pe_read or fcf_yield_read) or None
+    return {"header": header, "by_key": by_key}
+
+
+def _empty_reads() -> dict[str, Any]:
+    return {"header": None, "by_key": {k: None for k in _READ_KEYS}}
+
+
+class _Pt:
+    __slots__ = ("value",)
+
+    def __init__(self, value):
+        self.value = value
+
+
+def _pct(industry: dict | None):
+    return industry.get("favorable_percentile") if industry else None
+
+
+# -- queries -----------------------------------------------------------------
+
+async def _ticker_by_symbol(db, symbol: str) -> Ticker | None:
+    return (await db.execute(
+        select(Ticker).where(Ticker.symbol == symbol.strip().upper())
+    )).scalar_one_or_none()
+
+
+async def _snapshots_for(db, ciks) -> dict[str, list]:
+    out: dict[str, list] = defaultdict(list)
+    if not ciks:
+        return out
+    rows = (await db.execute(
+        select(FundamentalSnapshot).where(FundamentalSnapshot.cik.in_(list(ciks)))
+    )).scalars().all()
+    for r in rows:
+        out[r.cik].append(r)
+    return out
+
+
+async def _peer_group(db, two: str, subject_cik: str, subject_tid: int) -> dict[str, int]:
+    """{cik: representative ticker_id} for tracked issuers in the 2-digit SIC group,
+    CIK-deduplicated. Each issuer's representative is its lexicographically-smallest
+    symbol (deterministic), EXCEPT the subject issuer, which uses the requested
+    ticker — so a multi-class subject (GOOGL) is priced by the requested class, not
+    an arbitrary sibling (GOOG)."""
+    rows = (await db.execute(
+        select(Ticker.cik, Ticker.id, Ticker.symbol)
+        .where(Ticker.cik.is_not(None), func.substr(Ticker.sic, 1, 2) == two)
+    )).all()
+    rep: dict[str, tuple[int, str]] = {}
+    for cik, tid, sym in rows:
+        key = sym or ""
+        if cik not in rep or key < rep[cik][1]:
+            rep[cik] = (tid, key)
+    group = {cik: tid for cik, (tid, _) in rep.items()}
+    if subject_cik in group:
+        group[subject_cik] = subject_tid  # requested ticker prices the subject
+    return group
+
+
+async def _latest_closes(db, ticker_ids: set[int]) -> dict[int, tuple[float, date]]:
+    if not ticker_ids:
+        return {}
+    latest = (
+        select(OHLCVRecord.ticker_id, func.max(OHLCVRecord.date).label("d"))
+        .where(OHLCVRecord.ticker_id.in_(list(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.d)
+        )
+    )).all()
+    return {tid: (close, d) for tid, close, d in rows}
+
+
+async def _latest_close(db, ticker_id: int) -> tuple[float, date] | None:
+    return (await _latest_closes(db, {ticker_id})).get(ticker_id)
+
+
+# -- helpers -----------------------------------------------------------------
+
+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]
+
+
+def _empty_earnings() -> dict[str, Any]:
+    return {"next": None, "recent": []}
+
+
+def _p(price_tuple):
+    return price_tuple[0] if price_tuple else None
+
+
+def _finite(v) -> bool:
+    return isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v)
+
+
+def _round(v, ndigits):
+    return round(v, ndigits) if _finite(v) else None
+
+
+def _iso(d) -> str | None:
+    return d.isoformat() if d else None
+
+
+def _ny_today() -> date:
+    """Today's New York calendar date — the market's day, not the server's."""
+    return datetime.now(ZoneInfo("America/New_York")).date()

app/services/fundamentals_peers.py

@@ -14,11 +14,18 @@ a peer percentile** — leverage is compared via net_debt_to_ebitda.
 
 from __future__ import annotations
 
+import math
 import statistics
 from dataclasses import dataclass
+from typing import Any
 
 MIN_PEERS = 5
 
+
+def _finite(v: Any) -> bool:
+    """True for a finite number — excludes None, bool, NaN, ±inf (plan: null/invalid)."""
+    return isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v)
+
 # Metric -> is a higher value more favorable? (Peer-eligible metrics only;
 # absolute net_debt is intentionally absent — size-dependent.)
 HIGHER_IS_BETTER: dict[str, bool] = {
@@ -50,18 +57,33 @@ def peer_stat(
     """Median + favorable percentile for ``subject`` within its group.
 
     ``group_values`` is every issuer's value for the metric (including the
-    subject), CIK-deduplicated by the caller. Nulls are excluded. Returns None
-    when fewer than ``min_peers`` valid values exist, or the subject is null.
+    subject), CIK-deduplicated by the caller. Null/invalid (non-finite) values are
+    excluded. Returns None when fewer than ``min_peers`` valid values exist, or
+    the subject is null/invalid.
+
+    The percentile is a **tie-aware rank against the other issuers** —
+    ``(worse + 0.5·tied) / (peers − 1)`` — so a whole group of equal values maps
+    to 50, not 100, and the median maps to 50.
     """
-    valid = [v for v in group_values if v is not None]
-    if subject is None or len(valid) < min_peers:
+    valid = [v for v in group_values if _finite(v)]
+    if not _finite(subject) or len(valid) < min_peers:
         return None
     median = statistics.median(valid)
+
+    others = valid.copy()
+    try:
+        others.remove(subject)  # rank the subject against the OTHER issuers
+    except ValueError:
+        pass
+    denom = len(others)
+    if denom == 0:
+        return None
     if higher_is_better:
-        favorable = sum(1 for v in valid if v <= subject)
+        worse = sum(1 for v in others if v < subject)
     else:
-        favorable = sum(1 for v in valid if v >= subject)
-    percentile = round(favorable / len(valid) * 100)
+        worse = sum(1 for v in others if v > subject)
+    tied = sum(1 for v in others if v == subject)
+    percentile = round((worse + 0.5 * tied) / denom * 100)
     return PeerStat(median=median, favorable_percentile=percentile, peer_count=len(valid))
 
 

app/services/fundamentals_reads.py

@@ -22,13 +22,23 @@ PEER_FAVORABLE = 60
 PEER_ADVERSE = 40
 
 
-def _history_values(history: list[Any]) -> list[float]:
-    return [p.value for p in history if p.value is not None]
+def _latest_run(history: list[Any]) -> list[float]:
+    """The consecutive non-null values ending at the latest point (oldest->newest).
+    A null latest, or an internal gap, truncates the run — so a read never reflects
+    a period whose displayed value is n/a."""
+    run: list[float] = []
+    for p in reversed(history):
+        if p.value is None:
+            break
+        run.append(p.value)
+    run.reverse()
+    return run
 
 
 def growth_read(history: list[Any]) -> str | None:
-    """Change in a YoY-growth series: latest − prior. Needs >= 3 periods."""
-    vals = _history_values(history)
+    """Change in a YoY-growth series: latest − prior. Needs >= 3 consecutive
+    non-null values ending at the latest point."""
+    vals = _latest_run(history)
     if len(vals) < MIN_PERIODS:
         return None
     delta = vals[-1] - vals[-2]
@@ -40,8 +50,9 @@ def growth_read(history: list[Any]) -> str | None:
 
 
 def margin_read(history: list[Any]) -> str | None:
-    """Latest margin vs the mean of prior periods (pp). Needs >= 3 periods."""
-    vals = _history_values(history)
+    """Latest margin vs the mean of prior periods (pp). Needs >= 3 consecutive
+    non-null values ending at the latest point."""
+    vals = _latest_run(history)
     if len(vals) < MIN_PERIODS:
         return None
     delta = vals[-1] - mean(vals[:-1])

deploy/provision_fundamentals.sh

@@ -0,0 +1,100 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# One-time production provisioning for the shadow fundamentals sources.
+# Run as root to install; run with --check as the deploy user for a read-only
+# preflight. Version upgrades are intentional code changes, never "latest".
+DOLT_VERSION="2.2.0"
+DOLT_BINARY="${DOLT_BINARY:-/usr/local/bin/dolt}"
+DOLT_DATA_DIR="${DOLT_DATA_DIR:-/var/lib/signal-platform/dolt}"
+DOLT_EARNINGS_SUBDIR="${DOLT_EARNINGS_SUBDIR:-earnings}"
+APP_USER="${APP_USER:-deploy}"
+APP_GROUP="${APP_GROUP:-deploy}"
+ENV_FILE="${ENV_FILE:-/opt/signalplatform/.env}"
+MIN_FREE_GB="${DOLT_MIN_FREE_DISK_GB:-5}"
+EARNINGS_DIR="${DOLT_DATA_DIR}/${DOLT_EARNINGS_SUBDIR}"
+
+fail() {
+  echo "ERROR: $*" >&2
+  exit 1
+}
+
+version_ok() {
+  local output
+  output="$("$DOLT_BINARY" version 2>/dev/null || true)"
+  grep -Eq "(^|[[:space:]])v?${DOLT_VERSION}([[:space:]]|$)" <<<"$output"
+}
+
+check_free_space() {
+  local available_kb
+  available_kb="$(df -Pk "$DOLT_DATA_DIR" | awk 'NR == 2 {print $4}')"
+  [[ "$available_kb" =~ ^[0-9]+$ ]] || fail "could not read free space for $DOLT_DATA_DIR"
+  if ! awk -v available="$available_kb" -v minimum_gb="$MIN_FREE_GB" \
+    'BEGIN { exit !(available >= minimum_gb * 1024 * 1024) }'; then
+    fail "$DOLT_DATA_DIR has less than ${MIN_FREE_GB} GB free"
+  fi
+}
+
+check_env() {
+  [[ -f "$ENV_FILE" ]] || fail "missing environment file: $ENV_FILE"
+  grep -Fqx "DOLT_BINARY=$DOLT_BINARY" "$ENV_FILE" \
+    || fail "set DOLT_BINARY=$DOLT_BINARY in $ENV_FILE"
+  grep -Fqx "DOLT_DATA_DIR=$DOLT_DATA_DIR" "$ENV_FILE" \
+    || fail "set DOLT_DATA_DIR=$DOLT_DATA_DIR in $ENV_FILE"
+  grep -Fqx "DOLT_EARNINGS_SUBDIR=$DOLT_EARNINGS_SUBDIR" "$ENV_FILE" \
+    || 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"
+}
+
+check_all() {
+  id "$APP_USER" >/dev/null 2>&1 || fail "missing service user: $APP_USER"
+  [[ -x "$DOLT_BINARY" ]] || fail "missing Dolt binary: $DOLT_BINARY"
+  version_ok || fail "expected Dolt $DOLT_VERSION at $DOLT_BINARY"
+  [[ -d "$EARNINGS_DIR/.dolt" ]] \
+    || fail "missing earnings clone: $EARNINGS_DIR"
+  if [[ "$(id -un)" == "$APP_USER" ]]; then
+    [[ -r "$EARNINGS_DIR/.dolt" ]] \
+      || fail "earnings clone is not readable by $APP_USER"
+  elif command -v runuser >/dev/null 2>&1; then
+    runuser -u "$APP_USER" -- test -r "$EARNINGS_DIR/.dolt" \
+      || fail "earnings clone is not readable by $APP_USER"
+  else
+    fail "run --check as $APP_USER (or install runuser)"
+  fi
+  check_free_space
+  check_env
+  echo "OK: Dolt $DOLT_VERSION and earnings clone are provisioned"
+}
+
+if [[ "${1:-}" == "--check" ]]; then
+  check_all
+  exit 0
+fi
+
+[[ "$EUID" -eq 0 ]] || fail "run provisioning as root (or use --check)"
+command -v curl >/dev/null 2>&1 || fail "curl is required"
+command -v runuser >/dev/null 2>&1 || fail "runuser is required"
+id "$APP_USER" >/dev/null 2>&1 || fail "missing service user: $APP_USER"
+
+if ! version_ok; then
+  installer="$(mktemp)"
+  trap 'rm -f "$installer"' EXIT
+  curl -fsSL \
+    "https://github.com/dolthub/dolt/releases/download/v${DOLT_VERSION}/install.sh" \
+    -o "$installer"
+  bash "$installer"
+fi
+version_ok || fail "Dolt $DOLT_VERSION installation failed"
+
+install -d -o "$APP_USER" -g "$APP_GROUP" -m 0750 "$DOLT_DATA_DIR"
+check_free_space
+
+if [[ ! -d "$EARNINGS_DIR/.dolt" ]]; then
+  [[ ! -e "$EARNINGS_DIR" ]] \
+    || fail "$EARNINGS_DIR exists but is not a Dolt clone"
+  runuser -u "$APP_USER" -- \
+    "$DOLT_BINARY" clone post-no-preference/earnings "$EARNINGS_DIR"
+fi
+
+check_all

docs/dolt-integration-plan.md

@@ -109,12 +109,12 @@ reviewed separately if B begins.
   share count — both consumers (est. market cap, YoY dilution) want a
   point-in-time value. **Nothing
   derived is frozen into a row:** discrete quarters (10-Q YTD deltas, Q4 = FY −
-  Q1..Q3), TTM, YoY and the quarter-tape series are all computed **at read time** by
+  Q1..Q3), TTM, YoY and the four-period metric histories are all computed **at read time** by
   picking the newest valid accepted_at snapshot for *each* required period — so
   non-calendar fiscal years resolve correctly and a later amendment to a prior
   quarter is reflected automatically without ever storing a stale derived quarter.
   Readers pick the newest valid accepted_at per period; history powers the UI
-  quarter tape.
+  reference comparisons and deterministic reads.
 - Keep `fundamental_data` (`app/models/fundamental.py`) as the latest-value compat
   cache, repopulated by the daily SEC job — but only after the phase-A5 parity
   gate.
@@ -164,8 +164,11 @@ calls. Check free disk space before pulling; alert and skip if below threshold.
 **Deployment constraints:** deploy is `rsync --delete` of the repo tree
 (`.gitea/workflows/deploy.yml:127`), so clones and archives must live **outside the
 deployment path** — an env-configured persistent directory (e.g.
-`DOLT_DATA_DIR=/var/lib/signal-platform/dolt`), backed up. The dolt binary is a new
-prod runtime dependency: install with a pinned version in the deploy workflow.
+`DOLT_DATA_DIR=/var/lib/signal-platform/dolt`). The dolt binary is a new prod
+runtime dependency: install it once with the version-pinned provisioner in
+`deploy/provision_fundamentals.sh`; operational steps are in
+`docs/fundamentals-deployment.md`. The clone is reproducible from DoltHub; the
+normalized PostgreSQL rows remain part of the normal database backup.
 
 **Validation gates (block promotion, raise an alert via the existing system-events
 path):** source freshness as expected; tracked-universe coverage; no duplicate
@@ -193,13 +196,14 @@ Workstream A:
 
 - Dolt earnings import: daily ~02:30 ET (with the future-row replacement above).
 - **SEC fundamentals job: daily ~04:00 ET.** One job, three steps:
-  (a) freshness check **using conditional HTTP metadata (ETag/Last-Modified /
-  If-None-Match) before downloading** — an unchanged archive is a `no_op` without
-  pulling the multi-GB file; verify by SHA-256 after any actual download;
-  (b) when changed, **parse and write only tracked-universe CIKs** through
-  staging→promotion (the tracked universe is `ticker_universe_service`'s set;
-  CIKs are resolved from `company_tickers.json` each run, so a newly added ticker
-  self-resolves on its next SEC run — until then its metrics are simply null);
+  (a) detect a composite revision from the latest EDGAR daily-index date, the
+  exact tracked index rows, and the tracked-universe fingerprint — an unchanged
+  revision is a `no_op` before Company Facts are fetched;
+  (b) when changed, fetch and parse Company Facts only for tracked-universe CIKs
+  that filed, plus full available history for the first run or a newly added
+  issuer, through validation→atomic promotion. Universe resolution and the exact
+  index inputs are cached during revision detection and reused during staging;
+  CIKs are resolved from `company_tickers.json` without writes until promotion;
   (c) **always, locally, and only after production activation** (the phase-A5
   parity approval): refresh the legacy `fundamental_data` fields and mark affected
   cached fundamental scores stale. **Sources differ per field** — do not assume all
@@ -336,37 +340,41 @@ changes.
 
 ## UI — `frontend/src/components/ticker/FundamentalsPanel.tsx`
 
-One distinctive visual device — the **quarter tape** — in an otherwise restrained
+One distinctive visual device — the **Reference Rails** — in an otherwise restrained
 panel. Preserve the app's dark glass styling and numeric typography.
 
 ```
-Fundamentals                     Quality improving · valuation rich
-Next earnings  Aug 3 · AMC       Last 4:  beat  beat  miss  beat
+Fundamentals
+Growth accelerating · margins improving · valuation priced above peers
+Next earnings  Aug 3 · AMC       EPS surprises  ▂ ▅ ▃ ▆
 
-Quarter tape          Q−3    Q−2    Q−1    Latest      Read
-Revenue growth         8%     11%    15%     18%       accelerating
-Operating margin      19%     20%    20%     22%       improving
-FCF margin            12%     10%    14%     16%       above own average
-Share count change                                     1.8% dilution
+Operating trend              less favorable ← ref → more favorable
+Revenue growth                                             18%
+───────────────│━━━━●                   +3pp vs prior · accelerating
+Share count YoY                                           −1.7%
+───────────────│━━━━●                                    buying back
 
-Balance & valuation
-Net debt / EBITDA      1.4×      Industry median 2.1×   healthy leverage
-P/E                   29.2×      Industry median 23.5×  priced above peers
-FCF yield              3.8%      Industry median 3.1%   above peers
+Valuation & balance           less favorable ← median → more favorable
+P/E                                                       29.2×
+────●━━━━━━━━━━│────                  priced above peers · median 23.5× · 12 peers
 ```
 
-- Growth, margins, share count: latest four periods as a compact four-cell tape
-  (or sparkline) plus a deterministic text read (rules below).
-- P/E, FCF yield, leverage: horizontal industry-percentile strip with a median
-  marker. Hidden entirely when `industry` is null (< 5 peer issuers).
-- Earnings: four bars around a zero baseline — green beats, red misses, gray
+- Growth and margins: horizontal rails compare the latest value with the prior
+  quarter or prior-period average; share-count YoY compares with zero. The rail
+  is normalized so right is always more favorable, including buybacks.
+- P/E, FCF yield and leverage: horizontal favorable-percentile rails with a
+  peer-median marker. No decorative rail when `industry` is null (< 5 peers).
+- Every row keeps the exact value and one deterministic comparison caption;
+  missing values render `n/a`, and insufficient peers render `peers n/a`.
+- Earnings: four bars around a shared zero baseline — cyan beats, coral misses, gray
   unavailable — plus next date and BMO/AMC session countdown.
-- Accessibility: color always paired with text or arrows; neutral/ambiguous stays
-  gray; green/red only when a read is genuinely favorable/adverse.
-- Remove the hard-coded "FMP" source label — provenance is per metric.
+- Accessibility: color is always paired with text; neutral/ambiguous stays gray;
+  rails and earnings bars expose complete ARIA descriptions.
+- Remove the hard-coded "FMP" source label; surface filing and price-date
+  provenance in the footer.
 
 **Deterministic reads — one shared rule set.** Implement as a single function with
-named constants; the tape reads and the header sentence use identical outputs. No
+named constants; the metric reads and the header sentence use identical outputs. No
 LLM, no new composite score. Defaults (tunable constants, not scattered literals):
 
 - A series read requires ≥ 3 periods; otherwise show "—" and no read.
@@ -401,9 +409,9 @@ workstream B — Alpaca remains the price source throughout.
 **Workstream A:**
 
 - A0. License review **DONE** (earnings approved for private/internal use under
-  CC BY-SA 4.0, no redistribution — see Licensing above); still pending at deploy
-  time: dolt binary pinned in deploy; `DOLT_DATA_DIR` outside the rsync tree
-  (earnings clone only — small), provisioned and backed up.
+  CC BY-SA 4.0, no redistribution — see Licensing above). The Dolt version,
+  persistent `DOLT_DATA_DIR`, clone, and production checks are captured in
+  `deploy/provision_fundamentals.sh` and `docs/fundamentals-deployment.md`.
 - A1. Migration 026, import-run framework.
 - A2. Earnings ingestion in shadow (writes `earnings_events`, prod untouched);
   verify forward-calendar coverage and rescheduling behavior.

docs/fundamentals-deployment.md

@@ -0,0 +1,133 @@
+# 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.
+
+## 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.
+- 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
+  event. A failed validation does not promote partial data.
+
+The systemd service uses one application worker. The import framework also holds
+a PostgreSQL advisory lock per source, so an overlapping manual/scheduled run is
+skipped safely.
+
+## Prerequisites
+
+The production `.env` at `/opt/signalplatform/.env` must contain:
+
+```dotenv
+DOLT_BINARY=/usr/local/bin/dolt
+DOLT_DATA_DIR=/var/lib/signal-platform/dolt
+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
+```
+
+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.
+
+## One-time provisioning
+
+First deploy the commit containing this bundle to production. Then SSH to the
+server and run:
+
+```bash
+cd /opt/signalplatform
+sudo bash ./deploy/provision_fundamentals.sh
+sudo -u deploy bash ./deploy/provision_fundamentals.sh --check
+sudo systemctl restart signalplatform.service
+curl -fsS http://127.0.0.1:8998/api/v1/health
+```
+
+The provisioner is idempotent. It installs the pinned Dolt version, creates the
+persistent directory as `deploy:deploy`, clones
+`post-no-preference/earnings`, verifies free space and `.env`, and refuses an
+unexpected Dolt version. It does not modify PostgreSQL or start an import.
+
+Do not replace the pinned version with `latest`. A future Dolt upgrade should be
+a reviewed change to `DOLT_VERSION`, followed by the same provision/check flow.
+
+## First-run verification
+
+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
+   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.
+4. Confirm the next-run times correspond to 02:30 and 04:00 New York time.
+5. Open several ticker pages and confirm the fundamentals panel has populated
+   data and still handles partial/missing issuers cleanly.
+
+Optional database verification:
+
+```sql
+SELECT source, status, revision, source_max_date, started_at, completed_at,
+       validation_json
+FROM data_import_runs
+WHERE source IN ('dolt_earnings', 'sec_facts')
+ORDER BY id DESC
+LIMIT 10;
+
+SELECT count(*) FROM earnings_events WHERE source = 'dolt_earnings';
+SELECT count(*), count(DISTINCT cik) FROM fundamental_snapshots;
+```
+
+During the longer first SEC run, execute the following in a second SSH session.
+It opens an independent database connection and attempts the same source lock:
+
+```bash
+cd /opt/signalplatform
+sudo -u deploy .venv/bin/python - <<'PY'
+import asyncio
+
+from sqlalchemy import text
+
+from app.database import engine
+from app.services.data_import import _advisory_key
+
+
+async def main():
+    key = _advisory_key("sec_facts")
+    async with engine.connect() as connection:
+        acquired = await connection.scalar(
+            text("SELECT pg_try_advisory_lock(:key)"), {"key": key}
+        )
+        print("UNEXPECTED: lock acquired" if acquired else "OK: source lock is busy")
+        if acquired:
+            await connection.execute(
+                text("SELECT pg_advisory_unlock(:key)"), {"key": key}
+            )
+
+
+asyncio.run(main())
+PY
+```
+
+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.
+
+## Failure and rollback
+
+- Disable the failing shadow job in Admin → Jobs. This stops scheduled imports
+  without changing existing data or the legacy scoring path.
+- 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
+  binary, clone, permission, disk, or environment failures.
+- 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.

frontend/harness.html

@@ -0,0 +1,18 @@
+<!doctype html>
+<html lang="en" class="dark">
+  <head>
+    <meta charset="UTF-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <title>FundamentalsPanel harness</title>
+    <link rel="preconnect" href="https://fonts.googleapis.com" />
+    <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin />
+    <link
+      href="https://fonts.googleapis.com/css2?family=Space+Grotesk:wght@400;500;600;700&family=Instrument+Sans:wght@400;500;600;700&family=IBM+Plex+Mono:wght@400;500;600&display=swap"
+      rel="stylesheet"
+    />
+  </head>
+  <body class="bg-[#0a0b11] text-gray-100 font-sans">
+    <div id="root"></div>
+    <script type="module" src="/src/dev/harness.tsx"></script>
+  </body>
+</html>

frontend/src/components/admin/ScheduleSettings.tsx

@@ -6,10 +6,12 @@ import { SkeletonTable } from '../ui/Skeleton';
 const DEFAULTS: ScheduleConfig = {
   schedule_timezone: 'America/New_York',
   schedule_daily_pipeline_cron: '0 2 * * *',
-  schedule_near_close_pipeline_cron: '30 15 * * 1-5',
-  schedule_after_close_pipeline_cron: '45 16 * * 1-5',
-  schedule_intraday_pipeline_cron: '0 10-15 * * 1-5',
-  schedule_fundamentals_cron: '0 1 * * 1',
+  schedule_dolt_earnings_cron: '30 2 * * *',
+  schedule_sec_fundamentals_cron: '0 4 * * *',
+  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',
+  schedule_fundamentals_cron: '0 1 * * mon',
 };
 
 const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: boolean }[] = [
@@ -24,6 +26,18 @@ const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: b
     hint: 'OHLCV → benchmark → sentiment → regime → alerts (no R:R scan). Default 02:00 ET so regime-quadrant changes hit Telegram in the morning.',
     mono: true,
   },
+  {
+    key: 'schedule_dolt_earnings_cron',
+    label: 'Dolt earnings (shadow)',
+    hint: 'Pull and import earnings dates/results daily at 02:30 ET. Live scoring remains untouched before A5.',
+    mono: true,
+  },
+  {
+    key: 'schedule_sec_fundamentals_cron',
+    label: 'SEC fundamentals (shadow)',
+    hint: 'Import tracked-universe SEC facts daily at 04:00 ET. Unchanged revisions become no-op runs.',
+    mono: true,
+  },
   {
     key: 'schedule_near_close_pipeline_cron',
     label: 'Near-close pipeline (scan + alert)',
@@ -44,8 +58,8 @@ const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: b
   },
   {
     key: 'schedule_fundamentals_cron',
-    label: 'Fundamentals (weekly)',
-    hint: 'Slow, rate-limited. Default early Monday ET.',
+    label: 'Legacy fundamentals (weekly)',
+    hint: 'Existing provider chain retained until the A5 parity approval and A6 removal.',
     mono: true,
   },
 ];

frontend/src/components/ticker/FundamentalsPanel.tsx

@@ -1,157 +1,397 @@
-import { useMemo, useState } from 'react';
-import { formatPercent, formatLargeNumber } from '../../lib/format';
-import {
-  fundamentalScore,
-  metricStatus,
-  overallFundamentalStatus,
-} from '../../lib/fundamentals';
-import type { FundamentalResponse } from '../../lib/types';
+import { useMemo, type ReactNode } from 'react';
+import type {
+  EarningsRecent,
+  FundamentalResponse,
+  MetricItem,
+  MetricIndustry,
+} from '../../lib/types';
 
 interface FundamentalsPanelProps {
   data: FundamentalResponse;
 }
 
-const FIELD_LABELS: Record<string, string> = {
-  pe_ratio: 'P/E Ratio',
-  revenue_growth: 'Revenue Growth',
-  earnings_surprise: 'Earnings Surprise',
-  market_cap: 'Market Cap',
-};
+/** Favorable / neutral / adverse — always paired with the read text. */
+type Tone = 'good' | 'flat' | 'bad';
 
-type MetricKey = 'pe_ratio' | 'revenue_growth' | 'earnings_surprise' | 'market_cap';
+// Horizon tokens.
+const HZ = {
+  text: '#EDEEF3',
+  muted: '#9AA0B0',
+  track: '#5D6373',
+  fav: '#6EC9DB', // cyan
+  adv: '#EF9182', // coral
+};
+const toneColor = (t: Tone) => (t === 'good' ? HZ.fav : t === 'bad' ? HZ.adv : HZ.muted);
+
+const POSITIVE_READS = new Set([
+  'accelerating', 'improving', 'above peers', 'above own average', 'buying back',
+  'attractively valued', 'conservative leverage',
+]);
+const NEGATIVE_READS = new Set([
+  'decelerating', 'deteriorating', 'priced above peers', 'elevated leverage', 'below peers',
+]);
+function readTone(read: string | null | undefined): Tone {
+  if (!read) return 'flat';
+  if (POSITIVE_READS.has(read)) return 'good';
+  if (NEGATIVE_READS.has(read)) return 'bad';
+  if (read.includes('dilution')) return 'bad';
+  return 'flat';
+}
+
+function pct(v: number | null | undefined): string {
+  if (v == null || !Number.isFinite(v)) return 'n/a';
+  return `${Math.round(v * 10) / 10}%`;
+}
+function mult(v: number | null | undefined): string {
+  if (v == null || !Number.isFinite(v)) return 'n/a';
+  return `${v.toFixed(1)}×`;
+}
+function money(v: number | null | undefined): string {
+  if (v == null || !Number.isFinite(v)) return 'n/a';
+  const abs = Math.abs(v);
+  if (abs >= 1e12) return `$${(v / 1e12).toFixed(1)}T`;
+  if (abs >= 1e9) return `$${(v / 1e9).toFixed(1)}B`;
+  if (abs >= 1e6) return `$${(v / 1e6).toFixed(1)}M`;
+  return `$${v.toFixed(0)}`;
+}
+function signedPp(v: number): string {
+  return `${v >= 0 ? '+' : ''}${Math.round(v * 10) / 10}pp`;
+}
+function capitalize(s: string): string {
+  return s.length ? s[0].toUpperCase() + s.slice(1) : s;
+}
+function finiteOrNull(v: number | null | undefined): number | null {
+  return v != null && Number.isFinite(v) ? v : null;
+}
+function latestHistory(metric: MetricItem | undefined): number[] {
+  const run: number[] = [];
+  const history = metric?.history ?? [];
+  for (let i = history.length - 1; i >= 0; i -= 1) {
+    const value = finiteOrNull(history[i].value);
+    if (value == null) break;
+    run.unshift(value);
+  }
+  return run;
+}
+
+/** Parse YYYY-MM-DD as a LOCAL calendar date (no UTC off-by-one west of UTC). */
+function parseLocalDate(s: string): Date {
+  const [y, m, d] = s.split('-').map(Number);
+  return new Date(y, (m ?? 1) - 1, d ?? 1);
+}
+function shortDate(s: string): string {
+  return parseLocalDate(s).toLocaleDateString(undefined, { month: 'short', day: 'numeric' });
+}
 
 export function FundamentalsPanel({ data }: FundamentalsPanelProps) {
-  const [expanded, setExpanded] = useState<boolean>(false);
+  const metrics = useMemo(
+    () => Object.fromEntries((data.metrics ?? []).map((m) => [m.key, m])),
+    [data.metrics],
+  ) as Record<string, MetricItem | undefined>;
+  const reads = data.reads?.by_key ?? {};
+  const val = data.valuation;
+  const earnings = data.earnings;
+  const provenance = (data.metrics ?? []).find((m) => m.period_end) ?? null;
 
-  const score = useMemo(
-    () =>
-      fundamentalScore({
-        pe_ratio: data.pe_ratio,
-        revenue_growth: data.revenue_growth,
-        earnings_surprise: data.earnings_surprise,
-      }),
-    [data.pe_ratio, data.revenue_growth, data.earnings_surprise],
-  );
-  const overall = overallFundamentalStatus(score);
+  const hasAny = (data.metrics ?? []).some((m) => m.value != null) || !!val || !!earnings?.next
+    || (earnings?.recent?.length ?? 0) > 0;
 
-  const items: {
-    key: MetricKey;
-    label: string;
-    value: number | null;
-    format: (v: number) => string;
-  }[] = [
-    { key: 'pe_ratio', label: 'P/E Ratio', value: data.pe_ratio, format: (v) => v.toFixed(2) },
-    { key: 'revenue_growth', label: 'Revenue Growth', value: data.revenue_growth, format: formatPercent },
-    { key: 'earnings_surprise', label: 'Earnings Surprise', value: data.earnings_surprise, format: formatPercent },
-    { key: 'market_cap', label: 'Market Cap', value: data.market_cap, format: formatLargeNumber },
+  const trendRows: { key: string; label: string; kind: 'growth' | 'margin' | 'share' }[] = [
+    { key: 'revenue_growth_yoy', label: 'Revenue growth', kind: 'growth' },
+    { key: 'eps_growth_yoy', label: 'EPS growth', kind: 'growth' },
+    { key: 'operating_margin', label: 'Operating margin', kind: 'margin' },
+    { key: 'fcf_margin', label: 'FCF margin', kind: 'margin' },
+    { key: 'share_count_change_yoy', label: 'Share count YoY', kind: 'share' },
   ];
 
-  const unavailableEntries = Object.entries(data.unavailable_fields ?? {});
+  const valueRows: {
+    label: string; value: number | null; industry: MetricIndustry | null;
+    readKey: string; fmt: (v: number | null) => string;
+  }[] = [
+    { label: 'Net debt / EBITDA', value: metrics.net_debt_to_ebitda?.value ?? null,
+      industry: metrics.net_debt_to_ebitda?.industry ?? null, readKey: 'net_debt_to_ebitda', fmt: mult },
+    { label: 'P/E', value: val?.pe ?? null, industry: val?.pe_industry ?? null, readKey: 'pe', fmt: mult },
+    { label: 'FCF yield', value: val?.fcf_yield ?? null, industry: val?.fcf_yield_industry ?? null,
+      readKey: 'fcf_yield', fmt: pct },
+  ];
 
   return (
-    <div className="glass p-5">
-      <div className="mb-3 flex items-baseline justify-between gap-2">
-        <h3 className="text-xs font-medium uppercase tracking-widest text-gray-500">Fundamentals</h3>
-        {score != null && (
-          <span className="num text-[10px] text-gray-600" title="Equal average of available P/E, growth, and surprise (need 2+)">
-            score {score.toFixed(0)}
-          </span>
-        )}
-      </div>
+    <section className="glass p-5" aria-label="Fundamentals">
+      <h3 className="text-[10px] font-medium uppercase tracking-widest" style={{ color: HZ.muted }}>Fundamentals</h3>
+      {data.reads?.header ? (
+        <p className="mt-0.5 text-[15px] leading-snug" style={{ color: HZ.text }}>
+          {capitalize(data.reads.header)}
+        </p>
+      ) : !hasAny ? (
+        <p className="mt-1 text-[15px] leading-snug" style={{ color: HZ.muted }}>
+          No fundamentals reported yet.
+        </p>
+      ) : null}
 
-      <p className={`text-sm font-semibold ${overall.tone}`}>{overall.text}</p>
+      {hasAny && (
+        <>
+          <EarningsStrip earnings={earnings} />
 
-      <div className="mt-3 space-y-3 text-sm">
-        {items.map((item) => {
-          const reason = data.unavailable_fields?.[item.key];
-          const status = item.value !== null ? metricStatus(item.key, item.value) : null;
-          let display: React.ReactNode;
-          let valueClass = 'text-gray-200';
-
-          if (item.value !== null) {
-            display = item.format(item.value);
-          } else if (reason) {
-            display = reason;
-            valueClass = 'text-amber-400';
-          } else {
-            display = '—';
-          }
-
-          return (
-            <div key={item.key} className="flex items-start justify-between gap-3">
-              <span className="text-gray-400">{item.label}</span>
-              <div className="min-w-0 text-right">
-                <div className={`num ${valueClass}`}>{display}</div>
-                {status && (
-                  <div className={`mt-0.5 text-[11.5px] font-medium ${status.tone}`}>{status.text}</div>
-                )}
+          <div className="mt-4 grid gap-x-8 gap-y-5 sm:grid-cols-2">
+            <div>
+              <SectionHead label="Operating trend" axis="less favorable ← ref → more favorable" />
+              <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]} />
+                ))}
               </div>
             </div>
-          );
-        })}
-      </div>
-
-      <p className="mt-4 text-[11px] leading-relaxed text-gray-500">
-        Score = average of available P/E, revenue growth, and earnings surprise (need 2+).
-        {' '}P/E: lower scores higher (≈15 best, ≈45 worst).
-        {' '}Growth / surprise: 0% is neutral; stronger positives lift the score.
-        {' '}Market cap is size context only — not scored.
-      </p>
-
-      <button
-        type="button"
-        onClick={() => setExpanded((prev) => !prev)}
-        className="mt-3 flex w-full items-center justify-center gap-1 text-xs text-gray-500 transition-colors hover:text-gray-300"
-        aria-expanded={expanded}
-        aria-label={expanded ? 'Collapse details' : 'Expand details'}
-      >
-        <svg
-          className={`h-4 w-4 transition-transform ${expanded ? 'rotate-180' : ''}`}
-          fill="none"
-          viewBox="0 0 24 24"
-          stroke="currentColor"
-          strokeWidth={2}
-        >
-          <path strokeLinecap="round" strokeLinejoin="round" d="M19 9l-7 7-7-7" />
-        </svg>
-      </button>
-
-      {expanded && (
-        <div className="mt-3 space-y-3 border-t border-white/10 pt-3">
-          <div className="space-y-1 text-sm">
-            <div className="flex justify-between">
-              <span className="text-gray-500">Data Source</span>
-              <span className="text-gray-300">FMP</span>
-            </div>
-            {data.fetched_at && (
-              <div className="flex justify-between">
-                <span className="text-gray-500">Fetched</span>
-                <span className="text-gray-300">{new Date(data.fetched_at).toLocaleString()}</span>
+            <div>
+              <SectionHead label="Valuation & balance" axis="less favorable ← median → more favorable" />
+              <div className="mt-2.5 space-y-3.5">
+                {valueRows.map((r) => (
+                  <ValueRow key={r.label} {...r} read={reads[r.readKey]} />
+                ))}
               </div>
-            )}
+            </div>
           </div>
 
-          {unavailableEntries.length > 0 && (
-            <div>
-              <span className="text-xs font-medium uppercase tracking-widest text-gray-500">Unavailable Fields</span>
-              <ul className="mt-1 space-y-1">
-                {unavailableEntries.map(([field, reason]) => (
-                  <li key={field} className="flex justify-between text-sm">
-                    <span className="text-gray-400">{FIELD_LABELS[field] ?? field}</span>
-                    <span className="text-amber-400">{reason}</span>
-                  </li>
-                ))}
-              </ul>
-            </div>
-          )}
-        </div>
+          <Provenance provenance={provenance} priceDate={val?.price_date ?? null}
+                      marketCap={val?.market_cap_est ?? null} />
+        </>
       )}
+    </section>
+  );
+}
 
-      {!expanded && data.fetched_at && (
-        <p className="mt-2 text-xs text-gray-500">
-          Updated {new Date(data.fetched_at).toLocaleDateString()}
-        </p>
-      )}
+function SectionHead({ label, axis }: { label: string; axis: string }) {
+  return (
+    <div className="flex flex-wrap items-baseline justify-between gap-x-2">
+      <span className="text-[10px] font-medium uppercase tracking-widest" style={{ color: HZ.muted }}>{label}</span>
+      <span className="text-[11px]" style={{ color: HZ.track }}>{axis}</span>
     </div>
   );
 }
+
+function Bullet({ label, value, rail, comparison }: {
+  label: string; value: ReactNode; rail: ReactNode | null; comparison: ReactNode;
+}) {
+  return (
+    <div>
+      <div className="flex items-baseline justify-between gap-2">
+        <span className="truncate text-sm" style={{ color: HZ.muted }}>{label}</span>
+        <span className="num text-[15px]" style={{ color: HZ.text }}>{value}</span>
+      </div>
+      {rail && <div className="mt-1.5">{rail}</div>}
+      <div className={rail ? 'mt-1 text-[11.5px] leading-snug' : 'mt-1.5 text-[11.5px] leading-snug'}>
+        {comparison}
+      </div>
+    </div>
+  );
+}
+
+// ---- operating-trend row (delta vs reference, favorable = right) ------------
+
+function TrendRow({ label, kind, metric, read }: {
+  label: string; kind: 'growth' | 'margin' | 'share';
+  metric: MetricItem | undefined; read: string | null | undefined;
+}) {
+  const tone = readTone(read);
+  const value = finiteOrNull(metric?.value);
+  const history = latestHistory(metric);
+
+  let ref: number | null = null;
+  let refWord = '';
+  let halfRange = 8;
+  let neutral = 2;
+  let favSign = 1; // +1: higher is favorable; -1: lower is favorable
+  if (kind === 'growth') {
+    ref = history.length >= 2 ? history[history.length - 2] : null;
+    refWord = 'prior'; halfRange = 8; neutral = 2; favSign = 1;
+  } else if (kind === 'margin') {
+    const prior = history.slice(0, -1);
+    ref = prior.length ? prior.reduce((a, b) => a + b, 0) / prior.length : null;
+    refWord = 'avg'; halfRange = 4; neutral = 1; favSign = 1;
+  } else {
+    ref = 0; refWord = ''; halfRange = 5; neutral = 1; favSign = -1; // buyback (negative) is favorable
+  }
+  const delta = value != null && ref != null ? value - ref : null;
+
+  const comparison = value == null ? (
+    <span style={{ color: HZ.track }}>n/a</span>
+  ) : delta == null ? (
+    <span style={{ color: HZ.track }}>history n/a</span>
+  ) : (
+    <span style={{ color: toneColor(tone) }}>
+      {kind !== 'share' && (
+        <span className="num">{signedPp(delta)} vs {refWord} · </span>
+      )}
+      {read ?? '—'}
+    </span>
+  );
+
+  const rail = delta != null
+    ? <DeltaRail favOffset={favSign * delta} halfRange={halfRange} neutral={neutral} tone={tone}
+                 ariaLabel={`${label} ${pct(value)}, ${delta != null ? `${signedPp(delta)} vs reference` : ''}, ${read ?? 'no read'}`} />
+    : null;
+
+  return <Bullet label={label} value={pct(value)} rail={rail} comparison={comparison} />;
+}
+
+/** Comparison rail centered on a reference line (not a progress bar). favOffset > 0
+ * is favorable and moves the dot RIGHT for every metric. */
+function DeltaRail({ favOffset, halfRange, neutral, tone, ariaLabel }: {
+  favOffset: number; halfRange: number; neutral: number; tone: Tone; ariaLabel: string;
+}) {
+  const clamped = Math.max(-halfRange, Math.min(halfRange, favOffset));
+  const pos = 50 + (clamped / halfRange) * 50;
+  const barLeft = Math.min(50, pos);
+  const barWidth = Math.abs(pos - 50);
+  const bandHalf = (neutral / halfRange) * 50;
+  return (
+    <span className="relative block h-1.5 min-w-[7rem] flex-1 rounded-full" role="img" aria-label={ariaLabel}
+          style={{ background: 'rgba(93,99,115,0.22)' }}>
+      <span className="absolute inset-y-0 rounded-full" aria-hidden
+            style={{ left: `${50 - bandHalf}%`, width: `${2 * bandHalf}%`, background: 'rgba(93,99,115,0.4)' }} />
+      <span className="absolute inset-y-[-2px] w-px" aria-hidden style={{ left: '50%', background: 'rgba(237,238,243,0.55)' }} />
+      <span className="absolute inset-y-0 rounded-full" aria-hidden style={{ left: `${barLeft}%`, width: `${barWidth}%`, background: toneColor(tone) }} />
+      <Dot pos={pos} tone={tone} />
+    </span>
+  );
+}
+
+// ---- valuation/balance row (favorable percentile vs median) ----------------
+
+function ValueRow({ label, value, industry, read, fmt }: {
+  label: string; value: number | null; industry: MetricIndustry | null;
+  read: string | null | undefined; fmt: (v: number | null) => string;
+}) {
+  const tone = readTone(read);
+  const safeValue = finiteOrNull(value);
+  const rail = safeValue == null || !industry
+    ? null
+    : <PercentileRail percentile={industry.favorable_percentile} tone={tone}
+        ariaLabel={`${label}: ${industry.favorable_percentile}th favorable percentile vs ${industry.label}, median ${fmt(industry.median)}, ${industry.peer_count} peers`} />;
+  const comparison = safeValue == null ? (
+    <span style={{ color: HZ.track }}>n/a</span>
+  ) : industry ? (
+    <span style={{ color: toneColor(tone) }}>
+      {read ?? 'in line'}<span style={{ color: HZ.muted }}> · median {fmt(industry.median)} · {industry.peer_count} peers</span>
+    </span>
+  ) : (
+    <span style={{ color: HZ.track }}>peers n/a</span>
+  );
+  return <Bullet label={label} value={fmt(safeValue)} rail={rail} comparison={comparison} />;
+}
+
+/** 0-100 favorable-percentile rail with the peer median fixed at 50; right = more favorable. */
+function PercentileRail({ percentile, tone, ariaLabel }: {
+  percentile: number; tone: Tone; ariaLabel: string;
+}) {
+  const p = Math.max(0, Math.min(100, percentile));
+  const barLeft = Math.min(50, p);
+  const barWidth = Math.abs(p - 50);
+  return (
+    <span className="relative block h-1.5 min-w-[7rem] flex-1 rounded-full" role="img" aria-label={ariaLabel}
+          style={{ background: 'rgba(93,99,115,0.22)' }}>
+      <span className="absolute inset-y-[-2px] w-px" aria-hidden style={{ left: '50%', background: 'rgba(237,238,243,0.55)' }} />
+      <span className="absolute inset-y-0 rounded-full" aria-hidden style={{ left: `${barLeft}%`, width: `${barWidth}%`, background: toneColor(tone) }} />
+      <Dot pos={p} tone={tone} />
+    </span>
+  );
+}
+
+function Dot({ pos, tone }: { pos: number; tone: Tone }) {
+  return (
+    <span className="absolute top-1/2 h-2.5 w-2.5 -translate-x-1/2 -translate-y-1/2 rounded-full" aria-hidden
+          style={{ left: `${pos}%`, background: toneColor(tone), boxShadow: '0 0 0 2px #11131C' }} />
+  );
+}
+
+// ---- earnings + provenance -------------------------------------------------
+
+function Provenance({ provenance, priceDate, marketCap }: {
+  provenance: MetricItem | null; priceDate: string | null; marketCap: number | null;
+}) {
+  if (!provenance?.period_end && !priceDate) return null;
+  return (
+    <p className="mt-4 border-t border-white/10 pt-2 text-[11px] leading-relaxed" style={{ color: HZ.track }}>
+      {provenance?.period_end && (
+        <>SEC filings · latest {shortDate(provenance.period_end)}
+          {provenance.filed_date && <> (filed {shortDate(provenance.filed_date)})</>}</>
+      )}
+      {priceDate && (
+        <>{provenance?.period_end ? ' · ' : ''}Valuation at {shortDate(priceDate)} close · market cap {money(marketCap)} est.</>
+      )}
+    </p>
+  );
+}
+
+function EarningsStrip({ earnings }: { earnings: FundamentalResponse['earnings'] }) {
+  const next = earnings?.next;
+  const recent = earnings?.recent ?? [];
+  const when = next ? (next.days_until === 0 ? 'today' : `${next.days_until}d`) : null;
+  return (
+    <div className="mt-2 flex flex-wrap items-center justify-between gap-x-4 gap-y-1.5">
+      <span className="text-sm" style={{ color: HZ.text }}>
+        <span style={{ color: HZ.muted }}>Next earnings </span>
+        {next ? (
+          <>
+            {shortDate(next.date)}
+            {' · '}
+            <span className="uppercase">{next.session === 'unknown' ? 'TBD' : next.session}</span>
+            <span style={{ color: HZ.muted }}> · {when}</span>
+          </>
+        ) : (
+          <span style={{ color: HZ.muted }}>no date</span>
+        )}
+      </span>
+      {recent.length > 0 && <SurpriseSpark recent={recent} />}
+    </div>
+  );
+}
+
+/** Four tiny diverging bars around a zero baseline: beat above (cyan), miss below
+ * (coral), height ~ |surprise %|. Reads as a beat/miss history at a glance. */
+function SurpriseSpark({ recent }: { recent: EarningsRecent[] }) {
+  const ordered = recent.slice().reverse();
+  const description = ordered.map((e) => {
+    const surprise = e.surprise_pct;
+    const amount = surprise != null ? ` ${surprise > 0 ? '+' : ''}${surprise}%` : '';
+    return `${e.announce_date} ${surpriseLabel(e)}${amount}`;
+  }).join(', ');
+  return (
+    <span className="flex items-center gap-2">
+      <span className="text-[10px] uppercase tracking-widest" style={{ color: HZ.track }}>
+        EPS surprises
+      </span>
+      <span
+        className="relative flex h-7 items-center gap-1.5 px-0.5"
+        role="img"
+        tabIndex={0}
+        title={description}
+        aria-label={`Recent EPS surprises, oldest to newest: ${description}`}
+      >
+        <span className="absolute inset-x-0 top-1/2 h-px" aria-hidden style={{ background: 'rgba(93,99,115,0.5)' }} />
+        {ordered.map((e, i) => <SurpriseBar key={i} e={e} />)}
+      </span>
+    </span>
+  );
+}
+
+function surpriseLabel(e: EarningsRecent): string {
+  const beat = e.eps_actual != null && e.eps_estimate != null ? e.eps_actual - e.eps_estimate : null;
+  return beat == null ? 'n/a' : beat > 0 ? 'beat' : beat < 0 ? 'miss' : 'in line';
+}
+
+function SurpriseBar({ e }: { e: EarningsRecent }) {
+  const beat = e.eps_actual != null && e.eps_estimate != null ? e.eps_actual - e.eps_estimate : null;
+  const tone: Tone = beat == null ? 'flat' : beat > 0 ? 'good' : beat < 0 ? 'bad' : 'flat';
+  const s = e.surprise_pct;
+  const mag = s != null ? Math.min(Math.abs(s), 15) / 15 : 0; // cap at 15%
+  const h = beat == null ? 2 : 3 + mag * 9; // px
+  const up = (beat ?? 0) >= 0;
+  return (
+    <span className="relative z-[1] block h-7 w-2"
+          title={`${e.announce_date}: ${surpriseLabel(e)}${s != null ? ` ${s > 0 ? '+' : ''}${s}%` : ''}`}>
+      <span className="absolute inset-x-0 rounded-[1px]" aria-hidden
+            style={{ height: h, background: toneColor(tone), ...(up ? { bottom: '50%' } : { top: '50%' }) }} />
+    </span>
+  );
+}

frontend/src/dev/harness.tsx

@@ -0,0 +1,131 @@
+/* Dev-only visual harness for FundamentalsPanel. Served at /harness.html by
+ * `vite`. Not imported by the app. Renders the three key states so desktop and
+ * mobile can be eyeballed with representative fixtures. */
+import { createRoot } from 'react-dom/client';
+import '../styles/globals.css';
+import { FundamentalsPanel } from '../components/ticker/FundamentalsPanel';
+import type { FundamentalResponse, MetricItem } from '../lib/types';
+
+function h(period: string, value: number | null) {
+  return { period_end: period, value };
+}
+const P = ['2025-06-30', '2025-09-30', '2025-12-31', '2026-03-28'];
+
+function dateFromToday(days: number): string {
+  const date = new Date();
+  date.setHours(12, 0, 0, 0);
+  date.setDate(date.getDate() + days);
+  return [
+    date.getFullYear(),
+    String(date.getMonth() + 1).padStart(2, '0'),
+    String(date.getDate()).padStart(2, '0'),
+  ].join('-');
+}
+
+function metric(key: string, value: number | null, hist: (number | null)[],
+                industry: MetricItem['industry'] = 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',
+  };
+}
+const ind = (median: number, favorable_percentile: number) =>
+  ({ label: 'SIC 35 peers', median, favorable_percentile, peer_count: 12 });
+
+const legacy = {
+  pe_ratio: null, revenue_growth: null, earnings_surprise: null, market_cap: null,
+  next_earnings_date: null, fetched_at: null, unavailable_fields: {},
+};
+
+const full: FundamentalResponse = {
+  symbol: 'AAPL', ...legacy,
+  earnings: {
+    next: { date: dateFromToday(12), session: 'amc', days_until: 12 },
+    recent: [
+      { announce_date: '2025-08-01', period_end: '2025-06-30', eps_estimate: 1.4, eps_actual: 1.6, surprise_pct: 14.3 },
+      { announce_date: '2025-11-01', period_end: '2025-09-30', eps_estimate: 1.7, eps_actual: 1.9, surprise_pct: 11.8 },
+      { announce_date: '2026-02-01', period_end: '2025-12-31', eps_estimate: 2.6, eps_actual: 2.4, surprise_pct: -7.7 },
+      { announce_date: '2026-05-01', period_end: '2026-03-28', eps_estimate: 1.5, eps_actual: 1.65, surprise_pct: 10.0 },
+    ],
+  },
+  metrics: [
+    metric('revenue_growth_yoy', 18, [8, 11, 15, 18], ind(11, 82)),
+    metric('eps_growth_yoy', 24, [10, 18, 22, 24], ind(15, 70)),
+    metric('operating_margin', 32, [30, 31, 31, 32], ind(22, 88)),
+    metric('fcf_margin', 28, [24, 25, 27, 28], ind(18, 80)),
+    metric('net_debt', 16.2e9, [46e9, 44e9, 24e9, 16.2e9], null),
+    metric('net_debt_to_ebitda', 1.4, [1.9, 1.7, 1.5, 1.4], ind(2.1, 68)),
+    metric('share_count_change_yoy', -1.7, [-2.4, -2.2, -2.3, -1.7], null),
+  ],
+  valuation: {
+    pe: 29.2, fcf_yield: 3.8, market_cap_est: 3.2e12,
+    pe_industry: ind(23.5, 30), fcf_yield_industry: ind(3.1, 70), price_date: '2026-05-01',
+  },
+  reads: {
+    header: 'growth accelerating · margins improving · valuation priced above peers',
+    by_key: {
+      revenue_growth_yoy: 'accelerating', eps_growth_yoy: 'accelerating',
+      operating_margin: 'improving', fcf_margin: 'improving',
+      share_count_change_yoy: 'buying back', net_debt_to_ebitda: 'conservative leverage',
+      pe: 'priced above peers', fcf_yield: 'above peers', net_debt: null,
+    },
+  },
+};
+
+const partial: FundamentalResponse = {
+  symbol: 'NEWCO', ...legacy,
+  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('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),
+  ],
+  valuation: {
+    pe: 15.2, fcf_yield: null, market_cap_est: 5.4e8,
+    pe_industry: null, fcf_yield_industry: null, price_date: '2026-05-01',
+  },
+  reads: {
+    header: 'growth steady · margins stable',
+    by_key: {
+      revenue_growth_yoy: 'steady', operating_margin: 'stable',
+      share_count_change_yoy: '2.1% dilution', net_debt_to_ebitda: null,
+      pe: null, fcf_yield: null, eps_growth_yoy: null, fcf_margin: null, net_debt: null,
+    },
+  },
+};
+
+const empty: FundamentalResponse = {
+  symbol: 'ADR', ...legacy,
+  earnings: { next: null, recent: [] },
+  metrics: [
+    'revenue_growth_yoy', 'eps_growth_yoy', 'operating_margin', 'fcf_margin',
+    'net_debt', 'net_debt_to_ebitda', 'share_count_change_yoy',
+  ].map((k) => metric(k, null, [])),
+  valuation: null,
+  reads: { header: null, by_key: {} },
+};
+
+function Case({ title, data }: { title: string; data: FundamentalResponse }) {
+  return (
+    <div>
+      <div className="mb-1.5 text-[11px] uppercase tracking-widest text-gray-500">{title}</div>
+      <FundamentalsPanel data={data} />
+    </div>
+  );
+}
+
+createRoot(document.getElementById('root')!).render(
+  <div className="mx-auto max-w-3xl space-y-8 p-6">
+    <p className="text-[11px] uppercase tracking-widest text-gray-500">
+      Desktop width (~768px, two columns). Resize the browser to ~390px to check mobile (single column).
+    </p>
+    <Case title="Full" data={full} />
+    <Case title="Partial · insufficient peers" data={partial} />
+    <Case title="Empty" data={empty} />
+  </div>,
+);

frontend/src/lib/types.ts

@@ -191,6 +191,8 @@ export interface ActivationConfig {
 export interface ScheduleConfig {
   schedule_timezone: string;
   schedule_daily_pipeline_cron: string;
+  schedule_dolt_earnings_cron: string;
+  schedule_sec_fundamentals_cron: string;
   schedule_near_close_pipeline_cron: string;
   schedule_after_close_pipeline_cron: string;
   schedule_intraday_pipeline_cron: string;
@@ -703,8 +705,74 @@ export interface SentimentResponse {
 }
 
 // Fundamentals
+export interface MetricIndustry {
+  label: string;
+  median: number;
+  favorable_percentile: number; // 0-100, polarity-aware (higher = more favorable)
+  peer_count: number;
+}
+
+export interface MetricHistoryPoint {
+  period_end: string | null; // YYYY-MM-DD
+  value: number | null;
+}
+
+export type MetricKey =
+  | 'revenue_growth_yoy'
+  | 'eps_growth_yoy'
+  | 'operating_margin'
+  | 'fcf_margin'
+  | 'net_debt'
+  | 'net_debt_to_ebitda'
+  | 'share_count_change_yoy';
+
+export interface MetricItem {
+  key: MetricKey;
+  value: number | null;
+  history: MetricHistoryPoint[];
+  industry: MetricIndustry | null;
+  period_end: string | null;
+  filed_date: string | null;
+  source: string; // 'sec' | 'legacy_api'
+}
+
+export interface EarningsNext {
+  date: string;
+  session: string; // bmo | amc | unknown
+  days_until: number;
+}
+
+export interface EarningsRecent {
+  announce_date: string;
+  period_end: string | null;
+  eps_estimate: number | null;
+  eps_actual: number | null;
+  surprise_pct: number | null;
+}
+
+export interface EarningsObject {
+  next: EarningsNext | null;
+  recent: EarningsRecent[];
+}
+
+export interface Valuation {
+  pe: number | null;
+  fcf_yield: number | null;
+  market_cap_est: number | null;
+  pe_industry: MetricIndustry | null;
+  fcf_yield_industry: MetricIndustry | null;
+  price_date: string | null;
+}
+
+export interface FundamentalsReads {
+  header: string | null;
+  // fixed map over every metric key plus 'pe' and 'fcf_yield'; null when unavailable
+  by_key: Record<string, string | null>;
+}
+
 export interface FundamentalResponse {
   symbol: string;
+  // legacy fields (unchanged)
   pe_ratio: number | null;
   revenue_growth: number | null;
   earnings_surprise: number | null;
@@ -712,6 +780,11 @@ export interface FundamentalResponse {
   next_earnings_date: string | null;
   fetched_at: string | null;
   unavailable_fields: Record<string, string>;
+  // additive v1 (SEC/Dolt-derived) — present, with null/empty when unavailable
+  earnings: EarningsObject | null;
+  metrics: MetricItem[] | null;
+  valuation: Valuation | null;
+  reads: FundamentalsReads | null;
 }
 
 // Indicators

tests/unit/test_fundamentals_api.py

@@ -0,0 +1,252 @@
+"""Integration tests for the additive fundamentals API v1 assembly."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from datetime import date, datetime, timezone
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from app.database import Base
+import app.models  # noqa: F401
+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.schemas.fundamental import FundamentalResponse
+from app.services.fundamentals_api_service import METRIC_KEYS, build_fundamentals_v1
+
+UTC = timezone.utc
+TODAY = date(2026, 10, 15)
+
+
+@pytest.fixture
+async def factory():
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    eng = create_async_engine(f"sqlite+aiosqlite:///{path}")
+    async with eng.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    try:
+        yield async_sessionmaker(eng, class_=AsyncSession, expire_on_commit=False)
+    finally:
+        await eng.dispose()
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+
+
+_MONTHS = [3, 6, 9, 12]
+_FP = ["Q1", "Q2", "Q3", "FY"]
+
+
+async def _seed_issuer(s, symbol, cik, sic, rev_base, price, *, eps_base=1.0, snapshots=True):
+    t = Ticker(symbol=symbol, cik=cik, sic=sic)
+    s.add(t)
+    await s.flush()
+    if snapshots:
+        # three fiscal years so YoY growth reads have a >=3 consecutive run
+        for fy, mult in [(2024, 0.9), (2025, 1.0), (2026, 1.1)]:
+            shares = {2024: 1050, 2025: 1000, 2026: 950}[fy]  # steady buyback
+            rev = [rev_base * mult * x for x in (1.0, 1.05, 1.1, 1.15)]
+            eps = [eps_base * mult * x for x in (1.0, 1.05, 1.1, 1.15)]
+            for i, fp in enumerate(_FP):
+                pe = date(fy, _MONTHS[i], 28)
+                s.add(FundamentalSnapshot(
+                    cik=cik, accession=f"{cik}-{fy}-{fp}", form="10-K" if fp == "FY" else "10-Q",
+                    filed_date=pe, accepted_at=datetime(fy, _MONTHS[i], 28, tzinfo=UTC),
+                    period_end=pe, fiscal_year=fy, fiscal_period=fp,
+                    revenue=sum(rev[: i + 1]), operating_income=sum(rev[: i + 1]) * 0.2,
+                    diluted_eps=sum(eps[: i + 1]), cfo=sum(rev[: i + 1]) * 0.25,
+                    capex=sum(rev[: i + 1]) * 0.05, depreciation_amortization=sum(rev[: i + 1]) * 0.05,
+                    cash_and_st_investments=40, total_debt=100, shares_outstanding=shares))
+    s.add(OHLCVRecord(ticker_id=t.id, date=date(2026, 10, 1), open=price, high=price, low=price, close=price, volume=1000))
+    return t.id
+
+
+async def _seed_group(factory):
+    async with factory() as s:
+        aapl = await _seed_issuer(s, "AAPL", "0000000001", "3571", rev_base=1000, price=200, eps_base=2.0)
+        for i in range(5):  # 5 peers in SIC 35xx so the group has >= 5 valid issuers
+            await _seed_issuer(s, f"PEER{i}", f"000000010{i}", "3572", rev_base=500 + i * 100, price=50 + i * 10)
+        # AAPL earnings: one upcoming, one past with a surprise
+        s.add(EarningsEvent(ticker_id=aapl, announce_date=date(2026, 11, 1), session="amc", source="dolt_earnings"))
+        s.add(EarningsEvent(ticker_id=aapl, announce_date=date(2026, 8, 1), session="amc",
+                            period_end=date(2026, 6, 30), eps_estimate=2.0, eps_actual=2.2, source="dolt_earnings"))
+        await s.commit()
+    return aapl
+
+
+async def test_full_assembly(factory):
+    await _seed_group(factory)
+    async with factory() as s:
+        v1 = await build_fundamentals_v1(s, "AAPL", today=TODAY)
+
+    # earnings
+    assert v1["earnings"]["next"] == {"date": "2026-11-01", "session": "amc", "days_until": 17}
+    recent = v1["earnings"]["recent"]
+    assert recent and recent[0]["surprise_pct"] == pytest.approx(10.0)
+
+    # metrics — fixed key set, all present
+    assert [m["key"] for m in v1["metrics"]] == list(METRIC_KEYS)
+    by_key = {m["key"]: m for m in v1["metrics"]}
+    assert by_key["revenue_growth_yoy"]["value"] is not None
+    assert by_key["revenue_growth_yoy"]["source"] == "sec"
+    assert len(by_key["operating_margin"]["history"]) >= 3
+    # peer industry present for eligible metric (6 issuers), absent for size-dependent net_debt
+    assert by_key["operating_margin"]["industry"] is not None
+    assert by_key["operating_margin"]["industry"]["peer_count"] == 6
+    assert by_key["operating_margin"]["industry"]["label"] == "SIC 35 peers"
+    assert by_key["net_debt"]["industry"] is None
+
+    # valuation computed at request time
+    val = v1["valuation"]
+    assert val["pe"] is not None and val["market_cap_est"] is not None
+    assert val["price_date"] == "2026-10-01"
+    assert val["pe_industry"] is not None
+
+    # reads: header string + fixed by_key map (every metric + pe + fcf_yield)
+    assert v1["reads"]["header"]
+    assert set(v1["reads"]["by_key"]) == set(METRIC_KEYS) | {"pe", "fcf_yield"}
+    data = FundamentalResponse(symbol="AAPL", pe_ratio=12.3, **v1)  # legacy + v1 additive
+    dumped = data.model_dump()
+    assert dumped["pe_ratio"] == 12.3  # legacy preserved untouched
+    assert dumped["metrics"][0]["key"] == "revenue_growth_yoy"
+
+
+async def test_same_day_earnings_is_next_with_zero_days(factory):
+    async with factory() as s:
+        t = Ticker(symbol="TDY", cik=None)
+        s.add(t)
+        await s.flush()
+        s.add(EarningsEvent(ticker_id=t.id, announce_date=TODAY, session="bmo", source="dolt_earnings"))
+        s.add(EarningsEvent(ticker_id=t.id, announce_date=date(2026, 9, 1), session="amc",
+                            eps_estimate=1.0, eps_actual=1.1, source="dolt_earnings"))
+        await s.commit()
+    async with factory() as s:
+        v1 = await build_fundamentals_v1(s, "TDY", today=TODAY)
+    assert v1["earnings"]["next"] == {"date": TODAY.isoformat(), "session": "bmo", "days_until": 0}
+    # the same-day event is upcoming, not in recent
+    assert all(r["announce_date"] != TODAY.isoformat() for r in v1["earnings"]["recent"])
+
+
+async def test_eps_growth_read_is_populated(factory):
+    await _seed_group(factory)
+    async with factory() as s:
+        v1 = await build_fundamentals_v1(s, "AAPL", today=TODAY)
+    assert v1["reads"]["by_key"]["eps_growth_yoy"] is not None  # EPS read now computed
+
+
+async def test_non_positive_price_guards_valuation(factory):
+    async with factory() as s:
+        t = Ticker(symbol="ZERO", cik="0000000055", sic="3571")
+        s.add(t)
+        await s.flush()
+        s.add(FundamentalSnapshot(cik="0000000055", accession="z", form="10-K", filed_date=date(2026, 1, 1),
+                                  accepted_at=datetime(2026, 1, 1, tzinfo=UTC), period_end=date(2025, 12, 31),
+                                  fiscal_year=2025, fiscal_period="FY", diluted_eps=5.0, shares_outstanding=1000))
+        s.add(OHLCVRecord(ticker_id=t.id, date=date(2026, 1, 2), open=0, high=0, low=0, close=0, volume=1))
+        await s.commit()
+    async with factory() as s:
+        v1 = await build_fundamentals_v1(s, "ZERO", today=TODAY)
+    assert v1["valuation"] is None  # close of 0 is not a usable price
+
+
+async def test_no_cik_ticker_yields_null_metrics(factory):
+    async with factory() as s:
+        s.add(Ticker(symbol="ADR", cik=None))  # no SEC identity
+        await s.commit()
+    async with factory() as s:
+        v1 = await build_fundamentals_v1(s, "ADR", today=TODAY)
+    assert v1["valuation"] is None
+    assert all(m["value"] is None and m["industry"] is None for m in v1["metrics"])
+    assert v1["reads"]["header"] is None
+    assert v1["reads"]["by_key"] == {k: None for k in list(METRIC_KEYS) + ["pe", "fcf_yield"]}
+
+
+async def test_industry_omitted_below_five_peers(factory):
+    async with factory() as s:
+        await _seed_issuer(s, "SOLO", "0000000009", "9999", rev_base=1000, price=100, eps_base=2.0)
+        await s.commit()
+    async with factory() as s:
+        v1 = await build_fundamentals_v1(s, "SOLO", today=TODAY)
+    # only 1 issuer in the group -> below MIN_PEERS -> every industry omitted
+    assert all(m["industry"] is None for m in v1["metrics"])
+    assert v1["valuation"]["pe_industry"] is None
+    # but the subject's own valuation still computes
+    assert v1["valuation"]["pe"] is not None
+
+
+async def test_valuation_guarded_without_price(factory):
+    async with factory() as s:
+        t = Ticker(symbol="NOPX", cik="0000000077", sic="3571")
+        s.add(t)
+        await s.flush()
+        # snapshots but NO ohlcv close
+        s.add(FundamentalSnapshot(cik="0000000077", accession="a", form="10-K", filed_date=date(2026, 1, 1),
+                                  accepted_at=datetime(2026, 1, 1, tzinfo=UTC), period_end=date(2025, 12, 31),
+                                  fiscal_year=2025, fiscal_period="FY", diluted_eps=5.0, shares_outstanding=1000))
+        await s.commit()
+    async with factory() as s:
+        v1 = await build_fundamentals_v1(s, "NOPX", today=TODAY)
+    # no usable price -> valuation is null under the approved contract
+    assert v1["valuation"] is None
+
+
+async def test_multiclass_subject_priced_by_requested_ticker(factory):
+    cik = "0001652044"
+    async with factory() as s:
+        # GOOGL and GOOG share one CIK/snapshots but trade at different prices
+        await _seed_issuer(s, "GOOGL", cik, "7372", rev_base=1000, price=200, eps_base=2.0)
+        # add a second class sharing the CIK: same snapshots exist; just its own ticker+price
+        goog = Ticker(symbol="GOOG", cik=cik, sic="7372")
+        s.add(goog)
+        await s.flush()
+        s.add(OHLCVRecord(ticker_id=goog.id, date=date(2026, 10, 1), open=100, high=100, low=100, close=100, volume=1))
+        for i in range(4):  # peers so the group has >= 5 valid issuers
+            await _seed_issuer(s, f"P{i}", f"000000020{i}", "7373", rev_base=600 + i * 50, price=40 + i * 5)
+        await s.commit()
+
+    async with factory() as s:
+        googl = await build_fundamentals_v1(s, "GOOGL", today=TODAY)
+        goog_v = await build_fundamentals_v1(s, "GOOG", today=TODAY)
+
+    # subject P/E uses the REQUESTED class's price (200 vs 100), not an arbitrary sibling
+    assert googl["valuation"]["pe"] == pytest.approx(goog_v["valuation"]["pe"] * 2, rel=1e-6)
+
+
+async def test_endpoint_merges_legacy_and_v1(client, db_session):
+    from datetime import timezone as _tz
+
+    from app.dependencies import require_access
+    from app.main import app
+    from app.models.fundamental import FundamentalData
+
+    app.dependency_overrides[require_access] = lambda: None
+    try:
+        t = Ticker(symbol="AAPL", cik="0000000001", sic="3571")
+        db_session.add(t)
+        await db_session.flush()
+        db_session.add(FundamentalData(ticker_id=t.id, pe_ratio=12.3, revenue_growth=5.0,
+                                       fetched_at=datetime(2026, 1, 1, tzinfo=_tz.utc)))
+        db_session.add(FundamentalSnapshot(cik="0000000001", accession="a", form="10-K",
+                                           filed_date=date(2026, 1, 1), accepted_at=datetime(2026, 1, 1, tzinfo=_tz.utc),
+                                           period_end=date(2025, 12, 31), fiscal_year=2025, fiscal_period="FY",
+                                           diluted_eps=5.0, shares_outstanding=1000))
+        db_session.add(OHLCVRecord(ticker_id=t.id, date=date(2026, 1, 2), open=100, high=100, low=100, close=100, volume=1))
+        await db_session.flush()
+
+        resp = await client.get("/api/v1/fundamentals/AAPL")
+        assert resp.status_code == 200
+        data = resp.json()["data"]
+        assert data["pe_ratio"] == 12.3            # legacy preserved
+        assert data["revenue_growth"] == 5.0
+        assert len(data["metrics"]) == 7           # additive v1
+        assert data["earnings"] is not None
+        assert "by_key" in data["reads"]
+        assert data["valuation"]["price_date"] == "2026-01-02"
+    finally:
+        app.dependency_overrides.pop(require_access, None)

tests/unit/test_fundamentals_peers.py

@@ -2,24 +2,30 @@
 
 from __future__ import annotations
 
+import math
+
 from app.services import fundamentals_peers as pr
 
 
-def test_peer_stat_higher_is_better_percentile_and_median():
+def test_median_ranks_at_50_tie_aware():
     s = pr.peer_stat(3, [1, 2, 3, 4, 5], higher_is_better=True)
     assert s.median == 3
-    assert s.favorable_percentile == 60  # beats/ties 3 of 5
+    assert s.favorable_percentile == 50  # tie-aware rank of the median
     assert s.peer_count == 5
 
 
+def test_all_equal_peers_rank_at_50():
+    s = pr.peer_stat(3, [3, 3, 3, 3, 3], higher_is_better=True)
+    assert s.favorable_percentile == 50  # not 100 — ties don't get full credit
+
+
 def test_peer_stat_lower_is_better_flips_direction():
-    s = pr.peer_stat(3, [1, 2, 3, 4, 5], higher_is_better=False)
-    assert s.favorable_percentile == 60  # 3 of 5 are >= 3
+    assert pr.peer_stat(3, [1, 2, 3, 4, 5], higher_is_better=False).favorable_percentile == 50
 
 
-def test_peer_stat_top_and_bottom():
+def test_peer_stat_unique_top_and_bottom():
     assert pr.peer_stat(5, [1, 2, 3, 4, 5], higher_is_better=True).favorable_percentile == 100
-    assert pr.peer_stat(1, [1, 2, 3, 4, 5], higher_is_better=True).favorable_percentile == 20
+    assert pr.peer_stat(1, [1, 2, 3, 4, 5], higher_is_better=True).favorable_percentile == 0
 
 
 def test_peer_stat_requires_min_valid_peers():
@@ -27,9 +33,11 @@ def test_peer_stat_requires_min_valid_peers():
     assert pr.peer_stat(None, [1, 2, 3, 4, 5], higher_is_better=True) is None  # null subject
 
 
-def test_peer_stat_excludes_nulls_from_group():
-    s = pr.peer_stat(3, [1, 2, 3, 4, 5, None, None], higher_is_better=True)
-    assert s.peer_count == 5  # nulls dropped
+def test_peer_stat_excludes_null_and_non_finite():
+    s = pr.peer_stat(3, [1, 2, 3, 4, 5, None, math.nan, math.inf, -math.inf], higher_is_better=True)
+    assert s.peer_count == 5  # nulls + NaN/inf dropped
+    # a non-finite subject is invalid
+    assert pr.peer_stat(math.nan, [1, 2, 3, 4, 5], higher_is_better=True) is None
 
 
 def test_net_debt_is_not_peer_eligible():

tests/unit/test_fundamentals_reads.py

@@ -18,6 +18,16 @@ def test_growth_read_boundaries():
     assert rd.growth_read(_hist(5, 6)) is None                   # < 3 periods
 
 
+def test_reads_use_latest_nonnull_suffix():
+    # latest displayed value is n/a -> no read (never reflect a null latest)
+    assert rd.growth_read(_hist(5, 6, 8, None)) is None
+    assert rd.margin_read(_hist(19, 20, 22, None)) is None
+    # an internal gap truncates the run -> fewer than 3 consecutive -> no read
+    assert rd.growth_read(_hist(5, 6, None, 8)) is None
+    # a clean 3-run after an older gap still reads
+    assert rd.growth_read(_hist(None, 5, 6, 8)) == "accelerating"
+
+
 def test_margin_read_vs_mean_of_prior():
     # prior mean = (19+20)/2 = 19.5; latest 21 -> +1.5 -> improving
     assert rd.margin_read(_hist(19, 20, 21)) == "improving"

tests/unit/test_schedule_config.py

@@ -80,6 +80,28 @@ class TestTradingDayCrons:
         )
         assert fire.strftime("%a") == "Mon"
 
+    @pytest.mark.parametrize(
+        ("key", "hour", "minute"),
+        (
+            ("schedule_dolt_earnings_cron", 2, 30),
+            ("schedule_sec_fundamentals_cron", 4, 0),
+        ),
+    )
+    def test_shadow_imports_run_daily_at_expected_et_time(
+        self, key: str, hour: int, minute: int
+    ):
+        from datetime import datetime
+
+        from apscheduler.triggers.cron import CronTrigger
+
+        trigger = CronTrigger.from_crontab(
+            SCHEDULE_DEFAULTS[key], timezone=SCHEDULE_DEFAULTS["schedule_timezone"]
+        )
+        fire = trigger.get_next_fire_time(
+            None, datetime(2026, 7, 19, tzinfo=trigger.timezone)
+        )
+        assert (fire.hour, fire.minute) == (hour, minute)
+
 
 class TestScheduleConfig:
     async def test_defaults_when_unset(self, session: AsyncSession):

tests/unit/test_scheduler.py

@@ -1,5 +1,7 @@
 """Unit tests for app.scheduler module."""
 
+from types import SimpleNamespace
+
 import pytest
 
 from app.scheduler import (
@@ -8,7 +10,9 @@ from app.scheduler import (
     _parse_frequency,
     _resume_tickers,
     _last_successful,
+    _run_shadow_import,
     configure_scheduler,
+    get_job_runtime_snapshot,
     queue_backtest_options,
     queue_backtest_target_model,
     scheduler,
@@ -106,6 +110,8 @@ class TestConfigureScheduler:
             "benchmark_collector",
             "sentiment_collector",
             "fundamental_collector",
+            "dolt_earnings_import",
+            "sec_fundamentals_import",
             "rr_scanner",
             "shadow_book",
             "ticker_universe_sync",
@@ -137,6 +143,8 @@ class TestConfigureScheduler:
             "data_collector",
             "data_backfill",
             "fundamental_collector",
+            "dolt_earnings_import",
+            "sec_fundamentals_import",
             "market_regime",
             "near_close_pipeline",
             "regime_monitor",
@@ -147,3 +155,91 @@ class TestConfigureScheduler:
             "shadow_book",
             "ticker_universe_sync",
         ])
+
+
+class _SessionContext:
+    async def __aenter__(self):
+        return object()
+
+    async def __aexit__(self, *exc):
+        return None
+
+
+class TestShadowImportJobs:
+    @staticmethod
+    def _session_factory():
+        return _SessionContext()
+
+    async def test_promoted_run_surfaces_completion(self, monkeypatch):
+        async def enabled(db, job_name):
+            return True
+
+        async def imported(importer):
+            return SimpleNamespace(
+                status="promoted", revision="abcdef1234567890", error_details=None
+            )
+
+        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)
+
+        await _run_shadow_import("dolt_earnings_import", object())
+
+        runtime = get_job_runtime_snapshot("dolt_earnings_import")
+        assert runtime["status"] == "completed"
+        assert runtime["processed"] == 1
+        assert runtime["message"] == "promoted · abcdef123456"
+
+    async def test_failed_run_surfaces_error(self, monkeypatch):
+        async def enabled(db, job_name):
+            return True
+
+        async def imported(importer):
+            return SimpleNamespace(
+                status="failed", revision=None, error_details="validation failed"
+            )
+
+        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)
+
+        await _run_shadow_import("sec_fundamentals_import", object())
+
+        runtime = get_job_runtime_snapshot("sec_fundamentals_import")
+        assert runtime["status"] == "error"
+        assert runtime["processed"] == 0
+        assert runtime["message"] == "validation failed"
+
+    async def test_source_lock_surfaces_skipped(self, monkeypatch):
+        async def enabled(db, job_name):
+            return True
+
+        async def imported(importer):
+            return None
+
+        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)
+
+        await _run_shadow_import("dolt_earnings_import", object())
+
+        runtime = get_job_runtime_snapshot("dolt_earnings_import")
+        assert runtime["status"] == "skipped"
+        assert "already running" in runtime["message"]
+
+    async def test_disabled_job_never_runs_importer(self, monkeypatch):
+        async def disabled(db, job_name):
+            return False
+
+        async def should_not_run(importer):
+            raise AssertionError("disabled job ran importer")
+
+        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)
+
+        await _run_shadow_import("sec_fundamentals_import", object())
+
+        runtime = get_job_runtime_snapshot("sec_fundamentals_import")
+        assert runtime["status"] == "skipped"
+        assert runtime["message"] == "Disabled"