โ† ะะฐะทะฐะด
""" Step 8: Robustness check โ€” run top 5 candidates on 14d data. Compares 7d vs 14d performance to check if results hold on longer period. Also adds "bonus" candidates from golden combos (different Z/R:R for diversity). Usage: python3 backtests/step8_robustness.py Output: backtests/robustness_report.txt + robustness_results.json """ import os import sys import json import pickle sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from backtests.backtest_core import run_backtest BASE = os.path.dirname(__file__) def load_cache(days): path = os.path.join(BASE, f"data_cache_{days}d.pkl") if not os.path.exists(path): print(f"โŒ Not found: {path}") print(f" Run: python3 backtests/download_data.py --days {days}") sys.exit(1) with open(path, "rb") as f: cache = pickle.load(f) print(f"Loaded {days}d cache: {len(cache['symbols'])} symbols, date {cache['date']}") return cache def main(): # Load both caches cache_7d = load_cache(7) cache_14d = load_cache(14) # Load 7d candidates cand_path = os.path.join(BASE, "candidates_7d.json") with open(cand_path) as f: candidates = json.load(f) # Add bonus diverse candidates from golden combos analysis bonus = [ # Z=3.0 cluster (higher PF, fewer trades) {"z_entry": 3.0, "z_max": 3.5, "natr_min": 0.5, "natr_max": 2.0, "chop_min": 45, "tp_pct": 4.5, "sl_pct": 1.5, "rr": "3:1", "label": "Z3.0 wide CHOP"}, {"z_entry": 3.0, "z_max": 3.5, "natr_min": 0.5, "natr_max": 2.0, "chop_min": 45, "tp_pct": 3.0, "sl_pct": 1.0, "rr": "3:1", "label": "Z3.0 tighter TP/SL"}, # 2:1 R:R (higher WR) {"z_entry": 3.0, "z_max": 3.5, "natr_min": 0.5, "natr_max": 2.0, "chop_min": 45, "tp_pct": 3.0, "sl_pct": 1.5, "rr": "2:1", "label": "Z3.0 2:1 high WR"}, # 4:1 R:R {"z_entry": 3.0, "z_max": 3.5, "natr_min": 0.5, "natr_max": 2.0, "chop_min": 45, "tp_pct": 4.0, "sl_pct": 1.0, "rr": "4:1", "label": "Z3.0 4:1"}, # 5:1 R:R (high risk high reward) {"z_entry": 3.0, "z_max": 3.5, "natr_min": 0.5, "natr_max": 2.0, "chop_min": 45, "tp_pct": 5.0, "sl_pct": 1.0, "rr": "5:1", "label": "Z3.0 5:1"}, ] all_candidates = [] for i, c in enumerate(candidates): c["label"] = f"Top{i+1} from 7d" all_candidates.append(c) all_candidates.extend(bonus) print(f"\nTesting {len(all_candidates)} candidates on 7d AND 14d data...\n") report = [] report.append("=" * 90) report.append(" STEP 8: ROBUSTNESS CHECK โ€” 7d vs 14d") report.append("=" * 90) report.append("") results_json = [] header = ( f"{'#':<3} {'Label':<22} {'R:R':<4} {'TP/SL':<8} " f"{'Z':>7} {'NATR':>9} {'CH':>3} " f"| {'7d Tr':>5} {'WR':>5} {'PnL':>7} {'PF':>5} " f"| {'14d Tr':>5} {'WR':>5} {'PnL':>7} {'PF':>5} " f"| {'ฮ” PnL':>6} {'Stable?':>7}" ) report.append(header) report.append("-" * len(header)) for idx, c in enumerate(all_candidates, 1): params = { "z_entry": c["z_entry"], "z_max": c["z_max"], "natr_min": c["natr_min"], "natr_max": c["natr_max"], "chop_min": c["chop_min"], "tp_pct": c["tp_pct"], "sl_pct": c["sl_pct"], } r7 = run_backtest(cache_7d["symbols"], params) r14 = run_backtest(cache_14d["symbols"], params) # Normalize 14d PnL to 7d equivalent for fair comparison pnl_14d_per_week = r14["pnl"] / 2 # 14d = 2 weeks # Stability check: PF stays above 1.2 on 14d AND PnL positive stable = "โœ…" if r14["pnl"] > 0 and r14["pf"] >= 1.2 else "โŒ" rr = c.get("rr", "?") label = c.get("label", "") line = ( f"{idx:<3} {label:<22} {rr:<4} {c['tp_pct']}/{c['sl_pct']:<4} " f"{c['z_entry']:>4.1f}/{c['z_max']:<3.1f} " f"{c['natr_min']:.2f}-{c['natr_max']:.1f} " f"{c['chop_min']:>3} " f"| {r7['trades']:>5} {r7['wr']:>4.1f}% {r7['pnl']:>+6.1f} {r7['pf']:>5.2f} " f"| {r14['trades']:>5} {r14['wr']:>4.1f}% {r14['pnl']:>+6.1f} {r14['pf']:>5.2f} " f"| {r14['pnl'] - r7['pnl']:>+5.1f} {stable:>7}" ) report.append(line) results_json.append({ "label": label, **params, "rr": rr, "7d": r7, "14d": r14, "14d_pnl_per_week": round(pnl_14d_per_week, 2), "stable": r14["pnl"] > 0 and r14["pf"] >= 1.2, }) # Summary report.append("") report.append("=" * 90) report.append(" INTERPRETATION") report.append("=" * 90) stable_count = sum(1 for r in results_json if r["stable"]) report.append(f"\n Stable candidates (PnL>0 AND PFโ‰ฅ1.2 on 14d): {stable_count}/{len(results_json)}") # Best on 14d best_14d = max(results_json, key=lambda x: x["14d"]["pnl"]) report.append(f"\n Best 14d PnL: {best_14d['label']} โ†’ PnL={best_14d['14d']['pnl']:+.1f}, " f"PF={best_14d['14d']['pf']:.2f}, {best_14d['14d']['trades']} trades") # Best PF on 14d (min 30 trades) qualified = [r for r in results_json if r["14d"]["trades"] >= 30] if qualified: best_pf = max(qualified, key=lambda x: x["14d"]["pf"]) report.append(f" Best 14d PF: {best_pf['label']} โ†’ PF={best_pf['14d']['pf']:.2f}, " f"PnL={best_pf['14d']['pnl']:+.1f}, {best_pf['14d']['trades']} trades") # Consistency: similar PF on 7d and 14d report.append(f"\n Consistency check (PF drift 7dโ†’14d):") for r in results_json: pf7 = r["7d"]["pf"] pf14 = r["14d"]["pf"] drift = pf14 - pf7 emoji = "๐ŸŸข" if abs(drift) < 0.1 else ("๐ŸŸก" if abs(drift) < 0.2 else "๐Ÿ”ด") report.append(f" {emoji} {r['label']:<22} PF: {pf7:.2f} โ†’ {pf14:.2f} (ฮ”{drift:+.2f})") report.append(f"\n >>> Stable candidates go to Step 9 (DCA overlay)") report_text = "\n".join(report) print(report_text) # Save report_path = os.path.join(BASE, "robustness_report.txt") with open(report_path, "w") as f: f.write(report_text) print(f"\n๐Ÿ“„ Report: {report_path}") json_path = os.path.join(BASE, "robustness_results.json") with open(json_path, "w") as f: json.dump(results_json, f, indent=2) print(f"๐Ÿ“‹ JSON: {json_path}") if __name__ == "__main__": main()