diff --git a/analyze_trades.py b/analyze_trades.py new file mode 100644 index 0000000..e345c17 --- /dev/null +++ b/analyze_trades.py @@ -0,0 +1,240 @@ +#!/usr/bin/env python3 +""" +Phase 1 analysis: post-hoc filters on an existing trades.csv. + +Asks "which simple filters, applied to the trades the engine already +produced, would have moved total R toward zero or positive?" — without +re-running the backtest engine. + +Discipline: candidate filters are derived on the **first 70% of trades +by entry_time** (train slice), then evaluated on the **last 30%** (test +slice). This is the same time-based split the meta-model uses, so the +numbers are comparable. + +Outputs (printed + saved to backtest_results_all/phase1_summary.txt): + - Session-of-day breakdown + - Day-of-week breakdown + - Pair × regime cross-tab + - Per-pair contribution + - Combined-filter sweep (drop-worst-hours / drop-worst-pairs / both) + +Usage: + python analyze_trades.py + python analyze_trades.py --trades backtest_results_all/trades.csv +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +import pandas as pd + + +def fmt(df: pd.DataFrame) -> pd.DataFrame: + return df.round(3) + + +def split(df: pd.DataFrame, train_frac: float = 0.7): + df = df.sort_values("entry_time").reset_index(drop=True) + cutoff = int(len(df) * train_frac) + return df.iloc[:cutoff].copy(), df.iloc[cutoff:].copy() + + +def stats(df: pd.DataFrame) -> dict: + if len(df) == 0: + return {"n": 0, "wr": 0.0, "avg_r": 0.0, "total_r": 0.0, "pf": 0.0} + wins = (df["r_multiple"] > 0).sum() + losses = (df["r_multiple"] <= 0).sum() + gross_win = df.loc[df["r_multiple"] > 0, "r_multiple"].sum() + gross_loss = -df.loc[df["r_multiple"] <= 0, "r_multiple"].sum() + return { + "n": len(df), + "wr": wins / len(df) * 100, + "avg_r": df["r_multiple"].mean(), + "total_r": df["r_multiple"].sum(), + "pf": gross_win / gross_loss if gross_loss > 0 else float("inf"), + } + + +def stats_line(label: str, df: pd.DataFrame) -> str: + s = stats(df) + return ( + f"{label:<40s}" + f" n={s['n']:>5} WR={s['wr']:5.1f}%" + f" avgR={s['avg_r']:+6.3f} totalR={s['total_r']:+9.2f} PF={s['pf']:.2f}" + ) + + +def main() -> int: + ap = argparse.ArgumentParser(description="Phase 1 post-hoc analysis on backtest trades.") + ap.add_argument("--trades", default="backtest_results_all/trades.csv") + ap.add_argument("--out", default="backtest_results_all/phase1_summary.txt") + ap.add_argument("--train-frac", type=float, default=0.7) + args = ap.parse_args() + + p = Path(args.trades) + if not p.exists(): + print(f"❌ {p} not found") + return 1 + + df = pd.read_csv(p) + df["entry_time"] = pd.to_datetime(df["entry_time"]) + df["hour"] = df["entry_time"].dt.hour + df["dow"] = df["entry_time"].dt.dayofweek + df["winner"] = (df["r_multiple"] > 0).astype(int) + + train, test = split(df, args.train_frac) + out_lines = [] + + def emit(line: str = ""): + print(line) + out_lines.append(line) + + emit("=" * 92) + emit(f"PHASE 1 ANALYSIS — {len(df):,} trades total") + emit(f" full: {df['entry_time'].min().date()} → {df['entry_time'].max().date()}") + emit(f" train: {train['entry_time'].min().date()} → {train['entry_time'].max().date()} ({len(train):,} trades)") + emit(f" test: {test['entry_time'].min().date()} → {test['entry_time'].max().date()} ({len(test):,} trades)") + emit("=" * 92) + emit() + emit(stats_line("BASELINE full", df)) + emit(stats_line("BASELINE train", train)) + emit(stats_line("BASELINE test", test)) + emit() + + # ── Per-hour (training set, used for filter selection) ────────── + emit("─" * 92) + emit("HOUR (UTC) — train slice (used for filter selection)") + emit("─" * 92) + hr_stats = train.groupby("hour").apply( + lambda x: pd.Series(stats(x)), include_groups=False, + ).round(3) + emit(hr_stats.to_string()) + emit() + + # Bottom-K hours by avg_r + bad_hours_4 = hr_stats.sort_values("avg_r").head(4).index.tolist() + bad_hours_6 = hr_stats.sort_values("avg_r").head(6).index.tolist() + emit(f" Worst 4 hours by avg_r (train): {sorted(bad_hours_4)}") + emit(f" Worst 6 hours by avg_r (train): {sorted(bad_hours_6)}") + emit() + + # ── Per-DOW (training set) ────────────────────────────────────── + emit("─" * 92) + emit("DAY OF WEEK — train slice (0=Mon, 4=Fri)") + emit("─" * 92) + dow_stats = train.groupby("dow").apply( + lambda x: pd.Series(stats(x)), include_groups=False, + ).round(3) + emit(dow_stats.to_string()) + emit() + bad_dow = dow_stats.sort_values("avg_r").head(2).index.tolist() + emit(f" Worst 2 days by avg_r (train): {sorted(bad_dow)}") + emit() + + # ── Per-pair (training set) ───────────────────────────────────── + emit("─" * 92) + emit("PAIR — train slice") + emit("─" * 92) + pair_stats = train.groupby("pair").apply( + lambda x: pd.Series(stats(x)), include_groups=False, + ).round(3) + emit(pair_stats.to_string()) + emit() + bad_pairs = pair_stats.sort_values("total_r").head(3).index.tolist() + emit(f" Worst 3 pairs by total_r (train): {bad_pairs}") + emit() + + # ── Per-regime × direction (training set) ─────────────────────── + emit("─" * 92) + emit("REGIME × DIRECTION — train slice") + emit("─" * 92) + reg_stats = train.groupby(["regime", "direction"]).apply( + lambda x: pd.Series(stats(x)), include_groups=False, + ).round(3) + emit(reg_stats.to_string()) + emit() + + # ── Filter sweep ON THE TEST SET (rule selected on train) ─────── + emit("=" * 92) + emit("FILTER SWEEP — applied to TEST slice (rule chosen from train, no leakage)") + emit("=" * 92) + test_filtered_dropworst4hours = test[~test["hour"].isin(bad_hours_4)] + test_filtered_dropworst6hours = test[~test["hour"].isin(bad_hours_6)] + test_filtered_droppairs = test[~test["pair"].isin(bad_pairs)] + test_filtered_droprangingsell = test[~((test["regime"] == "ranging") & (test["direction"] == "SELL"))] + test_filtered_only_ranging = test[test["regime"] == "ranging"] + test_filtered_drop_dow = test[~test["dow"].isin(bad_dow)] + + test_combo_hr_pair = test[ + (~test["hour"].isin(bad_hours_4)) & (~test["pair"].isin(bad_pairs)) + ] + test_combo_hr_pair_dow = test[ + (~test["hour"].isin(bad_hours_4)) + & (~test["pair"].isin(bad_pairs)) + & (~test["dow"].isin(bad_dow)) + ] + + emit() + emit(stats_line("test baseline (no filter)", test)) + emit(stats_line("test drop 4 worst hours", test_filtered_dropworst4hours)) + emit(stats_line("test drop 6 worst hours", test_filtered_dropworst6hours)) + emit(stats_line("test drop 3 worst pairs", test_filtered_droppairs)) + emit(stats_line("test drop 2 worst DOWs", test_filtered_drop_dow)) + emit(stats_line("test ONLY ranging regime", test_filtered_only_ranging)) + emit(stats_line("test hours+pairs (combined)", test_combo_hr_pair)) + emit(stats_line("test hours+pairs+DOW (combined)", test_combo_hr_pair_dow)) + emit() + + # ── Same filters applied to TRAIN slice for reference ─────────── + emit("─" * 92) + emit("Same filters — TRAIN slice for reference (these are the in-sample numbers)") + emit("─" * 92) + train_drop_hours = train[~train["hour"].isin(bad_hours_4)] + train_drop_pairs = train[~train["pair"].isin(bad_pairs)] + train_combo = train[(~train["hour"].isin(bad_hours_4)) & (~train["pair"].isin(bad_pairs))] + + emit(stats_line("train baseline (no filter)", train)) + emit(stats_line("train drop 4 worst hours", train_drop_hours)) + emit(stats_line("train drop 3 worst pairs", train_drop_pairs)) + emit(stats_line("train hours+pairs (combined)", train_combo)) + emit() + + # ── Verdict ───────────────────────────────────────────────────── + emit("=" * 92) + emit("VERDICT") + emit("=" * 92) + test_base = stats(test) + test_combo = stats(test_combo_hr_pair_dow) + delta_r = test_combo["total_r"] - test_base["total_r"] + delta_avgr = test_combo["avg_r"] - test_base["avg_r"] + emit(f" Best combined filter (hours+pairs+DOW) on TEST set:") + emit(f" {test_base['n']} → {test_combo['n']} trades " + f"({test_combo['n']/test_base['n']*100:.0f}% retained)") + emit(f" total R: {test_base['total_r']:+.2f} → {test_combo['total_r']:+.2f} " + f"(Δ {delta_r:+.2f})") + emit(f" avg R: {test_base['avg_r']:+.3f} → {test_combo['avg_r']:+.3f} " + f"(Δ {delta_avgr:+.3f})") + emit(f" win rate: {test_base['wr']:.1f}% → {test_combo['wr']:.1f}%") + emit() + if test_combo["total_r"] > 0: + emit(" ✅ Filtered subset is profitable on the held-out test set.") + emit(" → Phase 2 (geometry grid) may be unnecessary; this filter alone may suffice.") + elif test_combo["avg_r"] > test_base["avg_r"]: + emit(" ⚠️ Filtered subset improves but is still negative.") + emit(" → Phase 2 (geometry grid) needed to find profitable structure.") + else: + emit(" ❌ Filters do not survive the train/test split — rejected as noise.") + emit(" → Skip Phase 1's filter list; rely on Phase 2 (geometry) instead.") + + Path(args.out).parent.mkdir(parents=True, exist_ok=True) + Path(args.out).write_text("\n".join(out_lines)) + emit() + emit(f"📁 Saved: {args.out}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/geometry_grid.py b/geometry_grid.py new file mode 100644 index 0000000..b6ce65c --- /dev/null +++ b/geometry_grid.py @@ -0,0 +1,361 @@ +#!/usr/bin/env python3 +""" +Phase 2: re-simulate the existing trade pool under different SL/TP geometries +without re-running the engine. + +The engine already gave us, per trade: entry_time, entry_price, direction, +and the original SL distance (= 1.5 × ATR by default). That distance lets +us recover ATR and re-derive new SL/TP for any (sl_mult, rr) combo. We +then walk forward through the saved OHLCV bars, applying first-touch +simulation with worst-case (SL first if both touch same bar). + +This is faster than re-running the engine because the signal output (when +to trade, in which direction, at what price) is independent of SL/TP +geometry — only the exits change. + +Discipline: we report TRAIN and TEST stats for every cell, time-split at +70/30 by entry_time. A cell wins only if it's positive on the held-out +TEST slice. + +Usage: + python geometry_grid.py + python geometry_grid.py --apply-phase1-filters + python geometry_grid.py --pairs USDJPY # single-pair grid +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path +from typing import Dict, Tuple + +import numpy as np +import pandas as pd + +# Reuse history loader so we don't re-implement +from backtest import load_history +from config import Timeframe + +# trades.csv stores TF as Twelve-Data interval string ("15min", "1h", "1d", …) +_TFSTR_TO_ENUM = { + "1min": Timeframe.M1, "5min": Timeframe.M5, + "15min": Timeframe.M15, "30min": Timeframe.M30, + "1h": Timeframe.H1, "4h": Timeframe.H4, + "1d": Timeframe.D1, "1wk": Timeframe.W1, +} + +# Filters carried over from Phase 1 +PHASE1_BAD_HOURS = {10, 17, 19, 21} +PHASE1_BAD_PAIRS = {"NZD/USD", "AUD/USD", "GBP/USD"} +PHASE1_BAD_DOWS = {5, 6} # Sat / Sun + + +def simulate_one( + direction: str, + entry: float, + new_sl: float, + new_tp: float, + forward_bars: pd.DataFrame, + max_bars: int, + spread_pips: float, + pip_value: float, +) -> Tuple[float, str]: + """First-touch simulation. Returns (r_multiple, outcome).""" + sl_distance = abs(entry - new_sl) + if sl_distance == 0: + return 0.0, "INVALID" + risk_pips = sl_distance / pip_value + spread_r = spread_pips / risk_pips if risk_pips > 0 else 0.0 + rr_target = abs(new_tp - entry) / sl_distance + + bars_examined = 0 + for ts, bar in forward_bars.iterrows(): + bars_examined += 1 + if bars_examined > max_bars: + break + bar_high = bar["High"] + bar_low = bar["Low"] + if direction == "BUY": + sl_hit = bar_low <= new_sl + tp_hit = bar_high >= new_tp + else: + sl_hit = bar_high >= new_sl + tp_hit = bar_low <= new_tp + if sl_hit: + return -1.0 - spread_r, "LOSS_SL" + if tp_hit: + return rr_target - spread_r, "WIN_TP1" + + if bars_examined == 0: + return 0.0, "INVALID" + last = forward_bars.iloc[bars_examined - 1] + pnl = (last["Close"] - entry) if direction == "BUY" else (entry - last["Close"]) + return (pnl / sl_distance) - spread_r, "TIMEOUT" + + +def pip_value_for(symbol: str) -> float: + """Return pip value: 0.01 for JPY pairs, 0.0001 elsewhere.""" + return 0.01 if "JPY" in symbol else 0.0001 + + +def grid_run( + df: pd.DataFrame, + history: Dict[Tuple[str, str], pd.DataFrame], + sl_mults: list, + rr_ratios: list, + max_hold: int = 100, + spread_pips: float = 1.0, +) -> pd.DataFrame: + """For each (sl_mult, rr) cell, re-simulate every trade and report stats.""" + # Pre-index history by (symbol, tf_str) for fast lookup + rows = [] + n_total = len(df) * len(sl_mults) * len(rr_ratios) + print(f" evaluating {n_total:,} simulations " + f"({len(df):,} trades × {len(sl_mults)} SL × {len(rr_ratios)} RR)…") + + for sl_mult in sl_mults: + for rr in rr_ratios: + r_multiples = [] + outcomes = [] + entry_times = [] + pairs = [] + tfs = [] + for trade in df.itertuples(index=False): + # Recover ATR from original SL distance (engine used 1.5×ATR) + atr = abs(trade.entry - trade.sl) / 1.5 + new_sl_dist = sl_mult * atr + if trade.direction == "BUY": + new_sl = trade.entry - new_sl_dist + new_tp = trade.entry + new_sl_dist * rr + else: + new_sl = trade.entry + new_sl_dist + new_tp = trade.entry - new_sl_dist * rr + + hist_df = history.get((trade.symbol, trade.tf)) + if hist_df is None: + continue + ts = pd.Timestamp(trade.entry_time) + # Forward bars (strictly after entry) + forward = hist_df[hist_df.index > ts].head(max_hold) + if forward.empty: + continue + + pip = pip_value_for(trade.symbol) + r, outcome = simulate_one( + trade.direction, trade.entry, + new_sl, new_tp, forward, + max_bars=max_hold, spread_pips=spread_pips, + pip_value=pip, + ) + r_multiples.append(r) + outcomes.append(outcome) + entry_times.append(ts) + pairs.append(trade.pair) + tfs.append(trade.tf) + + arr = np.array(r_multiples) if r_multiples else np.array([0.0]) + wins = (arr > 0).sum() + losses = (arr <= 0).sum() + gross_win = arr[arr > 0].sum() + gross_loss = -arr[arr <= 0].sum() + pf = gross_win / gross_loss if gross_loss > 0 else float("inf") + cum = np.cumsum(arr) + running_peak = np.maximum.accumulate(cum) + max_dd = (cum - running_peak).min() + + rows.append({ + "sl_mult": sl_mult, + "rr": rr, + "n": len(arr), + "wins": int(wins), + "wr": wins / len(arr) * 100 if len(arr) else 0, + "avg_r": arr.mean() if len(arr) else 0, + "total_r": arr.sum(), + "pf": pf, + "max_dd": max_dd, + # Carry the per-trade results so we can split train/test downstream + "_arr": arr, + "_entry_times": entry_times, + "_pairs": pairs, + "_outcomes": outcomes, + }) + return pd.DataFrame(rows) + + +def split_summary(cell: pd.Series, train_frac: float = 0.7) -> dict: + """Split a cell's per-trade results into train/test by entry_time.""" + arr = cell["_arr"] + times = cell["_entry_times"] + if len(arr) == 0: + return {"train_total": 0, "train_wr": 0, "test_total": 0, "test_wr": 0} + order = np.argsort(times) + cutoff = int(len(arr) * train_frac) + train_idx = order[:cutoff] + test_idx = order[cutoff:] + train_arr = arr[train_idx] + test_arr = arr[test_idx] + return { + "train_n": len(train_arr), + "train_wr": (train_arr > 0).mean() * 100 if len(train_arr) else 0, + "train_avg": train_arr.mean() if len(train_arr) else 0, + "train_total": train_arr.sum(), + "test_n": len(test_arr), + "test_wr": (test_arr > 0).mean() * 100 if len(test_arr) else 0, + "test_avg": test_arr.mean() if len(test_arr) else 0, + "test_total": test_arr.sum(), + } + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--trades", default="backtest_results_all/trades.csv") + ap.add_argument("--out", default="backtest_results_all/geometry_grid.txt") + ap.add_argument("--max-hold", type=int, default=100) + ap.add_argument("--spread-pips", type=float, default=1.0) + ap.add_argument("--apply-phase1-filters", action="store_true", + help="Drop weekend trades, worst 4 hours, worst 3 pairs before grid") + ap.add_argument("--sl-mults", nargs="+", type=float, + default=[1.0, 1.5, 2.0, 2.5, 3.0]) + ap.add_argument("--rr-ratios", nargs="+", type=float, + default=[1.0, 1.5, 2.0, 2.5]) + args = ap.parse_args() + + p = Path(args.trades) + if not p.exists(): + print(f"❌ {p} not found") + return 1 + + df = pd.read_csv(p) + df["entry_time"] = pd.to_datetime(df["entry_time"]) + df["hour"] = df["entry_time"].dt.hour + df["dow"] = df["entry_time"].dt.dayofweek + + print(f"📊 Loaded {len(df):,} trades from {p}") + + if args.apply_phase1_filters: + before = len(df) + df = df[~df["hour"].isin(PHASE1_BAD_HOURS)] + df = df[~df["pair"].isin(PHASE1_BAD_PAIRS)] + df = df[~df["dow"].isin(PHASE1_BAD_DOWS)] + df = df.reset_index(drop=True) + print(f" Phase 1 filters applied: {before:,} → {len(df):,} trades " + f"({len(df)/before*100:.0f}% retained)") + + # Load history for forward simulation + print(f"📁 Loading history…") + pairs = sorted(df["symbol"].unique().tolist()) + tfs = sorted({_TFSTR_TO_ENUM[t] for t in df["tf"].unique() if t in _TFSTR_TO_ENUM}, + key=lambda x: x.value) + history_raw = load_history(pairs, tfs) + # Rekey by (symbol, tf_string) to match trades.csv + history = {} + for (sym, tf_enum), hdf in history_raw.items(): + history[(sym, tf_enum.value)] = hdf + + print(f"\n🧮 Grid search:") + print(f" SL multipliers (× ATR): {args.sl_mults}") + print(f" RR ratios: {args.rr_ratios}\n") + + grid = grid_run( + df, history, args.sl_mults, args.rr_ratios, + max_hold=args.max_hold, spread_pips=args.spread_pips, + ) + + # ── Heatmap (full-data total R) ───────────────────────────────── + pivot_total = grid.pivot(index="sl_mult", columns="rr", values="total_r").round(1) + pivot_wr = grid.pivot(index="sl_mult", columns="rr", values="wr").round(1) + pivot_avg = grid.pivot(index="sl_mult", columns="rr", values="avg_r").round(3) + pivot_pf = grid.pivot(index="sl_mult", columns="rr", values="pf").round(2) + + out_lines = [] + def emit(line: str = ""): + print(line) + out_lines.append(line) + + emit("=" * 80) + emit(f"GEOMETRY GRID — {len(df):,} trades") + emit(f" filters applied: {args.apply_phase1_filters}") + emit(f" spread cost: {args.spread_pips} pip") + emit(f" max hold: {args.max_hold} bars") + emit("=" * 80) + emit() + emit("─── Total R (rows = SL multiplier × ATR; cols = RR ratio) ───") + emit(pivot_total.to_string()) + emit() + emit("─── Win rate % ───") + emit(pivot_wr.to_string()) + emit() + emit("─── Avg R per trade ───") + emit(pivot_avg.to_string()) + emit() + emit("─── Profit factor ───") + emit(pivot_pf.to_string()) + emit() + + # ── Train/Test discipline on top cells ────────────────────────── + grid_sorted = grid.sort_values("total_r", ascending=False) + emit("=" * 80) + emit("TOP 5 CELLS (by full-period total R) — with TRAIN / TEST split discipline") + emit("=" * 80) + for _, cell in grid_sorted.head(5).iterrows(): + s = split_summary(cell) + emit( + f" SL={cell['sl_mult']:.1f}×ATR RR=1:{cell['rr']:.1f} " + f"| full: {cell['n']} trades WR {cell['wr']:.1f}% " + f"avgR {cell['avg_r']:+.3f} totalR {cell['total_r']:+.1f} " + f"PF {cell['pf']:.2f}" + ) + emit( + f" train ({s['train_n']}): " + f"WR {s['train_wr']:.1f}% avgR {s['train_avg']:+.3f} totalR {s['train_total']:+.1f}" + ) + emit( + f" test ({s['test_n']}): " + f"WR {s['test_wr']:.1f}% avgR {s['test_avg']:+.3f} totalR {s['test_total']:+.1f}" + ) + emit() + + # ── Bottom 3 for contrast ─────────────────────────────────────── + emit("─── BOTTOM 3 cells (worst total R) ───") + for _, cell in grid_sorted.tail(3).iterrows(): + emit( + f" SL={cell['sl_mult']:.1f}×ATR RR=1:{cell['rr']:.1f} " + f"totalR {cell['total_r']:+.1f} WR {cell['wr']:.1f}%" + ) + emit() + + # ── Verdict ───────────────────────────────────────────────────── + emit("=" * 80) + emit("VERDICT") + emit("=" * 80) + best = grid_sorted.iloc[0] + s = split_summary(best) + emit( + f" Best cell: SL={best['sl_mult']:.1f}×ATR, RR=1:{best['rr']:.1f} " + f"(full totalR {best['total_r']:+.1f})" + ) + emit( + f" Test slice: {s['test_n']} trades, WR {s['test_wr']:.1f}%, " + f"totalR {s['test_total']:+.1f}, avgR {s['test_avg']:+.3f}" + ) + if s["test_total"] > 0: + emit(" ✅ Best geometry is profitable on held-out test set.") + emit(" → Phase 3A: hardcode in _generate_trade_setup, re-run end-to-end.") + elif s["test_avg"] > -0.05: + emit(" ⚠️ Best geometry is near-break-even on test set.") + emit(" → Phase 3A worth attempting; combine with Phase 1 filters for margin.") + else: + emit(" ❌ No geometry is profitable on held-out test set.") + emit(" → Phase 3B: M15 confluence is structurally not edge-bearing.") + emit(" Consider H4 primary, mean-reversion-only, or indicator-level meta-model.") + + Path(args.out).parent.mkdir(parents=True, exist_ok=True) + Path(args.out).write_text("\n".join(out_lines)) + emit() + emit(f"📁 Saved: {args.out}") + return 0 + + +if __name__ == "__main__": + sys.exit(main())