fix(fundamentals): API v1 review — multi-class pricing, reads contract, guards

1. Multi-class subject is priced by the REQUESTED ticker: the peer group's
   representative for the subject CIK is overridden to the requested ticker_id
   (other issuers pick a deterministic-by-symbol rep), so GOOGL's P/E uses
   GOOGL's price, not GOOG's. Differing-price GOOG/GOOGL test added.
2. reads matches the selected contract: header is null when there is no read;
   by_key is a fixed map over every metric key plus pe and fcf_yield, null when
   unavailable (was a sparse dict).
3. Earnings use the New York calendar date; same-day is UPCOMING (days_until 0),
   recent is strictly earlier.
4. Valuation is null when there is no usable price (> 0 required for P/E and
   market cap); when present, price_date is non-null.

Added a real router/API-envelope test with a seeded legacy record (the endpoint,
not just the schema merge). 6 tests pass.

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

app/schemas/fundamental.py

@@ -58,10 +58,13 @@ class Valuation(BaseModel):
 
 
 class FundamentalsReads(BaseModel):
-    """Deterministic text outputs, separate from the numeric metrics."""
+    """Deterministic text outputs, separate from the numeric metrics.
 
-    header: str = ""
-    metrics: dict[str, str] = {}  # {metric_key: read}
+    ``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):

app/services/fundamentals_api_service.py

@@ -11,8 +11,9 @@ from __future__ import annotations
 
 import math
 from collections import defaultdict
-from datetime import date
+from datetime import date, datetime
 from typing import Any
+from zoneinfo import ZoneInfo
 
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -33,14 +34,14 @@ METRIC_KEYS = (
 
 
 async def build_fundamentals_v1(db: AsyncSession, symbol: str, *, today: date | None = None) -> dict[str, Any]:
-    today = today or date.today()
+    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": {"header": "", "metrics": {}}}
+                "reads": _empty_reads()}
 
     subject_cik = ticker.cik
     derived = deriv.derive((await _snapshots_for(db, [subject_cik])).get(subject_cik, []))
@@ -49,7 +50,9 @@ async def build_fundamentals_v1(db: AsyncSession, symbol: str, *, today: date |
     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}
+        # 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()))
@@ -68,8 +71,9 @@ 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)
+    # 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:
@@ -129,6 +133,8 @@ def _build_valuation(derived, subject_price, peer_derived, peer_price_by_cik, tw
         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)
@@ -155,13 +161,13 @@ def _build_valuation(derived, subject_price, peer_derived, peer_price_by_cik, tw
 
 
 def _pe(price, ttm_eps):
-    if not _finite(price) or not _finite(ttm_eps) or ttm_eps <= 0:
+    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 not _finite(shares) or shares <= 0:
+    if not _finite(price) or price <= 0 or not _finite(shares) or shares <= 0:
         return None
     return price * shares
 
@@ -182,35 +188,40 @@ def _industry(key, subject, group_values, two):
 
 # -- reads -------------------------------------------------------------------
 
+_READ_KEYS = METRIC_KEYS + ("pe", "fcf_yield")
+
+
 def _build_reads(metrics: list[dict], valuation: dict | None) -> dict[str, Any]:
-    by_key = {m["key"]: m for m in metrics}
+    by_metric = {m["key"]: m for m in metrics}
 
     def hist(key):
-        return [_Pt(p["value"]) for p in by_key.get(key, {}).get("history", [])]
+        return [_Pt(p["value"]) for p in by_metric.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")))
+    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
 
-    # 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 {
+    # 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,
         "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}
+        "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:
@@ -244,15 +255,25 @@ async def _snapshots_for(db, ciks) -> dict[str, list]:
     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)."""
+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, func.min(Ticker.id))
+        select(Ticker.cik, Ticker.id, Ticker.symbol)
         .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}
+    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]]:
@@ -301,3 +322,8 @@ def _round(v, ndigits):
 
 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()