diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9fa3c6f..66a9aeb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,3 +19,5 @@ jobs: run: python -m pytest -q - name: Validate example signal 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 diff --git a/README.md b/README.md index eca393c..f46471c 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,18 @@ Validate the promoted latest artifact when it exists: python scripts/validate_latest_signal.py ``` +Run the synthetic overlay replay: + +```bash +python scripts/backtest_signal_overlay.py \ + --prices examples/price_history.example.csv \ + --signals examples/signal_history \ + --symbol QQQ +``` + +The replay tests a deterministic risk-reducing overlay only. It does not call +AI models and does not treat the example as production evidence. + ## Artifact Contract The latest artifact path is: @@ -87,3 +99,16 @@ data/output/signal_history/YYYY-MM-DD.json All artifacts must remain shadow-only. They cannot encode broker orders, target quantities, or live allocation overrides. + +## Replay Contract + +Historical validation should replay stored signal artifacts instead of asking a +model to re-create old judgments. The current example policy is intentionally +conservative: + +- no active signal: keep baseline exposure +- `confidence < 0.55`: no-op +- `risk_off`: reduce exposure to `0.5` +- `mixed`: reduce exposure to `0.8` +- severe risk flags such as `liquidity_stress` cap exposure at `0.6` +- the overlay never increases exposure above the baseline diff --git a/docs/architecture.md b/docs/architecture.md index d4e454b..f6484f2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -31,9 +31,21 @@ order routing. ## Validation Strategy -The current minimum check is schema validation for `latest_signal.json`. Future -promotion should add replay tests that consume stored artifacts without calling -model APIs. +The current minimum checks are schema validation and deterministic overlay +replay. Replay consumes stored signal artifacts from `signal_history` and maps +them through a fixed risk-reducing policy. It must never ask a model to recreate +old judgments. + +The first overlay harness intentionally measures only: + +- final equity +- total return +- maximum drawdown +- average exposure +- exposure turnover + +This is enough to identify whether the stored AI context would have reduced +risk or created unacceptable opportunity cost before any runtime integration. ## Risk Notes diff --git a/examples/price_history.example.csv b/examples/price_history.example.csv new file mode 100644 index 0000000..0f5370d --- /dev/null +++ b/examples/price_history.example.csv @@ -0,0 +1,19 @@ +date,symbol,close +2026-01-02,QQQ,100.00 +2026-01-09,QQQ,104.00 +2026-01-16,QQQ,108.00 +2026-01-23,QQQ,106.00 +2026-01-30,QQQ,110.00 +2026-02-06,QQQ,112.00 +2026-02-13,QQQ,109.00 +2026-02-20,QQQ,105.00 +2026-02-27,QQQ,101.00 +2026-03-06,QQQ,96.00 +2026-03-13,QQQ,94.00 +2026-03-20,QQQ,98.00 +2026-03-27,QQQ,102.00 +2026-04-03,QQQ,105.00 +2026-04-10,QQQ,108.00 +2026-04-17,QQQ,111.00 +2026-04-24,QQQ,113.00 +2026-05-01,QQQ,115.00 diff --git a/examples/signal_history/2026-01-02.json b/examples/signal_history/2026-01-02.json new file mode 100644 index 0000000..172b9e2 --- /dev/null +++ b/examples/signal_history/2026-01-02.json @@ -0,0 +1,26 @@ +{ + "schema_version": "1", + "as_of": "2026-01-02", + "generated_at": "2026-01-02T22:00:00Z", + "mode": "shadow", + "horizon": "1-3 months", + "universe": ["QQQ", "BIL", "BOXX"], + "regime": "risk_on", + "risk_flags": [], + "candidate_bias": { + "QQQ": "positive", + "BIL": "neutral", + "BOXX": "neutral" + }, + "confidence": 0.7, + "evidence": { + "sources": ["examples/context_bundle.example.json"], + "summary": "Synthetic risk-on example for overlay replay.", + "data_gaps": ["Synthetic example only."] + }, + "expires_at": "2026-02-05", + "policy": { + "execution_allowed": false, + "downstream_use": "Shadow replay only; deterministic policy controls any exposure mapping." + } +} diff --git a/examples/signal_history/2026-02-06.json b/examples/signal_history/2026-02-06.json new file mode 100644 index 0000000..6e255ea --- /dev/null +++ b/examples/signal_history/2026-02-06.json @@ -0,0 +1,26 @@ +{ + "schema_version": "1", + "as_of": "2026-02-06", + "generated_at": "2026-02-06T22:00:00Z", + "mode": "shadow", + "horizon": "1-3 months", + "universe": ["QQQ", "BIL", "BOXX"], + "regime": "risk_off", + "risk_flags": ["liquidity_stress"], + "candidate_bias": { + "QQQ": "avoid", + "BIL": "neutral", + "BOXX": "neutral" + }, + "confidence": 0.8, + "evidence": { + "sources": ["examples/context_bundle.example.json"], + "summary": "Synthetic risk-off example for overlay replay.", + "data_gaps": ["Synthetic example only."] + }, + "expires_at": "2026-03-19", + "policy": { + "execution_allowed": false, + "downstream_use": "Shadow replay only; deterministic policy controls any exposure mapping." + } +} diff --git a/examples/signal_history/2026-03-20.json b/examples/signal_history/2026-03-20.json new file mode 100644 index 0000000..0d1d75b --- /dev/null +++ b/examples/signal_history/2026-03-20.json @@ -0,0 +1,26 @@ +{ + "schema_version": "1", + "as_of": "2026-03-20", + "generated_at": "2026-03-20T22:00:00Z", + "mode": "shadow", + "horizon": "1-3 months", + "universe": ["QQQ", "BIL", "BOXX"], + "regime": "mixed", + "risk_flags": ["earnings_concentration"], + "candidate_bias": { + "QQQ": "watch", + "BIL": "neutral", + "BOXX": "neutral" + }, + "confidence": 0.65, + "evidence": { + "sources": ["examples/context_bundle.example.json"], + "summary": "Synthetic mixed-regime example for overlay replay.", + "data_gaps": ["Synthetic example only."] + }, + "expires_at": "2026-04-16", + "policy": { + "execution_allowed": false, + "downstream_use": "Shadow replay only; deterministic policy controls any exposure mapping." + } +} diff --git a/examples/signal_history/2026-04-17.json b/examples/signal_history/2026-04-17.json new file mode 100644 index 0000000..eda8d90 --- /dev/null +++ b/examples/signal_history/2026-04-17.json @@ -0,0 +1,26 @@ +{ + "schema_version": "1", + "as_of": "2026-04-17", + "generated_at": "2026-04-17T22:00:00Z", + "mode": "shadow", + "horizon": "1-3 months", + "universe": ["QQQ", "BIL", "BOXX"], + "regime": "risk_on", + "risk_flags": [], + "candidate_bias": { + "QQQ": "positive", + "BIL": "neutral", + "BOXX": "neutral" + }, + "confidence": 0.72, + "evidence": { + "sources": ["examples/context_bundle.example.json"], + "summary": "Synthetic recovery example for overlay replay.", + "data_gaps": ["Synthetic example only."] + }, + "expires_at": "2026-05-31", + "policy": { + "execution_allowed": false, + "downstream_use": "Shadow replay only; deterministic policy controls any exposure mapping." + } +} diff --git a/scripts/backtest_signal_overlay.py b/scripts/backtest_signal_overlay.py new file mode 100644 index 0000000..2863d54 --- /dev/null +++ b/scripts/backtest_signal_overlay.py @@ -0,0 +1,51 @@ +#!/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.overlay_backtest import ( # noqa: E402 + OverlayPolicy, + backtest_overlay, + load_price_history, + load_signals, +) + + +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("--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) + parser.add_argument("--mixed-exposure", type=float, default=0.8) + parser.add_argument("--risk-off-exposure", type=float, default=0.5) + parser.add_argument("--severe-flag-exposure", type=float, default=0.6) + parser.add_argument("--output", default="", help="Optional JSON summary path") + args = parser.parse_args() + + policy = OverlayPolicy( + min_confidence=args.min_confidence, + mixed_exposure=args.mixed_exposure, + risk_off_exposure=args.risk_off_exposure, + severe_flag_exposure=args.severe_flag_exposure, + ) + prices = load_price_history(Path(args.prices), symbol=args.symbol) + signals = load_signals(Path(args.signals)) + summary = backtest_overlay(prices, signals, policy=policy) + summary["symbol"] = args.symbol.upper() + + text = json.dumps(summary, ensure_ascii=True, indent=2) + "\n" + if args.output: + Path(args.output).write_text(text, encoding="utf-8") + print(text, end="") + 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 9ef54a8..128ae8d 100644 --- a/src/ai_long_horizon_signal_pipelines/__init__.py +++ b/src/ai_long_horizon_signal_pipelines/__init__.py @@ -1,5 +1,6 @@ """Shadow-only long-horizon AI signal artifact helpers.""" +from .overlay_backtest import OverlayPolicy, backtest_overlay from .schema import SignalValidationError, validate_signal -__all__ = ["SignalValidationError", "validate_signal"] +__all__ = ["OverlayPolicy", "SignalValidationError", "backtest_overlay", "validate_signal"] diff --git a/src/ai_long_horizon_signal_pipelines/overlay_backtest.py b/src/ai_long_horizon_signal_pipelines/overlay_backtest.py new file mode 100644 index 0000000..b7c4af5 --- /dev/null +++ b/src/ai_long_horizon_signal_pipelines/overlay_backtest.py @@ -0,0 +1,166 @@ +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 .schema import validate_signal + + +@dataclass(frozen=True) +class PricePoint: + date: dt.date + close: float + + +@dataclass(frozen=True) +class OverlayPolicy: + min_confidence: float = 0.55 + risk_on_exposure: float = 1.0 + mixed_exposure: float = 0.8 + risk_off_exposure: float = 0.5 + severe_flag_exposure: float = 0.6 + severe_flags: tuple[str, ...] = ( + "credit_stress", + "liquidity_stress", + "macro_shock", + "market_structure_stress", + "earnings_concentration", + ) + + +def parse_date(value: str) -> dt.date: + return dt.date.fromisoformat(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) + if len(rows) < 2: + raise ValueError(f"price history for {symbol} requires at least two rows") + return rows + + +def load_signals(path: Path) -> list[dict[str, Any]]: + signal_paths = sorted(path.glob("*.json")) if path.is_dir() else [path] + signals: list[dict[str, Any]] = [] + for signal_path in signal_paths: + payload = json.loads(signal_path.read_text(encoding="utf-8")) + validate_signal(payload) + signals.append(payload) + signals.sort(key=lambda item: parse_date(str(item["as_of"]))) + return signals + + +def signal_active_on(signal: dict[str, Any], date: dt.date) -> bool: + as_of = parse_date(str(signal["as_of"])) + expires_at = parse_date(str(signal["expires_at"])) + return as_of <= date <= expires_at + + +def signal_for_date(signals: list[dict[str, Any]], date: dt.date) -> dict[str, Any] | None: + active = [signal for signal in signals if signal_active_on(signal, date)] + return active[-1] if active else None + + +def exposure_for_signal(signal: dict[str, Any] | None, policy: OverlayPolicy) -> float: + if signal is None: + return policy.risk_on_exposure + confidence = float(signal["confidence"]) + if confidence < policy.min_confidence: + return policy.risk_on_exposure + + regime = str(signal["regime"]) + if regime == "risk_off": + exposure = policy.risk_off_exposure + elif regime == "mixed": + exposure = policy.mixed_exposure + else: + exposure = policy.risk_on_exposure + + risk_flags = {str(flag) for flag in signal.get("risk_flags", [])} + if risk_flags.intersection(policy.severe_flags): + exposure = min(exposure, policy.severe_flag_exposure) + + # This overlay is risk-reducing only. It can never increase baseline exposure. + return max(0.0, min(policy.risk_on_exposure, exposure)) + + +def max_drawdown(equity_curve: list[float]) -> float: + peak = equity_curve[0] + worst = 0.0 + for value in equity_curve: + peak = max(peak, value) + if peak > 0: + worst = min(worst, value / peak - 1.0) + return worst + + +def backtest_overlay( + prices: list[PricePoint], + signals: list[dict[str, Any]], + *, + policy: OverlayPolicy | None = None, +) -> dict[str, Any]: + policy = policy or OverlayPolicy() + baseline_equity = 1.0 + overlay_equity = 1.0 + baseline_curve = [baseline_equity] + overlay_curve = [overlay_equity] + exposures: list[float] = [] + turnover = 0.0 + previous_exposure = policy.risk_on_exposure + + for previous, current in zip(prices, prices[1:]): + exposure = exposure_for_signal(signal_for_date(signals, previous.date), policy) + daily_return = current.close / previous.close - 1.0 + baseline_equity *= 1.0 + daily_return + overlay_equity *= 1.0 + exposure * daily_return + baseline_curve.append(baseline_equity) + overlay_curve.append(overlay_equity) + exposures.append(exposure) + turnover += abs(exposure - previous_exposure) + previous_exposure = exposure + + avg_exposure = sum(exposures) / len(exposures) if exposures else policy.risk_on_exposure + return { + "periods": len(prices) - 1, + "start_date": prices[0].date.isoformat(), + "end_date": prices[-1].date.isoformat(), + "baseline": { + "final_equity": baseline_equity, + "total_return": baseline_equity - 1.0, + "max_drawdown": max_drawdown(baseline_curve), + }, + "overlay": { + "final_equity": overlay_equity, + "total_return": overlay_equity - 1.0, + "max_drawdown": max_drawdown(overlay_curve), + "avg_exposure": avg_exposure, + "turnover": turnover, + }, + "policy": { + "min_confidence": policy.min_confidence, + "risk_on_exposure": policy.risk_on_exposure, + "mixed_exposure": policy.mixed_exposure, + "risk_off_exposure": policy.risk_off_exposure, + "severe_flag_exposure": policy.severe_flag_exposure, + "severe_flags": list(policy.severe_flags), + }, + } diff --git a/tests/test_overlay_backtest.py b/tests/test_overlay_backtest.py new file mode 100644 index 0000000..2c5fa7e --- /dev/null +++ b/tests/test_overlay_backtest.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import datetime as dt +from pathlib import Path + +from ai_long_horizon_signal_pipelines.overlay_backtest import ( + OverlayPolicy, + backtest_overlay, + exposure_for_signal, + load_price_history, + load_signals, + signal_for_date, +) + + +ROOT = Path(__file__).resolve().parents[1] + + +def test_exposure_for_signal_is_risk_reducing_only() -> None: + policy = OverlayPolicy() + assert exposure_for_signal(None, policy) == 1.0 + assert exposure_for_signal({"confidence": 0.3, "regime": "risk_off", "risk_flags": []}, policy) == 1.0 + assert exposure_for_signal({"confidence": 0.8, "regime": "risk_off", "risk_flags": []}, policy) == 0.5 + assert exposure_for_signal({"confidence": 0.8, "regime": "mixed", "risk_flags": ["liquidity_stress"]}, policy) == 0.6 + assert exposure_for_signal({"confidence": 0.8, "regime": "risk_on", "risk_flags": []}, policy) == 1.0 + + +def test_signal_for_date_uses_latest_active_signal() -> None: + signals = load_signals(ROOT / "examples" / "signal_history") + + assert signal_for_date(signals, dt.date(2026, 1, 10))["regime"] == "risk_on" + assert signal_for_date(signals, dt.date(2026, 3, 1))["regime"] == "risk_off" + assert signal_for_date(signals, dt.date(2026, 4, 1))["regime"] == "mixed" + assert signal_for_date(signals, dt.date(2026, 6, 1)) is None + + +def test_backtest_overlay_reduces_drawdown_on_synthetic_path() -> None: + prices = load_price_history(ROOT / "examples" / "price_history.example.csv", symbol="QQQ") + signals = load_signals(ROOT / "examples" / "signal_history") + + summary = backtest_overlay(prices, signals) + + assert summary["periods"] == len(prices) - 1 + assert summary["baseline"]["max_drawdown"] < 0 + assert summary["overlay"]["max_drawdown"] > summary["baseline"]["max_drawdown"] + assert 0 < summary["overlay"]["avg_exposure"] <= 1.0 + assert summary["overlay"]["turnover"] > 0