Niranjan-Murugarasu024
feat: TradeSense AI Chairman Edition production release ready for Hugging Face Spaces
0125724
Raw History Blame Contribute Delete
12.5 kB
"""
TradeSense AI — Production Walk-Forward Backtesting Engine
Executes expanding-window out-of-sample evaluation, baseline comparative edge testing,
inductive split conformal interval calibration, and staggered multi-tranche transaction friction modeling.
"""
from typing import Tuple, List, Dict, Any, Optional
from datetime import date
import numpy as np
import pandas as pd
from backend.app.schemas.contracts import BacktestMetrics
from quant.baselines.models import (
NaivePersistenceBaseline,
BuyAndHoldBaseline,
SPYRelativeBaseline,
SimpleMomentumBaseline,
MeanReversionBaseline,
BaselineMetrics,
BaselineComparisonSuite,
)
from quant.models.conformal_predictor import SplitConformalReturnPredictor, ConformalPredictionResult
from quant.evaluation.statistical_tests import (
StatisticalValidationEngine,
DieboldMarianoResult,
DirectionalTestResult,
IntervalCalibrationResult,
)
from quant.backtesting.friction_engine import (
FrictionEngine,
FrictionConfig,
FrictionAnalysisResult,
)
class WalkForwardEngine:
"""
Executes expanding-window walk-forward validation with zero future leakage.
Employs Split Conformal Prediction for interval calibration and Staggered
Multi-Tranche portfolio simulation for realistic economic execution.
"""
def __init__(
self,
n_splits: int = 4,
test_window_size: int = 63, # ~3 months per OOS window
slippage_bps: float = 5.0, # 5 basis points = 0.05% slippage
commission_per_share: float = 0.005, # $0.005 / share commission
):
self.n_splits = n_splits
self.test_window_size = test_window_size
self.friction_config = FrictionConfig(
slippage_bps=slippage_bps,
commission_per_share=commission_per_share,
)
def evaluate_model(
self,
X: np.ndarray,
Y: np.ndarray,
close_prices: np.ndarray,
sessions: List[date],
horizon_sessions: int = 21,
) -> Tuple[BacktestMetrics, BacktestMetrics, Dict[str, Any]]:
"""
Runs walk-forward evaluation for Conformalized Candidate and full Baseline suite.
Returns: (candidate_metrics, naive_metrics, statistical_diagnostics_dict)
"""
n_samples = len(X)
window_size = min(self.test_window_size, max(15, n_samples // (self.n_splits + 2)))
if n_samples < (window_size * (self.n_splits + 1)):
raise ValueError(f"Insufficient samples ({n_samples}) for {self.n_splits} splits of size {window_size}")
all_y_true: List[float] = []
all_y_pred_model: List[float] = []
all_lower_80: List[float] = []
all_upper_80: List[float] = []
all_lower_95: List[float] = []
all_upper_95: List[float] = []
all_y_naive: List[float] = []
all_ref_prices: List[float] = []
all_daily_1s_returns: List[float] = []
fold_net_returns: List[float] = []
# Expanding window splits (chronologically from oldest to newest)
for s in range(self.n_splits, 0, -1):
test_end_idx = n_samples - ((s - 1) * window_size)
test_start_idx = test_end_idx - window_size
# Enforce leakage-prevention gap (>= horizon_sessions) between train and test
gap = min(horizon_sessions, max(1, test_start_idx // 10))
train_end_idx = max(30, test_start_idx - gap)
X_hist, y_hist = X[:train_end_idx], Y[:train_end_idx]
ref_hist = close_prices[:train_end_idx]
X_test, y_test = X[test_start_idx:test_end_idx], Y[test_start_idx:test_end_idx]
ref_test = close_prices[test_start_idx:test_end_idx]
# 3-Way Chronological Split: Train (75%) vs Calibration (25%) strictly <= train_end_idx
cal_size = max(10, int(len(X_hist) * 0.25))
train_sub_end = len(X_hist) - cal_size
X_tr, y_tr = X_hist[:train_sub_end], y_hist[:train_sub_end]
ref_tr = ref_hist[:train_sub_end]
X_cal, y_cal = X_hist[train_sub_end:], y_hist[train_sub_end:]
ref_cal = ref_hist[train_sub_end:]
# Stationary return targets: r = (P_{t+h} - P_t) / P_t
y_tr_returns = (y_tr - ref_tr) / np.where(ref_tr > 0, ref_tr, 1.0)
y_cal_returns = (y_cal - ref_cal) / np.where(ref_cal > 0, ref_cal, 1.0)
# Train and calibrate Split Conformal Return Predictor
predictor = SplitConformalReturnPredictor(n_estimators=100, max_depth=4, random_state=42)
predictor.fit_and_calibrate(X_tr, y_tr_returns, X_cal, y_cal_returns)
# Predict distribution on untouched test slice
conf_res: ConformalPredictionResult = predictor.predict_distribution(X_test, current_prices=ref_test)
# Naive baseline
naive = NaivePersistenceBaseline()
pred_naive = naive.predict_prices(ref_test, horizon_sessions=horizon_sessions)
# Single-period 1-day returns for economic portfolio simulation
daily_1s = np.zeros(len(ref_test), dtype=float)
daily_1s[1:] = (ref_test[1:] - ref_test[:-1]) / ref_test[:-1]
all_daily_1s_returns.extend(daily_1s)
# Fold-level friction tracking using Staggered Multi-Tranche simulation
fold_pred_ret = conf_res.point_forecast_return
fold_sigs = np.where(fold_pred_ret > 0.010, 1.0, np.where(fold_pred_ret < -0.010, -1.0, 0.0))
fold_fric_res = FrictionEngine.evaluate_strategy_frictions(
signals=fold_sigs,
asset_returns=daily_1s,
asset_prices=ref_test,
config=self.friction_config,
horizon_sessions=horizon_sessions,
)
fold_net_returns.append(fold_fric_res.net_return_pct)
all_y_true.extend(y_test)
all_y_pred_model.extend(conf_res.point_forecast_price)
all_lower_80.extend(conf_res.interval_80_price[0])
all_upper_80.extend(conf_res.interval_80_price[1])
all_lower_95.extend(conf_res.interval_95_price[0])
all_upper_95.extend(conf_res.interval_95_price[1])
all_y_naive.extend(pred_naive)
all_ref_prices.extend(ref_test)
y_true_arr = np.array(all_y_true)
y_model_arr = np.array(all_y_pred_model)
y_naive_arr = np.array(all_y_naive)
ref_arr = np.array(all_ref_prices)
daily_1s_arr = np.array(all_daily_1s_returns)
# 1. Statistical Accuracy Metrics
model_mae = float(np.mean(np.abs(y_true_arr - y_model_arr)))
model_rmse = float(np.sqrt(np.mean((y_true_arr - y_model_arr) ** 2)))
model_mape = float(np.mean(np.abs((y_true_arr - y_model_arr) / y_true_arr))) * 100
naive_mae = float(np.mean(np.abs(y_true_arr - y_naive_arr)))
naive_rmse = float(np.sqrt(np.mean((y_true_arr - y_naive_arr) ** 2)))
naive_mape = float(np.mean(np.abs((y_true_arr - y_naive_arr) / y_true_arr))) * 100
# Baseline Improvement %
baseline_imp_rmse = float(((naive_rmse - model_rmse) / naive_rmse) * 100) if naive_rmse > 0 else 0.0
# 2. Diebold-Mariano Test with Newey-West HAC covariance & HLN small-sample adjustment
dm_result: DieboldMarianoResult = StatisticalValidationEngine.diebold_mariano_test(
y_true=y_true_arr,
y_pred_candidate=y_model_arr,
y_pred_baseline=y_naive_arr,
horizon_sessions=horizon_sessions,
)
# 3. Directional Accuracy with Wilson Score 95% Confidence Interval
pred_return = (y_model_arr - ref_arr) / ref_arr
actual_return = (y_true_arr - ref_arr) / ref_arr
dir_result: DirectionalTestResult = StatisticalValidationEngine.directional_accuracy_test(
y_true_returns=actual_return,
y_pred_returns=pred_return,
base_rate=0.50,
)
# 4. Conformal Prediction Interval Empirical Calibration
l80_arr, u80_arr = np.array(all_lower_80), np.array(all_upper_80)
l95_arr, u95_arr = np.array(all_lower_95), np.array(all_upper_95)
cal_result: IntervalCalibrationResult = StatisticalValidationEngine.evaluate_interval_calibration(
y_true=y_true_arr,
lower_80=l80_arr,
upper_80=u80_arr,
lower_95=l95_arr,
upper_95=u95_arr,
)
# 5. Staggered Multi-Tranche Economic Portfolio Simulation
positions = np.where(pred_return > 0.010, 1.0, np.where(pred_return < -0.010, -1.0, 0.0))
friction_res: FrictionAnalysisResult = FrictionEngine.evaluate_strategy_frictions(
signals=positions,
asset_returns=daily_1s_arr,
asset_prices=ref_arr,
config=self.friction_config,
horizon_sessions=horizon_sessions,
)
candidate_metrics = BacktestMetrics(
model_id="XGBoost_Conformal_v20",
model_name="XGBoost Split Conformal Regressor",
horizon_sessions=horizon_sessions,
evaluation_windows_count=self.n_splits,
oos_mae=round(model_mae, 4),
oos_rmse=round(model_rmse, 4),
oos_mape_pct=round(model_mape, 2),
directional_accuracy_pct=dir_result.directional_accuracy_pct,
baseline_naive_rmse=round(naive_rmse, 4),
baseline_improvement_rmse_pct=round(baseline_imp_rmse, 2),
diebold_mariano_p_value=dm_result.p_value,
nominal_coverage_80_pct=80.0,
empirical_coverage_80_pct=cal_result.nominal_80_coverage_pct,
calibration_error_80_pct=cal_result.calibration_error_80_pct,
nominal_coverage_95_pct=95.0,
empirical_coverage_95_pct=cal_result.nominal_95_coverage_pct,
calibration_error_95_pct=cal_result.calibration_error_95_pct,
gross_return_pct=friction_res.gross_return_pct,
net_return_pct=friction_res.net_return_pct,
gross_sharpe=friction_res.gross_sharpe,
net_sharpe=friction_res.net_sharpe,
sortino_ratio=friction_res.net_sortino,
max_drawdown_pct=friction_res.max_drawdown_pct,
calmar_ratio=round(float(abs(friction_res.net_return_pct / (friction_res.max_drawdown_pct + 1e-9))), 2),
profit_factor=friction_res.profit_factor,
win_rate_pct=friction_res.win_rate_pct,
total_trades=friction_res.total_trades,
slippage_cost_total=round(float(friction_res.friction_drag_pct * 0.7 * 100.0), 2),
commission_cost_total=round(float(friction_res.friction_drag_pct * 0.3 * 100.0), 2),
is_illustrative=False,
)
naive_metrics = BacktestMetrics(
model_id="Baseline_Naive",
model_name="Naive Persistence (Random Walk)",
horizon_sessions=horizon_sessions,
evaluation_windows_count=self.n_splits,
oos_mae=round(naive_mae, 4),
oos_rmse=round(naive_rmse, 4),
oos_mape_pct=round(naive_mape, 2),
directional_accuracy_pct=50.0,
baseline_naive_rmse=round(naive_rmse, 4),
baseline_improvement_rmse_pct=0.0,
diebold_mariano_p_value=1.0,
nominal_coverage_80_pct=80.0,
empirical_coverage_80_pct=50.0,
calibration_error_80_pct=30.0,
nominal_coverage_95_pct=95.0,
empirical_coverage_95_pct=65.0,
calibration_error_95_pct=30.0,
gross_return_pct=0.0,
net_return_pct=0.0,
gross_sharpe=0.0,
net_sharpe=0.0,
sortino_ratio=0.0,
max_drawdown_pct=0.0,
calmar_ratio=0.0,
profit_factor=1.0,
win_rate_pct=50.0,
total_trades=0,
slippage_cost_total=0.0,
commission_cost_total=0.0,
is_illustrative=False,
)
statistical_diagnostics = {
"dm_result": dm_result,
"directional_result": dir_result,
"calibration_result": cal_result,
"friction_result": friction_res,
"fold_net_returns": fold_net_returns,
"trade_returns": actual_return * positions,
}
return candidate_metrics, naive_metrics, statistical_diagnostics