""" simulate.py Historical replay of a real trading day through the ACTUAL live code -- smart_engine.py's evaluate(), exit.py's evaluate(), stream.py's own SymbolBuffer -- fed by Alpaca's historical tick trades/quotes instead of the live SIP websocket. Not a reimplementation of the strategy: the exact same decision functions monitor.py calls are called here, so a replay run with the same max_positions monitor.py actually used should reproduce that day's real trades almost exactly (see --sanity). Built to answer one concrete question (2026-09-18): today's live run traded with max_positions=3 the whole session (the config change to 6 was saved mid-day but a running monitor.py never reloads config, so it only takes effect on the next restart) -- what would today have looked like if 6 slots had been live from the open? Memory design: this host has well under 1GB free RAM, and a full session's tick-level quotes for 30 symbols does not fit in memory at once (confirmed: one active symbol's full-day quotes alone ran ~150k-190k rows). So data is fetched and cached to disk PER SYMBOL, PER HOUR CHUNK (bounding peak memory to one small chunk regardless of total day volume), then the replay phase streams each symbol's cache file line-by-line rather than holding it in memory -- SymbolBuffer itself is already bounded (390-bar rolling window, 90s trade log), so total replay memory stays small and flat regardless of session length. Usage: python simulate.py --date 2026-09-18 --max-positions 3 --sanity Replays with 3 slots and diffs the result against the real data/trades/_trades.jsonl -- run this FIRST to trust the replay's fidelity before trusting any hypothetical slot count. Fetches+caches tick data on first run; later runs (same date) reuse the cache instantly. python simulate.py --date 2026-09-18 --max-positions 6 The actual question: what the day looks like with 6 slots live from the open. Reuses the cache from the --sanity run above. Known simplifications: - Position sizing uses ONE fixed equity snapshot (fetched live at simulate.py start), not the real intraday-fluctuating equity PositionManager.calculate_qty() reads every entry -- share counts and dollar P&L will be close but not identical to a real run even at the same slot count; %% P&L per trade is unaffected. - 1-min bars are built purely from replayed trade ticks (same SymbolBuffer._accumulate_bar() logic the live bot uses from its own trade stream), not Alpaca's separately-aggregated official bars. Expected to be extremely close (same trades, same VWAP/OHLC math) but not bit-exact. """ import argparse import gc import json import sys import time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta from pathlib import Path from zoneinfo import ZoneInfo from config_loader import get_config from alpaca_client import get_client from stream import SymbolBuffer import smart_engine import exit as exit_engine import exit_giveback_room import exit_resistance_stall import volatility import reentry from entry_rules import ShadowEntryRules, Bar as ShadowBar, record_decision import exit_giveback_experimental as giveback_exit import exit_fast_slope_experimental as fast_slope_exit import exit_prf_experimental as prf_exit import exit_resistance_timer_experimental as resistance_timer_exit import exit_velocity_experimental as velocity_exit import exit_cents_grace_experimental as cents_grace_exit import exit_step_lock_experimental as step_lock_exit BASE_DIR = Path(__file__).resolve().parent UTC = ZoneInfo("UTC") def _tz(): return ZoneInfo(get_config()["schedule"]["timezone"]) def _et_dt(date_str: str, hhmmss: str) -> datetime: h, m, s = (int(x) for x in hhmmss.split(":")) y, mo, d = (int(x) for x in date_str.split("-")) return datetime(y, mo, d, h, m, s, tzinfo=_tz()) def load_daily_atr(client, symbols: list, date_str: str, cache_dir: Path) -> dict: """[2026-09-23] Daily ATR(14) per symbol from sessions BEFORE date_str (what the live scanner would have seen that morning), cached next to the tick cache.""" path = cache_dir / "daily_atr.json" if path.exists(): return json.loads(path.read_text()) day = _et_dt(date_str, "00:00:00").astimezone(UTC) bars = client.get_daily_bars_bulk(symbols, day - timedelta(days=45), day - timedelta(seconds=1)) out = {} for sym, bl in bars.items(): done = [b for b in bl if b["t"] < day] datr = volatility.daily_atr(done) if datr: out[sym] = datr path.write_text(json.dumps(out)) return out class _ShadowTap: """[2026-09-26] Forwards replayed ticks to the SymbolBuffer AND the shadow recorder (entry_rules.py), exactly as stream.py's handlers do live.""" def __init__(self, symbol, buf, shadow): self.symbol, self.buf, self.shadow = symbol, buf, shadow def on_trade(self, price, size, ts): self.buf.on_trade(price, size, ts) self.shadow.on_trade(self.symbol, ts, price, size) def on_quote(self, bid, ask, ts): self.buf.on_quote(bid, ask, ts) self.shadow.on_quote(self.symbol, ts, bid, ask) def _shadow_push_bars(shadow, symbols, buffers, last_bar, bench_bars, bench_pos, t_utc): """Feed newly COMPLETED 1-min bars (and benchmark bars) to the shadow.""" for sym in symbols: bars = buffers[sym].get_bars(include_forming=False) for b in bars[-3:]: if last_bar.get(sym) is None or b["t"] > last_bar[sym]: shadow.on_bar(sym, ShadowBar(b["t"], b["o"], b["h"], b["l"], b["c"], b["v"])) last_bar[sym] = b["t"] for sym, bl in bench_bars.items(): i = bench_pos.get(sym, 0) while i < len(bl) and bl[i].ts + timedelta(minutes=1) <= t_utc: shadow.on_bar(sym, bl[i]); i += 1 bench_pos[sym] = i def _bench_bars(client, date_str): out = {} for sym in ("SPY", "IWM"): raw = client.get_minute_bars(sym, start=_et_dt(date_str, "09:30:00"), end=_et_dt(date_str, "16:00:00"), limit=1000) out[sym] = [ShadowBar(b.timestamp, float(b.open), float(b.high), float(b.low), float(b.close), float(b.volume)) for b in raw] return out def _load_fixed(path): if not path: return None trades = json.loads(Path(path).read_text())["trades"] out = [{"symbol": t["symbol"], "stop_price": t["stop_price"], "reasons": t.get("reasons"), "_t": datetime.fromisoformat(t["entry_time"])} for t in trades] return sorted(out, key=lambda x: x["_t"]) def load_ref5d(client, symbols: list, date_str: str, cache_dir: Path) -> dict: """[2026-09-25] 5-day reference frame per symbol as of date_str (last 5 sessions BEFORE it), cached next to the tick cache.""" path = cache_dir / "ref5d.json" if path.exists(): return json.loads(path.read_text()) day = _et_dt(date_str, "00:00:00").astimezone(UTC) bars = client.get_daily_bars_bulk(symbols, day - timedelta(days=15), day - timedelta(seconds=1)) out = {sym: volatility.five_day_reference([b for b in bl if b["t"] < day]) for sym, bl in bars.items()} out = {k: v for k, v in out.items() if v} path.write_text(json.dumps(out)) return out def load_candidates(date_str: str) -> list: path = BASE_DIR / "data" / "candidates" / f"{date_str}_scanner.json" with open(path) as f: return json.load(f) def load_real_poll_times(date_str: str): """[FIDELITY 2026-09-18] The real monitor.py loop's actual poll timestamps (data_store.append_decision_record logs one record per symbol evaluated, every poll) -- when available, replay drives its clock off these EXACT real timestamps instead of a synthetic 5s grid. Matters more than the ~5s cadence itself suggests: a synthetic grid starting at market_open_time on the dot vs. the real loop's actual first-poll offset (seen 2026-09-18: real polls landed at :06.03, not :00.00) is enough drift, by the time several candidates' 18s Stage-2 holds are converging within a poll or two of each other near the open, to flip which of two near-simultaneous candidates wins a shared slot -- confirmed: the first synthetic-grid replay of 2026-09-18 diverged from the real trade sequence starting at exactly that kind of close race (RUM vs AGEN for the 3rd slot). Returns None (caller falls back to the synthetic grid) if no decisions log exists for this date, e.g. simulating a day this tool didn't run against. """ path = BASE_DIR / "data" / "decisions" / f"{date_str}.jsonl" if not path.exists(): return None times = set() with open(path) as f: for line in f: times.add(json.loads(line)["timestamp"]) dts = sorted(datetime.fromisoformat(t).replace(tzinfo=UTC) for t in times) clusters = [dts[0]] for d in dts[1:]: if (d - clusters[-1]).total_seconds() > 1.0: clusters.append(d) return clusters # ---------------------------------------------------------------------- # Fetch + disk cache (bounded memory: one symbol, one hour chunk, at a time) # ---------------------------------------------------------------------- def _chunks(start: datetime, end: datetime, minutes: int = 20): cur = start step = timedelta(minutes=minutes) while cur < end: nxt = min(cur + step, end) yield cur, nxt cur = nxt def cache_symbol(client, symbol: str, start: datetime, end: datetime, cache_path: Path): """Low-memory by construction: fetches+writes one small (20-min) chunk at a time and drops it before the next -- this host has under 1GB total RAM and a full-day chunk for an active symbol alone was enough to get OOM-killed with only 3 fetches in flight at once.""" tmp_path = cache_path.with_suffix(".jsonl.tmp") n_events = 0 with open(tmp_path, "w") as f: for chunk_start, chunk_end in _chunks(start, end): trades = client.get_historical_trades(symbol, chunk_start, chunk_end) quotes = client.get_historical_quotes(symbol, chunk_start, chunk_end) events = [(t["t"], "trade", t["p"], t["s"]) for t in trades] events += [(q["t"], "quote", q["b"], q["a"]) for q in quotes] events.sort(key=lambda e: e[0]) for ts, kind, a, b in events: f.write(json.dumps([ts.isoformat(), kind, a, b]) + "\n") n_events += len(events) del trades, quotes, events gc.collect() tmp_path.rename(cache_path) return n_events def ensure_cache(client, symbols: list, start: datetime, end: datetime, cache_dir: Path, workers: int = 3): cache_dir.mkdir(parents=True, exist_ok=True) todo = [] for s in symbols: p = cache_dir / f"{s}.jsonl" if p.exists(): print(f" [cache hit] {s}", file=sys.stderr) else: todo.append(s) if not todo: return t0 = time.time() with ThreadPoolExecutor(max_workers=workers) as pool: futures = {pool.submit(cache_symbol, client, s, start, end, cache_dir / f"{s}.jsonl"): s for s in todo} done = 0 for fut in as_completed(futures): symbol = futures[fut] try: n = fut.result() except Exception as e: print(f" [FAILED] {symbol}: {e}", file=sys.stderr) n = 0 done += 1 print(f" [{done}/{len(todo)}] {symbol}: {n} events cached", file=sys.stderr) print(f" fetched {len(todo)} symbols in {time.time() - t0:.1f}s", file=sys.stderr) class SymbolEventReader: """Streams one symbol's cached JSONL file line-by-line -- never holds more than the current line in memory.""" def __init__(self, path: Path): self._f = open(path, "r") if path.exists() else None self._peeked = None self._advance_peek() def _advance_peek(self): if self._f is None: self._peeked = None return line = self._f.readline() if not line: self._peeked = None return ts_iso, kind, a, b = json.loads(line) self._peeked = (datetime.fromisoformat(ts_iso), kind, a, b) def drain_up_to(self, upto_ts, buf: SymbolBuffer): while self._peeked is not None and self._peeked[0] <= upto_ts: ts, kind, a, b = self._peeked if kind == "trade": buf.on_trade(a, b, ts) else: buf.on_quote(a, b, ts) self._advance_peek() def close(self): if self._f: self._f.close() # ---------------------------------------------------------------------- # Simulated position manager (local math only, no broker calls) # ---------------------------------------------------------------------- class SimPositionManager: def __init__(self, equity: float, trading_cfg: dict, max_positions: int): self.equity = equity self.cfg = trading_cfg self.max_positions = max_positions self.positions = {} self.trades = [] def has_available_slot(self) -> bool: return len(self.positions) < self.max_positions def is_symbol_open(self, symbol: str) -> bool: return symbol in self.positions def calculate_qty(self, entry_price: float, stop_price: float) -> int: risk_per_share = entry_price - stop_price if risk_per_share <= 0 or not entry_price: return 0 risk_dollars = self.equity * self.cfg["account_risk_pct_per_trade"] / 100.0 qty = int(risk_dollars / risk_per_share) notional_cap = self.equity * self.cfg["max_position_notional_pct_of_equity"] / 100.0 qty = min(qty, int(notional_cap / entry_price)) return qty if qty >= self.cfg["min_shares"] else 0 def enter(self, symbol: str, entry_price: float, stop_price: float, ts, reasons): if self.is_symbol_open(symbol) or not self.has_available_slot(): return False qty = self.calculate_qty(entry_price, stop_price) if qty <= 0: return False self.positions[symbol] = { "symbol": symbol, "qty": qty, "entry_price": entry_price, "stop_price": stop_price, "entry_time": ts.isoformat(), "reasons": reasons or [], } return True def exit(self, symbol: str, exit_price: float, reason: str, ts): p = self.positions.get(symbol) if not p: return pl_pct = (exit_price - p["entry_price"]) / p["entry_price"] * 100.0 if p["entry_price"] else 0.0 pl_dollars = (exit_price - p["entry_price"]) * p["qty"] # [2026-09-23] Exit-quality metrics: how far the trade ran for us # (peak) and against us (low) while held, and how much of the peak # gain the exit handed back. peak = max(p.get("peak_price", p["entry_price"]), exit_price) low = min(p.get("low_price", p["entry_price"]), exit_price) mfe_dollars = (peak - p["entry_price"]) * p["qty"] trade = {**p, "exit_price": exit_price, "exit_time": ts.isoformat(), "exit_reason": reason, "status": "closed", "pl_pct": round(pl_pct, 3), "pl_dollars": round(pl_dollars, 2), "peak_price": peak, "low_price": low, "mfe_cents": round((peak - p["entry_price"]) * 100, 2), "mae_cents": round((p["entry_price"] - low) * 100, 2), "mfe_dollars": round(mfe_dollars, 2), "giveback_dollars": round(mfe_dollars - pl_dollars, 2)} self.trades.append(trade) del self.positions[symbol] # ---------------------------------------------------------------------- # Replay engine -- mirrors monitor.py's SessionOrchestrator loop exactly, # fed by SymbolEventReader instead of a live StreamManager. # ---------------------------------------------------------------------- def get_buy_sell_volumes(buf: SymbolBuffer, window_seconds: float = 20.0): """Raw (buy_vol, sell_vol) over the trailing window_seconds -- SymbolBuffer.get_trade_imbalance() only exposes the ratio, but the giveback-exit experiment needs the two sides separately (see exit_giveback_experimental.py's demand-exhaustion vote). Same windowing logic as get_trade_imbalance() itself, just not collapsed into a single ratio.""" log = buf._trade_log if not log: return 0.0, 0.0 from datetime import timedelta as _td now_ts = log[-1][0] cutoff = now_ts - _td(seconds=window_seconds) buy_vol = sell_vol = 0.0 for ts, side, size in reversed(log): if ts < cutoff: break if side > 0: buy_vol += size elif side < 0: sell_vol += size return buy_vol, sell_vol def run_simulation(date_str: str, max_positions: int, candidates: list, cache_dir: Path, equity: float, poll_times=None, trace_symbols=None, use_giveback_exit=False, use_fast_slope_exit=False, use_prf_exit=False, use_resistance_timer_exit=False, resistance_timer_cfg=None, use_velocity_exit=False, velocity_cfg=None, use_cents_grace_exit=False, cents_grace_cfg=None, use_step_lock_exit=False, step_lock_cfg=None, step_lock_use_atr_distance=False, use_giveback_room_exit=False, giveback_room_cfg=None, smart_engine_cfg=None, scanner_stop_cents=None, resistance_stall_cfg=None, daily_atr=None, ref5d=None, reentry_cfg=None, fixed_entries=None, watch=None, shadow=None, bench_bars=None, shadow_gate=None) -> SimPositionManager: trace_symbols = set(trace_symbols or []) trace = [] cfg = get_config() sched = cfg["schedule"] trading_cfg = cfg["trading"] market_open = _et_dt(date_str, sched["market_open_time"]) no_new_entries = _et_dt(date_str, sched["no_new_entries_after"]).astimezone(UTC) force_liquidate = _et_dt(date_str, sched["force_liquidate_time"]) poll_interval = sched["poll_interval_seconds"] if poll_times is None: poll_times = [] t = market_open fl = force_liquidate while t <= fl: poll_times.append(t.astimezone(UTC)) t += timedelta(seconds=poll_interval) resistance_levels = {} avg_vol_baseline = {} for c in candidates: m = c["metrics"] resistance_levels[c["symbol"]] = { "premarket_high": m.get("premarket_high"), "prev_day_high": m.get("previous_day_high"), "range_20d_high": m.get("range_20d_high"), "daily_atr": (daily_atr or {}).get(c["symbol"]), **((ref5d or {}).get(c["symbol"], {})), } avg_vol_baseline[c["symbol"]] = m.get("avg_daily_volume") entry_imbalance_window = cfg.get("smart_engine", {}).get("imbalance_window_seconds", 20) exit_imbalance_window = cfg.get("exit", {}).get("imbalance_window_seconds", 20) symbols = [c["symbol"] for c in candidates] buffers = {s: SymbolBuffer(maxlen=390) for s in symbols} readers = {s: SymbolEventReader(cache_dir / f"{s}.jsonl") for s in symbols} pm = SimPositionManager(equity, trading_cfg, max_positions) entry_persistence = {} exit_confirmation = {} giveback_state = {} prf_state = {} resistance_timer_state = {} velocity_state = {} cents_grace_state = {} step_lock_state = {} giveback_room_state = {} scanner_stop_peak = {} # Keyed by (symbol, entry_time) so a re-entry never inherits a closed # position's armed/extension state. resistance_stall_state = {} last_exit_healthy = {} shadow_keys, shadow_last_bar, bench_pos = {}, {}, {} gate_blocked = {} shadow_taps = {sym: _ShadowTap(sym, buffers[sym], shadow) for sym in symbols} if shadow is not None else {} fixed_min_stop = smart_engine._merge_cfg(smart_engine_cfg)["min_stop"] for t_utc in poll_times: if t_utc > force_liquidate.astimezone(UTC): break for s in symbols: readers[s].drain_up_to(t_utc, shadow_taps[s] if shadow is not None else buffers[s]) if shadow is not None: _shadow_push_bars(shadow, symbols, buffers, shadow_last_bar, bench_bars or {}, bench_pos, t_utc) # exits for symbol in list(pm.positions.keys()): buf = buffers[symbol] bars = buf.get_bars() if not bars: continue _pos = pm.positions[symbol] _pos["peak_price"] = max(_pos.get("peak_price", _pos["entry_price"]), bars[-1]["c"]) _pos["low_price"] = min(_pos.get("low_price", _pos["entry_price"]), bars[-1]["c"]) # Checked FIRST, ahead of exit.py's three layers -- mirrors # monitor.py's real wiring (exit_giveback_room.evaluate() runs # before exit_engine.evaluate() and, if it fires, exit.py is # never even consulted that poll). if use_giveback_room_exit: gr_decision = exit_giveback_room.evaluate( symbol, bars, pm.positions[symbol]["entry_price"], giveback_room_state.get(symbol, {}), cfg=giveback_room_cfg) giveback_room_state[symbol] = gr_decision.state_out if symbol in trace_symbols: trace.append({"timestamp": t_utc.isoformat(), "symbol": symbol, "kind": "giveback_room_exit", "state": gr_decision.state, "reason": gr_decision.reason, "metrics": gr_decision.metrics}) if gr_decision.should_exit: current_price = bars[-1]["c"] pm.exit(symbol, current_price, gr_decision.reason, t_utc) exit_confirmation.pop(symbol, None) giveback_state.pop(symbol, None) prf_state.pop(symbol, None) resistance_timer_state.pop(symbol, None) velocity_state.pop(symbol, None) cents_grace_state.pop(symbol, None) step_lock_state.pop(symbol, None) giveback_room_state.pop(symbol, None) continue # Mirrors monitor.py: exit_resistance_stall runs after giveback-room, # ahead of exit.py. if resistance_stall_cfg is not None: pos = pm.positions[symbol] key = (symbol, pos["entry_time"]) rs_decision = exit_resistance_stall.evaluate( symbol, bars, pos["entry_time"], t_utc, resistance_stall_state.get(key, {}), cfg=resistance_stall_cfg, slots_free=pm.has_available_slot(), healthy=last_exit_healthy.get(symbol, True)) resistance_stall_state[key] = rs_decision.state_out if symbol in trace_symbols: trace.append({"timestamp": t_utc.isoformat(), "symbol": symbol, "kind": "resistance_stall", "state": rs_decision.state, "reason": rs_decision.reason, "metrics": rs_decision.metrics}) if rs_decision.should_exit: pm.exit(symbol, bars[-1]["c"], rs_decision.reason, t_utc) exit_confirmation.pop(symbol, None) giveback_room_state.pop(symbol, None) resistance_stall_state.pop(key, None) continue # Mirrors monitor.py's scanner_stop_phase: until the peak since # entry has run up scanner_stop_cents, only the entry stop exits. if scanner_stop_cents is not None: pos = pm.positions[symbol] current_price = bars[-1]["c"] peak = max(scanner_stop_peak.get(symbol, pos["entry_price"]), current_price) scanner_stop_peak[symbol] = peak peak_gain_cents = (peak - pos["entry_price"]) * 100.0 if peak_gain_cents < scanner_stop_cents: if current_price <= pos["stop_price"]: pm.exit(symbol, current_price, f"scanner stop: price ${current_price:.4f} <= stop ${pos['stop_price']:.4f}", t_utc) exit_confirmation.pop(symbol, None) giveback_room_state.pop(symbol, None) scanner_stop_peak.pop(symbol, None) continue trade_imbalance = buf.get_trade_imbalance(exit_imbalance_window) decision = exit_engine.evaluate( symbol, bars, pm.positions[symbol]["entry_price"], exit_confirmation.get(symbol, {}), trade_imbalance=trade_imbalance, stop_price=pm.positions[symbol]["stop_price"]) exit_confirmation[symbol] = decision.confirmation last_exit_healthy[symbol] = decision.reason.startswith("healthy") if symbol in trace_symbols: trace.append({"timestamp": t_utc.isoformat(), "symbol": symbol, "kind": "exit", "state": decision.state, "reason": decision.reason, "metrics": decision.metrics}) gb_decision = None if use_giveback_exit and not decision.should_exit: buy_vol, sell_vol = get_buy_sell_volumes(buf, exit_imbalance_window) gb_decision = giveback_exit.evaluate( symbol, bars, buy_vol, sell_vol, giveback_state.get(symbol, {})) giveback_state[symbol] = gb_decision.state_out if symbol in trace_symbols: trace.append({"timestamp": t_utc.isoformat(), "symbol": symbol, "kind": "giveback_exit", "state": gb_decision.state, "reason": gb_decision.reason, "metrics": gb_decision.metrics}) fs_decision = None if use_fast_slope_exit and not decision.should_exit and not (gb_decision and gb_decision.should_exit): sub_bars = buf.get_bars_sub() fs_decision = fast_slope_exit.evaluate(symbol, bars, sub_bars) if symbol in trace_symbols: trace.append({"timestamp": t_utc.isoformat(), "symbol": symbol, "kind": "fast_slope_exit", "state": fs_decision.state, "reason": fs_decision.reason, "metrics": fs_decision.metrics}) prf_decision = None if (use_prf_exit and not decision.should_exit and not (gb_decision and gb_decision.should_exit) and not (fs_decision and fs_decision.should_exit)): prf_decision = prf_exit.evaluate( symbol, bars, pm.positions[symbol]["entry_price"], trade_imbalance, prf_state.get(symbol, {})) prf_state[symbol] = prf_decision.state_out if symbol in trace_symbols: trace.append({"timestamp": t_utc.isoformat(), "symbol": symbol, "kind": "prf_exit", "state": prf_decision.state, "reason": prf_decision.reason, "metrics": prf_decision.metrics}) rt_decision = None if (use_resistance_timer_exit and not decision.should_exit and not (gb_decision and gb_decision.should_exit) and not (fs_decision and fs_decision.should_exit) and not (prf_decision and prf_decision.should_exit)): rt_decision = resistance_timer_exit.evaluate( symbol, bars, pm.positions[symbol]["entry_price"], trade_imbalance, t_utc, resistance_timer_state.get(symbol, {}), cfg=resistance_timer_cfg) resistance_timer_state[symbol] = rt_decision.state_out if symbol in trace_symbols: trace.append({"timestamp": t_utc.isoformat(), "symbol": symbol, "kind": "resistance_timer_exit", "state": rt_decision.state, "reason": rt_decision.reason, "metrics": rt_decision.metrics}) vel_decision = None if (use_velocity_exit and not decision.should_exit and not (gb_decision and gb_decision.should_exit) and not (fs_decision and fs_decision.should_exit) and not (prf_decision and prf_decision.should_exit) and not (rt_decision and rt_decision.should_exit)): sub_bars = buf.get_bars_sub() vel_decision = velocity_exit.evaluate( symbol, bars, sub_bars, pm.positions[symbol]["entry_price"], velocity_state.get(symbol, {}), cfg=velocity_cfg) velocity_state[symbol] = vel_decision.state_out if symbol in trace_symbols: trace.append({"timestamp": t_utc.isoformat(), "symbol": symbol, "kind": "velocity_exit", "state": vel_decision.state, "reason": vel_decision.reason, "metrics": vel_decision.metrics}) cg_decision = None if (use_cents_grace_exit and not decision.should_exit and not (gb_decision and gb_decision.should_exit) and not (fs_decision and fs_decision.should_exit) and not (prf_decision and prf_decision.should_exit) and not (rt_decision and rt_decision.should_exit) and not (vel_decision and vel_decision.should_exit)): cg_decision = cents_grace_exit.evaluate( symbol, bars, pm.positions[symbol]["entry_price"], cents_grace_state.get(symbol, {}), cfg=cents_grace_cfg) cents_grace_state[symbol] = cg_decision.state_out if symbol in trace_symbols: trace.append({"timestamp": t_utc.isoformat(), "symbol": symbol, "kind": "cents_grace_exit", "state": cg_decision.state, "reason": cg_decision.reason, "metrics": cg_decision.metrics}) sl_decision = None if (use_step_lock_exit and not decision.should_exit and not (gb_decision and gb_decision.should_exit) and not (fs_decision and fs_decision.should_exit) and not (prf_decision and prf_decision.should_exit) and not (rt_decision and rt_decision.should_exit) and not (vel_decision and vel_decision.should_exit) and not (cg_decision and cg_decision.should_exit)): call_cfg = step_lock_cfg if step_lock_use_atr_distance: pos = pm.positions[symbol] atr_stop_pct = ((pos["entry_price"] - pos["stop_price"]) / pos["entry_price"] * 100.0 if pos["entry_price"] else (step_lock_cfg or {}).get("initial_distance_pct")) call_cfg = {**(step_lock_cfg or {}), "initial_distance_pct": atr_stop_pct} sl_decision = step_lock_exit.evaluate( symbol, bars, pm.positions[symbol]["entry_price"], step_lock_state.get(symbol, {}), cfg=call_cfg) step_lock_state[symbol] = sl_decision.state_out if symbol in trace_symbols: trace.append({"timestamp": t_utc.isoformat(), "symbol": symbol, "kind": "step_lock_exit", "state": sl_decision.state, "reason": sl_decision.reason, "metrics": sl_decision.metrics}) def _pop_all_exit_state(): exit_confirmation.pop(symbol, None) giveback_state.pop(symbol, None) prf_state.pop(symbol, None) resistance_timer_state.pop(symbol, None) velocity_state.pop(symbol, None) cents_grace_state.pop(symbol, None) step_lock_state.pop(symbol, None) giveback_room_state.pop(symbol, None) if decision.should_exit: current_price = bars[-1]["c"] pm.exit(symbol, current_price, decision.reason, t_utc) _pop_all_exit_state() elif gb_decision is not None and gb_decision.should_exit: current_price = bars[-1]["c"] pm.exit(symbol, current_price, gb_decision.reason, t_utc) _pop_all_exit_state() elif fs_decision is not None and fs_decision.should_exit: current_price = bars[-1]["c"] pm.exit(symbol, current_price, fs_decision.reason, t_utc) _pop_all_exit_state() elif prf_decision is not None and prf_decision.should_exit: current_price = bars[-1]["c"] pm.exit(symbol, current_price, prf_decision.reason, t_utc) _pop_all_exit_state() elif rt_decision is not None and rt_decision.should_exit: current_price = bars[-1]["c"] pm.exit(symbol, current_price, rt_decision.reason, t_utc) _pop_all_exit_state() elif vel_decision is not None and vel_decision.should_exit: current_price = bars[-1]["c"] pm.exit(symbol, current_price, vel_decision.reason, t_utc) _pop_all_exit_state() elif cg_decision is not None and cg_decision.should_exit: current_price = bars[-1]["c"] pm.exit(symbol, current_price, cg_decision.reason, t_utc) _pop_all_exit_state() elif sl_decision is not None and sl_decision.should_exit: current_price = bars[-1]["c"] pm.exit(symbol, current_price, sl_decision.reason, t_utc) _pop_all_exit_state() # [2026-09-23] Exit-isolation mode: enter exactly the given trades # (symbol, time, stop) and nothing else, no slot limit -- so every # exit variant is scored on the same entries. if fixed_entries is not None: for fe in fixed_entries: if fe.get("_done") or fe["_t"] > t_utc: continue fe["_done"] = True buf = buffers[fe["symbol"]] bars = buf.get_bars() if not bars or pm.is_symbol_open(fe["symbol"]): fe["_skipped"] = True continue stop = fe["stop_price"] if fixed_min_stop["enabled"]: q = buf.latest_quote fl = volatility.stop_floor(bars[-1]["c"], resistance_levels[fe["symbol"]].get("daily_atr"), (q[1] - q[0]) if q else 0.0, fixed_min_stop) stop = min(stop, bars[-1]["c"] - fl["floor"]) pm.enter(fe["symbol"], bars[-1]["c"], stop, t_utc, fe.get("reasons")) continue # entries if pm.has_available_slot() and t_utc < no_new_entries: for c in candidates: symbol = c["symbol"] if pm.is_symbol_open(symbol): continue if not pm.has_available_slot(): break if watch is not None and not any(a <= t_utc < b for a, b in watch.get(symbol, [])): continue buf = buffers[symbol] bars = buf.get_bars() if not bars: continue if reentry_cfg: prior = [tr for tr in pm.trades if tr["symbol"] == symbol] allowed, _why = reentry.check(symbol, bars[-1]["c"], prior[-1] if prior else None, reentry_cfg) if not allowed: entry_persistence.pop(symbol, None) continue quote_tuple = buf.latest_quote quote = {"bid": quote_tuple[0], "ask": quote_tuple[1]} if quote_tuple else None session_high = max(b["h"] for b in bars) resistance = {**resistance_levels[symbol], "session_high": session_high} cumulative_volume = sum(b["v"] for b in bars) persistence_state = entry_persistence.get(symbol, {}) trade_imbalance = buf.get_trade_imbalance(entry_imbalance_window) decision = smart_engine.evaluate( symbol, bars, cumulative_volume, avg_vol_baseline.get(symbol), resistance, persistence_state, quote=quote, trade_imbalance=trade_imbalance, as_of=t_utc, cfg=smart_engine_cfg) entry_persistence[symbol] = decision.persistence if decision.should_enter and shadow_gate: # [2026-09-26] --shadow-gate: the bot's entry must ALSO get # an allowed entry_rules verdict at that moment; if not, no # buy -- persistence is kept, so it retries next poll verdict = shadow._build_snapshot(symbol, "GATE", t_utc, None).get("shadow", {}).get("decision") if verdict not in shadow_gate: gate_blocked[symbol] = gate_blocked.get(symbol, 0) + 1 decision.should_enter = False decision.state = f"GATE_{verdict or 'NO_DATA'}" if shadow is not None: record_decision(shadow, shadow_keys, symbol, decision, now=t_utc) if symbol in trace_symbols: trace.append({"timestamp": t_utc.isoformat(), "symbol": symbol, "kind": "entry", "state": decision.state, "reasons_for": decision.reasons_for, "reasons_against": decision.reasons_against, "metrics": decision.metrics}) if decision.should_enter: stop = decision.metrics.get("stage3", {}).get("stop") if stop is None: continue entered = pm.enter(symbol, decision.metrics["price"], stop, t_utc, decision.reasons_for) if entered: entry_persistence.pop(symbol, None) # EOD liquidation fl_utc = force_liquidate.astimezone(UTC) for symbol in list(pm.positions.keys()): readers[symbol].drain_up_to(fl_utc, buffers[symbol]) bars = buffers[symbol].get_bars() price = bars[-1]["c"] if bars else pm.positions[symbol]["entry_price"] pm.exit(symbol, price, "END_OF_DAY", fl_utc) for r in readers.values(): r.close() pm.trace = trace pm.gate_blocked = gate_blocked return pm def summarize(trades: list) -> dict: total_pl = sum(t["pl_dollars"] for t in trades) wins = sum(1 for t in trades if t["pl_dollars"] > 0) losses = sum(1 for t in trades if t["pl_dollars"] <= 0) return {"trades": len(trades), "wins": wins, "losses": losses, "total_pl": round(total_pl, 2)} def print_trades(trades: list, label: str): print(f"\n=== {label} ({len(trades)} trades) ===") for t in trades: print(f" {t['entry_time'][11:19]} {t['symbol']:6s} BUY {t['qty']:>4d} @ {t['entry_price']:.4f} " f"-> {t['exit_time'][11:19]} SELL @ {t['exit_price']:.4f} " f"P/L {t['pl_pct']:+.2f}% (${t['pl_dollars']:+.2f}) [{t['exit_reason'][:60]}]") s = summarize(trades) print(f" --- total: {s['trades']} trades, {s['wins']}W/{s['losses']}L, ${s['total_pl']:+.2f} ---") def sanity_check(sim_trades: list, date_str: str): real_path = BASE_DIR / "data" / "trades" / f"{date_str}_trades.jsonl" if not real_path.exists(): print(f"\n[sanity] no real trades file at {real_path}, skipping") return real_trades = [json.loads(line) for line in open(real_path)] print_trades(real_trades, "REAL (actual live run today)") real_syms = [t["symbol"] for t in real_trades] sim_syms = [t["symbol"] for t in sim_trades] print(f"\n[sanity] real sequence: {real_syms}") print(f"[sanity] simulated sequence: {sim_syms}") real_s = summarize(real_trades) sim_s = summarize(sim_trades) print(f"[sanity] real: {real_s}") print(f"[sanity] simulated: {sim_s}") def main(): ap = argparse.ArgumentParser() ap.add_argument("--date", required=True, help="YYYY-MM-DD") ap.add_argument("--max-positions", type=int, required=True) ap.add_argument("--sanity", action="store_true", help="diff against data/trades/_trades.jsonl") ap.add_argument("--workers", type=int, default=3, help="parallel fetch workers") ap.add_argument("--synthetic-grid", action="store_true", help="use an exact-5s poll grid instead of the real logged poll times " "(default prefers real times when data/decisions/.jsonl exists)") ap.add_argument("--trace-symbols", default="", help="comma-separated symbols to dump every entry/exit decision's full " "metrics dict for, e.g. BNC,GLOO") ap.add_argument("--giveback-exit", action="store_true", help="also check the experimental giveback-from-peak 4th exit layer " "(exit_giveback_experimental.py) alongside the real exit.py") ap.add_argument("--fast-slope-exit", action="store_true", help="also check the experimental fast sub-minute price-slope exit layer " "(exit_fast_slope_experimental.py) alongside the real exit.py") ap.add_argument("--prf-exit", action="store_true", help="also check the experimental Price Response Failure exit layer " "(exit_prf_experimental.py) alongside the real exit.py") ap.add_argument("--resistance-timer-exit", action="store_true", help="also check the experimental resistance + fixed recovery-window exit layer " "(exit_resistance_timer_experimental.py) alongside the real exit.py") ap.add_argument("--resistance-timer-window", type=float, default=60.0, help="recovery_window_seconds override for --resistance-timer-exit (default 60)") ap.add_argument("--resistance-timer-imbalance", type=float, default=0.0, help="resistance_imbalance_threshold override for --resistance-timer-exit (default 0.0, " "i.e. any positive imbalance counts -- PRF's own 'strong buying' bar is 0.70)") ap.add_argument("--velocity-exit", action="store_true", help="also check the experimental decline-velocity + participation-ratio exit layer " "(exit_velocity_experimental.py) alongside the real exit.py") ap.add_argument("--velocity-min-pct-per-min", type=float, default=6.0, help="min_decline_velocity_pct_per_min override for --velocity-exit (default 6.0)") ap.add_argument("--velocity-max-participation", type=float, default=1.0, help="max_participation_ratio_to_exit override for --velocity-exit (default 1.0)") ap.add_argument("--cents-grace-exit", action="store_true", help="also check the experimental flat-cents trailing stop + health-gated grace exit " "(exit_cents_grace_experimental.py, ported from the sibling premarket bot) " "alongside the real exit.py") ap.add_argument("--cents-grace-base", type=float, default=0.10, help="base_giveback_cents override for --cents-grace-exit (default 0.10)") ap.add_argument("--cents-grace-amount", type=float, default=0.05, help="grace_cents override for --cents-grace-exit (default 0.05)") ap.add_argument("--cents-grace-max-extensions", type=int, default=1, help="max_grace_extensions override for --cents-grace-exit (default 1)") ap.add_argument("--step-lock-exit", action="store_true", help="also check the experimental cent-for-cent-to-breakeven then %%-step-lock " "exit layer (exit_step_lock_experimental.py) alongside the real exit.py") ap.add_argument("--step-lock-distance", type=float, default=1.5, help="initial_distance_pct override for --step-lock-exit (default 1.5)") ap.add_argument("--step-lock-step", type=float, default=5.0, help="step_pct override for --step-lock-exit (default 5.0)") ap.add_argument("--step-lock-cap", type=float, default=None, help="giveback_cap_pct override for --step-lock-exit (default None, i.e. off)") ap.add_argument("--step-lock-tolerance", type=float, default=0.0, help="breakeven_tolerance_pct override for --step-lock-exit (default 0.0) -- " "the breakeven floor sits this %% below entry instead of exactly at it") ap.add_argument("--step-lock-use-atr-distance", action="store_true", help="for --step-lock-exit: use each trade's OWN entry-time ATR-based stop " "distance as initial_distance_pct instead of --step-lock-distance's fixed " "value -- ignores --step-lock-distance when set. Combine with " "--step-lock-step 0 for a pure breakeven-guarantee stop (no phase-2 lock).") ap.add_argument("--giveback-room-exit", action="store_true", help="also check the new production giveback-room exit layer " "(exit_giveback_room.py) -- checked FIRST, ahead of exit.py, " "same priority order as monitor.py's real live wiring") ap.add_argument("--giveback-room-min-cents", type=float, default=15.0, help="min_peak_gain_cents override for --giveback-room-exit (default 15.0)") ap.add_argument("--breakeven-lock", type=float, default=0.0, help="giveback-room: once peak gain >= this many cents, exit at entry + 1c (0 = off)") ap.add_argument("--giveback-no-slope", action="store_true", help="giveback-room: fire the giveback floor without waiting for a negative slope") ap.add_argument("--giveback-room-ratio", type=float, default=2.0 / 3.0, help="giveback_ratio override for --giveback-room-exit (default 0.6667)") ap.add_argument("--scanner-stop-cents", type=float, default=None, help="mirror monitor.py's scanner_stop_phase: until the peak since entry has run up " "this many cents, the entry stop is the only exit (default off)") ap.add_argument("--min-entry-score", type=float, default=None, help="smart_engine.min_entry_score override -- reject Stage-3-passed entries below " "this live entry_score (0-100, see smart_engine._compute_entry_score). " "Default None leaves config.json's value (0.0, i.e. off) in place.") ap.add_argument("--resistance-stall-exit", action="store_true", help="enable exit_resistance_stall (sell at the ceiling after ~60 min of failed " "tests with no breakout); settings from config.json's resistance_stall") ap.add_argument("--stall-low-ceiling", action="store_true", help="resistance_stall.low_ceiling_enabled: once armed, sell at the last-15-min ceiling " "when no slot is free or the position isn't healthy") ap.add_argument("--dead-money-minutes", type=float, default=None, help="resistance_stall.dead_money_minutes (with --stall-low-ceiling)") ap.add_argument("--exit-trend-bars", type=int, default=None, help="exit.deterioration_trend_bars: 'trend negative' sign over the last N bars") ap.add_argument("--stall-minutes", type=float, default=None, help="override resistance_stall.stall_minutes for --resistance-stall-exit") ap.add_argument("--min-stop-style", choices=["k", "tiers"], default=None, help="enable smart_engine.min_stop (volatility-scaled minimum stop distance)") ap.add_argument("--min-stop-k", type=float, default=0.15, help="for --min-stop-style k: floor = k x daily ATR") ap.add_argument("--reentry-after-loss", choices=["off", "block", "above_entry"], default=None, help="trading.reentry_after_loss.mode override (entry selectivity after a losing trade)") ap.add_argument("--fixed-entries", default=None, help="exit-isolation mode: path to a previous run's JSON; enter exactly its trades " "(same symbol/time/stop), nothing else, no slot limit") ap.add_argument("--max-ext-open", action="store_true", help="enable smart_engine.max_extension_from_open (3-5%% above today's open by volume/buying)") ap.add_argument("--shadow-gate", choices=["pass", "pass_incomplete"], default=None, help="[2026-09-26] only buy when entry_rules says PASS (or PASS/INCOMPLETE) at the " "moment the bot wants to enter; requires --shadow-log") ap.add_argument("--shadow-log", default=None, help="run the entry_rules.py shadow recorder during the replay, writing snapshots to this dir") ap.add_argument("--plan-5d-levels", action="store_true", help="setup_plan.five_day_levels: 5-day high as a level + realistic target cap (open + 1 daily ATR)") ap.add_argument("--plan-real-targets", action="store_true", help="setup_plan.real_level_targets: never target high_of_day / young ceilings") ap.add_argument("--plan-reclaim", action="store_true", help="setup_plan.reclaim_breakouts: count reclaim breakouts of a level") ap.add_argument("--watch-windows", default=None, help="JSON {symbol: {windows: [[start_iso, end_iso], ...], metrics: {...}}} -- replays " "intraday rescans: candidates = these symbols, each only evaluated for entry " "inside its windows") ap.add_argument("--plan-breakout-stop", choices=["pivot", "level"], default=None, help="override setup_plan.breakout_stop for this run") ap.add_argument("--plan-min-rr", type=float, default=None, help="override setup_plan.min_reward_risk for this run") ap.add_argument("--plan-no-context", action="store_true", help="force setup_plan.context_rules OFF for this run (overrides config.json)") ap.add_argument("--plan-context", action="store_true", help="enable setup_plan.context_rules (breakout needs rising slope; pullback needs " "reclaim, fading selling, support held; lower-highs reject waived for pullbacks)") ap.add_argument("--setup-plan", action="store_true", help="use setup_analyzer's per-tick trade plan in place of Stage 3 + entry_score " "(its stop replaces 3x ATR); settings from config.json's setup_plan") ap.add_argument("--entry-slope-bars", type=int, default=None, help="enable smart_engine's entry_slope_rule: the last N 1-min bars must slope up " "on every read of the hold (default off / config.json's value)") ap.add_argument("--entry-slope-mode", choices=["positive", "above_zero"], default="positive", help="for --entry-slope-bars: 'positive' = above the ATR-scaled flat band, " "'above_zero' = any slope > 0") ap.add_argument("--step-lock-handoff", action="store_true", help="for --step-lock-exit: only protect the opening leg (entry -> breakeven) -- " "once the peak clears initial_distance_pct, stop enforcing any stop here at " "all and let the real exit.py hard-stop + deterioration layers run the rest " "of the trade, instead of holding a flat breakeven floor forever") args = ap.parse_args() if args.shadow_gate and not args.shadow_log: ap.error("--shadow-gate needs --shadow-log") trace_symbols = [s.strip() for s in args.trace_symbols.split(",") if s.strip()] watch = None if args.watch_windows: wj = json.loads(Path(args.watch_windows).read_text()) candidates = [{"symbol": sym, "metrics": v["metrics"]} for sym, v in wj.items()] watch = {sym: [(datetime.fromisoformat(a), datetime.fromisoformat(b)) for a, b in v["windows"]] for sym, v in wj.items()} else: candidates = load_candidates(args.date) symbols = [c["symbol"] for c in candidates] print(f"Loaded {len(symbols)} candidates for {args.date}: {symbols}", file=sys.stderr) client = get_client() sched = get_config()["schedule"] start = _et_dt(args.date, sched["market_open_time"]).astimezone(UTC) end = _et_dt(args.date, sched["market_close_time"]).astimezone(UTC) cache_dir = BASE_DIR / "data" / "simulations" / "cache" / args.date print(f"Ensuring tick-data cache in {cache_dir} ...", file=sys.stderr) ensure_cache(client, symbols, start, end, cache_dir, workers=args.workers) account = client.get_account() equity = float(account.equity) print(f"Using fixed equity snapshot: ${equity:.2f}", file=sys.stderr) poll_times = None if args.synthetic_grid else load_real_poll_times(args.date) if poll_times: print(f"Driving replay off {len(poll_times)} REAL logged poll timestamps " f"(data/decisions/{args.date}.jsonl)", file=sys.stderr) else: print("No real decisions log found -- using a synthetic exact-5s poll grid", file=sys.stderr) # Overrides are merged ONTO config.json's smart_engine section -- passing # a bare {"min_entry_score": ..} used to replace the whole section, which # silently reverted atr_multiplier_stop etc. to smart_engine.py defaults. if args.plan_5d_levels: get_config().setdefault("setup_plan", {}).setdefault("five_day_levels", {})["enabled"] = True if args.plan_real_targets: get_config().setdefault("setup_plan", {})["real_level_targets"] = True if args.plan_reclaim: get_config().setdefault("setup_plan", {})["reclaim_breakouts"] = True if args.plan_breakout_stop: get_config().setdefault("setup_plan", {})["breakout_stop"] = args.plan_breakout_stop if args.plan_min_rr is not None: get_config().setdefault("setup_plan", {})["min_reward_risk"] = args.plan_min_rr if args.exit_trend_bars: get_config().setdefault("exit", {})["deterioration_trend_bars"] = args.exit_trend_bars if args.plan_no_context: # The live config may have context_rules ON (9/24 mid-day switch); # force it off so "without context" variants really are without it. get_config().setdefault("setup_plan", {}).setdefault("context_rules", {})["enabled"] = False if args.plan_context: # get_config() is cached, so this in-process override reaches # setup_analyzer (which reads setup_plan from config directly). get_config().setdefault("setup_plan", {}).setdefault("context_rules", {})["enabled"] = True se_overrides = {} if args.min_entry_score is not None: se_overrides["min_entry_score"] = args.min_entry_score if args.entry_slope_bars is not None: se_overrides["entry_slope_rule"] = {"enabled": args.entry_slope_bars > 0, "lookback_bars": args.entry_slope_bars or 5, "mode": args.entry_slope_mode} if args.setup_plan: se_overrides["use_setup_plan"] = True if args.max_ext_open: se_overrides["max_extension_from_open"] = {**get_config().get("smart_engine", {}).get("max_extension_from_open", {}), "enabled": True} if args.min_stop_style: se_overrides["min_stop"] = {**get_config().get("smart_engine", {}).get("min_stop", {}), "enabled": True, "style": args.min_stop_style, "k": args.min_stop_k} if se_overrides: se_overrides = {**get_config().get("smart_engine", {}), **se_overrides} pm = run_simulation(args.date, args.max_positions, candidates, cache_dir, equity, poll_times=poll_times, trace_symbols=trace_symbols, use_giveback_exit=args.giveback_exit, use_fast_slope_exit=args.fast_slope_exit, use_prf_exit=args.prf_exit, use_resistance_timer_exit=args.resistance_timer_exit, resistance_timer_cfg={"recovery_window_seconds": args.resistance_timer_window, "resistance_imbalance_threshold": args.resistance_timer_imbalance}, use_velocity_exit=args.velocity_exit, velocity_cfg={"min_decline_velocity_pct_per_min": args.velocity_min_pct_per_min, "max_participation_ratio_to_exit": args.velocity_max_participation}, use_cents_grace_exit=args.cents_grace_exit, cents_grace_cfg={"base_giveback_cents": args.cents_grace_base, "grace_cents": args.cents_grace_amount, "max_grace_extensions": args.cents_grace_max_extensions}, use_step_lock_exit=args.step_lock_exit, step_lock_cfg={"initial_distance_pct": args.step_lock_distance, "step_pct": args.step_lock_step, "giveback_cap_pct": args.step_lock_cap, "breakeven_tolerance_pct": args.step_lock_tolerance, "handoff_at_breakeven": args.step_lock_handoff}, step_lock_use_atr_distance=args.step_lock_use_atr_distance, use_giveback_room_exit=args.giveback_room_exit, giveback_room_cfg={**get_config().get("giveback_room", {}), "min_peak_gain_cents": args.giveback_room_min_cents, "giveback_ratio": args.giveback_room_ratio, "breakeven_lock_cents": args.breakeven_lock, "require_slope": not args.giveback_no_slope}, smart_engine_cfg=se_overrides or None, scanner_stop_cents=args.scanner_stop_cents, resistance_stall_cfg=( {**exit_resistance_stall.DEFAULT_CONFIG, **get_config().get("resistance_stall", {}), "enabled": True, **({"stall_minutes": args.stall_minutes} if args.stall_minutes else {}), **({"low_ceiling_enabled": True} if args.stall_low_ceiling else {}), **({"dead_money_minutes": args.dead_money_minutes} if args.dead_money_minutes else {})} if args.resistance_stall_exit else None), daily_atr=load_daily_atr(client, symbols, args.date, cache_dir), ref5d=load_ref5d(client, symbols, args.date, cache_dir), fixed_entries=_load_fixed(args.fixed_entries), watch=watch, shadow=(ShadowEntryRules({"log_dir": args.shadow_log, "track_outcomes": False}) if args.shadow_log else None), bench_bars=_bench_bars(client, args.date) if args.shadow_log else None, shadow_gate=({"pass": {"PASS"}, "pass_incomplete": {"PASS", "INCOMPLETE"}}[args.shadow_gate] if args.shadow_gate else None), reentry_cfg=({**get_config()["trading"].get("reentry_after_loss", {}), "mode": args.reentry_after_loss} if args.reentry_after_loss else get_config()["trading"].get("reentry_after_loss"))) label = (f"max_positions={args.max_positions}" + (f" +giveback-room-exit(min={args.giveback_room_min_cents:.1f}c," f"ratio={args.giveback_room_ratio:.3f})" if args.giveback_room_exit else "") + (f" +min-entry-score={args.min_entry_score:.1f}" if args.min_entry_score is not None else "") + (f" +resistance-stall({args.stall_minutes or get_config().get('resistance_stall', {}).get('stall_minutes', 60):.0f}m)" if args.resistance_stall_exit else "") + (" +setup-plan" if args.setup_plan else "") + (" +plan-context" if args.plan_context else "") + ((f" +min-stop(k={args.min_stop_k})" if args.min_stop_style == "k" else " +min-stop(tiers)") if args.min_stop_style else "") + (f" +entry-slope({args.entry_slope_bars} bars,{args.entry_slope_mode})" if args.entry_slope_bars is not None else "") + (f" +scanner-stop-phase({args.scanner_stop_cents:.0f}c)" if args.scanner_stop_cents is not None else "") + (" +giveback-exit" if args.giveback_exit else "") + (" +fast-slope-exit" if args.fast_slope_exit else "") + (" +prf-exit" if args.prf_exit else "") + (f" +resistance-timer-exit({args.resistance_timer_window:.0f}s," f"imb>={args.resistance_timer_imbalance:.2f})" if args.resistance_timer_exit else "") + (f" +velocity-exit(vel>={args.velocity_min_pct_per_min:.1f}%/min," f"part<{args.velocity_max_participation:.2f})" if args.velocity_exit else "") + (f" +cents-grace-exit(${args.cents_grace_base:.2f}+${args.cents_grace_amount:.2f}x" f"{args.cents_grace_max_extensions})" if args.cents_grace_exit else "") + (f" +step-lock-exit(D={'ATR@entry' if args.step_lock_use_atr_distance else f'{args.step_lock_distance:.1f}%'}" f",step={args.step_lock_step:.1f}%" + (f",cap={args.step_lock_cap:.1f}%" if args.step_lock_cap is not None else "") + (f",tol={args.step_lock_tolerance:.1f}%" if args.step_lock_tolerance else "") + (",handoff" if args.step_lock_handoff else "") + ")" if args.step_lock_exit else "")) print_trades(pm.trades, label) if args.sanity: sanity_check(pm.trades, args.date) out_dir = BASE_DIR / "data" / "simulations" out_dir.mkdir(parents=True, exist_ok=True) suffix = (("_giveback" if args.giveback_exit else "") + ("_fastslope" if args.fast_slope_exit else "") + ("_prf" if args.prf_exit else "") + ("_rt" if args.resistance_timer_exit else "") + ("_vel" if args.velocity_exit else "") + ("_cg" if args.cents_grace_exit else "") + ("_sl" if args.step_lock_exit else "") + ("_gr" if args.giveback_room_exit else "") + (f"_r{args.giveback_room_ratio * 100:.0f}" if args.giveback_room_exit and abs(args.giveback_room_ratio - 2.0 / 3.0) > 1e-3 else "") + (f"_mes{args.min_entry_score:.0f}" if args.min_entry_score is not None else "") + (f"_es{args.entry_slope_bars}{'p' if args.entry_slope_mode == 'positive' else 'z'}" if args.entry_slope_bars is not None else "") + (f"_ss{args.scanner_stop_cents:.0f}" if args.scanner_stop_cents is not None else "") + (f"_rs{int(args.stall_minutes or get_config().get('resistance_stall', {}).get('stall_minutes', 60))}" if args.resistance_stall_exit else "") + ("_plan" if args.setup_plan else "") + ("_ctx" if args.plan_context else "") + (f"_rr{int(args.plan_min_rr * 10)}" if args.plan_min_rr is not None else "") + (f"_bs{args.plan_breakout_stop}" if args.plan_breakout_stop else "") + (f"_re{args.reentry_after_loss}" if args.reentry_after_loss else "") + ("_fixed" if args.fixed_entries else "") + ("_rt" + str(int(args.plan_real_targets)) if args.plan_real_targets else "") + ("_rc" if args.plan_reclaim else "") + ("_ww" if args.watch_windows else "") + ("_ext" if args.max_ext_open else "") + ("_lc" if args.stall_low_ceiling else "") + ("_5d" if args.plan_5d_levels else "") + (f"_dm{int(args.dead_money_minutes)}" if args.dead_money_minutes else "") + (f"_et{args.exit_trend_bars}" if args.exit_trend_bars else "") + ((f"_msk{int(args.min_stop_k * 100)}" if args.min_stop_style == "k" else "_mst") if args.min_stop_style else "") + (f"_gate{args.shadow_gate}" if args.shadow_gate else "")) out_path = out_dir / f"{args.date}_sim_maxpos{args.max_positions}{suffix}.json" with open(out_path, "w") as f: json.dump({"date": args.date, "max_positions": args.max_positions, "equity_used": equity, "trades": pm.trades, "summary": summarize(pm.trades), "gate_blocked": getattr(pm, "gate_blocked", {})}, f, indent=2) if args.shadow_gate: print(f"shadow gate blocked entry attempts: {getattr(pm, 'gate_blocked', {})}") print(f"\nWrote {out_path}", file=sys.stderr) if trace_symbols: trace_path = out_dir / f"{args.date}_trace_maxpos{args.max_positions}.jsonl" with open(trace_path, "w") as f: for rec in pm.trace: f.write(json.dumps(rec) + "\n") print(f"Wrote {len(pm.trace)} trace records to {trace_path}", file=sys.stderr) if __name__ == "__main__": main()