"""TGTC single-day backtest simulator. Uses 5-min bars from IntradayCache and a synthetic gainer reconstruction (no Yahoo API) to simulate the TGTC VWAP pullback reclaim strategy on any historical trading day. """ from __future__ import annotations import datetime as dt import logging import math from dataclasses import dataclass, field from typing import Any log = logging.getLogger(__name__) from zoneinfo import ZoneInfo as _ZoneInfo _ET_ZONE = _ZoneInfo("America/New_York") def _et_to_utc_naive(date: dt.date, hour: int, minute: int) -> dt.datetime: """Convert an ET time on a given date to naive UTC (DST-aware).""" et_aware = dt.datetime(date.year, date.month, date.day, hour, minute, tzinfo=_ET_ZONE) return et_aware.astimezone(dt.timezone.utc).replace(tzinfo=None) def _bar_idx_at(bars: list[dict], cutoff_naive_utc: dt.datetime) -> int: """Return index of the last bar whose timestamp <= cutoff (or -1).""" from libs.tgtc.gainers_reconstruct import _bar_ts_naive_utc last = -1 for i, b in enumerate(bars): if _bar_ts_naive_utc(b) <= cutoff_naive_utc: last = i else: break return last @dataclass class TGTCTrade: symbol: str entry_price: float stop_price: float shares: int entry_bar_idx: int exit_price: float = 0.0 exit_bar_idx: int = -1 exit_reason: str = "" pnl: float = 0.0 r_multiple: float = 0.0 partial_taken: bool = False peak_price: float = 0.0 current_stop: float = 0.0 be_stop_active: bool = False status: str = "open" # open | closed entry_dt_utc: dt.datetime | None = None # set at entry for time-stop calculations partial_levels_taken: set = field(default_factory=set) # tracks which partial_levels fired tp_collision: bool = False # True if quick_tp fired AND stop was also touched on the same bar side: str = "long" # "long" or "short" # Candidate metadata (attached at entry for segmentation analysis) score: float = 0.0 rank_persistence: float = 0.0 rank_velocity: float = 0.0 price_structure: float = 0.0 volume_quality: float = 0.0 relative_strength: float = 0.0 pct_change_at_10: float = 0.0 dollar_volume_20d: float = 0.0 @dataclass class SimResult: date: str trades: list[TGTCTrade] = field(default_factory=list) candidates: list[dict[str, Any]] = field(default_factory=list) equity_curve: list[dict[str, Any]] = field(default_factory=list) initial_equity: float = 10000.0 final_equity: float = 10000.0 @property def total_pnl(self) -> float: return sum(t.pnl for t in self.trades if t.status == "closed") @property def total_return_pct(self) -> float: return self.total_pnl / self.initial_equity if self.initial_equity > 0 else 0.0 @property def win_rate(self) -> float: closed = [t for t in self.trades if t.status == "closed"] if not closed: return 0.0 return sum(1 for t in closed if t.pnl > 0) / len(closed) @property def n_trades(self) -> int: return len([t for t in self.trades if t.status == "closed"]) def run_tgtc_simulation( date: dt.date, bars_by_symbol: dict[str, list[dict]], prev_closes: dict[str, float], enrichment: dict[str, dict], # {symbol: {"atr_14": ..., "avg_dollar_vol_30d": ...}} qqq_pct_change_at_10: float | None = None, cfg: Any = None, # TGTCConfig ) -> SimResult: """Run a single-day TGTC simulation. Args: date: Trading date. bars_by_symbol: {symbol: [5m bar dicts]} filtered to this date's market hours. prev_closes: {symbol: float} prior-day close. enrichment: {symbol: {atr_14, avg_dollar_vol_30d, ...}} from enrich_daily_bars. qqq_pct_change_at_10: QQQ percent change at 10:00 ET (for RS score). cfg: TGTCConfig instance. Returns: SimResult with trade list, equity curve, candidates. """ raise NotImplementedError( "V1 simulator is deprecated in the TGTC V2 transition. " "V2 candidate selection and labeling is in scripts/build_tgtc_v2_top_gainer_events.py. " "V1 backtest results are archived in docs/tgtc_v1_development_report.md." ) # ── code below is V1 reference (not executed) ────────────────────────────── from libs.tgtc.domain import TGTCConfig from libs.tgtc.gainers_reconstruct import reconstruct_gainer_snapshots, _bar_ts_naive_utc from libs.tgtc.v2_features import compute_rank_features if cfg is None: cfg = TGTCConfig() params = cfg.tgtc_strategy flt = params.filters sw = params.score_weights ent = params.entry ex = params.exit rsk = params.risk date_str = date.isoformat() result = SimResult(date=date_str, initial_equity=rsk.initial_equity) equity = rsk.initial_equity daily_loss = 0.0 # QQQ regime gate if flt.min_qqq_pct_change_at_10 is not None and qqq_pct_change_at_10 is not None: if qqq_pct_change_at_10 < flt.min_qqq_pct_change_at_10: log.debug("TGTC %s: skip — QQQ %.2f%% < floor %.2f%%", date_str, qqq_pct_change_at_10*100, flt.min_qqq_pct_change_at_10*100) result.final_equity = equity return result # ── Step 1: Synthetic gainer snapshots (09:30–09:55) ────────────────────── snapshots = reconstruct_gainer_snapshots( bars_by_symbol=bars_by_symbol, prev_closes=prev_closes, date_str=date_str, min_pct_change=0.03, top_n=100, ) result.candidates = [] # will be filled after candidate selection if not snapshots: log.debug("TGTC %s: no synthetic snapshots (empty universe?)", date_str) result.final_equity = equity return result # ── Step 2: Rank features from collected snapshots ──────────────────────── rank_features = compute_rank_features(snapshots) # ── Step 3: At 10:00 ET, select candidates ───────────────────────────────── cutoff_10 = _et_to_utc_naive(date, 10, 0) candidates_scored: list[dict[str, Any]] = [] for sym, rf in rank_features.items(): bars = bars_by_symbol.get(sym, []) if not bars: continue bar_idx_10 = _bar_idx_at(bars, cutoff_10) if bar_idx_10 < 0: continue bar_10 = bars[bar_idx_10] prev_close = prev_closes.get(sym, 0.0) if prev_close <= 0: continue price_at_10 = float(bar_10["close"]) pct_change_at_10 = (price_at_10 - prev_close) / prev_close # Hard filters if price_at_10 < flt.min_price: continue if pct_change_at_10 < flt.min_day_change_at_10: continue if pct_change_at_10 > flt.max_day_change_at_10: continue # Dollar volume surrogate for market_cap filter enr = enrichment.get(sym, {}) avg_dv = enr.get("avg_dollar_vol_30d") or enr.get("avg_dollar_vol_20d") if avg_dv and avg_dv < flt.min_avg_dollar_volume_20d: continue # VWAP filter vwap_at_10 = get_bar_vwap(bars, bar_idx_10) if flt.must_be_above_vwap and (not vwap_at_10 or price_at_10 < vwap_at_10): continue # HOD pullback filter local_highs = [float(b["high"]) for b in bars[:bar_idx_10 + 1]] hod = max(local_highs) if local_highs else price_at_10 if hod > 0 and (hod - price_at_10) / hod > flt.max_pullback_from_hod: continue # Scores ps = compute_price_structure_score(bars, bar_idx_10) vq = compute_volume_quality(bars, bar_idx_10, avg_dv) rs = compute_relative_strength(pct_change_at_10, qqq_pct_change_at_10) tgtc_score = compute_tgtc_score( rank_persistence=rf["rank_persistence"], rank_velocity=max(0.0, rf["rank_velocity"]), price_structure=ps, volume_quality=vq, relative_strength=rs, weights=sw, ) atr_intraday = enr.get("atr_14") # use daily ATR as proxy for intraday candidates_scored.append({ "symbol": sym, "score": tgtc_score, "rank_persistence": rf["rank_persistence"], "rank_velocity": rf["rank_velocity"], "price_structure": ps, "volume_quality": vq, "relative_strength": rs, "pct_change_at_10": pct_change_at_10, "price_at_10": price_at_10, "vwap_at_10": vwap_at_10, "atr_intraday": atr_intraday, "avg_dv": avg_dv, "bar_idx_10": bar_idx_10, }) candidates_scored.sort(key=lambda c: c["score"], reverse=True) result.candidates = candidates_scored # ── Step 4: Entry scan (10:00–15:30), up to max_positions ───────────────── open_positions: list[TGTCTrade] = [] closed_positions: list[TGTCTrade] = [] # Define stop-check bar times: every 5 minutes from 10:00 to 15:50 stop_bars_utc: list[dt.datetime] = [] cur = dt.datetime(date.year, date.month, date.day, 10, 0) + dt.timedelta(hours=_ET_OFFSET) eod_utc = dt.datetime(date.year, date.month, date.day, 15, 55) + dt.timedelta(hours=_ET_OFFSET) while cur <= eod_utc: stop_bars_utc.append(cur) cur += dt.timedelta(minutes=5) entry_cutoff_utc = _et_to_utc_naive(date, 15, 30) # D3: time-of-day entry filter — cap entry cutoff if no_entry_after_et is set if ent.no_entry_after_et: h, m = ent.no_entry_after_et.split(":") tod_cutoff = _et_to_utc_naive(date, int(h), int(m)) entry_cutoff_utc = min(entry_cutoff_utc, tod_cutoff) limit = ent.max_candidates_to_scan if ent.max_candidates_to_scan is not None else rsk.max_positions * 3 candidates_to_scan = candidates_scored[:limit] equity_snapshots: list[dict[str, Any]] = [] for bar_dt_utc in stop_bars_utc: is_eod = bar_dt_utc >= eod_utc # EOD: close all if is_eod: for pos in list(open_positions): sym = pos.symbol bars = bars_by_symbol.get(sym, []) eod_idx = _bar_idx_at(bars, bar_dt_utc) exit_price = float(bars[eod_idx]["close"]) if eod_idx >= 0 else pos.entry_price is_short_eod = pos.side == "short" if is_short_eod: risk_per_share = pos.stop_price - pos.entry_price pnl = (pos.entry_price - exit_price) * pos.shares else: risk_per_share = pos.entry_price - pos.stop_price pnl = (exit_price - pos.entry_price) * pos.shares pos.exit_price = exit_price pos.exit_bar_idx = eod_idx pos.exit_reason = "eod_exit" pos.pnl = round(pnl, 2) pos.r_multiple = round(pnl / (risk_per_share * pos.shares), 2) if risk_per_share * pos.shares > 0 else 0.0 pos.status = "closed" equity += pnl daily_loss = min(daily_loss, pnl) closed_positions.append(pos) open_positions = [] break # Entry: look for new setups if below max_positions if bar_dt_utc <= entry_cutoff_utc: # no_entry_before_et: skip this bar if it's before the floor entry_before_floor = False if ent.no_entry_before_et: h_b, m_b = ent.no_entry_before_et.split(":") before_floor_utc = _et_to_utc_naive(date, int(h_b), int(m_b)) if bar_dt_utc < before_floor_utc: entry_before_floor = True if not entry_before_floor: for cand in candidates_to_scan: if len(open_positions) >= rsk.max_positions: break sym = cand["symbol"] # Skip already in position if any(p.symbol == sym for p in open_positions + closed_positions): continue bars = bars_by_symbol.get(sym, []) if not bars: continue as_of_idx = _bar_idx_at(bars, bar_dt_utc) if as_of_idx < 3: continue if ent.type == "hod_breakout": setup_fn = detect_hod_breakout elif ent.type == "fade_short": setup_fn = detect_fade_short else: setup_fn = detect_vwap_pullback_reclaim setup = setup_fn( bars=bars, start_bar_idx=cand["bar_idx_10"], as_of_bar_idx=as_of_idx, params=ent, prev_close=prev_closes.get(sym, 0.0), atr_intraday=cand.get("atr_intraday"), ) if setup is None: continue entry_price = setup["entry_price"] stop_price = setup["stop_price"] trade_side = setup.get("side", "long") if trade_side == "short": risk_per_share = stop_price - entry_price # stop above entry for shorts else: risk_per_share = entry_price - stop_price if risk_per_share <= 0: continue # Size by risk_per_trade_pct risk_dollars = equity * (rsk.risk_per_trade_pct / 100.0) shares = max(1, int(risk_dollars / risk_per_share)) cost = entry_price * shares # Daily loss limit if -daily_loss >= equity * (rsk.daily_loss_limit_pct / 100.0): break trade = TGTCTrade( symbol=sym, entry_price=entry_price, stop_price=stop_price, shares=shares, entry_bar_idx=setup["setup_bar_idx"], peak_price=entry_price, current_stop=stop_price, entry_dt_utc=bar_dt_utc, side=trade_side, # Candidate metadata for segmentation analysis score=cand.get("score") or 0.0, rank_persistence=cand.get("rank_persistence") or 0.0, rank_velocity=cand.get("rank_velocity") or 0.0, price_structure=cand.get("price_structure") or 0.0, volume_quality=cand.get("volume_quality") or 0.0, relative_strength=cand.get("relative_strength") or 0.0, pct_change_at_10=cand.get("pct_change_at_10") or 0.0, dollar_volume_20d=cand.get("avg_dv") or 0.0, ) open_positions.append(trade) log.debug("TGTC %s: ENTER %s @ %.2f stop=%.2f shares=%d", date_str, sym, entry_price, stop_price, shares) # Stop/exit management for open positions for pos in list(open_positions): sym = pos.symbol bars = bars_by_symbol.get(sym, []) as_of_idx = _bar_idx_at(bars, bar_dt_utc) if as_of_idx < 0: continue current_bar = bars[as_of_idx] current_price = float(current_bar["close"]) current_high = float(current_bar["high"]) current_low = float(current_bar["low"]) is_short = pos.side == "short" # Peak tracking (long only — shorts don't use peak for partials/BE) if not is_short and current_price > pos.peak_price: pos.peak_price = current_price # risk_per_share is always positive (magnitude of distance to stop) if is_short: risk_per_share = pos.stop_price - pos.entry_price # stop > entry for shorts else: risk_per_share = pos.entry_price - pos.stop_price # ── Quick Take Profit (BEFORE stop check) ────────── if ex.take_profit_pct is not None: if is_short: tp_target = pos.entry_price * (1.0 - ex.take_profit_pct) tp_hit = current_low <= tp_target stop_touched = current_high >= pos.current_stop if tp_hit: pos.tp_collision = bool(stop_touched) if ex.tp_requires_no_stop_touch and stop_touched: pass # let stop handler decide else: exit_price = tp_target pnl = (pos.entry_price - exit_price) * pos.shares # short P&L pos.exit_price = exit_price pos.exit_bar_idx = as_of_idx pos.exit_reason = "quick_tp" pos.pnl = round(pnl, 2) pos.r_multiple = round(pnl / (risk_per_share * pos.shares), 2) if risk_per_share * pos.shares > 0 else 0.0 pos.status = "closed" equity += pnl daily_loss = min(daily_loss, pnl) open_positions.remove(pos) closed_positions.append(pos) log.debug("TGTC %s: QUICK_TP(short) %s @ %.2f pnl=%.2f", date_str, sym, exit_price, pnl) continue else: tp_target = pos.entry_price * (1.0 + ex.take_profit_pct) if current_high >= tp_target: # Always record collision status (used for artifact analysis) pos.tp_collision = bool(current_low <= pos.current_stop) # Conservative ordering: skip TP if stop also touched this bar if ex.tp_requires_no_stop_touch and pos.tp_collision: pass # let stop_loss handler decide (next block) else: exit_price = tp_target pnl = (exit_price - pos.entry_price) * pos.shares pos.exit_price = exit_price pos.exit_bar_idx = as_of_idx pos.exit_reason = "quick_tp" pos.pnl = round(pnl, 2) pos.r_multiple = round(pnl / (risk_per_share * pos.shares), 2) if risk_per_share * pos.shares > 0 else 0.0 pos.status = "closed" equity += pnl daily_loss = min(daily_loss, pnl) open_positions.remove(pos) closed_positions.append(pos) log.debug("TGTC %s: QUICK_TP %s @ %.2f pnl=%.2f collision=%s", date_str, sym, exit_price, pnl, pos.tp_collision) continue # ── Partial exit at 1R (long only — shorts skip entirely) ── if not is_short and not pos.partial_taken and risk_per_share > 0: if ex.partial_levels: # Multi-level partial scale-out (uses bar.high) for i, level in enumerate(ex.partial_levels): if i in pos.partial_levels_taken: continue r_mult = level.get("r_multiple", 1.0) frac = level.get("fraction", 0.33) target = pos.entry_price + r_mult * risk_per_share if current_high >= target: partial_shares = max(1, int(pos.shares * frac)) if partial_shares >= pos.shares: partial_shares = pos.shares - 1 # keep at least 1 share if partial_shares > 0: partial_pnl = (target - pos.entry_price) * partial_shares equity += partial_pnl pos.shares -= partial_shares pos.partial_levels_taken.add(i) if not pos.partial_taken and ex.stop_to_be_after_1r and r_mult >= 1.0: pos.current_stop = pos.entry_price pos.be_stop_active = True pos.partial_taken = True # flag so be-stop only fires once log.debug("TGTC %s: PARTIAL_LEVEL[%d] %s +%d shares pnl=%.2f", date_str, i, sym, partial_shares, partial_pnl) elif not ex.disable_partial_at_1r: target_1r = pos.entry_price + risk_per_share if current_price >= target_1r: partial_shares = max(1, int(pos.shares * ex.partial_at_1r)) partial_pnl = (current_price - pos.entry_price) * partial_shares equity += partial_pnl pos.shares -= partial_shares pos.partial_taken = True if ex.stop_to_be_after_1r: pos.current_stop = pos.entry_price pos.be_stop_active = True log.debug("TGTC %s: PARTIAL %s +%d shares pnl=%.2f", date_str, sym, partial_shares, partial_pnl) # ── Stop hit ────────────────────────────────────────────────────── if is_short: stop_hit = current_high >= pos.current_stop else: stop_hit = current_low <= pos.current_stop if stop_hit: mode = getattr(ex, "stop_exit_mode", "conservative") slippage_bps = getattr(ex, "stop_slippage_bps", 10.0) if is_short: if mode == "optimistic": exit_price = pos.current_stop elif mode == "moderate": exit_price = pos.current_stop * (1.0 + slippage_bps / 10000.0) # worse for short else: exit_price = max(pos.current_stop, float(current_bar["open"])) pnl = (pos.entry_price - exit_price) * pos.shares # short P&L else: if mode == "optimistic": exit_price = pos.current_stop elif mode == "moderate": exit_price = pos.current_stop * (1.0 - slippage_bps / 10000.0) else: exit_price = min(pos.current_stop, float(current_bar["open"])) pnl = (exit_price - pos.entry_price) * pos.shares pos.exit_price = exit_price pos.exit_bar_idx = as_of_idx pos.exit_reason = "stop_loss" pos.pnl = round(pnl, 2) pos.r_multiple = round(pnl / (risk_per_share * pos.shares), 2) if risk_per_share * pos.shares > 0 else 0.0 pos.status = "closed" equity += pnl daily_loss = min(daily_loss, pnl) open_positions.remove(pos) closed_positions.append(pos) continue # ── Time-Stop (after stop check, before trend health) ──────────── if ex.force_exit_after_minutes is not None and pos.entry_dt_utc is not None: elapsed_min = (bar_dt_utc - pos.entry_dt_utc).total_seconds() / 60.0 if elapsed_min >= ex.force_exit_after_minutes: exit_price = current_price if is_short: pnl = (pos.entry_price - exit_price) * pos.shares else: pnl = (exit_price - pos.entry_price) * pos.shares pos.exit_price = exit_price pos.exit_bar_idx = as_of_idx pos.exit_reason = "time_stop" pos.pnl = round(pnl, 2) pos.r_multiple = round(pnl / (risk_per_share * pos.shares), 2) if risk_per_share * pos.shares > 0 else 0.0 pos.status = "closed" equity += pnl daily_loss = min(daily_loss, pnl) open_positions.remove(pos) closed_positions.append(pos) log.debug("TGTC %s: TIME_STOP %s @ %.2f elapsed=%.1fm pnl=%.2f", date_str, sym, exit_price, elapsed_min, pnl) continue # ── Trend health exit (long only — skip for shorts) ────────────── if not is_short: trend_score = compute_trend_health(bars, as_of_idx) if trend_score <= 1: exit_price = current_price pnl = (exit_price - pos.entry_price) * pos.shares pos.exit_price = exit_price pos.exit_bar_idx = as_of_idx pos.exit_reason = "trend_health_exit" pos.pnl = round(pnl, 2) pos.r_multiple = round(pnl / (risk_per_share * pos.shares), 2) if risk_per_share * pos.shares > 0 else 0.0 pos.status = "closed" equity += pnl daily_loss = min(daily_loss, pnl) open_positions.remove(pos) closed_positions.append(pos) continue # Equity snapshot unrealized = sum( ( (float(bars_by_symbol[p.symbol][_bar_idx_at(bars_by_symbol[p.symbol], bar_dt_utc)]["close"]) - p.entry_price) * p.shares if p.side == "long" else (p.entry_price - float(bars_by_symbol[p.symbol][_bar_idx_at(bars_by_symbol[p.symbol], bar_dt_utc)]["close"])) * p.shares ) if bars_by_symbol.get(p.symbol) and _bar_idx_at(bars_by_symbol[p.symbol], bar_dt_utc) >= 0 else 0.0 for p in open_positions ) equity_snapshots.append({ "ts_et": (bar_dt_utc - dt.timedelta(hours=_ET_OFFSET)).strftime("%H:%M"), "equity": round(equity + unrealized, 2), }) result.trades = closed_positions + open_positions result.equity_curve = equity_snapshots result.final_equity = round(equity, 2) return result