Add real price extraction for overlay replay (#2)

Author: Pigbibi

.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

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:

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,

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)

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())

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",
+]

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

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,
+    )

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]