diff --git a/app/services/sec_client.py b/app/services/sec_client.py
index 0f1bb9e..e4e1824 100644
--- a/app/services/sec_client.py
+++ b/app/services/sec_client.py
@@ -270,6 +270,7 @@ def _rows_from_arrays(arrays: dict[str, list]) -> list[dict[str, Any]]:
                 "accession": arrays["accessionNumber"][i],
                 "form": form,
                 "report_date": arrays["reportDate"][i] or None,
+                "filing_date": arrays["filingDate"][i] or None,
                 "acceptance_datetime": arrays["acceptanceDateTime"][i] or None,
                 "is_xbrl": bool(arrays.get("isXBRL", [0] * len(forms))[i]),
             }
diff --git a/app/services/sec_facts_parser.py b/app/services/sec_facts_parser.py
new file mode 100644
index 0000000..2c7e4d7
--- /dev/null
+++ b/app/services/sec_facts_parser.py
@@ -0,0 +1,289 @@
+"""Pure parser: SEC companyfacts -> fundamental_snapshots rows.
+
+Turns one issuer's `companyfacts` JSON (+ its submissions filing metadata) into
+per-accession snapshot rows for the filing's **primary period**, following the
+A3 design (docs/dolt-sec-a3-design.md). No I/O, no DB — unit-testable against a
+fixture and verifiable against a real companyfacts pull.
+
+The load-bearing rules (design Decision 2 + review):
+- Period identity comes from `end == submissions.reportDate`, never `fy/fp`
+  (fy/fp is the *filing's* context; comparatives inside a filing repeat it).
+- Duration facts are stored as **cumulative YTD**: pick the fact whose span
+  matches the fiscal-period-to-date length (Q1≈3mo … FY≈12mo) within tolerance.
+  If no YTD-length fact exists, store null — never a discrete masquerading as YTD.
+- Balance-sheet instants are taken at `end == reportDate`; `shares_outstanding`
+  is the cover-page `dei` fact whose own `end` (cover date) is stored separately.
+- Cash and debt composites are aggregate-first and mutually exclusive (each
+  source tag counted at most once).
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import date, datetime
+from typing import Any, NamedTuple
+
+logger = logging.getLogger(__name__)
+
+# Expected YTD span (days) per fiscal period; a duration fact must land within
+# tolerance of this to count as the period's cumulative value.
+_EXPECTED_YTD_DAYS = {"Q1": 91, "Q2": 182, "Q3": 273, "FY": 365}
+_YTD_TOLERANCE_DAYS = 20  # covers 52/53-week fiscal calendars
+
+# us-gaap duration concepts (money), priority order; first present wins.
+_DURATION_USD = {
+    "revenue": [
+        "RevenueFromContractWithCustomerExcludingAssessedTax",
+        "Revenues",
+        "SalesRevenueNet",
+    ],
+    "net_income": ["NetIncomeLoss"],
+    "operating_income": ["OperatingIncomeLoss"],
+    "cfo": [
+        "NetCashProvidedByUsedInOperatingActivities",
+        "NetCashProvidedByUsedInOperatingActivitiesContinuingOperations",
+    ],
+    "capex": [
+        "PaymentsToAcquirePropertyPlantAndEquipment",
+        "PaymentsToAcquireProductiveAssets",
+    ],
+    "depreciation_amortization": [
+        "DepreciationDepletionAndAmortization",
+        "DepreciationAmortizationAndAccretionNet",
+        "DepreciationAndAmortization",
+    ],
+}
+_EPS_CONCEPTS = ["EarningsPerShareDiluted"]  # unit USD/shares
+# us-gaap instant (balance-sheet) concepts, at end == reportDate.
+_CASH = ["CashAndCashEquivalentsAtCarryingValue"]
+_ST_INVESTMENTS = ["ShortTermInvestments", "MarketableSecuritiesCurrent"]  # pick one
+_LONG_TERM_DEBT_AGG = ["LongTermDebt"]
+_LONG_TERM_DEBT_PARTS = ["LongTermDebtNoncurrent", "LongTermDebtCurrent"]
+_SHORT_TERM_DEBT = ["ShortTermBorrowings", "CommercialPaper"]  # pick one
+
+
+class Fact(NamedTuple):
+    taxonomy: str
+    concept: str
+    unit: str
+    start: date | None  # None => instant
+    end: date
+    val: float
+    fy: int | None
+    fp: str | None
+
+
+@dataclass
+class SnapshotRow:
+    cik: str
+    accession: str
+    form: str
+    filed_date: date
+    accepted_at: datetime
+    period_end: date
+    fiscal_year: int
+    fiscal_period: str
+    period_start: date | None = None
+    revenue: float | None = None
+    net_income: float | None = None
+    operating_income: float | None = None
+    diluted_eps: float | None = None
+    cfo: float | None = None
+    capex: float | None = None
+    depreciation_amortization: float | None = None
+    cash_and_st_investments: float | None = None
+    total_debt: float | None = None
+    shares_outstanding: float | None = None
+    shares_outstanding_date: date | None = None
+
+
+@dataclass
+class FilingMeta:
+    report_date: date
+    filing_date: date
+    accepted_at: datetime
+    form: str
+
+
+def parse_snapshots(
+    companyfacts: dict[str, Any],
+    filings: dict[str, FilingMeta],
+    accessions: set[str],
+) -> tuple[list[SnapshotRow], list[dict[str, str]]]:
+    """Build snapshot rows for ``accessions`` (those with facts + filing meta).
+
+    Returns (rows, skips) where each skip is {accession, reason} for filings
+    with no usable period identity — the caller counts these in validation_json.
+    """
+    cik = f"{int(companyfacts['cik']):010d}"
+    by_accn = _index_by_accession(companyfacts)
+    rows: list[SnapshotRow] = []
+    skips: list[dict[str, str]] = []
+    for accn in accessions:
+        meta = filings.get(accn)
+        facts = by_accn.get(accn)
+        if meta is None or not facts:
+            skips.append({"accession": accn, "reason": "no facts or filing metadata"})
+            continue
+        row = _parse_one(cik, accn, facts, meta)
+        if row is None:
+            skips.append({"accession": accn, "reason": "no usable period identity"})
+            continue
+        rows.append(row)
+    return rows, skips
+
+
+def _index_by_accession(companyfacts: dict[str, Any]) -> dict[str, list[Fact]]:
+    """One pass over companyfacts -> {accession: [Fact, ...]}."""
+    out: dict[str, list[Fact]] = {}
+    for taxonomy, concepts in companyfacts.get("facts", {}).items():
+        for concept, body in concepts.items():
+            for unit, facts in body.get("units", {}).items():
+                for f in facts:
+                    accn = f.get("accn")
+                    if not accn:
+                        continue
+                    out.setdefault(accn, []).append(
+                        Fact(
+                            taxonomy=taxonomy,
+                            concept=concept,
+                            unit=unit,
+                            start=_d(f.get("start")),
+                            end=_d(f.get("end")),
+                            val=f.get("val"),
+                            fy=f.get("fy"),
+                            fp=f.get("fp"),
+                        )
+                    )
+    return out
+
+
+def _parse_one(cik: str, accn: str, facts: list[Fact], meta: FilingMeta) -> SnapshotRow | None:
+    fy, fp = _fiscal_context(facts)
+    if fy is None or fp not in _EXPECTED_YTD_DAYS:
+        return None
+
+    row = SnapshotRow(
+        cik=cik,
+        accession=accn,
+        form=meta.form,
+        filed_date=meta.filing_date,
+        accepted_at=meta.accepted_at,
+        period_end=meta.report_date,
+        fiscal_year=fy,
+        fiscal_period=fp,
+    )
+
+    # duration YTD facts (money) + EPS
+    for field_name, concepts in _DURATION_USD.items():
+        val, start = _select_ytd(facts, concepts, meta.report_date, fp, "USD")
+        setattr(row, field_name, val)
+        if field_name == "revenue" and start is not None:
+            row.period_start = start
+    eps, eps_start = _select_ytd(facts, _EPS_CONCEPTS, meta.report_date, fp, "USD/shares")
+    row.diluted_eps = eps
+    if row.period_start is None and eps_start is not None:
+        row.period_start = eps_start
+
+    # balance-sheet instants at reportDate
+    row.cash_and_st_investments = _compose_cash(facts, meta.report_date)
+    row.total_debt = _compose_debt(facts, meta.report_date)
+    row.shares_outstanding, row.shares_outstanding_date = _select_shares(facts)
+    return row
+
+
+def _fiscal_context(facts: list[Fact]) -> tuple[int | None, str | None]:
+    """A filing's own (fy, fp) — shared by all its facts; take the first set."""
+    for f in facts:
+        if f.fy is not None and f.fp:
+            return f.fy, f.fp
+    return None, None
+
+
+def _select_ytd(
+    facts: list[Fact], concepts: list[str], report_date: date, fp: str, unit: str
+) -> tuple[float | None, date | None]:
+    """First present concept whose duration fact ends at reportDate and whose span
+    matches the fiscal-period-to-date length. Returns (val, period_start)."""
+    expected = _EXPECTED_YTD_DAYS[fp]
+    for concept in concepts:
+        best: Fact | None = None
+        best_diff: int | None = None
+        for f in facts:
+            if (
+                f.concept != concept
+                or f.unit != unit
+                or f.start is None
+                or f.end != report_date
+                or f.val is None
+            ):
+                continue
+            diff = abs((f.end - f.start).days - expected)
+            if diff <= _YTD_TOLERANCE_DAYS and (best_diff is None or diff < best_diff):
+                best, best_diff = f, diff
+        if best is not None:
+            return float(best.val), best.start
+    return None, None
+
+
+def _select_instant(facts: list[Fact], concepts: list[str], report_date: date) -> float | None:
+    """First present instant (balance-sheet) fact at end == reportDate, unit USD."""
+    for concept in concepts:
+        for f in facts:
+            if (
+                f.concept == concept
+                and f.unit == "USD"
+                and f.start is None
+                and f.end == report_date
+                and f.val is not None
+            ):
+                return float(f.val)
+    return None
+
+
+def _compose_cash(facts: list[Fact], report_date: date) -> float | None:
+    cash = _select_instant(facts, _CASH, report_date)
+    st = _select_instant(facts, _ST_INVESTMENTS, report_date)  # first present of the two
+    if cash is None and st is None:
+        return None
+    return (cash or 0.0) + (st or 0.0)
+
+
+def _compose_debt(facts: list[Fact], report_date: date) -> float | None:
+    long_term = _select_instant(facts, _LONG_TERM_DEBT_AGG, report_date)
+    if long_term is None:
+        nc = _select_instant(facts, ["LongTermDebtNoncurrent"], report_date)
+        cur = _select_instant(facts, ["LongTermDebtCurrent"], report_date)
+        long_term = None if nc is None and cur is None else (nc or 0.0) + (cur or 0.0)
+    short_term = _select_instant(facts, _SHORT_TERM_DEBT, report_date)
+    if long_term is None and short_term is None:
+        return None
+    return (long_term or 0.0) + (short_term or 0.0)
+
+
+def _select_shares(facts: list[Fact]) -> tuple[float | None, date | None]:
+    """dei:EntityCommonStockSharesOutstanding — cover-page instant. Store its own
+    end (the cover date, which differs from period_end)."""
+    candidates = [
+        f
+        for f in facts
+        if f.taxonomy == "dei"
+        and f.concept == "EntityCommonStockSharesOutstanding"
+        and f.unit == "shares"
+        and f.start is None
+        and f.val is not None
+    ]
+    if not candidates:
+        return None, None
+    best = max(candidates, key=lambda f: f.end)
+    return float(best.val), best.end
+
+
+def _d(value: Any) -> date | None:
+    if not value:
+        return None
+    try:
+        return date.fromisoformat(str(value)[:10])
+    except ValueError:
+        return None
diff --git a/tests/unit/test_sec_client.py b/tests/unit/test_sec_client.py
index dc5d78b..f1317c7 100644
--- a/tests/unit/test_sec_client.py
+++ b/tests/unit/test_sec_client.py
@@ -31,6 +31,7 @@ SUBMISSIONS_BASE = {
             "accessionNumber": ["0000320193-26-000013", "0000320193-26-000006", "0000320193-26-000099"],
             "form": ["10-Q", "10-K", "8-K"],
             "reportDate": ["2026-03-28", "2025-09-27", "2026-04-01"],
+            "filingDate": ["2026-05-01", "2026-01-30", "2026-04-02"],
             "acceptanceDateTime": ["2026-05-01T10:01:00.000Z", "2025-10-31T10:01:26.000Z", "2026-04-02T09:00:00.000Z"],
             "isXBRL": [1, 1, 0],
         },
@@ -42,6 +43,7 @@ SUBMISSIONS_SHARD = {
     "accessionNumber": ["0000320193-94-000002"],
     "form": ["10-Q"],
     "reportDate": ["1993-12-31"],
+    "filingDate": ["1994-01-26"],
     "acceptanceDateTime": ["1994-01-26T05:00:00.000Z"],
     "isXBRL": [0],
 }
diff --git a/tests/unit/test_sec_facts_parser.py b/tests/unit/test_sec_facts_parser.py
new file mode 100644
index 0000000..0f326e9
--- /dev/null
+++ b/tests/unit/test_sec_facts_parser.py
@@ -0,0 +1,179 @@
+"""Tests for the companyfacts -> snapshot parser, on a realistic Apple-shaped
+fixture (the structure verified by live probe)."""
+
+from __future__ import annotations
+
+import os
+from datetime import date, datetime, timezone
+
+import pytest
+
+from app.services.sec_facts_parser import (
+    Fact,
+    FilingMeta,
+    _compose_cash,
+    _compose_debt,
+    parse_snapshots,
+)
+
+UTC = timezone.utc
+
+
+def _dur(start, end, val, accn, fy=2026, fp="Q2"):
+    return {"start": start, "end": end, "val": val, "fy": fy, "fp": fp, "accn": accn, "form": "10-Q"}
+
+
+def _inst(end, val, accn, fy=2026, fp="Q2"):
+    return {"end": end, "val": val, "fy": fy, "fp": fp, "accn": accn, "form": "10-Q"}
+
+
+COMPANYFACTS = {
+    "cik": 320193,
+    "facts": {
+        "us-gaap": {
+            "RevenueFromContractWithCustomerExcludingAssessedTax": {"units": {"USD": [
+                _dur("2025-09-28", "2025-12-27", 143756, "A", fp="Q1"),   # Q1 discrete == YTD
+                _dur("2025-09-28", "2026-03-28", 254940, "B"),            # Q2 YTD (181d)  <- want this
+                _dur("2025-12-28", "2026-03-28", 111184, "B"),            # Q2 discrete (90d)
+            ]}},
+            "NetIncomeLoss": {"units": {"USD": [_dur("2025-09-28", "2026-03-28", 40000, "B")]}},
+            # only a discrete-length fact for Q2 -> must be null, not the discrete
+            "OperatingIncomeLoss": {"units": {"USD": [_dur("2025-12-28", "2026-03-28", 30000, "B")]}},
+            "EarningsPerShareDiluted": {"units": {"USD/shares": [_dur("2025-09-28", "2026-03-28", 2.55, "B")]}},
+            "CashAndCashEquivalentsAtCarryingValue": {"units": {"USD": [_inst("2026-03-28", 30000, "B")]}},
+            "MarketableSecuritiesCurrent": {"units": {"USD": [_inst("2026-03-28", 20000, "B")]}},
+            "LongTermDebtNoncurrent": {"units": {"USD": [_inst("2026-03-28", 80000, "B")]}},
+            "LongTermDebtCurrent": {"units": {"USD": [_inst("2026-03-28", 10000, "B")]}},
+        },
+        "dei": {
+            "EntityCommonStockSharesOutstanding": {"units": {"shares": [
+                {"end": "2026-04-17", "val": 14687356000, "fy": 2026, "fp": "Q2", "accn": "B", "form": "10-Q"},
+            ]}},
+        },
+    },
+}
+
+FILINGS = {
+    "A": FilingMeta(date(2025, 12, 27), date(2026, 1, 30), datetime(2026, 1, 30, 11, 1, tzinfo=UTC), "10-Q"),
+    "B": FilingMeta(date(2026, 3, 28), date(2026, 5, 1), datetime(2026, 5, 1, 10, 1, tzinfo=UTC), "10-Q"),
+}
+
+
+def _by_accn(rows):
+    return {r.accession: r for r in rows}
+
+
+def test_parses_ytd_not_discrete_and_cover_date_shares():
+    rows, skips = parse_snapshots(COMPANYFACTS, FILINGS, {"A", "B"})
+    assert not skips
+    b = _by_accn(rows)["B"]
+
+    assert (b.cik, b.fiscal_year, b.fiscal_period) == ("0000320193", 2026, "Q2")
+    assert b.period_end == date(2026, 3, 28)
+    assert b.period_start == date(2025, 9, 28)  # YTD start (fiscal-year start)
+    assert b.revenue == 254940          # the 6-month YTD, NOT the 111184 discrete
+    assert b.net_income == 40000
+    assert b.operating_income is None   # only a discrete-length fact existed -> null
+    assert b.diluted_eps == 2.55
+    # cash + first-present ST investment (MarketableSecuritiesCurrent), each once
+    assert b.cash_and_st_investments == 50000
+    # long-term parts summed (no aggregate, no short-term)
+    assert b.total_debt == 90000
+    assert b.shares_outstanding == 14687356000
+    assert b.shares_outstanding_date == date(2026, 4, 17)  # cover date != period_end
+    assert b.filed_date == date(2026, 5, 1)
+    assert b.accepted_at == datetime(2026, 5, 1, 10, 1, tzinfo=UTC)
+
+
+def test_q1_discrete_is_the_ytd():
+    rows, _ = parse_snapshots(COMPANYFACTS, FILINGS, {"A"})
+    a = _by_accn(rows)["A"]
+    assert a.fiscal_period == "Q1"
+    assert a.revenue == 143756  # Q1 YTD == Q1 discrete
+    assert a.period_start == date(2025, 9, 28)
+
+
+def test_skips_filing_without_usable_period():
+    cf = {
+        "cik": 320193,
+        "facts": {"us-gaap": {"NetIncomeLoss": {"units": {"USD": [
+            {"start": "2025-09-28", "end": "2026-03-28", "val": 1, "fy": 2026, "fp": "H1", "accn": "X", "form": "10-Q"},
+        ]}}}},
+    }
+    filings = {"X": FilingMeta(date(2026, 3, 28), date(2026, 5, 1), datetime(2026, 5, 1, tzinfo=UTC), "10-Q")}
+    rows, skips = parse_snapshots(cf, filings, {"X"})
+    assert rows == []
+    assert skips == [{"accession": "X", "reason": "no usable period identity"}]
+
+
+def test_missing_accession_is_skipped():
+    rows, skips = parse_snapshots(COMPANYFACTS, FILINGS, {"NOPE"})
+    assert rows == []
+    assert skips == [{"accession": "NOPE", "reason": "no facts or filing metadata"}]
+
+
+def test_debt_prefers_aggregate_over_parts():
+    rd = date(2026, 3, 28)
+    facts = [
+        Fact("us-gaap", "LongTermDebt", "USD", None, rd, 95000, 2026, "Q2"),
+        Fact("us-gaap", "LongTermDebtNoncurrent", "USD", None, rd, 80000, 2026, "Q2"),
+        Fact("us-gaap", "LongTermDebtCurrent", "USD", None, rd, 10000, 2026, "Q2"),
+        Fact("us-gaap", "CommercialPaper", "USD", None, rd, 5000, 2026, "Q2"),
+    ]
+    # aggregate (95000) used, parts ignored; + one short-term pick (5000)
+    assert _compose_debt(facts, rd) == 100000
+
+
+def test_cash_picks_one_st_investment_source():
+    rd = date(2026, 3, 28)
+    facts = [
+        Fact("us-gaap", "CashAndCashEquivalentsAtCarryingValue", "USD", None, rd, 30000, 2026, "Q2"),
+        Fact("us-gaap", "ShortTermInvestments", "USD", None, rd, 15000, 2026, "Q2"),
+        Fact("us-gaap", "MarketableSecuritiesCurrent", "USD", None, rd, 20000, 2026, "Q2"),
+    ]
+    # ShortTermInvestments is first in priority -> 30000 + 15000 (not both ST tags)
+    assert _compose_cash(facts, rd) == 45000
+
+
+# Opt-in live check against real Apple companyfacts. Skips unless SEC_LIVE=1 and a
+# real SEC_USER_AGENT are set (network + fair-access contact email).
+@pytest.mark.skipif(
+    not (os.environ.get("SEC_LIVE") and os.environ.get("SEC_USER_AGENT")),
+    reason="set SEC_LIVE=1 + SEC_USER_AGENT to run the live SEC parser check",
+)
+async def test_live_apple_parse_invariants():
+    from app.services.sec_client import SecClient
+
+    def _dt(s):
+        return datetime.fromisoformat(s.replace("Z", "+00:00")) if s else None
+
+    async with SecClient(user_agent=os.environ["SEC_USER_AGENT"]) as c:
+        cf = await c.companyfacts(320193)
+        sub = await c.submissions(320193, include_history=False)
+
+    filings = {
+        f["accession"]: FilingMeta(
+            date.fromisoformat(f["report_date"]),
+            date.fromisoformat(f["filing_date"]),
+            _dt(f["acceptance_datetime"]),
+            f["form"],
+        )
+        for f in sub["filings"]
+        if f["report_date"] and f["filing_date"] and f["acceptance_datetime"]
+    }
+    rows, _ = parse_snapshots(cf, filings, set(filings))
+    assert len(rows) > 20
+    assert all(r.period_end and r.fiscal_year and r.fiscal_period for r in rows)
+    # YTD revenue is non-decreasing within a fiscal year
+    by_fy: dict[int, list] = {}
+    for r in rows:
+        if r.revenue is not None:
+            by_fy.setdefault(r.fiscal_year, []).append((r.fiscal_period, r.revenue))
+    order = {"Q1": 1, "Q2": 2, "Q3": 3, "FY": 4}
+    for fy, series in by_fy.items():
+        series.sort(key=lambda x: order[x[0]])
+        vals = [v for _, v in series]
+        assert vals == sorted(vals), f"YTD revenue not monotonic in FY{fy}: {series}"
+    # shares cover-date differs from period_end
+    latest = max(rows, key=lambda r: r.period_end)
+    assert latest.shares_outstanding_date != latest.period_end