JLaborda · JLaborda · Jul 1, 2026 · Jul 1, 2026 · Jul 1, 2026 · Jul 1, 2026
diff --git a/config/fundamentals/simfin_mapping_v1.yaml b/config/fundamentals/simfin_mapping_v1.yaml
@@ -16,6 +16,12 @@ fields:
   total_assets: "Total Assets"
   total_liabilities: "Total Liabilities"
   shares_outstanding: "Shares (Basic)"
+  # QV forensic / FS-Score inputs (Beneish YoY deltas, Gray/Carlisle components)
+  accounts_receivable: "Accounts Receivable"
+  cost_of_revenue: "Cost of Revenue"
+  depreciation_amortization: "Depreciation & Amortization"
+  sga_expense: "Selling, General & Administrative"
+  operating_cash_flow: "Net Cash from Operating Activities"
 
 meta:
   ticker: Ticker
@@ -28,5 +34,12 @@ meta:
   fiscal_year: "Fiscal Year"
   fiscal_period: "Fiscal Period"
 
+qv_required_fields:
+  - accounts_receivable
+  - cost_of_revenue
+  - depreciation_amortization
+  - sga_expense
+  - operating_cash_flow
+
 # ponytail: single lag when Publish Date is missing (quarterly default from architecture).
 missing_publish_lag_days: 45
diff --git a/spec/features/006-etl-data-lake/spec.md b/spec/features/006-etl-data-lake/spec.md
@@ -4,6 +4,8 @@
 
 **done** (demo slice) — SimFin connector + normalizer + shareprices snapshot + end-to-end orchestrator `run-demo-pipeline` ([#63](https://github.com/JLaborda/SmartWealthAI/issues/63)). SEC spike frozen ([ADR-0001](../../adr/0001-simfin-fundamentals-mvp.md)).
 
+**done** (phase 2 multi-period fundamentals) — annual/quarterly SimFin ingest, QV field mapping, PIT history interface ([#87](https://github.com/JLaborda/SmartWealthAI/issues/87)).
+
 ## Objective
 
 Build the module that downloads, validates, normalizes, and stores financial data so that every downstream module (universe construction, scoring, backtesting, sell-watch, portfolio evolution) can rely on a single trustworthy source. The data lake lives on AWS S3 and is queried with DuckDB. Point-in-time correctness and incremental refresh are mandatory.
@@ -24,6 +26,7 @@ Build the module that downloads, validates, normalizes, and stores financial dat
 
 ### Full MVP (phase 2 additions)
 
+- **Multi-period fundamentals** (annual/quarterly income, balance, cash flow) with PIT history for QV ([#87](https://github.com/JLaborda/SmartWealthAI/issues/87)).
 - SEC EDGAR ETL (frozen spike: `sec_client`, `download-fundamentals`).
 - Incremental per-CIK filing ingest when SEC normalizer ships.
 - S&P 500–scoped backfill policies.
@@ -40,9 +43,12 @@ Build the module that downloads, validates, normalizes, and stores financial dat
 
 | Input | Source | Notes |
 | --- | --- | --- |
-| Income statement (TTM) | SimFin bulk `income` variant `ttm` | EBIT, interest, revenue, net income. |
+| Income statement (TTM) | SimFin bulk `income` variant `ttm` | EBIT, interest, revenue, net income (Magic Formula demo). |
+| Income statement (annual/quarterly) | SimFin bulk `income` variants `annual`, `quarterly` | Phase 2 QV YoY deltas. |
 | Balance sheet (quarterly) | SimFin bulk `balance` variant `quarterly` | NWC, PP&E, debt, cash, shares. |
-| Cash flow (TTM) | SimFin bulk `cashflow` variant `ttm` | Phase 2 permanent-loss filter; ingest in demo for raw archive. |
+| Balance sheet (annual) | SimFin bulk `balance` variant `annual` | Phase 2 annual statement joins. |
+| Cash flow (TTM) | SimFin bulk `cashflow` variant `ttm` | Demo raw archive. |
+| Cash flow (annual/quarterly) | SimFin bulk `cashflow` variants `annual`, `quarterly` | CFO and D&A for Beneish / FS-Score. |
 | Publish / report / restated dates | SimFin statement rows | `Publish Date` → `as_of_date`. |
 | Company metadata | SimFin `companies` | `Ticker`, `CIK`, `IndustryId`, `SimFinId`. |
 | Industry labels | SimFin `industries` | Sector/industry names for exclusions CSV. |
@@ -184,6 +190,11 @@ Phase 2 yfinance cache semantics:
   (`poetry run download-prices`). Requires `shareprices/latest` from `download-simfin`.
   Writes `curated/prices/run_date=<date>/prices.parquet`. Hermetic tests in
   `tests/test_download_prices.py`.
+- Phase 2 multi-period fundamentals ([#87](https://github.com/JLaborda/SmartWealthAI/issues/87)):
+  `download-simfin` ingests annual/quarterly statement variants; `simfin_normalizer`
+  writes `statement_variant` rows and QV columns; `load_pit_fundamentals_history` in
+  `pit_fundamentals.py` returns one PIT row per fiscal period for annual/quarterly history.
+  Hermetic tests in `tests/test_simfin_normalizer.py` and `tests/test_pit_fundamentals.py`.
 
 **Operator sequence (demo pipeline):**
 
@@ -210,7 +221,8 @@ poetry run score-universe --run-date 2026-06-19
 | **`as_of_date` (demo)** | SimFin `Publish Date`; `Restated Date` → new `version_id`. |
 | **SEC normalizer (phase 2)** | `companyfacts` JSON from `raw/sec_edgar/...`; EDGAR acceptance as `as_of_date`. |
 | **Mapping** | `config/fundamentals/simfin_mapping_v1.yaml` (demo); `mapping_v1.yaml` (SEC phase 2). |
-| Canonical fields | Same ~12 curated columns for ROC, EY, and QC regardless of provider. |
+| Canonical fields | Core ~14 columns for ROC/EY/QC plus QV columns (`accounts_receivable`, `cost_of_revenue`, `depreciation_amortization`, `sga_expense`, `operating_cash_flow`) in `simfin_mapping_v1.yaml`. |
+| QV review queue | Annual/quarterly rows missing `qv_required_fields` → `curated/issues/` with `missing_qv_inputs`. |
 | Provenance | Per-field source column + `mapping_version`. |
 | Downstream contract | Scoring reads `curated/fundamentals` only — provider-agnostic schema. |
 | SEC spike | Frozen in repo; not deleted. |
@@ -220,6 +232,9 @@ poetry run score-universe --run-date 2026-06-19
 Downloads US fundamentals via the `simfin` Python package into `raw/simfin/`.
 Operator guide: [`spec/guides/download-simfin.md`](../../guides/download-simfin.md).
 
+Phase 2 adds five non-critical statement datasets (`income/annual`, `income/quarterly`,
+`balance/annual`, `cashflow/annual`, `cashflow/quarterly`) to the same CLI run.
+
 ### CLI
 
 ```bash
@@ -240,6 +255,7 @@ poetry run download-simfin --refresh-days 7 --force
 
 - [x] `simfin` dependency in `pyproject.toml`; API key from `SIMFIN_API_KEY`.
 - [x] CLI downloads all six demo datasets into stable `raw/simfin/` partitions.
+- [x] Phase 2: CLI also downloads annual/quarterly income, balance, and cashflow variants (non-critical).
 - [x] Re-run without `--force` skips datasets fresher than `refresh_days`; `--force` overwrites.
 - [x] Per-dataset failures recorded in run summary; batch continues when possible.
 - [x] Hermetic tests cover path building, skip/force logic, and mocked download.
@@ -288,6 +304,49 @@ poetry run normalize-simfin --snapshot-date 2026-06-18 --ticker AAPL --ticker MS
 poetry run normalize-simfin --snapshot-date 2026-06-18 --universe-run-date 2026-06-18 --quiet
 ```
 
+## Phase 2: Multi-period fundamentals and PIT history
+
+**Delivery:** phase 2 ([`roadmap.md`](../../constitution/roadmap.md) — issue [#87](https://github.com/JLaborda/SmartWealthAI/issues/87))
+
+### Objective
+
+Extend SimFin ingest and normalization so Quantitative Value downstream modules (Beneish, FS-Score) can load multi-period, point-in-time-correct fundamentals at any `decision_date`.
+
+### In scope
+
+- Download annual/quarterly income, balance, and cash-flow SimFin bulk variants into `raw/simfin/`.
+- Normalize to the same provider-agnostic `curated/fundamentals` schema with `statement_variant` metadata.
+- Map QV forensic / FS-Score input columns via `qv_required_fields` in `simfin_mapping_v1.yaml`.
+- `load_pit_fundamentals_history()` — one PIT row per fiscal period (annual/quarterly only).
+- Hermetic fixture tests proving no look-ahead (`as_of_date > decision_date` excluded).
+
+### Out of scope
+
+- Beneish M-Score or FS-Score calculators ([#89](https://github.com/JLaborda/SmartWealthAI/issues/89), [#91](https://github.com/JLaborda/SmartWealthAI/issues/91)).
+- Daily share prices or backtest engine ([#88](https://github.com/JLaborda/SmartWealthAI/issues/88)).
+- Changing Magic Formula demo scoring to use annual/quarterly rows.
+
+### Expected flow
+
+1. `download-simfin` fetches demo datasets plus phase 2 annual/quarterly statement variants.
+2. `normalize-simfin` processes three statement bundles: TTM (demo), annual, quarterly.
+3. Annual/quarterly rows join income + balance + cashflow on exact `Report Date`; missing QV inputs → review queue.
+4. Curated partitions store multiple `statement_variant` rows per `period=<YYYYQn>` parquet.
+5. `load_pit_fundamentals_history(ticker, decision_date)` returns annual/quarterly history with PIT selection per fiscal period.
+
+### Acceptance criteria (phase 2 multi-period)
+
+- [x] SimFin annual/quarterly variants downloaded to raw and normalized to provider-agnostic curated schema.
+- [x] PIT query returns multi-period history per ticker at a pinned decision date.
+- [x] Fixture test: row with `as_of_date > decision_date` never returned.
+- [x] Review queue rows written for missing Beneish/FS-Score inputs (`missing_qv_inputs`).
+- [x] `spec/features/006-etl-data-lake/spec.md` implementation status updated.
+
+### Risks
+
+- SimFin free-tier bulk size grows with extra variants; monitor download time on weekly refresh.
+- Same fiscal `period` partition holds multiple `statement_variant` rows; MF loaders filter to `ttm`.
+
 ## SEC fundamentals normalizer (phase 2)
 
 **GitHub issue:** [#52](https://github.com/JLaborda/SmartWealthAI/issues/52) — blocks [#44](https://github.com/JLaborda/SmartWealthAI/issues/44) (ROC/EY) and feeds [#50](https://github.com/JLaborda/SmartWealthAI/issues/50) (PIT selection).

diff --git a/spec/guides/download-simfin.md b/spec/guides/download-simfin.md
@@ -1,6 +1,6 @@
 # Download SimFin bulk fundamentals and prices (demo)
 
-Operator guide for the **SimFin bulk connector** on the June 30 demo path. Canonical spec: [`etl-data-lake.md`](../../006-etl-data-lake/spec.md).
+Operator guide for the **SimFin bulk connector** on the June 30 demo path. Canonical spec: [`spec/features/006-etl-data-lake/spec.md`](../../features/006-etl-data-lake/spec.md).
 
 ## Prerequisites
 
@@ -13,16 +13,21 @@ export SIMFIN_API_KEY="<from user secrets>"
 
 ## Download US bulk datasets
 
-Downloads six demo datasets into the raw lake under `data/raw/simfin/`:
-
-| Dataset | Variant | Lake partition |
-| --- | --- | --- |
-| `companies` | `default` | `dataset=companies/variant=default/market=us/` |
-| `industries` | `default` | `dataset=industries/variant=default/market=us/` |
-| `income` | `ttm` | `dataset=income/variant=ttm/market=us/` |
-| `balance` | `quarterly` | `dataset=balance/variant=quarterly/market=us/` |
-| `cashflow` | `ttm` | `dataset=cashflow/variant=ttm/market=us/` |
-| `shareprices` | `latest` | `dataset=shareprices/variant=latest/market=us/` |
+Downloads demo datasets plus phase 2 multi-period statement variants into the raw lake under `data/raw/simfin/`:
+
+| Dataset | Variant | Lake partition | Critical |
+| --- | --- | --- | --- |
+| `companies` | `default` | `dataset=companies/variant=default/market=us/` | yes |
+| `industries` | `default` | `dataset=industries/variant=default/market=us/` | yes |
+| `income` | `ttm` | `dataset=income/variant=ttm/market=us/` | yes |
+| `balance` | `quarterly` | `dataset=balance/variant=quarterly/market=us/` | yes |
+| `cashflow` | `ttm` | `dataset=cashflow/variant=ttm/market=us/` | no |
+| `shareprices` | `latest` | `dataset=shareprices/variant=latest/market=us/` | yes |
+| `income` | `annual` | `dataset=income/variant=annual/market=us/` | no |
+| `income` | `quarterly` | `dataset=income/variant=quarterly/market=us/` | no |
+| `balance` | `annual` | `dataset=balance/variant=annual/market=us/` | no |
+| `cashflow` | `annual` | `dataset=cashflow/variant=annual/market=us/` | no |
+| `cashflow` | `quarterly` | `dataset=cashflow/variant=quarterly/market=us/` | no |
 
 Each partition also includes `as_of_date=<YYYY-MM-DD>/` and the verbatim SimFin CSV filename (e.g. `us-income-ttm.csv`).
 
@@ -54,6 +59,8 @@ Pass `--ticker` to limit the normalize step to specific names (intersect univers
 
 `normalize-simfin` requires `--universe-run-date` (after `build-universe`) or `--ticker` for smoke tests. It does not process the full SimFin US table by default.
 
+Phase 2 annual/quarterly rows require the matching cashflow variant on disk; missing QV inputs route to `curated/issues/` with reason `missing_qv_inputs`.
+
 ## Refresh and cache behaviour
 
 - **Skip:** Re-run without `--force` when the on-disk lake copy is younger than `--refresh-days` (default `7`).
@@ -63,8 +70,8 @@ Pass `--ticker` to limit the normalize step to specific names (intersect univers
 ## Failures
 
 - A failure for one dataset does not stop the rest.
-- Non-critical dataset (`cashflow`) failure still exits `0` when critical datasets succeed.
-- Exit code `1` when all critical datasets (`companies`, `industries`, `income`, `balance`, `shareprices`) fail, or when `SIMFIN_API_KEY` is missing.
+- Non-critical datasets (`cashflow`, phase 2 statement variants) failure still exits `0` when critical datasets succeed.
+- Exit code `1` when all critical datasets (`companies`, `industries`, `income/ttm`, `balance/quarterly`, `shareprices`) fail, or when `SIMFIN_API_KEY` is missing.
 - Per-run errors are written to `raw/simfin/download_runs/as_of_date=<date>/errors.json` when any dataset fails.
 
 ## Module map

diff --git a/src/smartwealthai/download_simfin.py b/src/smartwealthai/download_simfin.py
@@ -3,7 +3,7 @@
 Orchestrates the demo SimFin connector documented in
 ``spec/features/006-etl-data-lake/spec.md``. Downloads ``companies``, ``industries``,
 ``income`` (TTM), ``balance`` (quarterly), ``cashflow`` (TTM), and ``shareprices``
-(latest) for ``market=us``.
+(latest) for ``market=us``. Phase 2 also downloads annual/quarterly statement variants.
 
 Usage::
 
@@ -81,6 +81,23 @@ class DatasetResult:
     SimFinDatasetSpec("shareprices", variant="latest", market="us"),
 )
 
+# Phase 2 multi-period fundamentals (issue #87); non-critical like demo cashflow.
+PHASE2_STATEMENT_DATASETS: tuple[SimFinDatasetSpec, ...] = (
+    SimFinDatasetSpec("income", variant="annual", market="us", critical=False),
+    SimFinDatasetSpec("income", variant="quarterly", market="us", critical=False),
+    SimFinDatasetSpec("balance", variant="annual", market="us", critical=False),
+    SimFinDatasetSpec("cashflow", variant="annual", market="us", critical=False),
+    SimFinDatasetSpec("cashflow", variant="quarterly", market="us", critical=False),
+)
+
+SIMFIN_DATASETS: tuple[SimFinDatasetSpec, ...] = DEMO_DATASETS + PHASE2_STATEMENT_DATASETS
+
+
+def dataset_spec_key(spec: SimFinDatasetSpec) -> str:
+    """Stable label for logs and run summaries (dataset + variant)."""
+    variant = spec.variant or "default"
+    return f"{spec.name}/{variant}"
+
 
 def download_dataset(
     spec: SimFinDatasetSpec,
@@ -100,8 +117,9 @@ def download_dataset(
         market=spec.market,
         as_of_date=as_of_date,
     )
+    label = dataset_spec_key(spec)
     if should_skip_dataset(lake_path, refresh_days=refresh_days, force=force, now=now):
-        return DatasetResult(name=spec.name, skipped=True)
+        return DatasetResult(name=label, skipped=True)
 
     try:
         source = fetch_csv(
@@ -113,8 +131,8 @@ def download_dataset(
         lake_path.parent.mkdir(parents=True, exist_ok=True)
         shutil.copy2(source, lake_path)
     except OSError as exc:
-        return DatasetResult(name=spec.name, failed=True, error=str(exc))
-    return DatasetResult(name=spec.name, downloaded=True)
+        return DatasetResult(name=label, failed=True, error=str(exc))
+    return DatasetResult(name=label, downloaded=True)
 
 
 def run_download(
@@ -140,8 +158,8 @@ def run_download(
         fetch_csv = fetch_dataset_csv
 
     results: list[DatasetResult] = []
-    for spec in DEMO_DATASETS:
-        logger.info("Processing SimFin dataset %s", spec.name)
+    for spec in SIMFIN_DATASETS:
+        logger.info("Processing SimFin dataset %s", dataset_spec_key(spec))
         result = download_dataset(
             spec,
             data_dir=data_dir,
@@ -161,8 +179,8 @@ def run_download(
     downloaded_total = sum(result.downloaded for result in results)
     skipped_total = sum(result.skipped for result in results)
     failures = [result for result in results if result.failed]
-    critical_specs = {spec.name: spec for spec in DEMO_DATASETS}
-    critical_failures = [result for result in failures if critical_specs[result.name].critical]
+    critical_by_key = {dataset_spec_key(spec): spec for spec in SIMFIN_DATASETS}
+    critical_failures = [result for result in failures if critical_by_key[result.name].critical]
 
     click.echo(f"Downloaded datasets: {downloaded_total}")
     click.echo(f"Skipped datasets: {skipped_total}")
@@ -175,7 +193,7 @@ def run_download(
         error_file.write_text(json.dumps(payload, indent=2))
         click.echo(f"Wrote error summary: {error_file}")
 
-    if len(critical_failures) == len([spec for spec in DEMO_DATASETS if spec.critical]):
+    if len(critical_failures) == len([spec for spec in SIMFIN_DATASETS if spec.critical]):
         return 1
     return 0