feat(etl): daily price history for backtest window (#88)

pyproject.toml

@@ -29,6 +29,7 @@ download-simfin = "smartwealthai.download_simfin:main"
 normalize-simfin = "smartwealthai.normalize_simfin:main"
 build-universe = "smartwealthai.build_universe:main"
 download-prices = "smartwealthai.download_prices:main"
+download-price-history = "smartwealthai.download_price_history:main"
 compute-metrics = "smartwealthai.compute_metrics:main"
 run-demo-pipeline = "smartwealthai.run_demo_pipeline:main"
 score-universe = "smartwealthai.score_universe:main"

spec/features/006-etl-data-lake/spec.md

@@ -195,6 +195,12 @@ Phase 2 yfinance cache semantics:
   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`.
+- Phase 2 daily price history ([#88](https://github.com/JLaborda/SmartWealthAI/issues/88)):
+  `price_history_ingest.py` and `download-price-history` CLI ingest SimFin
+  `shareprices/daily` into `curated/prices/ticker=<ticker>/year=<YYYY>/prices.parquet`.
+  `lookup_daily_adj_close()` supports run-date / rebalance-date joins for backtest NAV.
+  Hermetic tests in `tests/test_price_history_ingest.py`. Operator notes in
+  [`spec/guides/download-simfin.md`](../../guides/download-simfin.md).
 
 **Operator sequence (demo pipeline):**
 
@@ -347,6 +353,27 @@ Extend SimFin ingest and normalization so Quantitative Value downstream modules
 - 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`.
 
+## Daily price history (phase 2)
+
+**GitHub issue:** [#88](https://github.com/JLaborda/SmartWealthAI/issues/88) — feeds backtest NAV simulation ([#91](https://github.com/JLaborda/SmartWealthAI/issues/91)).
+
+Ingest SimFin bulk `shareprices/daily` for a configurable date window; write curated partitions by ticker and calendar year. Demo `shareprices/latest` path is unchanged.
+
+### Expected flow
+
+1. `download-price-history` downloads (or reuses) `raw/simfin/.../variant=daily/...`.
+2. Filter to universe tickers and `[start_date, end_date]`.
+3. Write `curated/prices/ticker=<ticker>/year=<YYYY>/prices.parquet`.
+4. Downstream backtests call `lookup_daily_adj_close()` for rebalance-date joins.
+
+### Acceptance criteria (phase 2 daily prices)
+
+- [x] Curated daily price parquet by ticker/year (`curated/prices/ticker=*/year=*/prices.parquet`).
+- [x] CLI populates prices for demo universe over configurable date range (`download-price-history`).
+- [x] `adj_close` usable for historical market cap and backtest NAV (`lookup_daily_adj_close`).
+- [x] Fixture tests for price ingest and run-date join (`tests/test_price_history_ingest.py`).
+- [x] Operator note in [`spec/guides/download-simfin.md`](../../guides/download-simfin.md) for SimFin free-tier limits.
+
 ## 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).

spec/guides/download-simfin.md

@@ -61,6 +61,29 @@ Pass `--ticker` to limit the normalize step to specific names (intersect univers
 
 Phase 2 annual/quarterly rows require the matching cashflow variant on disk; missing QV inputs route to `curated/issues/` with reason `missing_qv_inputs`.
 
+## Daily price history (phase 2)
+
+Backtests and historical market cap need **full daily adjusted prices**, not the demo `shareprices/latest` snapshot. Use `download-price-history` after `build-universe` (or pass `--ticker` for smoke tests).
+
+```bash
+poetry run download-price-history \
+  --universe-run-date 2026-06-18 \
+  --start-date 2016-01-01 \
+  --end-date 2026-06-18 \
+  --snapshot-date 2026-06-18
+```
+
+| Step | Output |
+| --- | --- |
+| SimFin bulk download | `raw/simfin/dataset=shareprices/variant=daily/market=us/as_of_date=<date>/us-shareprices-daily.csv` |
+| Normalize + partition | `curated/prices/ticker=<T>/year=<YYYY>/prices.parquet` |
+
+Curated columns: `ticker`, `price_date`, `close`, `adj_close`, `volume`. Use `lookup_daily_adj_close()` from `price_history_ingest` for the latest `adj_close` on or before a rebalance date.
+
+**Free-tier notes:** `shareprices/daily` is a large bulk file. Re-download weekly (`--refresh-days 7`, default) unless `--force`. The demo pipeline does **not** call this step. Per-ticker yfinance fallback is deferred; see phase 2 architecture when SimFin limits block a backfill.
+
+Flags: `--skip-download` (offline/tests), `--force` (overwrite raw + curated partitions), `--ticker` (repeatable).
+
 ## Refresh and cache behaviour
 
 - **Skip:** Re-run without `--force` when the on-disk lake copy is younger than `--refresh-days` (default `7`).
@@ -80,8 +103,10 @@ Phase 2 annual/quarterly rows require the matching cashflow variant on disk; mis
 | --- | --- |
 | `smartwealthai.simfin_client` | Configure API key, safe bulk download (zip-slip guarded), resolve cache CSV path. |
 | `smartwealthai.download_simfin` | CLI orchestration, skip/force logic, run summary. |
+| `smartwealthai.price_history_ingest` | Phase 2 daily price normalize + run-date lookup. |
+| `smartwealthai.download_price_history` | CLI for `shareprices/daily` → curated ticker/year partitions. |
 | `smartwealthai.lake_paths` | Raw lake path builders for `raw/simfin/`. |
 
 ## Tests
 
-Hermetic tests live in `tests/test_download_simfin.py`. They mock SimFin network calls; PR CI does not require a live API key.
+Hermetic tests live in `tests/test_download_simfin.py` and `tests/test_price_history_ingest.py`. They mock SimFin network calls; PR CI does not require a live API key.

src/smartwealthai/download_price_history.py

@@ -0,0 +1,173 @@
+"""CLI to ingest SimFin shareprices/daily into curated ticker/year partitions."""
+
+from __future__ import annotations
+
+import logging
+import os
+from datetime import UTC, date, datetime
+from pathlib import Path
+
+import click
+from click.testing import CliRunner
+
+from smartwealthai.price_history_ingest import (
+    ensure_raw_shareprices_daily,
+    resolve_history_tickers,
+    run_price_history_ingest,
+)
+from smartwealthai.simfin_client import configure_simfin
+
+logger = logging.getLogger(__name__)
+
+
+def resolve_tickers(
+    data_dir: Path,
+    *,
+    universe_run_date: date | None,
+    explicit_tickers: tuple[str, ...],
+) -> list[str]:
+    """Resolve ticker scope from CLI flags."""
+    try:
+        return resolve_history_tickers(
+            data_dir,
+            universe_run_date=universe_run_date,
+            explicit_tickers=explicit_tickers,
+        )
+    except (FileNotFoundError, ValueError) as exc:
+        raise click.ClickException(str(exc)) from exc
+
+
+@click.command(context_settings={"help_option_names": ["-h", "--help"]})
+@click.option(
+    "--data-dir",
+    type=click.Path(path_type=Path, file_okay=False),
+    default=Path("data"),
+    show_default=True,
+    help="Data lake root.",
+)
+@click.option(
+    "--start-date",
+    type=click.DateTime(formats=["%Y-%m-%d"]),
+    required=True,
+    help="Inclusive start of the curated price window.",
+)
+@click.option(
+    "--end-date",
+    type=click.DateTime(formats=["%Y-%m-%d"]),
+    required=True,
+    help="Inclusive end of the curated price window.",
+)
+@click.option(
+    "--snapshot-date",
+    type=click.DateTime(formats=["%Y-%m-%d"]),
+    default=lambda: datetime.now(tz=UTC).strftime("%Y-%m-%d"),
+    show_default="today (UTC)",
+    help="Raw SimFin shareprices/daily partition date.",
+)
+@click.option(
+    "--universe-run-date",
+    type=click.DateTime(formats=["%Y-%m-%d"]),
+    default=None,
+    help="Limit to tickers in this curated universe snapshot.",
+)
+@click.option(
+    "--ticker",
+    "tickers",
+    multiple=True,
+    help="Limit to specific tickers (repeatable; intersects universe when both set).",
+)
+@click.option(
+    "--refresh-days",
+    type=int,
+    default=7,
+    show_default=True,
+    help="Skip SimFin re-download when the raw partition is newer than this.",
+)
+@click.option(
+    "--force",
+    is_flag=True,
+    help="Re-download raw daily shareprices and overwrite curated partitions.",
+)
+@click.option(
+    "--skip-download",
+    is_flag=True,
+    help="Use existing raw shareprices/daily only (for tests and offline runs).",
+)
+def main(
+    data_dir: Path,
+    start_date: datetime,
+    end_date: datetime,
+    snapshot_date: datetime,
+    universe_run_date: datetime | None,
+    tickers: tuple[str, ...],
+    refresh_days: int,
+    force: bool,
+    skip_download: bool,
+) -> None:
+    """Populate curated daily prices from SimFin bulk shareprices/daily."""
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+
+    window_start = start_date.date()
+    window_end = end_date.date()
+    if window_start > window_end:
+        raise click.ClickException("--start-date must be on or before --end-date.")
+
+    snapshot = snapshot_date.date()
+    universe_date = universe_run_date.date() if universe_run_date is not None else None
+    ticker_list = resolve_tickers(
+        data_dir,
+        universe_run_date=universe_date,
+        explicit_tickers=tickers,
+    )
+
+    if not skip_download:
+        api_key = os.environ.get("SIMFIN_API_KEY")
+        if not api_key:
+            raise click.ClickException("SIMFIN_API_KEY is not set")
+        configure_simfin(api_key=api_key, cache_dir=data_dir / "cache" / "simfin")
+        try:
+            raw_path = ensure_raw_shareprices_daily(
+                data_dir,
+                snapshot_date=snapshot,
+                refresh_days=refresh_days,
+                force=force,
+            )
+        except OSError as exc:
+            raise click.ClickException(str(exc)) from exc
+        click.echo(f"Raw daily shareprices: {raw_path}")
+
+    try:
+        ingest_run = run_price_history_ingest(
+            data_dir=data_dir,
+            tickers=ticker_list,
+            start_date=window_start,
+            end_date=window_end,
+            snapshot_date=snapshot,
+            force=force,
+        )
+    except FileNotFoundError as exc:
+        raise click.ClickException(str(exc)) from exc
+
+    click.echo(f"Tickers: {len(ticker_list)}")
+    click.echo(f"Rows normalized: {ingest_run.row_count}")
+    click.echo(f"Partitions written: {len(ingest_run.partitions_written)}")
+    click.echo(f"Partitions skipped: {ingest_run.partitions_skipped}")
+    for path in ingest_run.partitions_written[:10]:
+        click.echo(f"  {path}")
+    if len(ingest_run.partitions_written) > 10:
+        logger.warning("... and %d more partitions", len(ingest_run.partitions_written) - 10)
+
+    if ingest_run.row_count == 0:
+        raise SystemExit(1)
+    raise SystemExit(0)
+
+
+def cli_run(argv: list[str] | None = None) -> int:
+    """Invoke the CLI programmatically (e.g. in tests)."""
+    runner = CliRunner()
+    result = runner.invoke(main, argv or [])
+    return result.exit_code
+
+
+if __name__ == "__main__":
+    main()