diff --git a/pyproject.toml b/pyproject.toml index aed564a..53da745 100644 --- a/pyproject.toml +++ b/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" diff --git a/spec/features/006-etl-data-lake/spec.md b/spec/features/006-etl-data-lake/spec.md index 55e8f6a..97c869e 100644 --- a/spec/features/006-etl-data-lake/spec.md +++ b/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=/year=/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=/year=/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). diff --git a/spec/guides/download-simfin.md b/spec/guides/download-simfin.md index 999505f..ae808dc 100644 --- a/spec/guides/download-simfin.md +++ b/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=/us-shareprices-daily.csv` | +| Normalize + partition | `curated/prices/ticker=/year=/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. diff --git a/src/smartwealthai/download_price_history.py b/src/smartwealthai/download_price_history.py new file mode 100644 index 0000000..3fb455b --- /dev/null +++ b/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() diff --git a/src/smartwealthai/price_history_ingest.py b/src/smartwealthai/price_history_ingest.py new file mode 100644 index 0000000..39e755f --- /dev/null +++ b/src/smartwealthai/price_history_ingest.py @@ -0,0 +1,233 @@ +"""Phase-2 daily price history from SimFin bulk shareprices/daily.""" + +from __future__ import annotations + +import shutil +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import date +from pathlib import Path + +import pandas as pd + +from smartwealthai.download_simfin import should_skip_dataset +from smartwealthai.lake_paths import curated_prices_path, simfin_bulk_path +from smartwealthai.price_ingest import load_universe_tickers, should_skip_artifact +from smartwealthai.simfin_client import fetch_dataset_csv + +CURATED_DAILY_PRICE_COLUMNS: tuple[str, ...] = ( + "ticker", + "price_date", + "close", + "adj_close", + "volume", +) + + +@dataclass +class PriceHistoryIngestRun: + """Outcome of one daily price history ingest run.""" + + partitions_written: list[Path] = field(default_factory=list) + partitions_skipped: int = 0 + row_count: int = 0 + + +def shareprices_daily_raw_path(data_dir: Path, *, snapshot_date: date) -> Path: + """Return the raw SimFin shareprices/daily lake path for ``snapshot_date``.""" + return simfin_bulk_path( + data_dir, + dataset="shareprices", + variant="daily", + market="us", + as_of_date=snapshot_date, + ) + + +def load_raw_shareprices_daily(data_dir: Path, *, snapshot_date: date) -> pd.DataFrame: + """Load the verbatim SimFin shareprices/daily CSV from the raw lake.""" + path = shareprices_daily_raw_path(data_dir, snapshot_date=snapshot_date) + if not path.exists(): + msg = f"SimFin shareprices/daily not found: {path}" + raise FileNotFoundError(msg) + return pd.read_csv(path, sep=";") + + +def ensure_raw_shareprices_daily( + data_dir: Path, + *, + snapshot_date: date, + refresh_days: int, + force: bool, + fetch_csv: Callable[..., Path] | None = None, +) -> Path: + """Download SimFin daily shareprices into the raw lake when missing or stale.""" + downloader = fetch_csv if fetch_csv is not None else fetch_dataset_csv + lake_path = shareprices_daily_raw_path(data_dir, snapshot_date=snapshot_date) + if should_skip_dataset(lake_path, refresh_days=refresh_days, force=force): + return lake_path + + source = downloader( + dataset="shareprices", + variant="daily", + market="us", + refresh_days=0 if force else refresh_days, + ) + lake_path.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, lake_path) + return lake_path + + +def build_daily_price_rows( + shareprices: pd.DataFrame, + tickers: list[str], + *, + start_date: date, + end_date: date, +) -> list[dict[str, object]]: + """Filter SimFin daily rows to ``tickers`` and the inclusive date window.""" + if shareprices.empty or not tickers: + return [] + + ticker_set = {ticker.upper() for ticker in tickers} + frame = shareprices.copy() + frame["ticker"] = frame["Ticker"].astype(str).str.upper() + frame["price_day"] = pd.to_datetime(frame["Date"]).dt.date + + eligible = frame.loc[ + frame["ticker"].isin(ticker_set) + & (frame["price_day"] >= start_date) + & (frame["price_day"] <= end_date) + ] + if eligible.empty: + return [] + + rows: list[dict[str, object]] = [] + for _, row in eligible.sort_values(["ticker", "price_day"]).iterrows(): + rows.append( + { + "ticker": row["ticker"], + "price_date": row["price_day"].isoformat(), + "close": float(row["Close"]), + "adj_close": float(row["Adj. Close"]), + "volume": float(row["Volume"]), + } + ) + return rows + + +def write_curated_daily_prices( + data_dir: Path, + rows: list[dict[str, object]], + *, + force: bool, +) -> tuple[list[Path], int]: + """Write daily price rows partitioned by ticker and calendar year.""" + if not rows: + return [], 0 + + frame = pd.DataFrame(rows, columns=list(CURATED_DAILY_PRICE_COLUMNS)) + frame["price_day"] = pd.to_datetime(frame["price_date"]).dt.date + frame["year"] = pd.to_datetime(frame["price_date"]).dt.year + + written: list[Path] = [] + skipped = 0 + for (ticker, year), group in frame.groupby(["ticker", "year"], sort=True): + path = curated_prices_path(data_dir, ticker=str(ticker), year=int(year)) + if should_skip_artifact(path, force=force): + skipped += 1 + continue + partition = group.drop(columns=["price_day", "year"]).reset_index(drop=True) + path.parent.mkdir(parents=True, exist_ok=True) + partition.to_parquet(path, index=False) + written.append(path) + + return written, skipped + + +def _load_daily_prices_for_years( + data_dir: Path, + *, + ticker: str, + years: tuple[int, ...], +) -> pd.DataFrame: + frames: list[pd.DataFrame] = [] + normalized = ticker.upper() + for year in years: + path = curated_prices_path(data_dir, ticker=normalized, year=year) + if path.exists(): + frames.append(pd.read_parquet(path)) + if not frames: + return pd.DataFrame(columns=list(CURATED_DAILY_PRICE_COLUMNS)) + return pd.concat(frames, ignore_index=True) + + +def lookup_daily_adj_close( + data_dir: Path, + *, + ticker: str, + as_of_date: date, +) -> float: + """Return ``adj_close`` on the latest trading day on or before ``as_of_date``.""" + # ponytail: reads at most two year partitions; upgrade path is DuckDB over curated/prices + years = (as_of_date.year - 1, as_of_date.year) + frame = _load_daily_prices_for_years(data_dir, ticker=ticker, years=years) + if frame.empty: + msg = f"No daily price history for {ticker!r}" + raise LookupError(msg) + + work = frame.copy() + work["price_day"] = pd.to_datetime(work["price_date"]).dt.date + eligible = work.loc[work["price_day"] <= as_of_date] + if eligible.empty: + msg = f"No daily price for {ticker!r} on or before {as_of_date}" + raise LookupError(msg) + + latest = eligible.sort_values("price_day").iloc[-1] + return float(latest["adj_close"]) + + +def run_price_history_ingest( + *, + data_dir: Path, + tickers: list[str], + start_date: date, + end_date: date, + snapshot_date: date, + force: bool = False, +) -> PriceHistoryIngestRun: + """Normalize SimFin shareprices/daily into curated ticker/year partitions.""" + shareprices = load_raw_shareprices_daily(data_dir, snapshot_date=snapshot_date) + rows = build_daily_price_rows( + shareprices, + tickers, + start_date=start_date, + end_date=end_date, + ) + written, skipped = write_curated_daily_prices(data_dir, rows, force=force) + return PriceHistoryIngestRun( + partitions_written=written, + partitions_skipped=skipped, + row_count=len(rows), + ) + + +def resolve_history_tickers( + data_dir: Path, + *, + universe_run_date: date | None, + explicit_tickers: tuple[str, ...], +) -> list[str]: + """Merge universe and explicit ticker filters for price history ingest.""" + ticker_set: set[str] | None = None + if universe_run_date is not None: + ticker_set = { + ticker.upper() for ticker in load_universe_tickers(data_dir, run_date=universe_run_date) + } + if explicit_tickers: + explicit = {ticker.upper() for ticker in explicit_tickers} + ticker_set = explicit if ticker_set is None else ticker_set & explicit + if ticker_set is None: + msg = "Pass universe_run_date or explicit_tickers to limit scope." + raise ValueError(msg) + return sorted(ticker_set) diff --git a/tests/fixtures/lake/MANIFEST.json b/tests/fixtures/lake/MANIFEST.json index 1172cdf..d35fec2 100644 --- a/tests/fixtures/lake/MANIFEST.json +++ b/tests/fixtures/lake/MANIFEST.json @@ -34,6 +34,7 @@ "raw/sec_edgar/cik=0000320193/endpoint=companyfacts/as_of_date=2026-05-21/response.json": "3f50c80631d8e18a40fc295a7933f3cea15e4a0e3d58e48ba210d633c593cb58", "raw/sec_edgar/cik=0000789019/endpoint=companyfacts/as_of_date=2026-05-21/response.json": "00dc380deb4359e5dedd3413d61a3d6f26cd0a71bda06badaba6f0fa8ced9dc0", "raw/yfinance/ticker=AAPL/endpoint=history/as_of_date=2026-05-21/history.json": "208c3352092938ed8addeea572a3ca972f5cb79ded68e83de0898d847dd6ea73", - "raw/yfinance/ticker=MSFT/endpoint=history/as_of_date=2026-05-21/history.json": "6eee67dec0113236a98fa61a4342626df2731d0d427585ab58b8a545f8292c70" + "raw/yfinance/ticker=MSFT/endpoint=history/as_of_date=2026-05-21/history.json": "6eee67dec0113236a98fa61a4342626df2731d0d427585ab58b8a545f8292c70", + "raw/simfin/dataset=shareprices/variant=daily/market=us/as_of_date=2026-06-18/us-shareprices-daily.csv": "097a9cfafc8d75e6aca48d1041d349ff4746b70bd0de5aab1fad15af3fbf2413" } } \ No newline at end of file diff --git a/tests/fixtures/lake/README.md b/tests/fixtures/lake/README.md index 42e269a..82a5c77 100644 --- a/tests/fixtures/lake/README.md +++ b/tests/fixtures/lake/README.md @@ -7,7 +7,7 @@ SimFin, and yfinance), then reduced and committed for hermetic tests. ## Directory layout - `raw/sec_edgar/...`: reduced real SEC companyfacts snapshots -- `raw/simfin/...`: reduced SimFin bulk CSV snapshots (income TTM, balance quarterly, companies, industries, shareprices/latest) +- `raw/simfin/...`: reduced SimFin bulk CSV snapshots (income TTM, balance quarterly, companies, industries, shareprices/latest, shareprices/daily) - `raw/yfinance/...`: reduced real yfinance history snapshots - `curated/fundamentals.csv`: stable derived dataset for PIT tests - `MANIFEST.json`: provenance and SHA-256 checksums @@ -50,6 +50,7 @@ against yfinance history snapshots: - `raw/sec_edgar/cik=0000320193/endpoint=companyfacts/as_of_date=2026-05-21/response.json` - `raw/sec_edgar/cik=0000789019/endpoint=companyfacts/as_of_date=2026-05-21/response.json` - `raw/simfin/dataset=shareprices/variant=latest/market=us/as_of_date=2026-06-18/us-shareprices-latest.csv` +- `raw/simfin/dataset=shareprices/variant=daily/market=us/as_of_date=2026-06-18/us-shareprices-daily.csv` - `raw/yfinance/ticker=AAPL/endpoint=history/as_of_date=2026-05-21/history.json` - `raw/yfinance/ticker=MSFT/endpoint=history/as_of_date=2026-05-21/history.json` diff --git a/tests/fixtures/lake/raw/simfin/dataset=shareprices/variant=daily/market=us/as_of_date=2026-06-18/us-shareprices-daily.csv b/tests/fixtures/lake/raw/simfin/dataset=shareprices/variant=daily/market=us/as_of_date=2026-06-18/us-shareprices-daily.csv new file mode 100644 index 0000000..c082f11 --- /dev/null +++ b/tests/fixtures/lake/raw/simfin/dataset=shareprices/variant=daily/market=us/as_of_date=2026-06-18/us-shareprices-daily.csv @@ -0,0 +1,7 @@ +Ticker;Date;Open;High;Low;Close;Adj. Close;Volume +AAPL;2025-12-31;270.0;275.0;269.0;274.0;270.0;50000000 +AAPL;2026-01-02;275.0;280.0;274.0;278.0;274.0;48000000 +AAPL;2026-06-16;268.0;272.0;267.0;271.0;270.5;45000000 +AAPL;2026-06-17;270.0;275.0;269.0;274.0;273.5;50000000 +MSFT;2025-12-31;410.0;425.0;408.0;420.0;45.0;30000000 +MSFT;2026-06-17;420.0;425.0;418.0;423.0;50.0;30000000 diff --git a/tests/test_price_history_ingest.py b/tests/test_price_history_ingest.py new file mode 100644 index 0000000..57ad1ed --- /dev/null +++ b/tests/test_price_history_ingest.py @@ -0,0 +1,459 @@ +"""Hermetic tests for SimFin shareprices/daily price history ingest (issue #88).""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import sys +from datetime import date +from pathlib import Path +from unittest.mock import patch + +import pandas as pd +import pytest +from click.testing import CliRunner + +from smartwealthai.download_price_history import cli_run, main +from smartwealthai.lake_paths import curated_prices_path +from smartwealthai.price_history_ingest import ( + CURATED_DAILY_PRICE_COLUMNS, + PriceHistoryIngestRun, + build_daily_price_rows, + ensure_raw_shareprices_daily, + load_raw_shareprices_daily, + lookup_daily_adj_close, + resolve_history_tickers, + run_price_history_ingest, + shareprices_daily_raw_path, + write_curated_daily_prices, +) +from smartwealthai.universe_builder import build_universe + +FIXTURE_LAKE = Path("tests/fixtures/lake") +RUN_DATE = date(2026, 6, 18) +SNAPSHOT_DATE = date(2026, 6, 18) +START_DATE = date(2026, 1, 1) +END_DATE = date(2026, 6, 30) + + +@pytest.fixture +def lake(tmp_path: Path) -> Path: + """Copy fixture lake and build a demo universe snapshot.""" + shutil.copytree(FIXTURE_LAKE, tmp_path / "lake") + root = tmp_path / "lake" + build_universe(root, run_date=RUN_DATE, snapshot_date=SNAPSHOT_DATE) + return root + + +def test_shareprices_daily_raw_path_matches_spec_layout() -> None: + path = shareprices_daily_raw_path(Path("data"), snapshot_date=date(2026, 6, 18)) + assert path == Path( + "data/raw/simfin/dataset=shareprices/variant=daily/market=us/" + "as_of_date=2026-06-18/us-shareprices-daily.csv" + ) + + +def test_load_raw_shareprices_daily_reads_fixture(lake: Path) -> None: + frame = load_raw_shareprices_daily(lake, snapshot_date=SNAPSHOT_DATE) + assert "AAPL" in frame["Ticker"].values + assert "MSFT" in frame["Ticker"].values + + +def test_build_daily_price_rows_filters_tickers_and_date_range(lake: Path) -> None: + shareprices = load_raw_shareprices_daily(lake, snapshot_date=SNAPSHOT_DATE) + rows = build_daily_price_rows( + shareprices, + ["AAPL", "MISSING"], + start_date=START_DATE, + end_date=END_DATE, + ) + + assert len(rows) == 3 + assert {row["ticker"] for row in rows} == {"AAPL"} + assert rows[-1]["adj_close"] == pytest.approx(273.5) + + +def test_write_curated_daily_prices_partitions_by_ticker_year(lake: Path) -> None: + shareprices = load_raw_shareprices_daily(lake, snapshot_date=SNAPSHOT_DATE) + rows = build_daily_price_rows( + shareprices, + ["AAPL", "MSFT"], + start_date=date(2025, 1, 1), + end_date=END_DATE, + ) + written, skipped = write_curated_daily_prices(lake, rows, force=False) + + assert skipped == 0 + assert len(written) == 4 + aapl_2026 = curated_prices_path(lake, ticker="AAPL", year=2026) + assert aapl_2026 in written + prices = pd.read_parquet(aapl_2026) + assert list(prices.columns) == list(CURATED_DAILY_PRICE_COLUMNS) + assert len(prices) == 3 + + +def test_write_curated_daily_prices_skips_existing_partitions(lake: Path) -> None: + shareprices = load_raw_shareprices_daily(lake, snapshot_date=SNAPSHOT_DATE) + rows = build_daily_price_rows( + shareprices, + ["AAPL"], + start_date=START_DATE, + end_date=END_DATE, + ) + write_curated_daily_prices(lake, rows, force=False) + _, skipped = write_curated_daily_prices(lake, rows, force=False) + assert skipped == 1 + + +def test_lookup_daily_adj_close_returns_latest_on_or_before_date(lake: Path) -> None: + run_price_history_ingest( + data_dir=lake, + tickers=["AAPL"], + start_date=date(2025, 1, 1), + end_date=END_DATE, + snapshot_date=SNAPSHOT_DATE, + ) + + assert lookup_daily_adj_close( + lake, ticker="AAPL", as_of_date=date(2026, 6, 17) + ) == pytest.approx(273.5) + assert lookup_daily_adj_close( + lake, ticker="AAPL", as_of_date=date(2026, 6, 18) + ) == pytest.approx(273.5) + + +def test_lookup_daily_adj_close_raises_when_no_history(lake: Path) -> None: + with pytest.raises(LookupError, match="No daily price history"): + lookup_daily_adj_close(lake, ticker="AAPL", as_of_date=RUN_DATE) + + +def test_run_price_history_ingest_end_to_end(lake: Path) -> None: + result = run_price_history_ingest( + data_dir=lake, + tickers=["AAPL", "MSFT"], + start_date=date(2025, 1, 1), + end_date=END_DATE, + snapshot_date=SNAPSHOT_DATE, + ) + + assert result.row_count == 6 + assert len(result.partitions_written) == 4 + assert curated_prices_path(lake, ticker="MSFT", year=2025).exists() + + +def test_cli_download_price_history_end_to_end(lake: Path) -> None: + exit_code = cli_run( + [ + "--data-dir", + str(lake), + "--universe-run-date", + RUN_DATE.isoformat(), + "--start-date", + START_DATE.isoformat(), + "--end-date", + END_DATE.isoformat(), + "--snapshot-date", + SNAPSHOT_DATE.isoformat(), + "--skip-download", + ] + ) + + assert exit_code == 0 + assert curated_prices_path(lake, ticker="AAPL", year=2026).exists() + + +def test_cli_fails_when_raw_daily_missing(lake: Path) -> None: + raw_path = shareprices_daily_raw_path(lake, snapshot_date=SNAPSHOT_DATE) + raw_path.unlink() + + exit_code = cli_run( + [ + "--data-dir", + str(lake), + "--ticker", + "AAPL", + "--start-date", + START_DATE.isoformat(), + "--end-date", + END_DATE.isoformat(), + "--snapshot-date", + SNAPSHOT_DATE.isoformat(), + "--skip-download", + ] + ) + + assert exit_code == 1 + + +def test_cli_downloads_raw_when_not_skipped(lake: Path, tmp_path: Path) -> None: + raw_path = shareprices_daily_raw_path(lake, snapshot_date=SNAPSHOT_DATE) + fixture_csv = raw_path.read_text() + raw_path.unlink() + + def fake_fetch(**_kwargs): + dest = tmp_path / "us-shareprices-daily.csv" + dest.write_text(fixture_csv) + return dest + + with patch.dict(os.environ, {"SIMFIN_API_KEY": "test-key"}): + with patch( + "smartwealthai.price_history_ingest.fetch_dataset_csv", + side_effect=fake_fetch, + ): + exit_code = cli_run( + [ + "--data-dir", + str(lake), + "--ticker", + "AAPL", + "--start-date", + START_DATE.isoformat(), + "--end-date", + END_DATE.isoformat(), + "--snapshot-date", + SNAPSHOT_DATE.isoformat(), + ] + ) + + assert exit_code == 0 + assert raw_path.exists() + + +def test_main_module_entrypoint(lake: Path) -> None: + result = subprocess.run( + [ + sys.executable, + "-m", + "smartwealthai.download_price_history", + "--data-dir", + str(lake), + "--ticker", + "AAPL", + "--start-date", + START_DATE.isoformat(), + "--end-date", + END_DATE.isoformat(), + "--snapshot-date", + SNAPSHOT_DATE.isoformat(), + "--skip-download", + ], + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0 + assert curated_prices_path(lake, ticker="AAPL", year=2026).exists() + + +def test_build_daily_price_rows_returns_empty_for_edge_cases(lake: Path) -> None: + shareprices = load_raw_shareprices_daily(lake, snapshot_date=SNAPSHOT_DATE) + + assert build_daily_price_rows(shareprices, [], start_date=START_DATE, end_date=END_DATE) == [] + assert ( + build_daily_price_rows(pd.DataFrame(), ["AAPL"], start_date=START_DATE, end_date=END_DATE) + == [] + ) + assert ( + build_daily_price_rows( + shareprices, + ["MISSING"], + start_date=START_DATE, + end_date=END_DATE, + ) + == [] + ) + + +def test_write_curated_daily_prices_empty_rows(lake: Path) -> None: + written, skipped = write_curated_daily_prices(lake, [], force=False) + assert written == [] + assert skipped == 0 + + +def test_ensure_raw_shareprices_daily_skips_fresh_copy(lake: Path) -> None: + raw_path = shareprices_daily_raw_path(lake, snapshot_date=SNAPSHOT_DATE) + assert raw_path.exists() + + def fail_fetch(**_kwargs): + msg = "network should not be called when raw partition is fresh" + raise AssertionError(msg) + + result = ensure_raw_shareprices_daily( + lake, + snapshot_date=SNAPSHOT_DATE, + refresh_days=7, + force=False, + fetch_csv=fail_fetch, + ) + assert result == raw_path + + +def test_lookup_daily_adj_close_raises_when_as_of_before_window(lake: Path) -> None: + run_price_history_ingest( + data_dir=lake, + tickers=["AAPL"], + start_date=date(2025, 1, 1), + end_date=END_DATE, + snapshot_date=SNAPSHOT_DATE, + ) + + with pytest.raises(LookupError, match="on or before"): + lookup_daily_adj_close(lake, ticker="AAPL", as_of_date=date(2025, 1, 1)) + + +def test_resolve_history_tickers_requires_scope(lake: Path) -> None: + with pytest.raises(ValueError, match="universe_run_date or explicit_tickers"): + resolve_history_tickers(lake, universe_run_date=None, explicit_tickers=()) + + +def test_cli_rejects_inverted_date_range(lake: Path) -> None: + runner = CliRunner() + result = runner.invoke( + main, + [ + "--data-dir", + str(lake), + "--ticker", + "AAPL", + "--start-date", + "2026-06-30", + "--end-date", + "2026-01-01", + "--snapshot-date", + SNAPSHOT_DATE.isoformat(), + "--skip-download", + ], + ) + assert result.exit_code != 0 + assert "--start-date must be on or before --end-date" in result.output + + +def test_cli_requires_simfin_api_key_when_downloading(lake: Path) -> None: + raw_path = shareprices_daily_raw_path(lake, snapshot_date=SNAPSHOT_DATE) + raw_path.unlink() + + with patch.dict(os.environ, {}, clear=True): + runner = CliRunner() + result = runner.invoke( + main, + [ + "--data-dir", + str(lake), + "--ticker", + "AAPL", + "--start-date", + START_DATE.isoformat(), + "--end-date", + END_DATE.isoformat(), + "--snapshot-date", + SNAPSHOT_DATE.isoformat(), + ], + ) + + assert result.exit_code != 0 + assert "SIMFIN_API_KEY is not set" in result.output + + +def test_cli_resolve_tickers_missing_universe(lake: Path) -> None: + runner = CliRunner() + result = runner.invoke( + main, + [ + "--data-dir", + str(lake), + "--universe-run-date", + "2099-01-01", + "--start-date", + START_DATE.isoformat(), + "--end-date", + END_DATE.isoformat(), + "--snapshot-date", + SNAPSHOT_DATE.isoformat(), + "--skip-download", + ], + ) + assert result.exit_code != 0 + assert "Universe not found" in result.output + + +def test_cli_handles_download_os_error(lake: Path) -> None: + with patch.dict(os.environ, {"SIMFIN_API_KEY": "test-key"}): + with patch( + "smartwealthai.download_price_history.ensure_raw_shareprices_daily", + side_effect=OSError("disk full"), + ): + runner = CliRunner() + result = runner.invoke( + main, + [ + "--data-dir", + str(lake), + "--ticker", + "AAPL", + "--start-date", + START_DATE.isoformat(), + "--end-date", + END_DATE.isoformat(), + "--snapshot-date", + SNAPSHOT_DATE.isoformat(), + ], + ) + + assert result.exit_code != 0 + assert "disk full" in result.output + + +def test_cli_fails_when_no_rows_normalized(lake: Path) -> None: + exit_code = cli_run( + [ + "--data-dir", + str(lake), + "--ticker", + "MISSING", + "--start-date", + START_DATE.isoformat(), + "--end-date", + END_DATE.isoformat(), + "--snapshot-date", + SNAPSHOT_DATE.isoformat(), + "--skip-download", + ] + ) + assert exit_code == 1 + + +def test_cli_warns_when_many_partitions_written( + lake: Path, caplog: pytest.LogCaptureFixture +) -> None: + many_paths = [curated_prices_path(lake, ticker=f"T{i:02d}", year=2026) for i in range(11)] + fake_run = PriceHistoryIngestRun( + partitions_written=many_paths, + partitions_skipped=0, + row_count=11, + ) + + with patch( + "smartwealthai.download_price_history.run_price_history_ingest", + return_value=fake_run, + ): + with caplog.at_level("WARNING"): + exit_code = cli_run( + [ + "--data-dir", + str(lake), + "--ticker", + "AAPL", + "--start-date", + START_DATE.isoformat(), + "--end-date", + END_DATE.isoformat(), + "--snapshot-date", + SNAPSHOT_DATE.isoformat(), + "--skip-download", + ] + ) + + assert exit_code == 0 + assert "... and 1 more partitions" in caplog.text