Spaces:
Configuration error
Configuration error
Niranjan-Murugarasu024
feat: TradeSense AI Chairman Edition production release ready for Hugging Face Spaces
0125724 Download quant/backtesting/engine.py from BlueMen/tradesenseAi: direct link, hf CLI and curl.
- Browser
- Download file 12.5 kB
-
https://huggingface.co/spaces/BlueMen/tradesenseAi/resolve/main/quant/backtesting/engine.py
- Command line
-
hf download hf://spaces/BlueMen/tradesenseAi/quant/backtesting/engine.py
-
curl -L -o engine.py https://huggingface.co/spaces/BlueMen/tradesenseAi/resolve/main/quant/backtesting/engine.py
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 | |