| 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
- Train return > 5%/trade but test is negative
- Works only on 1-2 symbols (overfitting)
- Requires > 10 hyperparameters
- Can't explain why signal should work
- 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 barsclose 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:
-
Trade Log CSV (27_nvda_trade_log.csv)
- Every trade with entry/exit times, prices, reasons
- All decision factors (iv_zscore, vxx signals, etc.)
-
Ledger CSV (27_nvda_ledger.csv)
- Capital before/after each trade
- Dollar PnL, shares, investment amounts
- Reconciliation verification
-
Proofs JSON (27_nvda_proofs.json)
- Detailed results of all audit proofs
- Sample data for manual verification
-
Summary JSON (27_nvda_summary.json)
- Train/test statistics
- B&H comparison
- Overfitting check ratio
-
Equity Chart (27_nvda_portfolio_equity.png)
- Visual equity curve
- Train/test split marker
Checklist for New Strategies
Before trusting any backtest results:
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
