🇪🇸 Leer en Español | 🇺🇸 English
A robust testing framework is the difference between a strategy that works in backtest and one that generates real alpha in production. This framework integrates statistical validation, robustness testing, and risk evaluation to ensure strategies are deployment-ready.
import pandas as pd
import numpy as np
from typing import Dict, List, Any, Tuple, Optional
from dataclasses import dataclass
from abc import ABC, abstractmethod
import warnings
@dataclass
class TestResult:
"""Standardized test result"""
test_name: str
passed: bool
score: float
details: Dict[str, Any]
recommendations: List[str]
critical: bool = False
class StrategyTester(ABC):
"""Base class for all strategy testers"""
@abstractmethod
def run_test(self, strategy_results: Dict[str, Any]) -> TestResult:
pass
class StrategyTestingFramework:
def __init__(self, strict_mode: bool = True):
self.strict_mode = strict_mode
self.testers = []
self.test_results = []
# Register default testers
self._register_default_testers()
def _register_default_testers(self):
"""Registers standard framework testers"""
self.register_tester(StatisticalSignificanceTester())
self.register_tester(RobustnessTester())
self.register_tester(RiskProfileTester())
self.register_tester(OverfittingTester())
self.register_tester(MarketRegimeTester())
self.register_tester(ExecutionFeasibilityTester())
self.register_tester(DrawdownTester())
self.register_tester(ConsistencyTester())
def register_tester(self, tester: StrategyTester):
"""Registers a new tester"""
self.testers.append(tester)
def run_full_testing_suite(self, strategy_results: Dict[str, Any]) -> Dict[str, Any]:
"""
Runs the full testing suite
"""
print("Testing Framework Starting...")
test_results = []
critical_failures = []
for tester in self.testers:
print(f" Running: {tester.__class__.__name__}")
try:
result = tester.run_test(strategy_results)
test_results.append(result)
if result.critical and not result.passed:
critical_failures.append(result)
except Exception as e:
error_result = TestResult(
test_name=tester.__class__.__name__,
passed=False,
score=0.0,
details={'error': str(e)},
recommendations=[f"Fix error in {tester.__class__.__name__}: {e}"],
critical=True
)
test_results.append(error_result)
critical_failures.append(error_result)
# Calculate overall score
overall_score = self._calculate_overall_score(test_results)
# Generate report
testing_report = self._generate_testing_report(test_results, overall_score)
# Deployment decision
deployment_ready = self._assess_deployment_readiness(
test_results, critical_failures, overall_score
)
return {
'test_results': test_results,
'critical_failures': critical_failures,
'overall_score': overall_score,
'testing_report': testing_report,
'deployment_ready': deployment_ready,
'recommendations': self._aggregate_recommendations(test_results)
}
def _calculate_overall_score(self, test_results: List[TestResult]) -> float:
"""Calculates weighted overall score"""
if not test_results:
return 0.0
# Weights by criticality
weights = []
scores = []
for result in test_results:
weight = 2.0 if result.critical else 1.0
weights.append(weight)
scores.append(result.score)
weighted_score = np.average(scores, weights=weights)
return float(weighted_score)
def _assess_deployment_readiness(self, test_results: List[TestResult],
critical_failures: List[TestResult],
overall_score: float) -> Dict[str, Any]:
"""Assesses whether the strategy is ready for deployment"""
# Deployment criteria
criteria = {
'no_critical_failures': len(critical_failures) == 0,
'minimum_score': overall_score >= 0.7, # 70% mínimo
'sufficient_tests_passed': sum(1 for r in test_results if r.passed) >= len(test_results) * 0.8
}
deployment_ready = all(criteria.values())
confidence_level = "HIGH" if overall_score >= 0.8 else "MEDIUM" if overall_score >= 0.6 else "LOW"
return {
'ready': deployment_ready,
'confidence': confidence_level,
'criteria_met': criteria,
'blocking_issues': [cf.test_name for cf in critical_failures]
}class StatisticalSignificanceTester(StrategyTester):
def __init__(self, min_trades: int = 30, confidence_level: float = 0.05):
self.min_trades = min_trades
self.confidence_level = confidence_level
def run_test(self, strategy_results: Dict[str, Any]) -> TestResult:
"""
Validates statistical significance of results
"""
trades = strategy_results.get('trades', [])
returns = strategy_results.get('returns', [])
if len(trades) < self.min_trades:
return TestResult(
test_name="Statistical Significance",
passed=False,
score=0.0,
details={'trades_count': len(trades), 'minimum_required': self.min_trades},
recommendations=[f"Need at least {self.min_trades} trades for statistical validity"],
critical=True
)
# Student t-test for returns
from scipy import stats
returns_array = np.array(returns)
# Normality test
normality_test = stats.jarque_bera(returns_array)
is_normal = normality_test.pvalue > self.confidence_level
# T-test for mean != 0
if is_normal:
t_stat, p_value = stats.ttest_1samp(returns_array, 0)
else:
# Use non-parametric test if not normal
t_stat, p_value = stats.wilcoxon(returns_array)
# Autocorrelation test (independence)
from statsmodels.stats.diagnostic import acorr_ljungbox
lb_test = acorr_ljungbox(returns_array, lags=[10], return_df=True)
independence = lb_test['lb_pvalue'].iloc[0] > self.confidence_level
# Sharpe ratio test
sharpe_ratio = np.mean(returns_array) / np.std(returns_array) * np.sqrt(252)
sharpe_se = np.sqrt((1 + 0.5 * sharpe_ratio**2) / len(returns_array))
sharpe_t_stat = sharpe_ratio / sharpe_se
sharpe_p_value = 2 * (1 - stats.norm.cdf(abs(sharpe_t_stat)))
# Composite score
tests_passed = sum([
p_value < self.confidence_level, # Retornos significativos
independence, # Independencia
sharpe_p_value < self.confidence_level # Sharpe significativo
])
score = tests_passed / 3.0
passed = score >= 0.67 # At least 2 of 3 tests
details = {
'n_trades': len(trades),
'mean_return': float(np.mean(returns_array)),
'return_t_stat': float(t_stat),
'return_p_value': float(p_value),
'is_normal': is_normal,
'independence': independence,
'sharpe_ratio': float(sharpe_ratio),
'sharpe_p_value': float(sharpe_p_value),
'tests_passed': tests_passed
}
recommendations = []
if not passed:
if p_value >= self.confidence_level:
recommendations.append("Returns are not statistically significant")
if not independence:
recommendations.append("Returns show autocorrelation - check for data snooping")
if sharpe_p_value >= self.confidence_level:
recommendations.append("Sharpe ratio is not statistically significant")
return TestResult(
test_name="Statistical Significance",
passed=passed,
score=score,
details=details,
recommendations=recommendations,
critical=True
)class RobustnessTester(StrategyTester):
def __init__(self, parameter_sensitivity_threshold: float = 0.3):
self.sensitivity_threshold = parameter_sensitivity_threshold
def run_test(self, strategy_results: Dict[str, Any]) -> TestResult:
"""
Evaluates strategy robustness against parameter changes
"""
# Get sensitivity analysis results if available
sensitivity_data = strategy_results.get('sensitivity_analysis', {})
if not sensitivity_data:
return self._run_basic_robustness_test(strategy_results)
# Analyze parameter sensitivity
param_robustness_scores = []
sensitive_params = []
for param_name, param_results in sensitivity_data.items():
if isinstance(param_results, dict) and 'sensitivity_score' in param_results:
sensitivity = param_results['sensitivity_score']
# Score: less sensitivity = more robust
robustness_score = max(0, 1 - sensitivity)
param_robustness_scores.append(robustness_score)
if sensitivity > self.sensitivity_threshold:
sensitive_params.append(param_name)
# Overall robustness score
if param_robustness_scores:
overall_robustness = np.mean(param_robustness_scores)
else:
overall_robustness = 0.5 # Neutral score if no data
# Temporal stability test
temporal_stability = self._test_temporal_stability(strategy_results)
# Combine scores
final_score = (overall_robustness * 0.6 + temporal_stability * 0.4)
passed = final_score >= 0.6 and len(sensitive_params) <= 2
details = {
'parameter_robustness': float(overall_robustness),
'temporal_stability': float(temporal_stability),
'sensitive_parameters': sensitive_params,
'parameter_scores': {
param: score for param, score in zip(sensitivity_data.keys(), param_robustness_scores)
}
}
recommendations = []
if not passed:
if overall_robustness < 0.6:
recommendations.append("Strategy is too sensitive to parameter changes")
if len(sensitive_params) > 2:
recommendations.append(f"Too many sensitive parameters: {sensitive_params}")
if temporal_stability < 0.5:
recommendations.append("Strategy shows temporal instability")
return TestResult(
test_name="Robustness",
passed=passed,
score=final_score,
details=details,
recommendations=recommendations,
critical=True
)
def _test_temporal_stability(self, strategy_results: Dict[str, Any]) -> float:
"""Temporal stability test using rolling windows"""
trades = strategy_results.get('trades', [])
if len(trades) < 50:
return 0.5 # Neutral score for few trades
# Split trades into periods
n_periods = 5
period_size = len(trades) // n_periods
period_returns = []
for i in range(n_periods):
start_idx = i * period_size
end_idx = start_idx + period_size if i < n_periods - 1 else len(trades)
period_trades = trades[start_idx:end_idx]
returns = [trade.get('return', 0) for trade in period_trades]
if returns:
period_return = np.mean(returns)
period_returns.append(period_return)
if len(period_returns) < 3:
return 0.5
# Calculate stability as inverse of coefficient of variation
cv = np.std(period_returns) / abs(np.mean(period_returns)) if np.mean(period_returns) != 0 else float('inf')
stability = max(0, 1 - cv) if cv != float('inf') else 0
return min(1.0, stability)class OverfittingTester(StrategyTester):
def __init__(self):
self.overfitting_indicators = {
'is_oos_degradation': 0.3, # Degradación > 30% en OOS
'parameter_count': 0.2, # Muchos parámetros
'complexity_score': 0.3, # Complejidad del modelo
'stability_score': 0.2 # Estabilidad temporal
}
def run_test(self, strategy_results: Dict[str, Any]) -> TestResult:
"""
Detects overfitting signals in the strategy
"""
# 1. In-Sample vs Out-of-Sample comparison
is_performance = strategy_results.get('in_sample_metrics', {})
oos_performance = strategy_results.get('out_of_sample_metrics', {})
degradation_score = self._calculate_degradation_score(is_performance, oos_performance)
# 2. Parameter complexity analysis
parameters = strategy_results.get('parameters', {})
parameter_complexity_score = self._assess_parameter_complexity(parameters)
# 3. Model complexity
model_complexity = strategy_results.get('model_complexity', {})
complexity_score = self._assess_model_complexity(model_complexity)
# 4. Parameter stability
param_evolution = strategy_results.get('parameter_evolution', [])
stability_score = self._assess_parameter_stability(param_evolution)
# Composite score (invertido: menos overfitting = mejor score)
overfitting_risk = (
degradation_score * self.overfitting_indicators['is_oos_degradation'] +
parameter_complexity_score * self.overfitting_indicators['parameter_count'] +
complexity_score * self.overfitting_indicators['complexity_score'] +
(1 - stability_score) * self.overfitting_indicators['stability_score']
)
final_score = max(0, 1 - overfitting_risk)
passed = final_score >= 0.6 # Low overfitting risk
details = {
'overfitting_risk': float(overfitting_risk),
'degradation_score': float(degradation_score),
'parameter_complexity': float(parameter_complexity_score),
'model_complexity': float(complexity_score),
'stability_score': float(stability_score),
'risk_breakdown': {
'is_oos_degradation': degradation_score,
'parameter_count': parameter_complexity_score,
'model_complexity': complexity_score,
'parameter_stability': 1 - stability_score
}
}
recommendations = []
if not passed:
if degradation_score > 0.5:
recommendations.append("Significant performance degradation in out-of-sample")
if parameter_complexity_score > 0.7:
recommendations.append("Too many parameters - consider simplification")
if complexity_score > 0.7:
recommendations.append("Model is too complex - risk of overfitting")
if stability_score < 0.5:
recommendations.append("Parameters are unstable across time periods")
return TestResult(
test_name="Overfitting Detection",
passed=passed,
score=final_score,
details=details,
recommendations=recommendations,
critical=True
)
def _calculate_degradation_score(self, is_metrics: Dict, oos_metrics: Dict) -> float:
"""Calculates IS vs OOS degradation score"""
if not is_metrics or not oos_metrics:
return 0.5 # Neutral score if no data
key_metrics = ['sharpe_ratio', 'total_return', 'win_rate']
degradations = []
for metric in key_metrics:
is_value = is_metrics.get(metric, 0)
oos_value = oos_metrics.get(metric, 0)
if is_value != 0:
degradation = max(0, (is_value - oos_value) / abs(is_value))
degradations.append(degradation)
if not degradations:
return 0.5
avg_degradation = np.mean(degradations)
return min(1.0, avg_degradation / 0.5) # Normalize: 50% degradation = score 1.0class MarketRegimeTester(StrategyTester):
def __init__(self):
self.required_regimes = ['bull', 'bear', 'sideways', 'high_volatility']
def run_test(self, strategy_results: Dict[str, Any]) -> TestResult:
"""
Evaluates performance across different market regimes
"""
regime_results = strategy_results.get('regime_analysis', {})
if not regime_results:
return self._perform_basic_regime_analysis(strategy_results)
# Evaluate performance in each regime
regime_scores = {}
regimes_tested = list(regime_results.keys())
for regime, results in regime_results.items():
regime_sharpe = results.get('sharpe_ratio', 0)
regime_return = results.get('total_return', 0)
regime_trades = results.get('trade_count', 0)
# Score per regime (metric combination)
regime_score = self._calculate_regime_score(regime_sharpe, regime_return, regime_trades)
regime_scores[regime] = regime_score
# Evaluate regime coverage
regime_coverage = len(regimes_tested) / len(self.required_regimes)
# Consistency across regimes
if regime_scores:
regime_consistency = 1 - (np.std(list(regime_scores.values())) / np.mean(list(regime_scores.values())))
regime_consistency = max(0, regime_consistency)
else:
regime_consistency = 0
# Final score
avg_regime_performance = np.mean(list(regime_scores.values())) if regime_scores else 0
final_score = (
avg_regime_performance * 0.5 +
regime_coverage * 0.3 +
regime_consistency * 0.2
)
passed = final_score >= 0.6 and regime_coverage >= 0.75
details = {
'regime_scores': regime_scores,
'regime_coverage': float(regime_coverage),
'regime_consistency': float(regime_consistency),
'regimes_tested': regimes_tested,
'missing_regimes': [r for r in self.required_regimes if r not in regimes_tested]
}
recommendations = []
if not passed:
if regime_coverage < 0.75:
recommendations.append(f"Insufficient regime coverage. Missing: {details['missing_regimes']}")
if avg_regime_performance < 0.5:
recommendations.append("Poor performance across market regimes")
if regime_consistency < 0.3:
recommendations.append("Inconsistent performance across regimes")
return TestResult(
test_name="Market Regime Analysis",
passed=passed,
score=final_score,
details=details,
recommendations=recommendations,
critical=False
)class ExecutionFeasibilityTester(StrategyTester):
def __init__(self):
self.feasibility_criteria = {
'max_position_size': 0.1, # 10% del daily volume
'max_slippage': 0.005, # 0.5% slippage máximo
'min_liquidity_ratio': 0.05, # 5% del ADV mínimo
'max_turnover': 5.0 # 5x turnover anual máximo
}
def run_test(self, strategy_results: Dict[str, Any]) -> TestResult:
"""
Evaluates execution feasibility in real markets
"""
trades = strategy_results.get('trades', [])
execution_analysis = strategy_results.get('execution_analysis', {})
# 1. Position size vs liquidity analysis
position_feasibility = self._assess_position_sizes(trades, execution_analysis)
# 2. Expected slippage analysis
slippage_analysis = self._assess_slippage_impact(trades, execution_analysis)
# 3. Turnover analysis
turnover_analysis = self._assess_turnover_requirements(trades)
# 4. Timing constraints
timing_feasibility = self._assess_timing_constraints(trades)
# Composite score
feasibility_scores = [
position_feasibility,
slippage_analysis,
turnover_analysis,
timing_feasibility
]
final_score = np.mean(feasibility_scores)
passed = final_score >= 0.7 # High standard for feasibility
details = {
'position_feasibility': float(position_feasibility),
'slippage_feasibility': float(slippage_analysis),
'turnover_feasibility': float(turnover_analysis),
'timing_feasibility': float(timing_feasibility),
'overall_feasibility': float(final_score)
}
recommendations = []
if not passed:
if position_feasibility < 0.7:
recommendations.append("Position sizes may be too large for available liquidity")
if slippage_analysis < 0.7:
recommendations.append("Expected slippage is too high for profitability")
if turnover_analysis < 0.7:
recommendations.append("Strategy requires excessive turnover")
if timing_feasibility < 0.7:
recommendations.append("Timing constraints may prevent execution")
return TestResult(
test_name="Execution Feasibility",
passed=passed,
score=final_score,
details=details,
recommendations=recommendations,
critical=True
)
def _assess_position_sizes(self, trades: List[Dict], execution_data: Dict) -> float:
"""Evaluates whether position sizes are executable"""
if not trades:
return 1.0
# Extract position sizes and volumes
position_sizes = []
liquidity_ratios = []
for trade in trades:
position_value = trade.get('position_value', 0)
daily_volume = trade.get('avg_daily_volume', 1) # Avoid division by zero
if daily_volume > 0:
liquidity_ratio = position_value / daily_volume
liquidity_ratios.append(liquidity_ratio)
if not liquidity_ratios:
return 0.5 # Neutral score
# Percentage of trades exceeding threshold
excessive_positions = sum(1 for ratio in liquidity_ratios if ratio > self.feasibility_criteria['max_position_size'])
excessive_ratio = excessive_positions / len(liquidity_ratios)
# Score: fewer excessive positions = better
return max(0, 1 - excessive_ratio * 2) # Penalize heavily
def _assess_slippage_impact(self, trades: List[Dict], execution_data: Dict) -> float:
"""Evaluates slippage impact on the strategy"""
if not trades:
return 1.0
total_gross_pnl = sum(trade.get('gross_pnl', 0) for trade in trades)
estimated_slippage_cost = sum(trade.get('estimated_slippage', 0) for trade in trades)
if total_gross_pnl <= 0:
return 0.0 # No profit to begin with
slippage_impact = estimated_slippage_cost / total_gross_pnl
# Score based on slippage impact
if slippage_impact <= 0.1: # ≤10% impact
return 1.0
elif slippage_impact <= 0.3: # ≤30% impact
return 0.7
elif slippage_impact <= 0.5: # ≤50% impact
return 0.4
else:
return 0.0 # >50% impact = not feasibleclass TestingReportGenerator:
def __init__(self):
self.report_template = """
# Strategy Testing Report
## Executive Summary
Strategy Testing Framework Results for: {strategy_name}
Overall Score: {overall_score:.2f}/1.00
Deployment Ready: {deployment_ready}
## Test Results Summary
| Test Name | Status | Score | Critical |
|-----------|--------|-------|----------|
{test_summary_table}
## Detailed Analysis
{detailed_analysis}
## Critical Issues
{critical_issues}
## Recommendations
{recommendations}
## Next Steps
{next_steps}
"""
def generate_comprehensive_report(self, testing_results: Dict[str, Any]) -> str:
"""Generates comprehensive testing report"""
# Extract data
test_results = testing_results['test_results']
overall_score = testing_results['overall_score']
deployment_ready = testing_results['deployment_ready']['ready']
# Generate test summary table
test_summary = self._generate_test_summary_table(test_results)
# Generate detailed analysis
detailed_analysis = self._generate_detailed_analysis(test_results)
# Generate critical issues
critical_issues = self._generate_critical_issues(testing_results['critical_failures'])
# Generate recommendations
recommendations = self._format_recommendations(testing_results['recommendations'])
# Generate next steps
next_steps = self._generate_next_steps(testing_results)
# Fill template
report = self.report_template.format(
strategy_name=testing_results.get('strategy_name', 'Unknown Strategy'),
overall_score=overall_score,
deployment_ready="✅ YES" if deployment_ready else "❌ NO",
test_summary_table=test_summary,
detailed_analysis=detailed_analysis,
critical_issues=critical_issues,
recommendations=recommendations,
next_steps=next_steps
)
return report
def _generate_test_summary_table(self, test_results: List[TestResult]) -> str:
"""Generates test summary table"""
table_rows = []
for result in test_results:
status = "✅ PASS" if result.passed else "❌ FAIL"
critical = "🔴 YES" if result.critical else "⚪ NO"
row = f"| {result.test_name} | {status} | {result.score:.2f} | {critical} |"
table_rows.append(row)
return "\n".join(table_rows)
def export_results_to_json(self, testing_results: Dict[str, Any], filepath: str):
"""Exports results to JSON for further analysis"""
import json
# Convert TestResult objects to dicts
serializable_results = {
'overall_score': testing_results['overall_score'],
'deployment_ready': testing_results['deployment_ready'],
'test_results': [
{
'test_name': tr.test_name,
'passed': tr.passed,
'score': tr.score,
'details': tr.details,
'recommendations': tr.recommendations,
'critical': tr.critical
}
for tr in testing_results['test_results']
],
'recommendations': testing_results['recommendations']
}
with open(filepath, 'w') as f:
json.dump(serializable_results, f, indent=2, default=str)# Complete framework usage example
def test_gap_and_go_strategy():
"""Complete testing example for Gap and Go strategy"""
# 1. Run full backtesting (simulated)
strategy_results = {
'trades': generate_sample_trades(), # List of trades
'returns': generate_sample_returns(), # List of returns
'in_sample_metrics': {'sharpe_ratio': 1.2, 'total_return': 0.15},
'out_of_sample_metrics': {'sharpe_ratio': 0.9, 'total_return': 0.12},
'parameters': {'min_gap': 0.02, 'volume_multiplier': 2.0},
'sensitivity_analysis': generate_sensitivity_data(),
'regime_analysis': generate_regime_data(),
'execution_analysis': generate_execution_data()
}
# 2. Initialize testing framework
testing_framework = StrategyTestingFramework(strict_mode=True)
# 3. Run tests
testing_results = testing_framework.run_full_testing_suite(strategy_results)
# 4. Generate report
report_generator = TestingReportGenerator()
report = report_generator.generate_comprehensive_report(testing_results)
# 5. Export results
report_generator.export_results_to_json(testing_results, 'gap_and_go_test_results.json')
# 6. Make decision
if testing_results['deployment_ready']['ready']:
print("✅ Strategy is ready for deployment!")
else:
print("❌ Strategy needs improvements before deployment")
print("Blocking issues:", testing_results['deployment_ready']['blocking_issues'])
return testing_results
if __name__ == "__main__":
results = test_gap_and_go_strategy()A robust testing framework is essential for separating strategies that work in backtest from those that will generate real alpha. This framework provides comprehensive validation including statistical significance, robustness, overfitting detection, and execution feasibility.