experiment-creation

name	experiment-creation
description	Template for creating Volatio research experiments with Optuna optimization and portfolio backtesting. Includes proper train/test splits, lookahead bias prevention, results saving, and validated patterns from Exp 051-053.

experiment-creation

Overview

This skill provides templates and patterns for creating research experiments in Volatio. Based on validated patterns from Experiments 051-055 (Laplace options research).

When to Use

• Exploring a new signal hypothesis with data analysis
• Creating a new hyperparameter optimization experiment
• Building a portfolio backtest for validated strategies
• Setting up train/test temporal splits
• Implementing trading simulations with proper exit logic

Experiment Lifecycle

┌─────────────────────────────────────────────────────────────────────────┐
│  PHASE 1: EXPLORATION                                                   │
│  ─────────────────────                                                  │
│  • Hypothesis formation (why should this work?)                         │
│  • Data availability check (what do we have?)                           │
│  • Exploratory queries & visualizations                                 │
│  • Correlation analysis with forward returns                            │
│  • Quick sanity checks (does signal exist?)                             │
│  Output: explore.py, notebooks, data samples                            │
├─────────────────────────────────────────────────────────────────────────┤
│  PHASE 2: OPTIMIZATION                                                  │
│  ─────────────────────                                                  │
│  • Define parameter search space                                        │
│  • Optuna optimization on TRAIN data only                               │
│  • Early stopping when converged                                        │
│  • Single evaluation on TEST (no peeking!)                              │
│  Output: run_optimization.py, results/final_results.json                │
├─────────────────────────────────────────────────────────────────────────┤
│  PHASE 3: PORTFOLIO BACKTEST                                            │
│  ─────────────────────                                                  │
│  • Load optimized parameters                                            │
│  • Simulate realistic execution (T+1 OPEN entry)                        │
│  • Position sizing, stops, transaction costs                            │
│  • Daily equity curve, trade log                                        │
│  Output: run_portfolio.py, results_portfolio/                           │
├─────────────────────────────────────────────────────────────────────────┤
│  PHASE 4: DOCUMENTATION                                                 │
│  ─────────────────────                                                  │
│  • Document in RESULTS.md                                               │
│  • Update anti-patterns if failed                                       │
│  • Capture learnings for future experiments                             │
│  Output: README.md, RESULTS entry                                       │
└─────────────────────────────────────────────────────────────────────────┘

Experiment Directory Structure

research/experiments/[family]/XXX-experiment-name/
├── explore.py             # Phase 1: Data exploration & hypothesis testing
├── run_optimization.py    # Phase 2: Optuna hyperparameter search
├── run_portfolio.py       # Phase 3: Portfolio backtest (after optimization)
├── helpers.py             # Data fetching, signal computation
├── universes.py           # Symbol universe definitions (if needed)
├── README.md              # Experiment documentation
└── results/               # Output directory (auto-created)
    ├── exploration/       # Plots, correlation matrices, data samples
    ├── intermediate_results.json
    ├── final_results.json
    └── optimization_results.json

Phase 1: Exploration Template

"""
XXX - [Experiment Name] - EXPLORATION

HYPOTHESIS:
===========
[Why should this signal work? What's the economic rationale?]

DATA REQUIRED:
==============
[What tables/columns do we need? Do we have it?]

QUESTIONS TO ANSWER:
====================
1. Does the signal exist in our data?
2. What's the correlation with forward returns?
3. How stable is the signal over time?
4. Are there obvious confounds or biases?

USAGE:
======
cd /Users/adamsohn/Projects/volatio
source research/.venv/bin/activate
python research/experiments/[family]/XXX-experiment-name/explore.py
"""

from __future__ import annotations

import warnings
from datetime import datetime
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import psycopg2

warnings.filterwarnings("ignore", message=".*pandas only supports SQLAlchemy.*")

# =============================================================================
# CONFIGURATION
# =============================================================================

DATABASE_URL = "postgresql://volatio:vibes123@localhost:5432/volatio_market"
RESULTS_DIR = Path(__file__).parent / "results" / "exploration"
RESULTS_DIR.mkdir(parents=True, exist_ok=True)

# Analysis period
START_DATE = "2023-01-01"
END_DATE = "2025-12-31"

# Symbols to explore (start small, expand if promising)
SYMBOLS = ["SPY", "QQQ", "AAPL", "NVDA", "TSLA"]


# =============================================================================
# DATABASE QUERIES
# =============================================================================

def get_connection():
    return psycopg2.connect(DATABASE_URL)


def explore_data_availability(conn) -> pd.DataFrame:
    """Check what data we have for the hypothesis."""
    query = """
    -- Customize this query for your hypothesis
    SELECT 
        underlying,
        COUNT(*) as records,
        MIN(time::date) as first_date,
        MAX(time::date) as last_date
    FROM options_minute_aggs
    WHERE underlying = ANY(%s)
    GROUP BY underlying
    ORDER BY records DESC
    """
    return pd.read_sql(query, conn, params=(SYMBOLS,))


def compute_signal(conn, symbol: str) -> pd.DataFrame:
    """
    Compute your candidate signal.
    
    CRITICAL: Use only data available at time T to compute signal for T.
    """
    # TODO: Replace with your signal computation
    query = """
    SELECT 
        time::date as trade_date,
        -- YOUR SIGNAL COMPUTATION HERE
        SUM(volume * close * 100) as total_flow
    FROM options_minute_aggs
    WHERE underlying = %s
      AND time >= %s AND time < %s
    GROUP BY time::date
    ORDER BY trade_date
    """
    return pd.read_sql(query, conn, params=(symbol, START_DATE, END_DATE))


def get_forward_returns(conn, symbol: str, horizons: list[int] = [1, 3, 5, 10]) -> pd.DataFrame:
    """Get forward returns for correlation analysis."""
    query = """
    SELECT 
        date as trade_date,
        close,
        close / LAG(close, 1) OVER (ORDER BY date) - 1 as ret_1d
    FROM daily_bars_rth
    WHERE symbol = %s
      AND date >= %s AND date < %s
    ORDER BY date
    """
    df = pd.read_sql(query, conn, params=(symbol, START_DATE, END_DATE))
    
    # Compute forward returns
    for h in horizons:
        df[f'fwd_ret_{h}d'] = df['close'].shift(-h) / df['close'] - 1
    
    return df


# =============================================================================
# ANALYSIS FUNCTIONS
# =============================================================================

def analyze_signal_distribution(signal_df: pd.DataFrame, signal_col: str):
    """Visualize signal distribution and stationarity."""
    fig, axes = plt.subplots(2, 2, figsize=(12, 8))
    
    # Distribution
    axes[0, 0].hist(signal_df[signal_col].dropna(), bins=50, edgecolor='black')
    axes[0, 0].set_title(f'{signal_col} Distribution')
    
    # Time series
    axes[0, 1].plot(signal_df['trade_date'], signal_df[signal_col])
    axes[0, 1].set_title(f'{signal_col} Over Time')
    
    # Rolling mean/std (stationarity check)
    rolling_mean = signal_df[signal_col].rolling(20).mean()
    rolling_std = signal_df[signal_col].rolling(20).std()
    axes[1, 0].plot(signal_df['trade_date'], rolling_mean, label='20d Mean')
    axes[1, 0].fill_between(signal_df['trade_date'], 
                            rolling_mean - rolling_std, 
                            rolling_mean + rolling_std, alpha=0.3)
    axes[1, 0].set_title('Rolling Mean ± Std')
    axes[1, 0].legend()
    
    # Autocorrelation
    from pandas.plotting import autocorrelation_plot
    autocorrelation_plot(signal_df[signal_col].dropna(), ax=axes[1, 1])
    axes[1, 1].set_title('Autocorrelation')
    
    plt.tight_layout()
    plt.savefig(RESULTS_DIR / f'{signal_col}_distribution.png', dpi=150)
    plt.close()


def analyze_forward_correlations(merged_df: pd.DataFrame, signal_col: str):
    """Compute correlation between signal and forward returns."""
    horizons = [c for c in merged_df.columns if c.startswith('fwd_ret_')]
    
    correlations = {}
    for h in horizons:
        corr = merged_df[signal_col].corr(merged_df[h])
        correlations[h] = corr
    
    print(f"\n📊 {signal_col} → Forward Return Correlations:")
    print("-" * 40)
    for h, corr in correlations.items():
        strength = "🟢" if abs(corr) > 0.05 else "🔴"
        print(f"  {h}: {corr:+.4f} {strength}")
    
    return correlations


def quick_backtest(merged_df: pd.DataFrame, signal_col: str, threshold: float):
    """
    Quick sanity check: does high signal predict positive returns?
    
    NOT for production - just to see if there's any signal at all.
    """
    df = merged_df.dropna(subset=[signal_col, 'fwd_ret_5d']).copy()
    
    # Z-score the signal
    df['signal_z'] = (df[signal_col] - df[signal_col].mean()) / df[signal_col].std()
    
    # Split by signal strength
    high_signal = df[df['signal_z'] > threshold]
    low_signal = df[df['signal_z'] < -threshold]
    neutral = df[abs(df['signal_z']) <= threshold]
    
    print(f"\n📈 Quick Backtest (threshold={threshold}):")
    print("-" * 50)
    print(f"  High signal ({len(high_signal)} days): {high_signal['fwd_ret_5d'].mean()*100:+.2f}% avg 5d return")
    print(f"  Low signal ({len(low_signal)} days):  {low_signal['fwd_ret_5d'].mean()*100:+.2f}% avg 5d return")
    print(f"  Neutral ({len(neutral)} days):       {neutral['fwd_ret_5d'].mean()*100:+.2f}% avg 5d return")
    
    spread = high_signal['fwd_ret_5d'].mean() - low_signal['fwd_ret_5d'].mean()
    print(f"\n  Long/Short Spread: {spread*100:+.2f}%")
    
    return {
        'high_return': high_signal['fwd_ret_5d'].mean(),
        'low_return': low_signal['fwd_ret_5d'].mean(),
        'spread': spread,
    }


# =============================================================================
# MAIN EXPLORATION
# =============================================================================

def main():
    print("=" * 70)
    print("XXX - [EXPERIMENT NAME] - EXPLORATION")
    print("=" * 70)
    
    conn = get_connection()
    
    # Step 1: Check data availability
    print("\n📋 Step 1: Data Availability")
    availability = explore_data_availability(conn)
    print(availability.to_string())
    
    # Step 2: Compute signal for each symbol
    print("\n📊 Step 2: Computing Signals")
    all_correlations = {}
    
    for symbol in SYMBOLS:
        print(f"\n--- {symbol} ---")
        
        signal_df = compute_signal(conn, symbol)
        returns_df = get_forward_returns(conn, symbol)
        
        if len(signal_df) == 0:
            print(f"  ⚠️ No signal data for {symbol}")
            continue
        
        # Merge signal with returns
        merged = signal_df.merge(returns_df, on='trade_date', how='inner')
        
        if len(merged) < 100:
            print(f"  ⚠️ Only {len(merged)} days of data, need 100+")
            continue
        
        # Analyze
        signal_col = 'total_flow'  # TODO: Replace with your signal column
        analyze_signal_distribution(merged, signal_col)
        corrs = analyze_forward_correlations(merged, signal_col)
        quick_backtest(merged, signal_col, threshold=1.5)
        
        all_correlations[symbol] = corrs
    
    # Step 3: Summary
    print("\n" + "=" * 70)
    print("EXPLORATION SUMMARY")
    print("=" * 70)
    
    # TODO: Add your conclusions
    print("\n🎯 Key Findings:")
    print("  1. [Finding 1]")
    print("  2. [Finding 2]")
    
    print("\n✅ Next Steps:")
    print("  - If promising: Create run_optimization.py")
    print("  - If weak: Document learnings and move on")
    
    conn.close()


if __name__ == "__main__":
    main()

Critical Anti-Patterns to AVOID

From lessons learned (see RESULTS_v2.md):

Anti-Pattern	Why It Fails	Example
Per-symbol optimization	Overfitting trap	Exp 029, 032
Kernel MMD / complex signals	Adds noise, hurts returns	Exp 025, 030
Short signals from options flow	Asymmetry is fundamental	Exp 054
Train results without OOS	Worthless validation	Exp 047
Universal parameters across all	Market structure varies	Exp 032

run_optimization.py Template

"""
XXX - [Experiment Name]

[Brief description of what this experiment tests]

METHODOLOGY:
============
1. Temporal train/test split: Train 2023-2024, Test 2025
2. Optuna optimizes ONLY on train data (no look-ahead bias)
3. Best train params evaluated ONCE on test data
4. Only report STABLE strategies (train→test gap < 50%)

RUNTIME ESTIMATE:
================
[Estimate based on # trials × # universes]

USAGE:
======
cd /Users/adamsohn/Projects/volatio
source research/.venv/bin/activate
python research/experiments/options/laplace/XXX-experiment-name/run_optimization.py
"""

from __future__ import annotations

import json
import sys
import warnings
from datetime import datetime
from pathlib import Path

import numpy as np
import optuna
import pandas as pd
from optuna.samplers import TPESampler

# Reuse helpers from 051 (or copy if modifications needed)
HELPERS_DIR = Path(__file__).parent.parent / "051-otm-call-optimization"
sys.path.insert(0, str(HELPERS_DIR))

from helpers import (
    get_connection,
    get_otm_call_flow_daily,
    get_daily_prices,
    compute_signals,
    simulate_trades,
    clear_caches,
)

warnings.filterwarnings("ignore", message=".*pandas only supports SQLAlchemy.*")
optuna.logging.set_verbosity(optuna.logging.WARNING)

# =============================================================================
# CONFIGURATION
# =============================================================================

RESULTS_DIR = Path(__file__).parent / "results"
RESULTS_DIR.mkdir(exist_ok=True)

# CRITICAL: Temporal split - NEVER optimize on test data
TRAIN_PERIODS = [("2023-01-01", "2023-12-31"), ("2024-01-01", "2024-12-31")]
TEST_PERIOD = ("2025-01-01", "2025-12-31")

# Optimization settings
N_TRIALS = 100          # Per universe - adjust based on param space
EARLY_STOP_PATIENCE = 30  # Stop if no improvement
MIN_TRADES = 15         # Minimum trades to consider valid

# Progress tracking
_trial_count = 0
_best_score = -100


# =============================================================================
# SYMBOL UNIVERSES
# =============================================================================

UNIVERSES = [
    {
        "name": "Universe-Name",
        "symbols": ["SPY", "QQQ", "TSLA", "NVDA"],  # Define your symbols
        "description": "Description of this universe",
    },
]


# =============================================================================
# EVALUATION FUNCTION
# =============================================================================

def evaluate_params(
    conn,
    symbols: list[str],
    # Add your hyperparameters here
    param1: float,
    param2: int,
    start_date: str,
    end_date: str,
) -> dict:
    """
    Evaluate strategy parameters on given symbols and date range.
    
    Returns metrics dict with avg_return, n_trades, win_rate, sharpe.
    """
    all_trades = []
    symbols_with_trades = 0
    
    for symbol in symbols:
        try:
            # Fetch data and generate signals
            # ... your signal logic here ...
            
            # Simulate trades
            # trades = simulate_trades(...)
            
            # if trades:
            #     all_trades.extend(trades)
            #     symbols_with_trades += 1
            pass
            
        except Exception:
            continue
    
    if len(all_trades) == 0:
        return {
            "n_trades": 0,
            "n_symbols": symbols_with_trades,
            "avg_return": -100,
            "win_rate": 0,
            "sharpe": 0,
        }
    
    returns = [t.return_pct for t in all_trades]
    avg_return = np.mean(returns)
    std_return = np.std(returns) if len(returns) > 1 else 1.0
    
    return {
        "n_trades": len(all_trades),
        "n_symbols": symbols_with_trades,
        "avg_return": float(avg_return),
        "win_rate": float(sum(1 for r in returns if r > 0) / len(returns) * 100),
        "sharpe": float(avg_return / std_return if std_return > 0 else 0),
    }


# =============================================================================
# OPTUNA OBJECTIVE (TRAIN ONLY)
# =============================================================================

def create_objective(conn, symbols: list[str]):
    """
    Create Optuna objective that optimizes on TRAIN data only.
    
    CRITICAL: No test data is used in scoring - prevents look-ahead bias.
    """
    global _trial_count, _best_score
    
    def objective(trial: optuna.Trial) -> float:
        global _trial_count, _best_score
        _trial_count += 1
        
        # Sample hyperparameters
        param1 = trial.suggest_float("param1", 1.0, 5.0, step=0.1)
        param2 = trial.suggest_int("param2", 5, 30, step=5)
        # Add more parameters...
        
        # === TRAIN EVALUATION ONLY ===
        train_returns = []
        train_trades = 0
        
        for train_start, train_end in TRAIN_PERIODS:
            result = evaluate_params(
                conn, symbols,
                param1, param2,
                train_start, train_end
            )
            if result["n_trades"] > 0:
                train_returns.append(result["avg_return"])
                train_trades += result["n_trades"]
        
        if len(train_returns) == 0 or train_trades < MIN_TRADES:
            return -100.0
        
        train_avg = float(np.mean(train_returns))
        
        if train_avg <= 0:
            return -100.0
        
        # Score based on TRAIN performance only
        trade_factor = min(1.0, np.sqrt(train_trades / 100))
        score = float(train_avg * trade_factor)
        
        # Progress logging
        if score > _best_score:
            _best_score = score
            print(f"  🎯 Trial {_trial_count}: NEW BEST score={score:.2f}")
        
        # Store for later retrieval
        trial.set_user_attr("train_avg", train_avg)
        trial.set_user_attr("train_trades", train_trades)
        
        return score
    
    return objective


# =============================================================================
# EARLY STOPPING
# =============================================================================

class EarlyStoppingCallback:
    """Stop optimization if no improvement in `patience` trials."""
    
    def __init__(self, patience: int = EARLY_STOP_PATIENCE):
        self.patience = patience
        self.best_value = float("-inf")
        self.trials_without_improvement = 0
    
    def __call__(self, study: optuna.Study, trial: optuna.trial.FrozenTrial) -> None:
        if trial.value is not None and trial.value > self.best_value:
            self.best_value = trial.value
            self.trials_without_improvement = 0
        else:
            self.trials_without_improvement += 1
        
        if self.trials_without_improvement >= self.patience:
            print(f"  ⏹️  Early stopping: no improvement in {self.patience} trials")
            study.stop()


# =============================================================================
# MAIN OPTIMIZATION LOOP
# =============================================================================

def optimize_universe(universe: dict, n_trials: int = N_TRIALS) -> dict:
    """
    Run optimization for a single universe.
    
    1. Optimize on TRAIN data (2023-2024)
    2. Evaluate best params on TEST data (2025) ONCE
    3. Return stability analysis
    """
    global _trial_count, _best_score
    
    name = universe["name"]
    symbols = universe["symbols"]
    
    print(f"\n{'='*70}")
    print(f"Optimizing: {name}")
    print(f"{'='*70}")
    
    conn = get_connection()
    _trial_count = 0
    _best_score = -100
    clear_caches()
    
    # Run optimization
    sampler = TPESampler(seed=42)
    study = optuna.create_study(direction="maximize", sampler=sampler)
    objective = create_objective(conn, symbols)
    early_stop = EarlyStoppingCallback()
    
    study.optimize(objective, n_trials=n_trials, show_progress_bar=False, callbacks=[early_stop])
    
    best = study.best_trial
    
    if best.value <= -100:
        conn.close()
        return {"name": name, "success": False, "error": "No profitable strategy"}
    
    # === SINGLE TEST EVALUATION (no look-ahead bias) ===
    print(f"\n--- Evaluating on TEST (2025) ---")
    test_result = evaluate_params(
        conn, symbols,
        best.params["param1"],
        best.params["param2"],
        TEST_PERIOD[0], TEST_PERIOD[1]
    )
    
    train_avg = best.user_attrs.get("train_avg", 0)
    test_avg = test_result["avg_return"]
    
    # Calculate stability gap
    gap = abs(train_avg - test_avg) / max(train_avg, test_avg, 1) * 100
    stable = train_avg > 0 and test_avg > 0 and gap < 50
    
    print(f"Train: {train_avg:.2f}% | Test: {test_avg:.2f}% | Gap: {gap:.1f}%")
    print(f"Status: {'✅ STABLE' if stable else '⚠️ UNSTABLE'}")
    
    conn.close()
    
    return {
        "name": name,
        "success": True,
        "stable": stable,
        "params": best.params,
        "train_avg": train_avg,
        "test_avg": test_avg,
        "gap": gap,
    }


def main():
    print("=" * 80)
    print("XXX - [EXPERIMENT NAME]")
    print("=" * 80)
    
    all_results = []
    
    for universe in UNIVERSES:
        result = optimize_universe(universe)
        all_results.append(result)
        
        # Save intermediate results
        with open(RESULTS_DIR / "intermediate_results.json", "w") as f:
            json.dump({"results": all_results}, f, indent=2, default=str)
    
    # Save final results
    with open(RESULTS_DIR / "final_results.json", "w") as f:
        json.dump({
            "experiment": "XXX-experiment-name",
            "timestamp": datetime.now().isoformat(),
            "train_periods": TRAIN_PERIODS,
            "test_period": TEST_PERIOD,
            "results": all_results,
        }, f, indent=2, default=str)
    
    print(f"\n✅ Results saved to {RESULTS_DIR}")


if __name__ == "__main__":
    main()

run_portfolio.py Template

"""
XXX - [Experiment Name] Portfolio Backtest

Run portfolio using optimized parameters from run_optimization.py.

Features:
- Loads strategies from results/final_results.json
- T+1 OPEN entry (no lookahead bias)
- Daily rebalancing with position limits
- Stop losses and max hold exits
- Transaction costs (commission + slippage)

LOOK-AHEAD BIAS VERIFICATION:
=============================
1. Signal uses .shift(1) - only prior day data
2. Entry on T+1 OPEN (day after signal)
3. All parameters from TRAIN optimization, not TEST
4. Price data queried only up to current simulation date

USAGE:
======
cd /Users/adamsohn/Projects/volatio
source research/.venv/bin/activate
python research/experiments/options/laplace/XXX-experiment-name/run_portfolio.py
"""

from __future__ import annotations

import json
import sys
import warnings
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

# Reuse helpers from 051
HELPERS_DIR = Path(__file__).parent.parent / "051-otm-call-optimization"
sys.path.insert(0, str(HELPERS_DIR))

from helpers import (
    get_connection,
    get_otm_call_flow_daily,
    get_daily_prices,
    compute_signals,
    clear_caches,
)

warnings.filterwarnings("ignore", message=".*pandas only supports SQLAlchemy.*")

# =============================================================================
# CONFIGURATION
# =============================================================================

RESULTS_DIR = Path(__file__).parent / "results_portfolio"
RESULTS_DIR.mkdir(exist_ok=True)

OPTIMIZATION_RESULTS = Path(__file__).parent / "results" / "final_results.json"

# Portfolio settings
INITIAL_CAPITAL = 100_000
POSITION_SIZE_PCT = 0.02    # 2% per position
MAX_POSITIONS = 50          # Max concurrent positions
MAX_POSITIONS_PER_STRATEGY = 10

# Transaction costs (REALISTIC values)
COMMISSION_PER_TRADE = 1.00   # $1 per trade
SLIPPAGE_PCT = 0.05           # 0.05% slippage

# Backtest period
BACKTEST_START = "2025-01-01"
BACKTEST_END = "2025-12-31"


# =============================================================================
# POSITION TRACKING
# =============================================================================

@dataclass
class Position:
    symbol: str
    strategy_name: str
    entry_date: str
    entry_price: float
    shares: int
    cost_basis: float
    stop_loss_price: float
    max_hold_date: str
    z_score: float


# =============================================================================
# LOAD STRATEGIES FROM OPTIMIZATION
# =============================================================================

def load_strategies() -> list[dict]:
    """Load stable strategies from optimization results."""
    if not OPTIMIZATION_RESULTS.exists():
        print(f"❌ Results not found: {OPTIMIZATION_RESULTS}")
        print("Run run_optimization.py first!")
        return []
    
    with open(OPTIMIZATION_RESULTS) as f:
        data = json.load(f)
    
    strategies = []
    
    for result in data.get("results", []):
        if not result.get("stable", False):
            continue
        
        params = result.get("params", {})
        strategies.append({
            "name": result["name"],
            "symbols": result.get("valid_symbols", result.get("symbols", [])),
            # Map your params here
            **params,
        })
    
    return strategies


# =============================================================================
# PORTFOLIO SIMULATION
# =============================================================================

def run_portfolio():
    """Main portfolio simulation loop."""
    conn = get_connection()
    strategies = load_strategies()
    
    if not strategies:
        print("No strategies to backtest!")
        return
    
    print(f"Loaded {len(strategies)} strategies")
    
    # Initialize
    equity = INITIAL_CAPITAL
    positions: dict[str, Position] = {}
    trades = []
    daily_equity = []
    
    # Generate trading dates
    trading_dates = pd.date_range(BACKTEST_START, BACKTEST_END, freq='B')
    
    for date in trading_dates:
        date_str = date.strftime("%Y-%m-%d")
        
        # 1. Check exits (stop loss, max hold)
        # 2. Generate new signals
        # 3. Enter new positions (within limits)
        # 4. Update equity
        
        daily_equity.append({
            "date": date_str,
            "equity": equity,
            "positions": len(positions),
        })
    
    # Save results
    pd.DataFrame(daily_equity).to_csv(RESULTS_DIR / "daily_equity.csv", index=False)
    pd.DataFrame(trades).to_csv(RESULTS_DIR / "trades.csv", index=False)
    
    # Calculate metrics
    equity_series = pd.DataFrame(daily_equity)
    total_return = (equity_series["equity"].iloc[-1] / INITIAL_CAPITAL - 1) * 100
    
    print(f"\n{'='*60}")
    print("PORTFOLIO RESULTS")
    print(f"{'='*60}")
    print(f"Total Return: {total_return:.2f}%")
    print(f"Total Trades: {len(trades)}")
    
    conn.close()


if __name__ == "__main__":
    run_portfolio()

Key Components Checklist

1. Temporal Split (MANDATORY)

TRAIN_PERIODS = [("2023-01-01", "2023-12-31"), ("2024-01-01", "2024-12-31")]
TEST_PERIOD = ("2025-01-01", "2025-12-31")

2. Lookahead Bias Prevention

• Signal z-score uses .shift(1) - only prior day data
• Entry on T+1 OPEN (day after signal fires)
• All params selected on TRAIN, evaluated ONCE on TEST
• Never peek at test data during optimization

3. Results Directory

RESULTS_DIR = Path(__file__).parent / "results"
RESULTS_DIR.mkdir(exist_ok=True)

4. Early Stopping

class EarlyStoppingCallback:
    def __init__(self, patience: int = 30):
        ...

5. Stability Validation

# Gap < 50% between train and test = STABLE
gap = abs(train_avg - test_avg) / max(train_avg, test_avg) * 100
stable = train_avg > 0 and test_avg > 0 and gap < 50

6. Position Management

@dataclass
class Position:
    symbol: str
    entry_date: str
    entry_price: float
    shares: int
    stop_loss_price: float
    max_hold_date: str

7. Transaction Costs

COMMISSION_PER_TRADE = 1.00   # $1 per trade
SLIPPAGE_PCT = 0.05           # 0.05% slippage

Helper Functions to Reuse

Located in 051-otm-call-optimization/helpers.py:

Function	Purpose
get_connection()	Database connection
get_otm_call_flow_daily()	Fetch OTM call flow with caching
get_daily_prices()	Fetch OHLC prices with caching
compute_signals()	Z-score signals with .shift(1)
simulate_trades()	Trade simulation with stop loss
clear_caches()	Clear data caches between runs

Running Experiments

# Activate research environment
cd /Users/adamsohn/Projects/volatio
source research/.venv/bin/activate

# Run optimization first
python research/experiments/options/laplace/XXX-experiment-name/run_optimization.py

# Run portfolio backtest (uses optimization results)
python research/experiments/options/laplace/XXX-experiment-name/run_portfolio.py

Documenting Results

After experiment completes, add to docs/temp/RESULTS_v2.md:

## Experiment XXX: [Title]

**Status:** ✅ Validated / ⚠️ Needs Work / ❌ Failed

**Objective:** [What you were testing]

**Methodology:**
- Train: 2023-2024
- Test: 2025
- [Key approach]

**Key Results:**
| Metric | Train | Test |
|--------|-------|------|
| Return | X.XX% | X.XX% |
| Trades | N | N |
| Sharpe | X.XX | X.XX |

**Conclusion:** [What you learned]

**Next Steps:** [What to try next]

Red Flags - Kill Experiment Early

1. Train return > 5%/trade but test is negative
2. Works only on 1-2 symbols (overfitting)
3. Requires > 10 hyperparameters
4. Can't explain why signal should work
5. Complexity keeps increasing to "fix" issues

Volatio Backtest Audit Framework

Overview

This document describes the comprehensive audit framework developed for validating trading strategy backtests. The framework ensures no look-ahead bias, correct calculations, and proper trade ledger reconciliation.

The framework was developed through iterative research in research/experiments/options/extrinsic/039-etf-dte-visualization/volitility_stock/NVDA/ spanning scripts 01-27.

---

Quick Start

Run a fully audited backtest:

cd research/experiments/options/extrinsic/039-etf-dte-visualization/volitility_stock/NVDA
python 27_portfolio_exact_script24.py

Expected output: 12/12 proofs passed

---

The 15 Audit Proofs

Phase 1: Data Integrity (Proofs 1-4)

These proofs validate the raw data before any calculations.

PROOF 1: RAW_DATA_INTEGRITY

Purpose: Verify database query returns valid OHLCV data.

Checks:

• No NULL values in close/high/low
• high >= low for all bars
• close within [low, high] range
• Sufficient trading days (>500 for multi-year backtest)

Implementation:

def proof_01_raw_data_integrity() -> Dict[str, Any]:
    query = """
    SELECT 
        COUNT(*) as total_bars,
        COUNT(DISTINCT DATE(time)) as trading_days,
        SUM(CASE WHEN close <= 0 THEN 1 ELSE 0 END) as invalid_prices,
        SUM(CASE WHEN high < low THEN 1 ELSE 0 END) as high_low_violations,
        SUM(CASE WHEN close > high OR close < low THEN 1 ELSE 0 END) as close_violations
    FROM minute_bars
    WHERE symbol = 'NVDA' AND DATE(time) >= '2022-01-01'
    """
    # ... execute and verify all counts are 0

PROOF 2: 15-MIN INTERVAL VERIFICATION

Purpose: Verify data is at expected intervals (0, 15, 30, 45 minutes).

Checks:

• Only valid minute values appear
• No unexpected timestamps

Implementation:

query = """
SELECT EXTRACT(MINUTE FROM time)::int as minute, COUNT(*) as bar_count
FROM minute_bars
WHERE symbol = 'NVDA'
  AND EXTRACT(MINUTE FROM time) IN (0, 15, 30, 45, 59)
GROUP BY EXTRACT(MINUTE FROM time)
"""

PROOF 3: MARKET_HOURS_FILTER

Purpose: Verify only regular market hours data (9:30 AM - 4:00 PM ET).

SQL Filter (use in ALL queries):

AND (
    (EXTRACT(HOUR FROM time) = 9 AND EXTRACT(MINUTE FROM time) >= 30) OR
    (EXTRACT(HOUR FROM time) BETWEEN 10 AND 15)
)

Why this matters: Pre-market/after-hours data can have different liquidity and price behavior.

PROOF 4: SPLIT_ADJUSTMENT_VERIFICATION

Purpose: Verify stock splits are correctly handled.

NVDA Example (10:1 split on 2024-06-10):

NVDA_SPLIT_DATE = pd.Timestamp("2024-06-10")
NVDA_SPLIT_RATIO = 10

# Apply adjustment to pre-split prices
mask = df['date'] < NVDA_SPLIT_DATE
for col in ['open', 'high', 'low', 'close']:
    df.loc[mask, col] = df.loc[mask, col] / NVDA_SPLIT_RATIO

Verification: Compare pre-split adjusted avg to post-split avg (ratio should be ~1.0).

---

Phase 2: Feature Causality (Proofs 5-6)

These proofs verify features use only PAST data (no look-ahead).

PROOF 5: IV_ZSCORE_PRIOR_DAY

Purpose: Verify IV z-score uses only prior day statistics.

Critical Pattern:

# Daily IV calculation
daily_iv = df.groupby('date')['intraday_range'].last()

# Rolling stats SHIFTED by 1 day to use PRIOR data only
daily_iv_mean = daily_iv.rolling(20, min_periods=5).mean().shift(1)
daily_iv_std = daily_iv.rolling(20, min_periods=5).std().shift(1)

# Z-score calculation
df['iv_zscore'] = (df['intraday_range'] - df['daily_iv_mean']) / df['daily_iv_std']

The .shift(1) is CRITICAL - without it, you're using today's data to trade today!

Manual Verification:

def proof_05_iv_zscore_prior_day(df):
    test_date = pd.Timestamp('2024-03-15')
    
    # Get 20 PRIOR days (excluding test_date)
    all_dates = sorted(df['date'].unique())
    date_idx = list(all_dates).index(test_date)
    prior_20 = all_dates[date_idx - 20 : date_idx]  # Excludes test_date!
    
    # Calculate manually
    prior_ivs = [df[df['date'] == d]['intraday_range'].iloc[-1] for d in prior_20]
    manual_mean = np.mean(prior_ivs)
    manual_std = np.std(prior_ivs, ddof=1)  # Use ddof=1 to match pandas
    
    # Compare to stored values
    stored_mean = df[df['date'] == test_date]['daily_iv_mean'].iloc[0]
    assert abs(manual_mean - stored_mean) < 0.1

PROOF 6: VXX_EMA_CAUSALITY

Purpose: Verify EMA calculations use only past data.

Why pandas ewm is safe:

# pandas ewm is CAUSAL by design - it only uses data up to current index
df['vxx_ema_fast'] = df['vxx_close'].ewm(span=5, adjust=False).mean()

Manual Verification:

def proof_06_vxx_ema_causality(df):
    test_idx = 100
    vxx_up_to = df['vxx_close'].iloc[:test_idx + 1].values
    
    # Manual EMA calculation
    alpha = 2 / (5 + 1)
    ema = vxx_up_to[0]
    for v in vxx_up_to[1:]:
        ema = alpha * v + (1 - alpha) * ema
    
    assert abs(ema - df['vxx_ema_fast'].iloc[test_idx]) < 0.01

---

Phase 3: Trade Execution (Proofs 7-11)

These proofs validate the trading logic itself.

PROOF 7: ENTRY_PRICE_VERIFICATION

Purpose: Confirm entry prices exactly match database close prices.

Implementation:

def proof_07_entry_price(df, trades):
    for t in trades[:10]:
        entry_time = pd.Timestamp(t['entry_time'])
        bar = df[df['time'] == entry_time]
        
        assert abs(bar['close'].iloc[0] - t['entry_price']) < 0.01

PROOF 8: EXIT_PRICE_IN_BAR_RANGE

Purpose: All exit prices must be within the bar's actual [low, high] range.

Critical Fix (the TSLA bug):

# WRONG - can return theoretical price outside bar range:
actual_exit = min(bar_high, tp_level)  # tp_level might > bar_high!

# CORRECT - always use actual bar price:
if bar_high >= tp_level:
    return bar_high, 'TAKE_PROFIT'  # Use the actual bar price

Verification:

def proof_08_exit_in_bar_range(trades):
    for t in trades:
        exit_price = t['exit_price']
        bar_low, bar_high = parse_bar_range(t['exit_bar_range'])
        
        assert bar_low - 0.01 <= exit_price <= bar_high + 0.01

PROOF 9: RETURN_CALCULATION

Purpose: Verify return formula: pnl = position × (exit - entry) / entry

def proof_09_return_calculation(trades):
    for t in trades:
        calculated = t['position'] * (t['exit_price'] - t['entry_price']) / t['entry_price'] * 100
        stored = t['pnl_pct']
        
        assert abs(calculated - stored) < 0.01

PROOF 10: ONE_TRADE_PER_DAY

Purpose: No duplicate entries on the same day.

entry_dates = [t['entry_date'] for t in trades]
assert len(entry_dates) == len(set(entry_dates))

PROOF 11: OVERNIGHT_HOLD_VERIFICATION (or SAME_DAY_EXIT)

Purpose: Verify exit date matches the strategy requirement.

For overnight strategies:

for t in trades:
    assert t['exit_date'] > t['entry_date']  # Must exit next day or later

For same-day strategies:

for t in trades:
    assert t['exit_date'] == t['entry_date']  # Must exit same day

---

Phase 4: Portfolio Reconciliation (Proofs 12-15)

These proofs validate the entire portfolio simulation.

PROOF 12: TRADE_LEDGER_RECONCILIATION

Purpose: Verify Final Capital = Initial Capital + Sum(all PnL)

Critical Fix (the rounding bug):

def proof_12_ledger_reconciliation(trades):
    capital = INITIAL_CAPITAL
    unrounded_pnl_sum = 0.0  # Track UNROUNDED for exact reconciliation
    
    for t in trades:
        investment = capital * POSITION_SIZE_PCT
        shares = investment / t['entry_price']
        dollar_pnl = shares * t['position'] * (t['exit_price'] - t['entry_price'])
        
        unrounded_pnl_sum += dollar_pnl  # Don't round!
        capital += dollar_pnl
    
    expected = INITIAL_CAPITAL + unrounded_pnl_sum
    actual = capital
    
    assert abs(expected - actual) < 0.001  # Should match exactly

PROOF 13: PERIOD_ISOLATION

Purpose: Train and test periods have no data overlap.

train_trades = [t for t in trades if t['entry_date'] < TEST_START]
test_trades = [t for t in trades if t['entry_date'] >= TEST_START]

# No train trade should exit in test period
train_exits_in_test = [t for t in train_trades if t['exit_date'] >= TEST_START]
assert len(train_exits_in_test) == 0

PROOF 14: BUY_AND_HOLD_COMPARISON

Purpose: Calculate underlying B&H as a baseline comparison.

def proof_14_buy_and_hold(df, period='train'):
    period_df = df[(df['date'] >= start) & (df['date'] <= end)]
    
    first_close = period_df.iloc[0]['close']
    last_close = period_df.iloc[-1]['close']
    
    bh_return = (last_close - first_close) / first_close * 100
    
    # Strategy should be compared to this baseline

PROOF 15: CUMULATIVE_RETURN_VERIFICATION

Purpose: Manual cumulative return matches pandas cumprod.

returns = [t['pnl_pct'] / 100 for t in trades]
manual_cumret = np.prod([1 + r for r in returns]) - 1

# Should match pandas calculation
pandas_cumret = (1 + pd.Series(returns)).cumprod().iloc[-1] - 1

assert abs(manual_cumret - pandas_cumret) < 0.0001

---

Common Look-Ahead Bias Patterns

Pattern 1: Using Today's Data to Trade Today

Bug:

# Signal uses today's close, then trades at today's close
signal = df['close'] > df['close'].shift(1)
entry_price = df['close']  # WRONG - can't trade at price you just used!

Fix:

# Signal uses yesterday's data
signal = df['close'].shift(1) > df['close'].shift(2)
entry_price = df['open']  # Enter at NEXT bar's open

Pattern 2: Missing .shift() on Rolling Features

Bug:

df['iv_mean'] = df['iv'].rolling(20).mean()
df['iv_zscore'] = (df['iv'] - df['iv_mean']) / df['iv_std']
# This uses TODAY's iv in the mean calculation!

Fix:

df['iv_mean'] = df['iv'].rolling(20).mean().shift(1)  # Use PRIOR day
df['iv_zscore'] = (df['iv'] - df['iv_mean']) / df['iv_std']

Pattern 3: Exit Price Outside Bar Range

Bug:

# Theoretical stop level might be below bar's actual low
stop_price = entry_price * (1 - stop_pct)
if bar_low <= stop_price:
    exit_price = stop_price  # WRONG - might be < bar_low!

Fix:

if bar_low <= stop_price:
    exit_price = bar_low  # Use actual bar price

---

Output Artifacts

A properly audited backtest produces:

1. Trade Log CSV (27_nvda_trade_log.csv)

   • Every trade with entry/exit times, prices, reasons
   • All decision factors (iv_zscore, vxx signals, etc.)

2. Ledger CSV (27_nvda_ledger.csv)

   • Capital before/after each trade
   • Dollar PnL, shares, investment amounts
   • Reconciliation verification

3. Proofs JSON (27_nvda_proofs.json)

   • Detailed results of all audit proofs
   • Sample data for manual verification

4. Summary JSON (27_nvda_summary.json)

   • Train/test statistics
   • B&H comparison
   • Overfitting check ratio

5. Equity Chart (27_nvda_portfolio_equity.png)

   • Visual equity curve
   • Train/test split marker

---

Checklist for New Strategies

Before trusting any backtest results:

• PROOF 1: Raw data has no NULL/invalid values
• PROOF 2: Data at expected intervals
• PROOF 3: Regular hours only (no pre/after market)
• PROOF 4: Stock splits correctly adjusted
• PROOF 5: Rolling features use .shift(1) for prior day
• PROOF 6: EMAs are causal (pandas ewm is safe)
• PROOF 7: Entry prices match database exactly
• PROOF 8: Exit prices within bar's [low, high] range
• PROOF 9: Return formula is correct
• PROOF 10: One trade per day only
• PROOF 11: Exit timing matches strategy (same-day vs overnight)
• PROOF 12: Ledger reconciles exactly
• PROOF 13: Train/test periods isolated
• PROOF 14: B&H baseline calculated
• PROOF 15: Cumulative return verified

---

Overfitting Detection

After all proofs pass, check for overfitting:

train_sharpe = train_stats['sharpe']
test_sharpe = test_stats['sharpe']
ratio = test_sharpe / train_sharpe

if ratio >= 0.8:
    print("✅ EXCELLENT: Minimal overfitting")
elif ratio >= 0.5:
    print("⚠️ ACCEPTABLE: Some overfitting")
else:
    print("❌ CONCERN: Likely overfit")

Target: Test/Train Sharpe ratio ≥ 0.5

---

Example: Full Audit Output

AUDIT PROOF SUMMARY
==========================================
proof_01: RAW_DATA_INTEGRITY              ✅ PASS
proof_02: SPLIT_ADJUSTMENT_VERIFICATION   ✅ PASS
proof_03: IV_ZSCORE_PRIOR_DAY             ✅ PASS
proof_04: VXX_EMA_CAUSALITY               ✅ PASS
proof_05: ENTRY_PRICE_VERIFICATION        ✅ PASS
proof_06: EXIT_PRICE_IN_BAR_RANGE         ✅ PASS
proof_07: RETURN_CALCULATION              ✅ PASS
proof_08: TRADE_LEDGER_RECONCILIATION     ✅ PASS
proof_09: PERIOD_ISOLATION                ✅ PASS
proof_10: BUY_AND_HOLD_COMPARISON (Train) ✅ PASS
proof_11: BUY_AND_HOLD_COMPARISON (Test)  ✅ PASS
proof_12: NO_FUTURE_DATA_LEAK             ✅ PASS

TOTAL: 12/12 proofs passed

🎉 ALL AUDIT PROOFS PASSED - NO LOOK-AHEAD BIAS DETECTED