diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 66a9aeb..bd7d520 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,3 +21,9 @@ jobs: run: python scripts/validate_latest_signal.py examples/latest_signal.example.json - name: Run example overlay replay run: python scripts/backtest_signal_overlay.py --prices examples/price_history.example.csv --signals examples/signal_history --symbol QQQ + - name: Run example price extraction + run: | + python scripts/extract_price_history.py \ + --source examples/price_history.example.csv \ + --target /tmp/qqq_overlay_prices.csv \ + --symbols QQQ diff --git a/README.md b/README.md index f46471c..d2a002d 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,28 @@ python scripts/backtest_signal_overlay.py \ The replay tests a deterministic risk-reducing overlay only. It does not call AI models and does not treat the example as production evidence. +Extract compact real-price input from an existing QuantStrategyLab price file: + +```bash +python scripts/extract_price_history.py \ + --source ../UsEquitySnapshotPipelines/data/output/tqqq_growth_income_real_full_archive_2026-05-26/price_history.csv \ + --target data/input/qqq_price_history.csv \ + --symbols QQQ +``` + +Then replay stored shadow signals against those prices: + +```bash +python scripts/backtest_signal_overlay.py \ + --prices data/input/qqq_price_history.csv \ + --signals data/output/signal_history \ + --symbol QQQ \ + --output data/output/tmp/replay_summary.json +``` + +The price loader accepts both this repository's compact `date,symbol,close` +schema and the existing QuantStrategyLab `symbol,as_of,close` schema. + ## Artifact Contract The latest artifact path is: diff --git a/docs/architecture.md b/docs/architecture.md index f6484f2..f5ab947 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -47,6 +47,11 @@ The first overlay harness intentionally measures only: This is enough to identify whether the stored AI context would have reduced risk or created unacceptable opportunity cost before any runtime integration. +The replay harness can read either compact `date,symbol,close` CSV files or the +existing QuantStrategyLab `symbol,as_of,close` price-history files. Large source +files should stay in their owning strategy repositories or object storage; this +repository only stores small extracted replay inputs when needed for research. + ## Risk Notes The artifact is research evidence, not a trading instruction. Missing evidence, diff --git a/scripts/backtest_signal_overlay.py b/scripts/backtest_signal_overlay.py index 2863d54..b9f5aa1 100644 --- a/scripts/backtest_signal_overlay.py +++ b/scripts/backtest_signal_overlay.py @@ -19,7 +19,11 @@ def main() -> int: parser = argparse.ArgumentParser(description="Replay shadow AI signals as a deterministic risk overlay.") - parser.add_argument("--prices", default="examples/price_history.example.csv", help="CSV with date,symbol,close") + parser.add_argument( + "--prices", + default="examples/price_history.example.csv", + help="CSV with symbol,close and date or as_of", + ) parser.add_argument("--signals", default="examples/signal_history", help="Signal JSON file or directory") parser.add_argument("--symbol", default="QQQ", help="Risk asset symbol to test") parser.add_argument("--min-confidence", type=float, default=0.55) diff --git a/scripts/extract_price_history.py b/scripts/extract_price_history.py new file mode 100644 index 0000000..63e93b8 --- /dev/null +++ b/scripts/extract_price_history.py @@ -0,0 +1,38 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) + +from ai_long_horizon_signal_pipelines.price_history import write_filtered_price_history # noqa: E402 + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Extract a compact date,symbol,close CSV for overlay replay." + ) + parser.add_argument("--source", required=True, help="Input CSV with symbol,close and date or as_of") + parser.add_argument("--target", default="data/input/price_history.csv", help="Output CSV path") + parser.add_argument("--symbols", default="QQQ", help="Comma-separated symbols to keep") + parser.add_argument("--start-date", help="Optional inclusive YYYY-MM-DD lower bound") + parser.add_argument("--end-date", help="Optional inclusive YYYY-MM-DD upper bound") + args = parser.parse_args() + + summary = write_filtered_price_history( + args.source, + args.target, + symbols=args.symbols, + start_date=args.start_date, + end_date=args.end_date, + ) + print(json.dumps(summary.__dict__, ensure_ascii=True, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/ai_long_horizon_signal_pipelines/__init__.py b/src/ai_long_horizon_signal_pipelines/__init__.py index 128ae8d..785cf79 100644 --- a/src/ai_long_horizon_signal_pipelines/__init__.py +++ b/src/ai_long_horizon_signal_pipelines/__init__.py @@ -1,6 +1,14 @@ """Shadow-only long-horizon AI signal artifact helpers.""" from .overlay_backtest import OverlayPolicy, backtest_overlay +from .price_history import PriceExtractionSummary, write_filtered_price_history from .schema import SignalValidationError, validate_signal -__all__ = ["OverlayPolicy", "SignalValidationError", "backtest_overlay", "validate_signal"] +__all__ = [ + "OverlayPolicy", + "PriceExtractionSummary", + "SignalValidationError", + "backtest_overlay", + "validate_signal", + "write_filtered_price_history", +] diff --git a/src/ai_long_horizon_signal_pipelines/overlay_backtest.py b/src/ai_long_horizon_signal_pipelines/overlay_backtest.py index b7c4af5..4f9edfa 100644 --- a/src/ai_long_horizon_signal_pipelines/overlay_backtest.py +++ b/src/ai_long_horizon_signal_pipelines/overlay_backtest.py @@ -1,12 +1,12 @@ from __future__ import annotations -import csv import datetime as dt import json from dataclasses import dataclass from pathlib import Path from typing import Any +from .price_history import parse_price_date, read_price_rows from .schema import validate_signal @@ -33,25 +33,14 @@ class OverlayPolicy: def parse_date(value: str) -> dt.date: - return dt.date.fromisoformat(value) + return parse_price_date(value) def load_price_history(path: Path, *, symbol: str) -> list[PricePoint]: - rows: list[PricePoint] = [] - with path.open(newline="", encoding="utf-8") as handle: - reader = csv.DictReader(handle) - required = {"date", "symbol", "close"} - missing = required.difference(reader.fieldnames or ()) - if missing: - raise ValueError(f"price history missing columns: {', '.join(sorted(missing))}") - for row in reader: - if str(row["symbol"]).strip().upper() != symbol.upper(): - continue - close = float(row["close"]) - if close <= 0: - raise ValueError(f"close must be positive for {symbol} on {row['date']}") - rows.append(PricePoint(date=parse_date(row["date"]), close=close)) - rows.sort(key=lambda item: item.date) + rows = [ + PricePoint(date=row.date, close=row.close) + for row in read_price_rows(path, symbols=[symbol]) + ] if len(rows) < 2: raise ValueError(f"price history for {symbol} requires at least two rows") return rows diff --git a/src/ai_long_horizon_signal_pipelines/price_history.py b/src/ai_long_horizon_signal_pipelines/price_history.py new file mode 100644 index 0000000..6516b2e --- /dev/null +++ b/src/ai_long_horizon_signal_pipelines/price_history.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +import csv +import datetime as dt +from dataclasses import dataclass +from pathlib import Path + + +DATE_COLUMNS = ("date", "as_of") + + +@dataclass(frozen=True) +class PriceRow: + date: dt.date + symbol: str + close: float + + +@dataclass(frozen=True) +class PriceExtractionSummary: + source: str + target: str + input_rows: int + output_rows: int + symbols: list[str] + start_date: str | None + end_date: str | None + + +def parse_price_date(value: object) -> dt.date: + text = str(value or "").strip() + if not text: + raise ValueError("price date is required") + if text.endswith("Z"): + text = f"{text[:-1]}+00:00" + try: + return dt.date.fromisoformat(text) + except ValueError: + return dt.datetime.fromisoformat(text).date() + + +def _resolve_date_column(fieldnames: list[str] | None) -> str: + fields = set(fieldnames or []) + for column in DATE_COLUMNS: + if column in fields: + return column + expected = " or ".join(DATE_COLUMNS) + raise ValueError(f"price history missing date column: expected {expected}") + + +def _normalize_symbols(symbols: list[str] | tuple[str, ...] | set[str] | str) -> list[str]: + if isinstance(symbols, str): + raw_symbols = symbols.split(",") + else: + raw_symbols = list(symbols) + normalized: list[str] = [] + for symbol in raw_symbols: + text = str(symbol or "").strip().upper() + if text and text not in normalized: + normalized.append(text) + if not normalized: + raise ValueError("at least one symbol is required") + return normalized + + +def read_price_rows( + path: str | Path, + *, + symbols: list[str] | tuple[str, ...] | set[str] | str, + start_date: str | None = None, + end_date: str | None = None, +) -> list[PriceRow]: + selected_symbols = set(_normalize_symbols(symbols)) + start = parse_price_date(start_date) if start_date else None + end = parse_price_date(end_date) if end_date else None + rows_by_key: dict[tuple[dt.date, str], PriceRow] = {} + + with Path(path).open(newline="", encoding="utf-8") as handle: + reader = csv.DictReader(handle) + required = {"symbol", "close"} + missing = required.difference(reader.fieldnames or ()) + if missing: + raise ValueError(f"price history missing columns: {', '.join(sorted(missing))}") + date_column = _resolve_date_column(reader.fieldnames) + + for raw_row in reader: + symbol = str(raw_row["symbol"]).strip().upper() + if symbol not in selected_symbols: + continue + row_date = parse_price_date(raw_row[date_column]) + if start and row_date < start: + continue + if end and row_date > end: + continue + close = float(raw_row["close"]) + if close <= 0: + raise ValueError(f"close must be positive for {symbol} on {row_date.isoformat()}") + rows_by_key[(row_date, symbol)] = PriceRow(date=row_date, symbol=symbol, close=close) + + return [rows_by_key[key] for key in sorted(rows_by_key)] + + +def write_filtered_price_history( + source: str | Path, + target: str | Path, + *, + symbols: list[str] | tuple[str, ...] | set[str] | str, + start_date: str | None = None, + end_date: str | None = None, +) -> PriceExtractionSummary: + selected_symbols = _normalize_symbols(symbols) + start = parse_price_date(start_date) if start_date else None + end = parse_price_date(end_date) if end_date else None + rows_by_key: dict[tuple[dt.date, str], PriceRow] = {} + input_rows = 0 + + with Path(source).open(newline="", encoding="utf-8") as handle: + reader = csv.DictReader(handle) + required = {"symbol", "close"} + missing = required.difference(reader.fieldnames or ()) + if missing: + raise ValueError(f"price history missing columns: {', '.join(sorted(missing))}") + date_column = _resolve_date_column(reader.fieldnames) + + for raw_row in reader: + input_rows += 1 + symbol = str(raw_row["symbol"]).strip().upper() + if symbol not in selected_symbols: + continue + row_date = parse_price_date(raw_row[date_column]) + if start and row_date < start: + continue + if end and row_date > end: + continue + close = float(raw_row["close"]) + if close <= 0: + raise ValueError(f"close must be positive for {symbol} on {row_date.isoformat()}") + rows_by_key[(row_date, symbol)] = PriceRow(date=row_date, symbol=symbol, close=close) + + output_path = Path(target) + output_path.parent.mkdir(parents=True, exist_ok=True) + rows = [rows_by_key[key] for key in sorted(rows_by_key)] + with output_path.open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=["date", "symbol", "close"]) + writer.writeheader() + for row in rows: + writer.writerow({"date": row.date.isoformat(), "symbol": row.symbol, "close": row.close}) + + return PriceExtractionSummary( + source=str(source), + target=str(target), + input_rows=input_rows, + output_rows=len(rows), + symbols=selected_symbols, + start_date=start.isoformat() if start else None, + end_date=end.isoformat() if end else None, + ) diff --git a/tests/test_overlay_backtest.py b/tests/test_overlay_backtest.py index 2c5fa7e..812d47f 100644 --- a/tests/test_overlay_backtest.py +++ b/tests/test_overlay_backtest.py @@ -45,3 +45,24 @@ def test_backtest_overlay_reduces_drawdown_on_synthetic_path() -> None: assert summary["overlay"]["max_drawdown"] > summary["baseline"]["max_drawdown"] assert 0 < summary["overlay"]["avg_exposure"] <= 1.0 assert summary["overlay"]["turnover"] > 0 + + +def test_load_price_history_accepts_quant_strategy_as_of_schema(tmp_path) -> None: + prices_path = tmp_path / "prices.csv" + prices_path.write_text( + "\n".join( + [ + "symbol,as_of,close,volume", + "QQQ,2026-01-02,100,1000", + "SPY,2026-01-02,90,1000", + "QQQ,2026-01-05,101,1000", + ] + ) + + "\n", + encoding="utf-8", + ) + + prices = load_price_history(prices_path, symbol="QQQ") + + assert [price.date.isoformat() for price in prices] == ["2026-01-02", "2026-01-05"] + assert [price.close for price in prices] == [100.0, 101.0] diff --git a/tests/test_price_history.py b/tests/test_price_history.py new file mode 100644 index 0000000..22efa24 --- /dev/null +++ b/tests/test_price_history.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +import csv + +from ai_long_horizon_signal_pipelines.price_history import ( + parse_price_date, + read_price_rows, + write_filtered_price_history, +) + + +def test_parse_price_date_accepts_timestamp_text() -> None: + assert parse_price_date("2026-01-02 00:00:00").isoformat() == "2026-01-02" + assert parse_price_date("2026-01-02T21:00:00Z").isoformat() == "2026-01-02" + + +def test_write_filtered_price_history_extracts_compact_schema(tmp_path) -> None: + source = tmp_path / "source.csv" + source.write_text( + "\n".join( + [ + "symbol,as_of,close,open", + "QQQ,2026-01-02,100,99", + "SPY,2026-01-02,90,89", + "QQQ,2026-01-05,101,100", + "QQQ,2026-01-06,102,101", + ] + ) + + "\n", + encoding="utf-8", + ) + target = tmp_path / "out" / "prices.csv" + + summary = write_filtered_price_history( + source, + target, + symbols="qqq", + start_date="2026-01-03", + end_date="2026-01-06", + ) + + with target.open(newline="", encoding="utf-8") as handle: + rows = list(csv.DictReader(handle)) + + assert summary.input_rows == 4 + assert summary.output_rows == 2 + assert summary.symbols == ["QQQ"] + assert rows == [ + {"date": "2026-01-05", "symbol": "QQQ", "close": "101.0"}, + {"date": "2026-01-06", "symbol": "QQQ", "close": "102.0"}, + ] + + +def test_read_price_rows_keeps_last_duplicate_date_symbol(tmp_path) -> None: + source = tmp_path / "source.csv" + source.write_text( + "\n".join( + [ + "date,symbol,close", + "2026-01-02,QQQ,100", + "2026-01-02,QQQ,101", + "2026-01-05,QQQ,102", + ] + ) + + "\n", + encoding="utf-8", + ) + + rows = read_price_rows(source, symbols=["QQQ"]) + + assert [row.close for row in rows] == [101.0, 102.0]