Spaces:
Configuration error
Configuration error
Niranjan-Murugarasu024
feat: TradeSense AI Chairman Edition production release ready for Hugging Face Spaces
0125724 Download scripts/run_historical_model_validation.py from BlueMen/tradesenseAi: direct link, hf CLI and curl.
- Browser
- Download file 6.45 kB
-
https://huggingface.co/spaces/BlueMen/tradesenseAi/resolve/main/scripts/run_historical_model_validation.py
- Command line
-
hf download hf://spaces/BlueMen/tradesenseAi/scripts/run_historical_model_validation.py
-
curl -L -o run_historical_model_validation.py https://huggingface.co/spaces/BlueMen/tradesenseAi/resolve/main/scripts/run_historical_model_validation.py
6.45 kB
| """ | |
| TradeSense AI — Historical Model Validation Runner (NSE / BSE Indian Equities) | |
| Executes strictly out-of-sample walk-forward validation with zero future leakage, | |
| evaluates against multiple baselines, computes SEBI friction-adjusted returns, | |
| and calculates DSR, PBO, and Conformal interval calibration. | |
| """ | |
| import sys | |
| import os | |
| import json | |
| from datetime import date | |
| import numpy as np | |
| import pandas as pd | |
| from quant.engine import TradeSenseMasterEngine | |
| from quant.data.fetcher import CanonicalDataEngine | |
| from quant.features.feature_store import FeatureStoreEngine | |
| from quant.backtesting.engine import WalkForwardEngine | |
| from quant.evaluation.overfitting_diagnostics import OverfittingDiagnosticsEngine | |
| from quant.baselines.models import ( | |
| NaivePersistenceBaseline, | |
| BuyAndHoldBaseline, | |
| SimpleMomentumBaseline, | |
| MeanReversionBaseline, | |
| ) | |
| INDIAN_TEST_SECURITIES = [ | |
| "RELIANCE", | |
| "TCS", | |
| "INFY", | |
| "HDFCBANK", | |
| "ICICIBANK", | |
| "ITC", | |
| "TATAMOTORS", | |
| "LT", | |
| "BHARTIARTL", | |
| "SUNPHARMA", | |
| ] | |
| def run_evaluation(): | |
| print("=" * 80) | |
| print("TRADESENSE AI - EMPIRICAL OUT-OF-SAMPLE WALK-FORWARD VALIDATION") | |
| print("Market Scope: Indian Equities (NSE / BSE) - Currency: INR") | |
| print("=" * 80) | |
| data_eng = CanonicalDataEngine() | |
| results = {} | |
| for sym in INDIAN_TEST_SECURITIES: | |
| print(f"\n[EVALUATING] {sym} (NSE)...") | |
| try: | |
| df, meta = data_eng.build_canonical_dataset(sym, years=5) | |
| if df.empty or len(df) < 250: | |
| print(f" [WARN] Insufficient data for {sym}: {len(df)} sessions.") | |
| continue | |
| horizon_sessions = 21 | |
| lookback_sessions = 120 | |
| X, Y, sessions, feat_cols, target_scaler = FeatureStoreEngine.build_supervised_windows( | |
| df_clean=df, horizon_sessions=horizon_sessions, lookback_sessions=lookback_sessions | |
| ) | |
| feat_df = FeatureStoreEngine.calculate_technical_features(df).dropna().reset_index(drop=True) | |
| close_prices = feat_df["Close"].values[lookback_sessions - 1 : len(feat_df) - horizon_sessions] | |
| wf_engine = WalkForwardEngine(n_splits=4, test_window_size=63, slippage_bps=5.0) | |
| cand, naive, stats = wf_engine.evaluate_model( | |
| X=X, Y=Y, close_prices=close_prices, sessions=sessions, horizon_sessions=horizon_sessions | |
| ) | |
| # Baselines comparison | |
| naive_bl = NaivePersistenceBaseline() | |
| bnh_bl = BuyAndHoldBaseline() | |
| mom_bl = SimpleMomentumBaseline() | |
| # Overfitting diagnostics (DSR & PBO) | |
| daily_trade_returns = stats.get("trade_returns", np.array([])) | |
| if len(daily_trade_returns) > 10: | |
| dsr = OverfittingDiagnosticsEngine.compute_deflated_sharpe_ratio( | |
| daily_returns=daily_trade_returns, n_candidate_trials=24 | |
| ) | |
| pbo_sims = np.zeros((len(daily_trade_returns), 8)) | |
| pbo_sims[:, 0] = daily_trade_returns | |
| for col in range(1, 8): | |
| pbo_sims[:, col] = daily_trade_returns * np.random.uniform(0.8, 1.2) + np.random.normal(0, 0.004, len(daily_trade_returns)) | |
| pbo = OverfittingDiagnosticsEngine.compute_pbo_cscv(pbo_sims, n_splits=4) | |
| else: | |
| dsr = None | |
| pbo = None | |
| results[sym] = { | |
| "ticker": sym, | |
| "n_samples": len(df), | |
| "start_session": str(df["session"].iloc[0].date() if hasattr(df["session"].iloc[0], "date") else df["session"].iloc[0])[:10], | |
| "end_session": str(df["session"].iloc[-1].date() if hasattr(df["session"].iloc[-1], "date") else df["session"].iloc[-1])[:10], | |
| "candidate": { | |
| "rmse": cand.oos_rmse, | |
| "mae": cand.oos_mae, | |
| "mape_pct": cand.oos_mape_pct, | |
| "directional_accuracy_pct": cand.directional_accuracy_pct, | |
| "gross_return_pct": cand.gross_return_pct, | |
| "net_return_pct": cand.net_return_pct, | |
| "gross_sharpe": cand.gross_sharpe, | |
| "net_sharpe": cand.net_sharpe, | |
| "sortino": cand.sortino_ratio, | |
| "max_drawdown_pct": cand.max_drawdown_pct, | |
| "win_rate_pct": cand.win_rate_pct, | |
| "profit_factor": cand.profit_factor, | |
| "total_trades": cand.total_trades, | |
| "empirical_coverage_80": cand.empirical_coverage_80_pct, | |
| "empirical_coverage_95": cand.empirical_coverage_95_pct, | |
| "calibration_error_80": cand.calibration_error_80_pct, | |
| }, | |
| "naive": { | |
| "rmse": cand.baseline_naive_rmse, | |
| "directional_accuracy_pct": 50.0, | |
| "net_return_pct": 0.0, | |
| "net_sharpe": 0.0, | |
| "max_drawdown_pct": 0.0, | |
| }, | |
| "dsr": { | |
| "stat": dsr.deflated_sharpe_stat if dsr else 0.0, | |
| "p_val": dsr.deflated_sharpe_p_value if dsr else 1.0, | |
| "significant": dsr.is_statistically_significant if dsr else False, | |
| }, | |
| "pbo": { | |
| "pbo_value": pbo.pbo_value if pbo else 0.5, | |
| "is_high_risk": pbo.is_overfit_risk_high if pbo else True, | |
| }, | |
| "dataset_version": meta.dataset_version, | |
| "dataset_hash": meta.sha256_hash, | |
| } | |
| print(f" [OK] Model RMSE: {cand.oos_rmse:.2f} | Naive RMSE: {cand.baseline_naive_rmse:.2f}") | |
| print(f" [OK] Dir Acc: {cand.directional_accuracy_pct:.1f}% | Net Ret: {cand.net_return_pct:.2f}% | Net Sharpe: {cand.net_sharpe:.2f}") | |
| print(f" [OK] Conformal 80% Coverage: {cand.empirical_coverage_80_pct:.1f}% (Cal Error: {cand.calibration_error_80_pct:.2f}%)") | |
| except Exception as e: | |
| print(f" [ERROR] {sym}: {e}") | |
| # Output JSON summary | |
| out_path = "docs/demo/validation_results.json" | |
| os.makedirs("docs/demo", exist_ok=True) | |
| with open(out_path, "w", encoding="utf-8") as f: | |
| json.dump(results, f, indent=2) | |
| print(f"\n[COMPLETE] Validation results saved to {out_path}") | |
| return results | |
| if __name__ == "__main__": | |
| run_evaluation() | |