"""What do violent bars do to OPEN trades, and can we protect them? Analysis tooling only. Replays every trade of a book on the raw bars with the backtester's exit model (SL-first, gap-through at open, close-only trail ratchet, 300-bar timeout, $0.65 cost) and compares: base : reproduce the book (validation: must match r_multiple for almost every trade) flat_data : exit at the OPEN of the first bar whose close-hour enters {12,13} UTC (US data) flat_asia : same for close-hour {0,1} UTC (Asian open) tight_data : entering {12,13}, hold a protective stop 0.5R below the best price for the window tight_asia : same for {0,1} react_exit : a burst bar (range >= 3x prior ATR14) closes AGAINST the trade -> exit next open react_tight : after an opposing burst, protective stop 0.5R below the best price for the rest Plus a damage table: what happens to a trade when the first burst hits it (with vs against). python backtest/research/open_trade_bursts.py BOOK.csv PARQUET BAR_MINUTES ARM DIST """ import sys import numpy as np, pandas as pd COST = 0.65; MAX_BARS = 300; K = 3.0; PROT = 0.5 def wilder_atr(df, period=14): pc = df["Close"].shift(1) tr = pd.concat([df["High"] - df["Low"], (df["High"] - pc).abs(), (df["Low"] - pc).abs()], axis=1).max(axis=1) return tr.ewm(alpha=1 / period, min_periods=period, adjust=False).mean() def simulate(bars, i0, direction, entry, sl, arm, dist, variant, hours_flat=(), hours_tight=(), flat_mask=None, tight_mask=None): """bars: dict of numpy arrays; i0 = index of the entry bar. Returns (exit_price, bars_held, reason, flag).""" O, H, L, C, burst, chour = bars["O"], bars["H"], bars["L"], bars["C"], bars["burst"], bars["chour"] d = 1.0 if direction == "BUY" else -1.0 risk = abs(entry - sl); stop = sl; peak = entry; flag = 0 n = len(O); tight = False; tight_until = -1; exit_next = False for j in range(i0, min(n, i0 + MAX_BARS)): if exit_next: return O[j], j - i0, "REACT", flag h = chour[j] in_flat = h in hours_flat; in_tight = h in hours_tight if variant.startswith("flat") and in_flat: return O[j], j - i0, "FLAT", 1 if variant == "mask_flat" and flat_mask is not None and flat_mask[j]: return O[j], j - i0, "FLAT", 1 if variant.startswith("tight") and in_tight: tight = True; tight_until = j; flag = 1 if variant == "mask_tight" and tight_mask is not None and tight_mask[j]: tight = True; tight_until = j; flag = 1 # stop check (SL first, gap-through at open) if d > 0: if O[j] <= stop: return O[j], j - i0, "SL", flag if L[j] <= stop: return stop, j - i0, "SL", flag else: if O[j] >= stop: return O[j], j - i0, "SL", flag if H[j] >= stop: return stop, j - i0, "SL", flag # update peak / trail at bar close peak = max(peak, H[j]) if d > 0 else min(peak, L[j]) fav = d * (peak - entry) if fav >= arm * risk: stop = max(stop, peak - dist * risk) if d > 0 else min(stop, peak + dist * risk) if tight and (variant == "react_tight" or j <= tight_until or in_tight): stop = max(stop, peak - PROT * risk) if d > 0 else min(stop, peak + PROT * risk) if variant.startswith("react") and burst[j] and d * (C[j] - O[j]) < 0: flag = 1 if variant == "react_exit": exit_next = True else: tight = True j = min(n - 1, i0 + MAX_BARS - 1) return C[j], j - i0, "TIMEOUT", flag def r_of(direction, entry, sl, exit_price): d = 1.0 if direction == "BUY" else -1.0 return (d * (exit_price - entry) - COST) / abs(entry - sl) def stats(r): r = np.asarray(r); eq = np.concatenate([[0.0], np.cumsum(r)]); dd = (np.maximum.accumulate(eq) - eq).max() pf = r[r > 0].sum() / max(-r[r < 0].sum(), 1e-9) return dict(n=len(r), total=r.sum(), mean=r.mean(), wr=100 * (r > 0).mean(), pf=pf, dd=dd) def main(): book_csv, pq, bm, arm, dist = sys.argv[1], sys.argv[2], int(sys.argv[3]), float(sys.argv[4]), float(sys.argv[5]) book = pd.read_csv(book_csv); book["entry_time"] = pd.to_datetime(book["entry_time"], utc=True, format="mixed") px = pd.read_parquet(pq) atr_prev = wilder_atr(px).shift(1) bars = dict(O=px["Open"].values, H=px["High"].values, L=px["Low"].values, C=px["Close"].values, burst=((px["High"] - px["Low"]) >= K * atr_prev).fillna(False).values, chour=(px.index + pd.Timedelta(minutes=bm)).hour.values) idx = pd.Index(px.index) pos = idx.get_indexer(book["entry_time"]) ok = pos >= 0 print(f"{book_csv}: {len(book)} trades, {ok.sum()} matched to bars; arm {arm} dist {dist}") book = book[ok].reset_index(drop=True); pos = pos[ok] half = np.where(np.arange(len(book)) < len(book) // 2, 1, 2) variants = {"base": {}, "flat_data": dict(hours_flat=(12, 13)), "flat_asia": dict(hours_flat=(0, 1)), "tight_data": dict(hours_tight=(12, 13)), "tight_asia": dict(hours_tight=(0, 1)), "react_exit": {}, "react_tight": {}} results = {} for v, kw in variants.items(): rs, flags = [], [] for i, t in book.iterrows(): xp, held, reason, flag = simulate(bars, pos[i], t.direction, t.entry, t.sl, arm, dist, v, **kw) rs.append(r_of(t.direction, t.entry, t.sl, xp)); flags.append(flag) results[v] = (np.array(rs), np.array(flags)) base = results["base"][0] match = np.isclose(base, book["r_multiple"].values, atol=0.02).mean() print(f"validation: replay reproduces the book's r_multiple (±0.02) on {100*match:.1f}% of trades; " f"sum replay {base.sum():+.2f} vs book {book.r_multiple.sum():+.2f}") print("\n== PROTECTION VARIANTS (whole book; 'affected' = trades where the rule fired) ==") print(f"{'variant':12s} {'n':>5s} {'affected':>8s} {'total R':>9s} {'Δ vs base':>9s} {'PF':>5s} {'maxDD':>6s} {'WR%':>5s} | affected only: base R -> variant R | Δ h1 / Δ h2") b = stats(base) for v, (r, f) in results.items(): s = stats(r); aff = f == 1 d1 = r[half == 1].sum() - base[half == 1].sum(); d2 = r[half == 2].sum() - base[half == 2].sum() print(f"{v:12s} {s['n']:5d} {aff.sum():8d} {s['total']:+9.2f} {s['total']-b['total']:+9.2f} {s['pf']:5.2f} {s['dd']:6.2f} {s['wr']:5.1f} | " f"{base[aff].sum():+8.2f} -> {r[aff].sum():+8.2f} (n={aff.sum()}) | {d1:+.2f} / {d2:+.2f}") # damage table: first burst while open (base path), with vs against print("\n== DAMAGE: first violent bar that hits an OPEN trade (base exits) ==") rows = [] for i, t in book.iterrows(): d = 1.0 if t.direction == "BUY" else -1.0; risk = abs(t.entry - t.sl) xp, held, reason, _ = simulate(bars, pos[i], t.direction, t.entry, t.sl, arm, dist, "base") j0 = pos[i]; j1 = j0 + held for j in range(j0 + 1, j1 + 1): # bursts after the entry bar, up to and including the exit bar if bars["burst"][j]: against = d * (bars["C"][j] - bars["O"][j]) < 0 mtm_before = (d * (bars["C"][j - 1] - t.entry) - COST) / risk mtm_after = (d * (bars["C"][j] - t.entry) - COST) / risk rows.append(dict(against=against, bars_open=j - j0, mtm_before=mtm_before, mtm_after=mtm_after, final=r_of(t.direction, t.entry, t.sl, xp), ended_here=(j == j1), hour=bars["chour"][j], sched=bars["chour"][j] in (0, 1, 12, 13))) break dmg = pd.DataFrame(rows) if len(dmg): g = dmg.groupby("against").agg(n=("final", "size"), bars_open_med=("bars_open", "median"), mtm_before=("mtm_before", "mean"), mtm_after=("mtm_after", "mean"), final=("final", "mean"), ended_on_that_bar_pct=("ended_here", lambda s: 100 * s.mean()), in_sched_hours_pct=("sched", lambda s: 100 * s.mean())) print(g.round(3).to_string()) ag = dmg[dmg.against] print(f"\nopposing bursts: {len(ag)} trades ({100*len(ag)/len(book):.1f}% of the book); " f"mean R just before {ag.mtm_before.mean():+.3f} -> just after {ag.mtm_after.mean():+.3f} -> final {ag.final.mean():+.3f}; " f"{100*(ag.final<0).mean():.0f}% end as losers; {100*ag.ended_here.mean():.0f}% are stopped ON that bar") print(f" those with an open PROFIT before the burst (mtm>0): n={int((ag.mtm_before>0).sum())}, final mean {ag[ag.mtm_before>0].final.mean():+.3f}, " f"{100*(ag[ag.mtm_before>0].final<0).mean():.0f}% end as losers") print(f" by hour of the opposing burst (n, mean final R): " + ", ".join(f"{h:02d}:{n}/{m:+.2f}" for h, n, m in ag.groupby("hour").final.agg(["size", "mean"]).reset_index().itertuples(index=False))) if __name__ == "__main__": main()