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:
+    legacy: dict = {}
-        data = FundamentalResponse(symbol=symbol.strip().upper())
+    if record is not None:
-    else:
+        legacy = dict(
-        data = FundamentalResponse(
-            symbol=symbol.strip().upper(),
             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/schemas/fundamental.py

@@ -7,8 +7,71 @@ 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."""
+
+    header: str = ""
+    metrics: dict[str, str] = {}  # {metric_key: read}
+
+
 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 +81,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/fundamentals_api_service.py

@@ -0,0 +1,303 @@
+"""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
+from typing import Any
+
+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 date.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": {"header": "", "metrics": {}}}
+
+    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:
+        group = await _peer_group(db, two)  # {cik: representative 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()
+    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
+
+    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 not _finite(ttm_eps) or ttm_eps <= 0:
+        return None
+    return price / ttm_eps
+
+
+def _market_cap(price, shares):
+    if not _finite(price) 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 -------------------------------------------------------------------
+
+def _build_reads(metrics: list[dict], valuation: dict | None) -> dict[str, Any]:
+    by_key = {m["key"]: m for m in metrics}
+
+    def hist(key):
+        return [_Pt(p["value"]) for p in by_key.get(key, {}).get("history", [])]
+
+    growth = reads.growth_read(hist("revenue_growth_yoy"))
+    op_margin = reads.margin_read(hist("operating_margin"))
+    fcf_margin = reads.margin_read(hist("fcf_margin"))
+    share = reads.share_count_read(by_key.get("share_count_change_yoy", {}).get("value"))
+    leverage = reads.peer_read("net_debt_to_ebitda", _pct(by_key.get("net_debt_to_ebitda", {}).get("industry")))
+
+    # valuation read: P/E peer read, fall back to FCF yield
+    val_read = None
+    if valuation:
+        val_read = reads.peer_read("pe", _pct(valuation.get("pe_industry")))
+        if val_read is None:
+            val_read = reads.peer_read("fcf_yield", _pct(valuation.get("fcf_yield_industry")))
+
+    header = reads.header_sentence(growth, op_margin, val_read)
+    metric_reads = {k: v for k, v in {
+        "revenue_growth_yoy": growth,
+        "operating_margin": op_margin,
+        "fcf_margin": fcf_margin,
+        "share_count_change_yoy": share,
+        "net_debt_to_ebitda": leverage,
+        "valuation": val_read,
+    }.items() if v is not None}
+    return {"header": header, "metrics": metric_reads}
+
+
+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) -> dict[str, int]:
+    """{cik: representative (min) ticker_id} for tracked issuers in the 2-digit SIC
+    group — CIK-deduplicated (multi-class tickers collapse to one issuer)."""
+    rows = (await db.execute(
+        select(Ticker.cik, func.min(Ticker.id))
+        .where(Ticker.cik.is_not(None), func.substr(Ticker.sic, 1, 2) == two)
+        .group_by(Ticker.cik)
+    )).all()
+    return {cik: tid for cik, tid in rows}
+
+
+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

tests/unit/test_fundamentals_api.py

@@ -0,0 +1,156 @@
+"""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:
+        for fy, mult in [(2025, 1.0), (2026, 1.1)]:
+            shares = 1000 if fy == 2025 else 950  # 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 present + additive merge validates against the schema
+    assert v1["reads"]["header"]
+    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_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": "", "metrics": {}}
+
+
+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)
+    val = v1["valuation"]
+    assert val is not None  # snapshots exist -> object present
+    assert val["pe"] is None and val["market_cap_est"] is None and val["price_date"] is None