From 4b2b863075fd66bff81a58a608f0bbcb8af0492f Mon Sep 17 00:00:00 2001 From: JLaborda <15078416+JLaborda@users.noreply.github.com> Date: Wed, 1 Jul 2026 10:42:48 +0000 Subject: [PATCH 1/4] docs(mvp): document phase 2 multi-period fundamentals etl Record issue #87 delivery: annual/quarterly SimFin variants, QV field mapping, PIT history interface, and updated operator download table. Co-authored-by: Cursor --- spec/features/006-etl-data-lake/spec.md | 65 +++++++++++++++++++++++-- spec/guides/download-simfin.md | 33 ++++++++----- 2 files changed, 82 insertions(+), 16 deletions(-) diff --git a/spec/features/006-etl-data-lake/spec.md b/spec/features/006-etl-data-lake/spec.md index ba93021..55e8f6a 100644 --- 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=/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=` 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 index d3c5362..999505f 100644 --- 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="" ## 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=/` 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=/errors.json` when any dataset fails. ## Module map From 28047a9bcf82a0e725edc6ba2d584bff36dd6e48 Mon Sep 17 00:00:00 2001 From: JLaborda <15078416+JLaborda@users.noreply.github.com> Date: Wed, 1 Jul 2026 10:42:56 +0000 Subject: [PATCH 2/4] feat(etl): add multi-period simfin ingest normalize and pit history Download annual/quarterly statement variants, normalize with QV field mapping and review-queue routing, and expose load_pit_fundamentals_history for per-period PIT queries without look-ahead. Closes #87 Co-authored-by: Cursor --- config/fundamentals/simfin_mapping_v1.yaml | 13 + src/smartwealthai/download_simfin.py | 36 ++- src/smartwealthai/pit_fundamentals.py | 87 ++++++- src/smartwealthai/simfin_normalizer.py | 274 +++++++++++++++++---- 4 files changed, 350 insertions(+), 60 deletions(-) diff --git a/config/fundamentals/simfin_mapping_v1.yaml b/config/fundamentals/simfin_mapping_v1.yaml index d409bad..cdf5ccd 100644 --- 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/src/smartwealthai/download_simfin.py b/src/smartwealthai/download_simfin.py index c4541b6..29d190a 100644 --- 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 diff --git a/src/smartwealthai/pit_fundamentals.py b/src/smartwealthai/pit_fundamentals.py index 4f26bb1..fe07c78 100644 --- a/src/smartwealthai/pit_fundamentals.py +++ b/src/smartwealthai/pit_fundamentals.py @@ -60,6 +60,28 @@ def _select_pit_fundamentals(frame: pd.DataFrame, as_of_date: date) -> pd.DataFr return eligible.sort_values(["cik", "_as_of", "version_id"]).groupby("cik", sort=False).tail(1) +def _select_pit_fundamentals_by_period(frame: pd.DataFrame, as_of_date: date) -> pd.DataFrame: + """Return the latest PIT row per (CIK, fiscal period) on or before ``as_of_date``.""" + if frame.empty: + return frame + + work = frame.copy() + work["_as_of"] = pd.to_datetime(work["as_of_date"]) + decision = pd.Timestamp(as_of_date) + eligible = work.loc[work["_as_of"] <= decision] + if eligible.empty: + return eligible.iloc[0:0] + + group_cols = ( + ["cik", "fiscal_period_end"] if "fiscal_period_end" in eligible.columns else ["cik"] + ) + return ( + eligible.sort_values([*group_cols, "_as_of", "version_id"]) + .groupby(group_cols, sort=False) + .tail(1) + ) + + def load_pit_fundamentals_row( data_dir: Path, *, @@ -75,8 +97,8 @@ def load_pit_fundamentals_row( rows: list[pd.Series] = [] for path in cik_dir.glob("period=*/fundamentals.parquet"): - frame = pd.read_parquet(path) - if frame.empty: + frame = _read_fundamentals_partition(path) + if frame is None: continue rows.append(frame.iloc[0]) @@ -92,6 +114,41 @@ def load_pit_fundamentals_row( return selected.iloc[0] +def load_pit_fundamentals_history( + data_dir: Path, + *, + ticker: str, + as_of_date: date, +) -> pd.DataFrame: + """Return PIT fundamentals history for ``ticker`` (one row per fiscal period).""" + cik = resolve_ticker_cik(data_dir, ticker=ticker, run_date=as_of_date) + paths = _fundamentals_paths_for_ciks(data_dir, {cik}) + if not paths: + msg = f"No curated fundamentals for ticker {ticker!r} (cik={cik})" + raise MetricsInputError(msg) + + frames: list[pd.DataFrame] = [] + for path in paths: + frame = _read_fundamentals_partition_all(path) + if frame is not None: + frames.append(frame) + + if not frames: + msg = f"No curated fundamentals for ticker {ticker!r}" + raise MetricsInputError(msg) + + history = pd.concat(frames, ignore_index=True) + if "statement_variant" in history.columns: + history = history.loc[history["statement_variant"].isin(("annual", "quarterly"))] + + selected = _select_pit_fundamentals_by_period(history, as_of_date) + if selected.empty: + msg = f"No PIT fundamentals history for {ticker!r} on or before {as_of_date}" + raise MetricsInputError(msg) + + return selected.sort_values("fiscal_period_end").reset_index(drop=True) + + def load_prices_by_ticker(data_dir: Path, *, run_date: date) -> dict[str, pd.Series]: """Load the full price snapshot indexed by upper-case ticker.""" path = curated_prices_snapshot_path(data_dir, run_date=run_date) @@ -140,10 +197,28 @@ def _read_fundamentals_partition(path: Path) -> pd.DataFrame | None: frame = pd.read_parquet(path) if frame.empty: return None - row = frame.iloc[[0]].copy() - if "cik" not in row.columns: - row["cik"] = _cik_from_partition_path(path) - return row + if "cik" not in frame.columns: + frame = frame.copy() + frame["cik"] = _cik_from_partition_path(path) + if "statement_variant" in frame.columns: + ttm = frame.loc[frame["statement_variant"] == "ttm"] + if not ttm.empty: + frame = ttm + else: + frame = frame.iloc[[-1]] + else: + frame = frame.iloc[[0]] + return frame + + +def _read_fundamentals_partition_all(path: Path) -> pd.DataFrame | None: + frame = pd.read_parquet(path) + if frame.empty: + return None + if "cik" not in frame.columns: + frame = frame.copy() + frame["cik"] = _cik_from_partition_path(path) + return frame def load_pit_fundamentals_bulk( diff --git a/src/smartwealthai/simfin_normalizer.py b/src/smartwealthai/simfin_normalizer.py index 6dc11c9..279acd7 100644 --- a/src/smartwealthai/simfin_normalizer.py +++ b/src/smartwealthai/simfin_normalizer.py @@ -42,6 +42,11 @@ "total_assets", "total_liabilities", "shares_outstanding", + "accounts_receivable", + "cost_of_revenue", + "depreciation_amortization", + "sga_expense", + "operating_cash_flow", ) DATE_COLUMNS = ("Report Date", "Publish Date", "Restated Date") @@ -52,6 +57,23 @@ logger = logging.getLogger(__name__) +@dataclass(frozen=True) +class StatementBundle: + """One income/balance/cashflow variant group for normalization.""" + + income_variant: str + balance_variant: str + cashflow_variant: str | None + enforce_qv_fields: bool + + +STATEMENT_BUNDLES: tuple[StatementBundle, ...] = ( + StatementBundle("ttm", "quarterly", None, enforce_qv_fields=False), + StatementBundle("annual", "annual", "annual", enforce_qv_fields=True), + StatementBundle("quarterly", "quarterly", "quarterly", enforce_qv_fields=True), +) + + @dataclass class NormalizeResult: """Outcome counters for one normalizer run.""" @@ -77,30 +99,94 @@ def normalize_simfin( """Transform raw SimFin bulk CSVs into curated fundamentals parquet.""" progress_enabled = _resolve_show_progress(show_progress) mapping = load_simfin_mapping(mapping_path) - paths = _raw_paths(data_dir, snapshot_date) + companies_path = simfin_bulk_path( + data_dir, + dataset="companies", + variant=None, + market="us", + as_of_date=snapshot_date, + ) + companies = pd.read_csv(companies_path, sep=";", dtype={"CIK": "string"}) + company_lookup = companies.set_index(mapping["meta"]["ticker"], drop=False) + + curated_rows: list[dict[str, object]] = [] + issue_rows: list[dict[str, object]] = [] + effective_run_date = run_date or snapshot_date + + for bundle in STATEMENT_BUNDLES: + bundle_rows, bundle_issues = _normalize_statement_bundle( + data_dir, + snapshot_date=snapshot_date, + bundle=bundle, + mapping=mapping, + company_lookup=company_lookup, + tickers=tickers, + show_progress=progress_enabled, + ) + curated_rows.extend(bundle_rows) + issue_rows.extend(bundle_issues) + + result = NormalizeResult() + result.written_rows = _write_curated_fundamentals( + data_dir, + curated_rows, + show_progress=progress_enabled, + ) + + if issue_rows: + issues_path = curated_issues_path(data_dir, run_date=effective_run_date) + issues_path.parent.mkdir(parents=True, exist_ok=True) + pd.DataFrame(issue_rows).to_parquet(issues_path, index=False) + result.issue_rows = len(issue_rows) + + return result + + +def _normalize_statement_bundle( + data_dir: Path, + *, + snapshot_date: date, + bundle: StatementBundle, + mapping: dict, + company_lookup: pd.DataFrame, + tickers: set[str] | None, + show_progress: bool, +) -> tuple[list[dict[str, object]], list[dict[str, object]]]: + paths = _raw_paths_for_bundle(data_dir, snapshot_date, bundle) + if not paths["income"].exists(): + return [], [] income = _read_simfin_csv(paths["income"]) - balance = _read_simfin_csv(paths["balance"]) - companies = pd.read_csv(paths["companies"], sep=";", dtype={"CIK": "string"}) + balance = _read_simfin_csv(paths["balance"]) if paths["balance"].exists() else pd.DataFrame() + cashflow = ( + _read_simfin_csv(paths["cashflow"]) + if paths["cashflow"] is not None and paths["cashflow"].exists() + else pd.DataFrame() + ) ticker_col = mapping["meta"]["ticker"] report_col = mapping["meta"]["report_date"] - company_lookup = companies.set_index(ticker_col, drop=False) curated_rows: list[dict[str, object]] = [] issue_rows: list[dict[str, object]] = [] - effective_run_date = run_date or snapshot_date if tickers is not None: income = income.loc[income[ticker_col].isin(tickers)] - balance_by_ticker = _index_balance_by_ticker( + balance_by_ticker = _index_statement_by_ticker( balance, ticker_col=ticker_col, report_col=report_col, ) + cashflow_by_ticker = _index_statement_by_ticker( + cashflow, + ticker_col=ticker_col, + report_col=report_col, + ) work_tickers = _work_tickers(income, ticker_col=ticker_col, tickers=tickers) - ticker_iter = _ticker_progress_iter(work_tickers, enabled=progress_enabled) + ticker_iter = _ticker_progress_iter(work_tickers, enabled=show_progress) + exact_balance_match = bundle.income_variant != "ttm" + for ticker in ticker_iter: ticker_income = income.loc[income[ticker_col] == ticker] for _, income_row in ticker_income.iterrows(): @@ -120,42 +206,43 @@ def normalize_simfin( continue report_date = income_row[report_col] - balance_row = _latest_balance_row( + balance_row = _matching_statement_row( balance_by_ticker, ticker=ticker, report_date=report_date, report_col=report_col, + exact_match=exact_balance_match, ) if balance_row is None: issue_rows.append(_issue_row(ticker=ticker, reason="missing_balance_row")) continue + cashflow_row = None + if paths["cashflow"] is not None: + cashflow_row = _matching_statement_row( + cashflow_by_ticker, + ticker=ticker, + report_date=report_date, + report_col=report_col, + exact_match=True, + ) + row, issues = _build_curated_row( income_row=income_row, balance_row=balance_row, + cashflow_row=cashflow_row, ticker=ticker, cik=cik, mapping=mapping, + statement_variant=bundle.income_variant, + enforce_qv_fields=bundle.enforce_qv_fields, ) if issues: issue_rows.extend(issues) if row is not None: curated_rows.append(row) - result = NormalizeResult() - result.written_rows = _write_curated_fundamentals( - data_dir, - curated_rows, - show_progress=progress_enabled, - ) - - if issue_rows: - issues_path = curated_issues_path(data_dir, run_date=effective_run_date) - issues_path.parent.mkdir(parents=True, exist_ok=True) - pd.DataFrame(issue_rows).to_parquet(issues_path, index=False) - result.issue_rows = len(issue_rows) - - return result + return curated_rows, issue_rows def _write_curated_fundamentals( @@ -184,8 +271,12 @@ def _write_curated_fundamentals( cik=str(cik), period=str(period), ) - # ponytail: last row wins on duplicate partition (matches sequential overwrite) - partitions.append((out_path, group.iloc[[-1]])) + deduped = ( + group.drop_duplicates(subset=["statement_variant"], keep="last") + if "statement_variant" in group.columns + else group.iloc[[-1]] + ) + partitions.append((out_path, deduped)) if len(partitions) == 1: partitions[0][1].to_parquet(partitions[0][0], index=False) @@ -251,29 +342,57 @@ def _write_partition(item: tuple[Path, pd.DataFrame]) -> None: frame.to_parquet(path, index=False) -def _raw_paths(data_dir: Path, snapshot_date: date) -> dict[str, Path]: +def _raw_paths_for_bundle( + data_dir: Path, + snapshot_date: date, + bundle: StatementBundle, +) -> dict[str, Path | None]: + cashflow_path = None + if bundle.cashflow_variant is not None: + cashflow_path = simfin_bulk_path( + data_dir, + dataset="cashflow", + variant=bundle.cashflow_variant, + market="us", + as_of_date=snapshot_date, + ) return { "income": simfin_bulk_path( data_dir, dataset="income", - variant="ttm", + variant=bundle.income_variant, market="us", as_of_date=snapshot_date, ), "balance": simfin_bulk_path( data_dir, dataset="balance", - variant="quarterly", - market="us", - as_of_date=snapshot_date, - ), - "companies": simfin_bulk_path( - data_dir, - dataset="companies", - variant=None, + variant=bundle.balance_variant, market="us", as_of_date=snapshot_date, ), + "cashflow": cashflow_path, + } + + +def _raw_paths(data_dir: Path, snapshot_date: date) -> dict[str, Path]: + """Demo TTM paths (tests and legacy callers).""" + bundle_paths = _raw_paths_for_bundle( + data_dir, + snapshot_date, + STATEMENT_BUNDLES[0], + ) + companies = simfin_bulk_path( + data_dir, + dataset="companies", + variant=None, + market="us", + as_of_date=snapshot_date, + ) + return { + "income": bundle_paths["income"], + "balance": bundle_paths["balance"], + "companies": companies, } @@ -281,42 +400,84 @@ def _read_simfin_csv(path: Path) -> pd.DataFrame: return pd.read_csv(path, sep=";", parse_dates=list(DATE_COLUMNS)) -def _index_balance_by_ticker( - balance: pd.DataFrame, +def _index_statement_by_ticker( + frame: pd.DataFrame, *, ticker_col: str, report_col: str, ) -> dict[str, pd.DataFrame]: - """Pre-group balance rows by ticker sorted by report date (ponytail: dict of frames).""" + """Pre-group statement rows by ticker sorted by report date.""" + if frame.empty: + return {} indexed: dict[str, pd.DataFrame] = {} - for ticker, group in balance.groupby(ticker_col, sort=False): + for ticker, group in frame.groupby(ticker_col, sort=False): indexed[str(ticker)] = group.sort_values(report_col) return indexed -def _latest_balance_row( - balance_by_ticker: dict[str, pd.DataFrame], +def _index_balance_by_ticker( + balance: pd.DataFrame, + *, + ticker_col: str, + report_col: str, +) -> dict[str, pd.DataFrame]: + """Alias kept for tests importing the demo helper name.""" + return _index_statement_by_ticker( + balance, + ticker_col=ticker_col, + report_col=report_col, + ) + + +def _matching_statement_row( + rows_by_ticker: dict[str, pd.DataFrame], *, ticker: str, report_date: pd.Timestamp, report_col: str, + exact_match: bool, ) -> pd.Series | None: - group = balance_by_ticker.get(ticker) + group = rows_by_ticker.get(ticker) if group is None: return None + if exact_match: + subset = group.loc[group[report_col] == report_date] + if subset.empty: + return None + return subset.iloc[-1] subset = group.loc[group[report_col] <= report_date] if subset.empty: return None return subset.iloc[-1] +def _latest_balance_row( + balance_by_ticker: dict[str, pd.DataFrame], + *, + ticker: str, + report_date: pd.Timestamp, + report_col: str, +) -> pd.Series | None: + """Demo TTM join: latest balance on or before income report date.""" + return _matching_statement_row( + balance_by_ticker, + ticker=ticker, + report_date=report_date, + report_col=report_col, + exact_match=False, + ) + + def _build_curated_row( *, income_row: pd.Series, balance_row: pd.Series, + cashflow_row: pd.Series | None, ticker: str, cik: str, mapping: dict, + statement_variant: str, + enforce_qv_fields: bool, ) -> tuple[dict[str, object] | None, list[dict[str, object]]]: meta = mapping["meta"] fields = mapping["fields"] @@ -356,19 +517,24 @@ def _build_curated_row( "as_of_date": as_of_date, "version_id": version_id, "period": period, + "statement_variant": statement_variant, "mapping_version": mapping["version"], "currency": "USD", "as_of_source": as_of_source, } + statement_sources = (income_row, balance_row, cashflow_row) for canonical, simfin_col in fields.items(): - source = income_row if simfin_col in income_row.index else balance_row - row[canonical] = _numeric(source.get(simfin_col)) + row[canonical] = _field_value(simfin_col, statement_sources) if _missing_mandatory(row): issues.append(_issue_row(ticker=ticker, reason="missing_mandatory_fields")) return None, issues + if enforce_qv_fields and _missing_qv_fields(row, mapping): + issues.append(_issue_row(ticker=ticker, reason="missing_qv_inputs")) + return None, issues + if as_of_date < fiscal_period_end: issues.append(_issue_row(ticker=ticker, reason="as_of_before_period_end")) return None, issues @@ -376,6 +542,13 @@ def _build_curated_row( return row, issues +def _field_value(simfin_col: str, sources: tuple[pd.Series | None, ...]) -> float | None: + for source in sources: + if source is not None and simfin_col in source.index: + return _numeric(source.get(simfin_col)) + return None + + def _resolve_as_of_date( *, publish_date: object, @@ -424,6 +597,11 @@ def _missing_mandatory(row: dict[str, object]) -> bool: return any(row.get(field) is None for field in mandatory) +def _missing_qv_fields(row: dict[str, object], mapping: dict) -> bool: + required = mapping.get("qv_required_fields", []) + return any(row.get(field) is None for field in required) + + def _issue_row(*, ticker: str, reason: str) -> dict[str, str]: return {"ticker": ticker, "reason": reason} @@ -431,7 +609,7 @@ def _issue_row(*, ticker: str, reason: str) -> dict[str, str]: def _load_mapping_yaml(path: Path) -> dict: """Parse the project mapping YAML (two-level dict + top-level scalars only).""" root: dict[str, object] = {} - section: dict[str, str] | None = None + section: dict[str, str] | list[str] | None = None for raw in path.read_text().splitlines(): line = raw.split("#", 1)[0].rstrip() if not line.strip(): @@ -443,6 +621,9 @@ def _load_mapping_yaml(path: Path) -> dict: if name in {"fields", "meta"}: section = {} root[name] = section + elif name == "qv_required_fields": + section = [] + root[name] = section else: section = None continue @@ -451,11 +632,14 @@ def _load_mapping_yaml(path: Path) -> dict: root[key.strip()] = parsed section = None continue + if isinstance(section, list) and stripped.startswith("- "): + section.append(stripped[2:].strip()) + continue key, sep, value = line.partition(":") if not sep: continue key = key.strip() value = value.strip().strip('"').strip("'") - if section is not None: + if isinstance(section, dict): section[key] = value return root From 8833d60374fe5810e8c17c5e75749d41a37524f2 Mon Sep 17 00:00:00 2001 From: JLaborda <15078416+JLaborda@users.noreply.github.com> Date: Wed, 1 Jul 2026 10:43:08 +0000 Subject: [PATCH 3/4] test(etl): add hermetic coverage for multi-period fundamentals Cover phase 2 download variants, quarterly QV field normalization, missing-input review queue routing, and PIT history look-ahead guards. Co-authored-by: Cursor --- .../us-balance-quarterly.csv | 13 ++-- .../us-cashflow-quarterly.csv | 3 + .../us-income-quarterly.csv | 3 + tests/test_download_simfin.py | 24 ++++++- tests/test_pit_fundamentals.py | 68 +++++++++++++++++++ tests/test_simfin_normalizer.py | 63 ++++++++++++++++- 6 files changed, 164 insertions(+), 10 deletions(-) create mode 100644 tests/fixtures/lake/raw/simfin/dataset=cashflow/variant=quarterly/market=us/as_of_date=2026-06-18/us-cashflow-quarterly.csv create mode 100644 tests/fixtures/lake/raw/simfin/dataset=income/variant=quarterly/market=us/as_of_date=2026-06-18/us-income-quarterly.csv diff --git a/tests/fixtures/lake/raw/simfin/dataset=balance/variant=quarterly/market=us/as_of_date=2026-06-18/us-balance-quarterly.csv b/tests/fixtures/lake/raw/simfin/dataset=balance/variant=quarterly/market=us/as_of_date=2026-06-18/us-balance-quarterly.csv index 8d19495..7b60d0b 100644 --- a/tests/fixtures/lake/raw/simfin/dataset=balance/variant=quarterly/market=us/as_of_date=2026-06-18/us-balance-quarterly.csv +++ b/tests/fixtures/lake/raw/simfin/dataset=balance/variant=quarterly/market=us/as_of_date=2026-06-18/us-balance-quarterly.csv @@ -1,6 +1,7 @@ -Ticker;SimFinId;Report Date;Publish Date;Restated Date;Currency;Fiscal Year;Fiscal Period;Total Current Assets;Total Current Liabilities;Cash, Cash Equivalents & Short Term Investments;Short Term Debt;Property, Plant & Equipment, Net;Long Term Debt;Preferred Equity;Minority Interest;Total Assets;Total Liabilities;Shares (Basic) -AAPL;111052;2024-09-28;2024-11-01;;USD;2024;Q4;152987000000;176000000000;29943000000;0;45680000000;95281000000;0;0;364980000000;308030000000;15115800000 -MSFT;222222;2024-06-30;2024-07-30;;USD;2024;Q4;100000000;50000000;20000000;10000000;80000000;100000000;0;0;200000000;120000000;20000000 -FOO;333333;2024-06-30;2024-07-30;;USD;2024;Q4;50000000;30000000;10000000;5000000;50000000;0;0;0;110000000;40000000;10000000 -BAR;444444;2024-06-30;2024-07-30;;USD;2024;Q4;40000000;20000000;5000000;0;30000000;10000000;0;0;80000000;35000000;5000000 -LOST;555555;2024-06-30;2024-07-30;;USD;2024;Q4;40000000;20000000;5000000;0;30000000;0;0;0;80000000;30000000;5000000 +Ticker;SimFinId;Report Date;Publish Date;Restated Date;Currency;Fiscal Year;Fiscal Period;Total Current Assets;Total Current Liabilities;Cash, Cash Equivalents & Short Term Investments;Short Term Debt;Property, Plant & Equipment, Net;Long Term Debt;Preferred Equity;Minority Interest;Total Assets;Total Liabilities;Shares (Basic);Accounts Receivable +AAPL;111052;2023-09-30;2023-11-03;;USD;2023;Q4;143566000000;145308000000;29965000000;0;43715000000;95281000000;0;0;352583000000;290437000000;15550000000;29508000000 +AAPL;111052;2024-09-28;2024-11-01;;USD;2024;Q4;152987000000;176000000000;29943000000;0;45680000000;95281000000;0;0;364980000000;308030000000;15115800000;33410000000 +MSFT;222222;2024-06-30;2024-07-30;;USD;2024;Q4;100000000;50000000;20000000;10000000;80000000;100000000;0;0;200000000;120000000;20000000;5000000 +FOO;333333;2024-06-30;2024-07-30;;USD;2024;Q4;50000000;30000000;10000000;5000000;50000000;0;0;0;110000000;40000000;10000000;2000000 +BAR;444444;2024-06-30;2024-07-30;;USD;2024;Q4;40000000;20000000;5000000;0;30000000;10000000;0;0;80000000;35000000;5000000;1000000 +LOST;555555;2024-06-30;2024-07-30;;USD;2024;Q4;40000000;20000000;5000000;0;30000000;0;0;0;80000000;30000000;5000000;1000000 diff --git a/tests/fixtures/lake/raw/simfin/dataset=cashflow/variant=quarterly/market=us/as_of_date=2026-06-18/us-cashflow-quarterly.csv b/tests/fixtures/lake/raw/simfin/dataset=cashflow/variant=quarterly/market=us/as_of_date=2026-06-18/us-cashflow-quarterly.csv new file mode 100644 index 0000000..b453c92 --- /dev/null +++ b/tests/fixtures/lake/raw/simfin/dataset=cashflow/variant=quarterly/market=us/as_of_date=2026-06-18/us-cashflow-quarterly.csv @@ -0,0 +1,3 @@ +Ticker;SimFinId;Report Date;Publish Date;Restated Date;Currency;Fiscal Year;Fiscal Period;Depreciation & Amortization;Net Cash from Operating Activities +AAPL;111052;2023-09-30;2023-11-03;;USD;2023;Q4;11519000000;110543000000 +AAPL;111052;2024-09-28;2024-11-01;;USD;2024;Q4;11445000000;118254000000 diff --git a/tests/fixtures/lake/raw/simfin/dataset=income/variant=quarterly/market=us/as_of_date=2026-06-18/us-income-quarterly.csv b/tests/fixtures/lake/raw/simfin/dataset=income/variant=quarterly/market=us/as_of_date=2026-06-18/us-income-quarterly.csv new file mode 100644 index 0000000..f72fa0a --- /dev/null +++ b/tests/fixtures/lake/raw/simfin/dataset=income/variant=quarterly/market=us/as_of_date=2026-06-18/us-income-quarterly.csv @@ -0,0 +1,3 @@ +Ticker;SimFinId;Report Date;Publish Date;Restated Date;Currency;Fiscal Year;Fiscal Period;Revenue;Operating Income (Loss);Net Income;Cost of Revenue;Selling, General & Administrative +AAPL;111052;2023-09-30;2023-11-03;;USD;2023;Q4;383285000000;114301000000;96995000000;214137000000;24932000000 +AAPL;111052;2024-09-28;2024-11-01;;USD;2024;Q4;391035000000;123216000000;93736000000;210352000000;26097000000 diff --git a/tests/test_download_simfin.py b/tests/test_download_simfin.py index 3039b32..0424826 100644 --- a/tests/test_download_simfin.py +++ b/tests/test_download_simfin.py @@ -17,8 +17,11 @@ from smartwealthai.download_simfin import ( DEMO_DATASETS, + PHASE2_STATEMENT_DATASETS, + SIMFIN_DATASETS, DatasetResult, cli_run, + dataset_spec_key, download_dataset, run_download, should_skip_dataset, @@ -94,6 +97,21 @@ def test_should_not_skip_when_lake_file_missing(tmp_path: Path) -> None: assert should_skip_dataset(target, refresh_days=7, force=False, now=1_700_000_000.0) is False +def test_phase2_statement_datasets_cover_annual_and_quarterly_variants() -> None: + keys = {dataset_spec_key(spec) for spec in PHASE2_STATEMENT_DATASETS} + assert keys == { + "income/annual", + "income/quarterly", + "balance/annual", + "cashflow/annual", + "cashflow/quarterly", + } + + +def test_simfin_datasets_include_demo_and_phase2() -> None: + assert len(SIMFIN_DATASETS) == len(DEMO_DATASETS) + len(PHASE2_STATEMENT_DATASETS) + + def test_download_dataset_copies_csv_into_lake_layout(tmp_path: Path) -> None: spec = DEMO_DATASETS[2] # income ttm cache_csv = tmp_path / "cache" / "us-income-ttm.csv" @@ -119,7 +137,7 @@ def fake_fetch(**_kwargs: object) -> Path: market="us", as_of_date=date(2026, 6, 18), ) - assert result == DatasetResult(name="income", downloaded=True) + assert result == DatasetResult(name="income/ttm", downloaded=True) assert lake_path.read_text() == cache_csv.read_text() @@ -150,7 +168,7 @@ def fail_fetch(**_kwargs: object) -> Path: fetch_csv=fail_fetch, now=now, ) - assert result == DatasetResult(name="income", skipped=True) + assert result == DatasetResult(name="income/ttm", skipped=True) def test_run_download_continues_after_noncritical_failure(tmp_path: Path) -> None: @@ -438,7 +456,7 @@ def fake_fetch( def test_run_download_skips_all_fresh_datasets(tmp_path: Path) -> None: data_dir = tmp_path / "lake" now = time.time() - for spec in DEMO_DATASETS: + for spec in SIMFIN_DATASETS: lake_path = simfin_bulk_path( data_dir, dataset=spec.name, diff --git a/tests/test_pit_fundamentals.py b/tests/test_pit_fundamentals.py index 0631aa8..a0c3dd8 100644 --- a/tests/test_pit_fundamentals.py +++ b/tests/test_pit_fundamentals.py @@ -18,10 +18,12 @@ _read_fundamentals_partition, _resolve_show_progress, _select_pit_fundamentals, + _select_pit_fundamentals_by_period, _ticker_progress_iter, compute_metrics_for_ticker, compute_metrics_for_tickers, load_pit_fundamentals_bulk, + load_pit_fundamentals_history, load_pit_fundamentals_row, load_ticker_price_row, resolve_ticker_cik, @@ -59,7 +61,9 @@ def _write_prices(lake: Path, *, rows: list[dict[str, object]]) -> None: def _base_fundamentals_row(**overrides: object) -> dict[str, object]: row = { "as_of_date": pd.Timestamp("2026-06-01"), + "fiscal_period_end": pd.Timestamp("2024-09-28"), "version_id": 1, + "statement_variant": "ttm", "ebit": 100.0, "current_assets": 200.0, "current_liabilities": 80.0, @@ -311,6 +315,70 @@ def test_compute_metrics_for_tickers_skips_missing_inputs(pit_lake: Path) -> Non assert [result.ticker for result in results] == ["AAPL"] +def test_select_pit_fundamentals_by_period_excludes_lookahead_rows() -> None: + frame = pd.DataFrame( + [ + { + "cik": CIK, + "fiscal_period_end": pd.Timestamp("2024-09-28"), + "as_of_date": pd.Timestamp("2024-11-01"), + "version_id": 1, + "statement_variant": "quarterly", + }, + { + "cik": CIK, + "fiscal_period_end": pd.Timestamp("2024-09-28"), + "as_of_date": pd.Timestamp("2027-01-01"), + "version_id": 2, + "statement_variant": "quarterly", + }, + ] + ) + + selected = _select_pit_fundamentals_by_period(frame, date(2026, 6, 18)) + + assert len(selected) == 1 + assert selected.iloc[0]["version_id"] == 1 + + +def test_load_pit_fundamentals_history_returns_multi_period_rows(pit_lake: Path) -> None: + _write_fundamentals( + pit_lake, + period="2023Q4", + row=_base_fundamentals_row( + as_of_date=pd.Timestamp("2023-11-03"), + fiscal_period_end=pd.Timestamp("2023-09-30"), + statement_variant="quarterly", + accounts_receivable=29_508_000_000, + ), + ) + _write_fundamentals( + pit_lake, + period="2024Q4", + row=_base_fundamentals_row( + statement_variant="quarterly", + accounts_receivable=33_410_000_000, + ), + ) + + history = load_pit_fundamentals_history(pit_lake, ticker="AAPL", as_of_date=RUN_DATE) + + assert len(history) == 2 + assert history.iloc[0]["fiscal_period_end"] == pd.Timestamp("2023-09-30") + assert history.iloc[1]["accounts_receivable"] == 33_410_000_000 + + +def test_load_pit_fundamentals_history_ignores_ttm_rows(pit_lake: Path) -> None: + _write_fundamentals( + pit_lake, + period="2024Q4", + row=_base_fundamentals_row(statement_variant="ttm"), + ) + + with pytest.raises(MetricsInputError, match="No PIT fundamentals history"): + load_pit_fundamentals_history(pit_lake, ticker="AAPL", as_of_date=RUN_DATE) + + def test_compute_metrics_for_tickers_loads_universe_when_cik_map_omitted(pit_lake: Path) -> None: results = compute_metrics_for_tickers( pit_lake, diff --git a/tests/test_simfin_normalizer.py b/tests/test_simfin_normalizer.py index 6098900..4a3f73d 100644 --- a/tests/test_simfin_normalizer.py +++ b/tests/test_simfin_normalizer.py @@ -632,6 +632,53 @@ def test_normalize_simfin_uses_calendar_period_when_fiscal_labels_missing( assert curated_fundamentals_path(lake_with_simfin, cik="0000320193", period="2024Q3").exists() +def test_load_simfin_mapping_parses_qv_required_fields() -> None: + mapping = load_simfin_mapping(DEFAULT_MAPPING_PATH) + + assert "accounts_receivable" in mapping["qv_required_fields"] + assert mapping["fields"]["operating_cash_flow"] == "Net Cash from Operating Activities" + + +def test_normalize_simfin_writes_quarterly_qv_fields(tmp_path: Path) -> None: + _copy_simfin_fixtures(tmp_path, include_multiperiod=True) + + result = normalize_simfin( + tmp_path, + snapshot_date=SIMFIN_FIXTURE_DATE, + tickers={"AAPL"}, + ) + + assert result.written_rows >= 2 + output = curated_fundamentals_path(tmp_path, cik="0000320193", period="2024Q4") + quarterly = ( + pd.read_parquet(output).loc[lambda frame: frame["statement_variant"] == "quarterly"].iloc[0] + ) + assert quarterly["accounts_receivable"] == 33_410_000_000 + assert quarterly["operating_cash_flow"] == 118_254_000_000 + + +def test_normalize_simfin_routes_missing_qv_inputs_to_issues(tmp_path: Path) -> None: + _copy_simfin_fixtures(tmp_path, include_multiperiod=True) + cashflow_path = simfin_bulk_path( + tmp_path, + dataset="cashflow", + variant="quarterly", + market="us", + as_of_date=SIMFIN_FIXTURE_DATE, + ) + cashflow_path.unlink() + + result = normalize_simfin( + tmp_path, + snapshot_date=SIMFIN_FIXTURE_DATE, + tickers={"AAPL"}, + ) + + assert result.issue_rows >= 1 + issues = pd.read_parquet(curated_issues_path(tmp_path, run_date=SIMFIN_FIXTURE_DATE)) + assert "missing_qv_inputs" in set(issues["reason"]) + + def test_load_simfin_mapping_parses_minimal_yaml(tmp_path: Path) -> None: mapping_path = tmp_path / "mapping.yaml" mapping_path.write_text( @@ -656,12 +703,26 @@ def _append_csv_line(path: Path, line: str) -> None: path.write_text(path.read_text().rstrip("\n") + "\n" + line + "\n") -def _copy_simfin_fixtures(data_dir: Path) -> None: +_SKIP_UNLESS_MULTIPERIOD = ( + "dataset=income/variant=quarterly/", + "dataset=income/variant=annual/", + "dataset=balance/variant=annual/", + "dataset=cashflow/variant=annual/", + "dataset=cashflow/variant=quarterly/", +) + + +def _copy_simfin_fixtures(data_dir: Path, *, include_multiperiod: bool = False) -> None: src = FIXTURES_DIR / "raw" / "simfin" dst = data_dir / "raw" / "simfin" for path in src.rglob("*"): if path.is_file(): rel = path.relative_to(src) + rel_str = f"{rel.parent}/" + if not include_multiperiod and any( + marker in rel_str for marker in _SKIP_UNLESS_MULTIPERIOD + ): + continue target = dst / rel target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(path.read_bytes()) From f1b9045a402269192d0bbe9939d74b3d807b2182 Mon Sep 17 00:00:00 2001 From: JLaborda <15078416+JLaborda@users.noreply.github.com> Date: Wed, 1 Jul 2026 11:03:05 +0000 Subject: [PATCH 4/4] test(etl): cover patch gaps in pit and simfin normalizer Exercise PIT history edge paths, partition readers, statement matching, and YAML list parsing so Codecov patch coverage meets the threshold. Co-authored-by: Cursor --- tests/test_pit_fundamentals.py | 168 ++++++++++++++++++++++++++++++++ tests/test_simfin_normalizer.py | 40 ++++++++ 2 files changed, 208 insertions(+) diff --git a/tests/test_pit_fundamentals.py b/tests/test_pit_fundamentals.py index a0c3dd8..c2907e2 100644 --- a/tests/test_pit_fundamentals.py +++ b/tests/test_pit_fundamentals.py @@ -16,6 +16,7 @@ _fundamentals_paths_for_ciks, _optional_float, _read_fundamentals_partition, + _read_fundamentals_partition_all, _resolve_show_progress, _select_pit_fundamentals, _select_pit_fundamentals_by_period, @@ -389,3 +390,170 @@ def test_compute_metrics_for_tickers_loads_universe_when_cik_map_omitted(pit_lak assert len(results) == 1 assert results[0].ticker == "AAPL" + + +def test_select_pit_fundamentals_without_cik_column() -> None: + frame = pd.DataFrame( + [ + {"as_of_date": pd.Timestamp("2024-01-01"), "version_id": 1}, + {"as_of_date": pd.Timestamp("2025-01-01"), "version_id": 2}, + ] + ) + + selected = _select_pit_fundamentals(frame, RUN_DATE) + + assert len(selected) == 1 + assert selected.iloc[0]["version_id"] == 2 + + +def test_select_pit_fundamentals_by_period_returns_empty_when_all_future() -> None: + frame = pd.DataFrame( + [ + { + "cik": CIK, + "fiscal_period_end": pd.Timestamp("2024-09-28"), + "as_of_date": pd.Timestamp("2027-01-01"), + "version_id": 1, + } + ] + ) + + assert _select_pit_fundamentals_by_period(frame, RUN_DATE).empty + + +def test_load_pit_fundamentals_history_raises_when_no_partitions(pit_lake: Path) -> None: + shutil.rmtree(pit_lake / "curated" / "fundamentals") + + with pytest.raises(MetricsInputError, match="No curated fundamentals for ticker"): + load_pit_fundamentals_history(pit_lake, ticker="AAPL", as_of_date=RUN_DATE) + + +def test_load_pit_fundamentals_history_raises_when_partitions_empty(pit_lake: Path) -> None: + shutil.rmtree(pit_lake / "curated" / "fundamentals" / f"cik={CIK}" / "period=2024Q4") + empty_path = ( + pit_lake + / "curated" + / "fundamentals" + / f"cik={CIK}" + / "period=2024Q3" + / "fundamentals.parquet" + ) + empty_path.parent.mkdir(parents=True, exist_ok=True) + pd.DataFrame(columns=["as_of_date", "version_id"]).to_parquet(empty_path, index=False) + + with pytest.raises(MetricsInputError, match="No curated fundamentals for ticker"): + load_pit_fundamentals_history(pit_lake, ticker="AAPL", as_of_date=RUN_DATE) + + +def test_read_fundamentals_partition_uses_last_row_when_no_ttm(pit_lake: Path) -> None: + path = ( + pit_lake + / "curated" + / "fundamentals" + / f"cik={CIK}" + / "period=2024Q4" + / "fundamentals.parquet" + ) + pd.DataFrame( + [ + _base_fundamentals_row(statement_variant="quarterly", ebit=50.0), + _base_fundamentals_row(statement_variant="annual", ebit=60.0), + ] + ).to_parquet(path, index=False) + + frame = _read_fundamentals_partition(path) + + assert frame is not None + assert frame.iloc[0]["statement_variant"] == "annual" + assert frame.iloc[0]["ebit"] == 60.0 + + +def test_read_fundamentals_partition_without_statement_variant_column(pit_lake: Path) -> None: + path = ( + pit_lake + / "curated" + / "fundamentals" + / f"cik={CIK}" + / "period=2024Q3" + / "fundamentals.parquet" + ) + path.parent.mkdir(parents=True, exist_ok=True) + row = _base_fundamentals_row(ebit=77.0) + row.pop("statement_variant") + pd.DataFrame([row]).to_parquet(path, index=False) + + frame = _read_fundamentals_partition(path) + + assert frame is not None + assert frame.iloc[0]["ebit"] == 77.0 + + +def test_read_fundamentals_partition_all_returns_none_for_empty_file(pit_lake: Path) -> None: + path = ( + pit_lake + / "curated" + / "fundamentals" + / f"cik={CIK}" + / "period=2024Q2" + / "fundamentals.parquet" + ) + path.parent.mkdir(parents=True, exist_ok=True) + pd.DataFrame(columns=["as_of_date"]).to_parquet(path, index=False) + + assert _read_fundamentals_partition_all(path) is None + + +def test_read_fundamentals_partition_all_injects_cik_from_path(pit_lake: Path) -> None: + path = ( + pit_lake + / "curated" + / "fundamentals" + / f"cik={CIK}" + / "period=2024Q1" + / "fundamentals.parquet" + ) + path.parent.mkdir(parents=True, exist_ok=True) + row = _base_fundamentals_row() + row.pop("cik", None) + pd.DataFrame([row]).to_parquet(path, index=False) + + frame = _read_fundamentals_partition_all(path) + + assert frame is not None + assert frame.iloc[0]["cik"] == CIK + + +def test_read_fundamentals_partition_all_keeps_existing_cik_column(pit_lake: Path) -> None: + path = ( + pit_lake + / "curated" + / "fundamentals" + / f"cik={CIK}" + / "period=2023Q4" + / "fundamentals.parquet" + ) + path.parent.mkdir(parents=True, exist_ok=True) + pd.DataFrame([{**_base_fundamentals_row(), "cik": CIK}]).to_parquet(path, index=False) + + frame = _read_fundamentals_partition_all(path) + + assert frame is not None + assert frame.iloc[0]["cik"] == CIK + + +def test_load_pit_fundamentals_history_without_statement_variant_column(pit_lake: Path) -> None: + shutil.rmtree(pit_lake / "curated" / "fundamentals" / f"cik={CIK}") + for period, as_of, fiscal_end in ( + ("2023Q4", "2023-11-03", "2023-09-30"), + ("2024Q4", "2026-06-01", "2024-09-28"), + ): + row = _base_fundamentals_row( + as_of_date=pd.Timestamp(as_of), + fiscal_period_end=pd.Timestamp(fiscal_end), + ) + row.pop("statement_variant") + _write_fundamentals(pit_lake, period=period, row=row) + + history = load_pit_fundamentals_history(pit_lake, ticker="AAPL", as_of_date=RUN_DATE) + + assert len(history) == 2 diff --git a/tests/test_simfin_normalizer.py b/tests/test_simfin_normalizer.py index 4a3f73d..9eb4ae0 100644 --- a/tests/test_simfin_normalizer.py +++ b/tests/test_simfin_normalizer.py @@ -27,6 +27,7 @@ DEFAULT_MAPPING_PATH, _index_balance_by_ticker, _latest_balance_row, + _matching_statement_row, _raw_paths, _read_simfin_csv, _resolve_show_progress, @@ -639,6 +640,39 @@ def test_load_simfin_mapping_parses_qv_required_fields() -> None: assert mapping["fields"]["operating_cash_flow"] == "Net Cash from Operating Activities" +def test_matching_statement_row_exact_match_returns_none_when_dates_differ( + lake_with_simfin: Path, +) -> None: + paths = _raw_paths(lake_with_simfin, SIMFIN_FIXTURE_DATE) + balance = _read_simfin_csv(paths["balance"]) + balance_by_ticker = _index_balance_by_ticker( + balance, + ticker_col="Ticker", + report_col="Report Date", + ) + + result = _matching_statement_row( + balance_by_ticker, + ticker="AAPL", + report_date=pd.Timestamp("1999-01-01"), + report_col="Report Date", + exact_match=True, + ) + + assert result is None + + +def test_normalize_simfin_processes_all_income_tickers_when_unscoped( + lake_with_simfin: Path, +) -> None: + result = normalize_simfin( + lake_with_simfin, + snapshot_date=SIMFIN_FIXTURE_DATE, + ) + + assert result.written_rows >= 1 + + def test_normalize_simfin_writes_quarterly_qv_fields(tmp_path: Path) -> None: _copy_simfin_fixtures(tmp_path, include_multiperiod=True) @@ -687,7 +721,12 @@ def test_load_simfin_mapping_parses_minimal_yaml(tmp_path: Path) -> None: " ebit: EBIT\n" "meta:\n" " ticker: Ticker\n" + "qv_required_fields:\n" + " - accounts_receivable\n" + " note without colon\n" + " orphan: value\n" "unknown_section:\n" + " ignored line without colon\n" "missing_publish_lag_days: 45\n" "note without colon\n" ) @@ -696,6 +735,7 @@ def test_load_simfin_mapping_parses_minimal_yaml(tmp_path: Path) -> None: assert mapping["version"] == "test_v1" assert mapping["fields"]["ebit"] == "EBIT" + assert mapping["qv_required_fields"] == ["accounts_receivable"] assert mapping["missing_publish_lag_days"] == 45