| 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.*")
DATABASE_URL = "postgresql://volatio:vibes123@localhost:5432/volatio_market"
RESULTS_DIR = Path(__file__).parent / "results" / "exploration"
RESULTS_DIR.mkdir(parents=True, exist_ok=True)
START_DATE = "2023-01-01"
END_DATE = "2025-12-31"
SYMBOLS = ["SPY", "QQQ", "AAPL", "NVDA", "TSLA"]
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.
"""
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))
for h in horizons:
df[f'fwd_ret_{h}d'] = df['close'].shift(-h) / df['close'] - 1
return df
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))
axes[0, 0].hist(signal_df[signal_col].dropna(), bins=50, edgecolor='black')
axes[0, 0].set_title(f'{signal_col} Distribution')
axes[0, 1].plot(signal_df['trade_date'], signal_df[signal_col])
axes[0, 1].set_title(f'{signal_col} Over Time')
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()
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()
df['signal_z'] = (df[signal_col] - df[signal_col].mean()) / df[signal_col].std()
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,
}
def main():
print("=" * 70)
print("XXX - [EXPERIMENT NAME] - EXPLORATION")
print("=" * 70)
conn = get_connection()
print("\n๐ Step 1: Data Availability")
availability = explore_data_availability(conn)
print(availability.to_string())
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
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
signal_col = 'total_flow'
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
print("\n" + "=" * 70)
print("EXPLORATION SUMMARY")
print("=" * 70)
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
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)
RESULTS_DIR = Path(__file__).parent / "results"
RESULTS_DIR.mkdir(exist_ok=True)
TRAIN_PERIODS = [("2023-01-01", "2023-12-31"), ("2024-01-01", "2024-12-31")]
TEST_PERIOD = ("2025-01-01", "2025-12-31")
N_TRIALS = 100
EARLY_STOP_PATIENCE = 30
MIN_TRADES = 15
_trial_count = 0
_best_score = -100
UNIVERSES = [
{
"name": "Universe-Name",
"symbols": ["SPY", "QQQ", "TSLA", "NVDA"],
"description": "Description of this universe",
},
]
def evaluate_params(
conn,
symbols: list[str],
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:
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),
}
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
param1 = trial.suggest_float("param1", 1.0, 5.0, step=0.1)
param2 = trial.suggest_int("param2", 5, 30, step=5)
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
trade_factor = min(1.0, np.sqrt(train_trades / 100))
score = float(train_avg * trade_factor)
if score > _best_score:
_best_score = score
print(f" ๐ฏ Trial {_trial_count}: NEW BEST score={score:.2f}")
trial.set_user_attr("train_avg", train_avg)
trial.set_user_attr("train_trades", train_trades)
return score
return objective
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()
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()
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"}
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"]
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)
with open(RESULTS_DIR / "intermediate_results.json", "w") as f:
json.dump({"results": all_results}, f, indent=2, default=str)
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
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.*")
RESULTS_DIR = Path(__file__).parent / "results_portfolio"
RESULTS_DIR.mkdir(exist_ok=True)
OPTIMIZATION_RESULTS = Path(__file__).parent / "results" / "final_results.json"
INITIAL_CAPITAL = 100_000
POSITION_SIZE_PCT = 0.02
MAX_POSITIONS = 50
MAX_POSITIONS_PER_STRATEGY = 10
COMMISSION_PER_TRADE = 1.00
SLIPPAGE_PCT = 0.05
BACKTEST_START = "2025-01-01"
BACKTEST_END = "2025-12-31"
@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
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", [])),
**params,
})
return strategies
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")
equity = INITIAL_CAPITAL
positions: dict[str, Position] = {}
trades = []
daily_equity = []
trading_dates = pd.date_range(BACKTEST_START, BACKTEST_END, freq='B')
for date in trading_dates:
date_str = date.strftime("%Y-%m-%d")
daily_equity.append({
"date": date_str,
"equity": equity,
"positions": len(positions),
})
pd.DataFrame(daily_equity).to_csv(RESULTS_DIR / "daily_equity.csv", index=False)
pd.DataFrame(trades).to_csv(RESULTS_DIR / "trades.csv", index=False)
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 = 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
SLIPPAGE_PCT = 0.05
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
cd /Users/adamsohn/Projects/volatio
source research/.venv/bin/activate
python research/experiments/options/laplace/XXX-experiment-name/run_optimization.py
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'
"""
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
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 = df.groupby('date')['intraday_range'].last()
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)
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')
all_dates = sorted(df['date'].unique())
date_idx = list(all_dates).index(test_date)
prior_20 = all_dates[date_idx - 20 : date_idx]
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)
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:
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
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):
actual_exit = min(bar_high, tp_level)
if bar_high >= tp_level:
return bar_high, 'TAKE_PROFIT'
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']
For same-day strategies:
for t in trades:
assert t['exit_date'] == t['entry_date']
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
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
capital += dollar_pnl
expected = INITIAL_CAPITAL + unrounded_pnl_sum
actual = capital
assert abs(expected - actual) < 0.001
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]
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
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
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 = df['close'] > df['close'].shift(1)
entry_price = df['close']
Fix:
signal = df['close'].shift(1) > df['close'].shift(2)
entry_price = df['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']
Fix:
df['iv_mean'] = df['iv'].rolling(20).mean().shift(1)
df['iv_zscore'] = (df['iv'] - df['iv_mean']) / df['iv_std']
Pattern 3: Exit Price Outside Bar Range
Bug:
stop_price = entry_price * (1 - stop_pct)
if bar_low <= stop_price:
exit_price = stop_price
Fix:
if bar_low <= stop_price:
exit_price = bar_low
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
