"""TGTC live trading engine. Phase sequence (dry-run, no Alpaca orders in V1): 09:20 ET pre_screen – prior-day enrichment + universe filter 09:29:30 collect_snapshot – Yahoo fetch loop (every 20s until 10:00) 10:00 finalize_cands – rank features, score, hard filter → DB candidates 10:00+ entry_check – every 5m: VWAP pullback reclaim signal 10:05+ stop_check – every 5m: trend_health, stop, partial exit 15:55 eod_exit – close all virtual positions 16:00 post_close – daily snapshot """ from __future__ import annotations import asyncio import datetime as dt import logging import uuid from typing import Any from zoneinfo import ZoneInfo from apps.tgtc_trader.models import ( TGTCCandidateRow, TGTCDailySnapshotRow, TGTCPositionRow, TGTCTradeRow, ) from apps.tgtc_trader.state import TGTCStateManager from libs.tgtc.domain import TGTCConfig log = logging.getLogger(__name__) _TZ_ET = ZoneInfo("America/New_York") class TGTCEngine: """Per-session TGTC intraday engine (dry-run V1).""" def __init__( self, session_id: str, config: TGTCConfig, state: TGTCStateManager, ) -> None: self._session_id = session_id self._cfg = config self._state = state self._params = config.tgtc_strategy self._log = logging.getLogger(f"tgtc.engine.{session_id}") self._snapshots: list[dict] = [] # accumulated during collection self._candidates: list[dict] = [] self._enrichment: dict[str, dict] = {} self._bars_by_symbol: dict[str, list[dict]] = {} self._prev_closes: dict[str, float] = {} self._open_positions: dict[str, TGTCPositionRow] = {} # symbol → position # ── Pre-screen ──────────────────────────────────────────────────────────── def run_pre_screen(self, date_str: str) -> None: """09:20 ET: load universe, run prior-day enrichment.""" self._state.update_phase(self._session_id, date_str, "pre_screen") self._log.info("TGTC pre_screen %s", date_str) try: self._do_pre_screen(date_str) except Exception as exc: self._log.error("TGTC pre_screen failed: %s", exc, exc_info=True) def _do_pre_screen(self, date_str: str) -> None: from apps.orb_trader.screener import load_universe from libs.intraday.cache import IntradayCache from libs.tgtc.gainers_reconstruct import _bar_ts_naive_utc universe_source = getattr(self._cfg, "universe", "midlarge") tickers = load_universe(universe_source) self._log.info("TGTC universe: %d tickers", len(tickers)) # Load prev_close and enrichment from IntradayCache (prior trading days' 5m bars). # This avoids relying on Alpaca daily bars (which don't exist on AlpacaBroker) # and daily Parquet cache (which may be empty). intraday_cache = IntradayCache() date = dt.date.fromisoformat(date_str) # Collect up to 20 prior trading days' intraday data for ATR + prev_close prior_days: list[str] = [] for delta in range(1, 30): d = (date - dt.timedelta(days=delta)).isoformat() prior_days.append(d) if len(prior_days) >= 20: break # 16:00 ET cutoff in naive UTC = 20:00 UTC (EDT = UTC-4) close_et_utc_hour = 20 enrichment_built: dict[str, dict] = {} prev_closes_built: dict[str, float] = {} miss_count = 0 for sym in tickers: day_closes: list[float] = [] # daily close prices from intraday last bar day_ranges: list[float] = [] # daily H-L range for ATR for d in prior_days: try: bars = intraday_cache.get(sym, d) if not bars: continue # Find last bar at or before 16:00 ET close d_parts = [int(x) for x in d.split("-")] close_cutoff = dt.datetime(d_parts[0], d_parts[1], d_parts[2], close_et_utc_hour, 0) close_bar = None hi = lo = None for b in bars: b_ts = _bar_ts_naive_utc(b) if b_ts <= close_cutoff: close_bar = b h = float(b.get("high") or 0) l = float(b.get("low") or 0) hi = max(hi, h) if hi is not None else h lo = min(lo, l) if lo is not None else l if close_bar: c = float(close_bar.get("close") or 0) if c > 0: day_closes.append(c) if hi and lo and hi > lo: day_ranges.append(hi - lo) except Exception: continue if not day_closes: miss_count += 1 continue prev_close = day_closes[0] # most recent prior day if prev_close <= 0: miss_count += 1 continue prev_closes_built[sym] = prev_close atr14 = (sum(day_ranges[:14]) / len(day_ranges[:14])) if day_ranges else None last30_closes = day_closes[:30] avg_dv = None # cannot compute dollar vol from intraday last-bar alone enrichment_built[sym] = { "atr_14": atr14, "avg_dollar_vol_30d": avg_dv, "avg_dollar_vol_20d": avg_dv, } self._enrichment = enrichment_built self._prev_closes = prev_closes_built self._log.info( "TGTC pre_screen: %d symbols enriched, %d prev_closes loaded, %d missing", len(self._enrichment), len(self._prev_closes), miss_count, ) # ── Snapshot collection ──────────────────────────────────────────────────── async def run_collect_snapshot(self, date_str: str) -> None: """09:29:30–10:00 ET: fetch Yahoo snapshots every interval_seconds.""" self._state.update_phase(self._session_id, date_str, "collect_snapshot") col = self._params.collection interval = col.interval_seconds self._snapshots = [] end_et_parts = [int(p) for p in col.end_et.split(":")] date = dt.date.fromisoformat(date_str) end_et = dt.datetime(date.year, date.month, date.day, end_et_parts[0], end_et_parts[1], tzinfo=_TZ_ET) self._log.info("TGTC collect_snapshot starting until %s ET", col.end_et) tick_count = 0 while True: now_et = dt.datetime.now(tz=_TZ_ET) if now_et >= end_et: break captured_at = now_et.strftime("%Y-%m-%dT%H:%M:%S") try: from apps.tgtc_trader.yahoo_gainers import fetch_day_gainers from apps.tgtc_trader.snapshot_store import save_snapshot_parquet quotes = await fetch_day_gainers() if quotes: tick_count += 1 rows = [] from apps.tgtc_trader.models import TGTCSnapshotRow for q in quotes: snap_row = TGTCSnapshotRow( session_id=self._session_id, date=date_str, captured_at=captured_at, symbol=q.symbol, rank=q.rank, price=q.price, pct_change=q.pct_change, volume=q.volume, market_cap=q.market_cap, ) self._snapshots.append({ "captured_at": captured_at, "symbol": q.symbol, "rank": q.rank, "price": q.price, "pct_change": q.pct_change, "volume": q.volume, "market_cap": q.market_cap, }) rows.append(snap_row) self._state.save_snapshot_batch(rows) save_snapshot_parquet(quotes, date_str, captured_at) self._log.debug("TGTC tick %d: %d gainers @ %s", tick_count, len(quotes), captured_at) except Exception as exc: self._log.warning("TGTC snapshot fetch error: %s", exc) await asyncio.sleep(interval) self._log.info("TGTC collect_snapshot done: %d ticks", tick_count) # ── Finalize candidates ──────────────────────────────────────────────────── def run_finalize_candidates(self, date_str: str) -> None: """10:00 ET: compute rank features, scores, apply hard filters.""" self._state.update_phase(self._session_id, date_str, "finalize_candidates") self._log.info("TGTC finalize_candidates %s (%d snapshot rows)", date_str, len(self._snapshots)) try: self._do_finalize_candidates(date_str) except Exception as exc: self._log.error("TGTC finalize_candidates failed: %s", exc, exc_info=True) def _do_finalize_candidates(self, date_str: str) -> None: # V1 rule-based signals removed in TGTC V2 transition. # V2 candidate scoring uses ML selector (libs/tgtc/v2_features.py). raise NotImplementedError( "V1 finalize_candidates uses deleted signals. " "V2 engine to be implemented post selector kill-test (Gate 1)." ) from libs.tgtc.v2_features import compute_rank_features from libs.tgtc.gainers_reconstruct import _bar_ts_naive_utc import datetime as dt if not self._snapshots: self._log.warning("TGTC finalize_candidates: no snapshots collected") return rank_features = compute_rank_features(self._snapshots) # Fetch intraday bars for rank_features symbols candidates_raw = list(rank_features.keys()) if not candidates_raw: return # Fetch 5m intraday bars: try IntradayCache first, then live oracle for today date = dt.date.fromisoformat(date_str) cutoff_utc = dt.datetime(date.year, date.month, date.day, 14, 0) # 10:00 ET ≈ 14:00 UTC syms_to_fetch = list(set(candidates_raw) | {"QQQ"}) bars_raw: dict[str, list[dict]] = {} today_str = dt.date.today().isoformat() try: from libs.intraday.cache import IntradayCache intraday_cache = IntradayCache() for sym in syms_to_fetch: bars = intraday_cache.get(sym, date_str) if bars: bars_raw[sym] = bars self._log.info("TGTC finalize: %d/%d symbols from intraday cache", len(bars_raw), len(syms_to_fetch)) except Exception as exc: self._log.warning("TGTC finalize: intraday cache failed: %s", exc) # For today's live bars not yet in cache, fetch from oracle missing = [s for s in syms_to_fetch if s not in bars_raw] if missing and date_str == today_str: try: from libs.oracle_client.alpaca import get_multi_intraday_bars_today live = get_multi_intraday_bars_today(tickers=missing) bars_raw.update(live) self._log.info("TGTC finalize: %d/%d symbols from live oracle", len(live), len(missing)) except Exception as exc: self._log.warning("TGTC finalize: live oracle fetch failed: %s", exc) self._bars_by_symbol = bars_raw # QQQ pct change at 10:00 qqq_bars = self._bars_by_symbol.get("QQQ", []) qqq_pct_at_10 = None qqq_prev = self._prev_closes.get("QQQ") if qqq_bars and qqq_prev and qqq_prev > 0: bar10 = None for b in qqq_bars: b_ts = _bar_ts_naive_utc(b) if b_ts <= cutoff_utc: bar10 = b if bar10: qqq_pct_at_10 = (float(bar10["close"]) - qqq_prev) / qqq_prev flt = self._params.filters sw = self._params.score_weights now_str = dt.datetime.now(tz=_TZ_ET).isoformat() candidate_rows: list[TGTCCandidateRow] = [] self._candidates = [] for sym, rf in rank_features.items(): bars = self._bars_by_symbol.get(sym, []) if not bars: continue prev_close = self._prev_closes.get(sym, 0.0) if prev_close <= 0: continue # Find 10:00 ET bar bar10 = None bar10_idx = -1 for i, b in enumerate(bars): b_ts = _bar_ts_naive_utc(b) if b_ts <= cutoff_utc: bar10 = b bar10_idx = i if bar10 is None: continue price_at_10 = float(bar10["close"]) pct_at_10 = (price_at_10 - prev_close) / prev_close # Hard filters if price_at_10 < flt.min_price: continue if pct_at_10 < flt.min_day_change_at_10 or pct_at_10 > flt.max_day_change_at_10: continue enr = self._enrichment.get(sym, {}) avg_dv = enr.get("avg_dollar_vol_30d") or enr.get("avg_dollar_vol_20d") or 0.0 vwap10 = get_bar_vwap(bars, bar10_idx) above_vwap = bool(vwap10 and price_at_10 >= vwap10) if flt.must_be_above_vwap and not above_vwap: continue hod = max(float(b["high"]) for b in bars[:bar10_idx + 1]) if bar10_idx >= 0 else price_at_10 if hod > 0 and (hod - price_at_10) / hod > flt.max_pullback_from_hod: continue ps = compute_price_structure_score(bars, bar10_idx) vq = compute_volume_quality(bars, bar10_idx, avg_dv if avg_dv > 0 else None) rs = compute_relative_strength(pct_at_10, qqq_pct_at_10) 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, ) candidate_rows.append(TGTCCandidateRow( session_id=self._session_id, date=date_str, symbol=sym, score=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_at_10, price_at_10=price_at_10, vwap_at_10=vwap10, above_vwap=above_vwap, decided_at=now_str, status="pending", )) self._candidates.append({ "symbol": sym, "score": 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_at_10, "price_at_10": price_at_10, "vwap_at_10": vwap10, "above_vwap": above_vwap, "bar_idx_10": bar10_idx, "atr_intraday": enr.get("atr_14"), }) candidate_rows.sort(key=lambda r: r.score or 0.0, reverse=True) self._candidates.sort(key=lambda c: c["score"], reverse=True) if candidate_rows: self._state.save_candidates(candidate_rows) self._log.info("TGTC finalize: %d candidates saved", len(candidate_rows)) # ── Entry check ─────────────────────────────────────────────────────────── def run_entry_check(self, date_str: str) -> None: """10:00–15:30 ET every 5m: look for VWAP pullback reclaim entries.""" try: self._do_entry_check(date_str) except Exception as exc: self._log.error("TGTC entry_check failed: %s", exc, exc_info=True) def _do_entry_check(self, date_str: str) -> None: # V1 entry signals removed in TGTC V2 transition. raise NotImplementedError( "V1 entry_check uses deleted signals. " "V2 entry to be implemented post selector kill-test (Gate 1)." ) from libs.tgtc.gainers_reconstruct import _bar_ts_naive_utc rsk = self._params.risk ent = self._params.entry now_et = dt.datetime.now(tz=_TZ_ET) date = dt.date.fromisoformat(date_str) cutoff_utc_now = dt.datetime( date.year, date.month, date.day, now_et.hour, now_et.minute ) + dt.timedelta(hours=4) # rough ET→UTC if len(self._open_positions) >= rsk.max_positions: return # Refresh intraday bars for candidates candidate_syms = [c["symbol"] for c in self._candidates[:rsk.max_positions * 3] if c["symbol"] not in self._open_positions] if not candidate_syms: return today_str = dt.date.today().isoformat() try: from libs.intraday.cache import IntradayCache intraday_cache = IntradayCache() from_cache: list[str] = [] not_in_cache: list[str] = [] for sym in candidate_syms: bars = intraday_cache.get(sym, date_str) if bars: self._bars_by_symbol[sym] = bars from_cache.append(sym) else: not_in_cache.append(sym) # For today's live bars, fall back to oracle if not_in_cache and date_str == today_str: from libs.oracle_client.alpaca import get_multi_intraday_bars_today live = get_multi_intraday_bars_today(tickers=not_in_cache) self._bars_by_symbol.update(live) except Exception as exc: self._log.warning("TGTC entry_check: bar refresh failed: %s", exc) return equity = self._state.get_equity(self._session_id) or rsk.initial_equity # Build set of symbols already touched today (open OR closed) — entry-once guard. # Mirrors libs/tgtc/simulator.py:281 logic. Checks both tgtc_positions and # tgtc_trades so a stop-out in a prior cycle cannot be re-entered. traded_today: set[str] = set() for p in self._state.get_all_positions(self._session_id): if p.get("date") == date_str: traded_today.add(p["symbol"]) for t in self._state.get_trades(self._session_id): if t.get("date") == date_str: traded_today.add(t["symbol"]) for cand in self._candidates: if len(self._open_positions) >= rsk.max_positions: break sym = cand["symbol"] if sym in self._open_positions or sym in traded_today: continue bars = self._bars_by_symbol.get(sym, []) if not bars: continue # as_of bar index up to now as_of_idx = -1 for i, b in enumerate(bars): if _bar_ts_naive_utc(b) <= cutoff_utc_now: as_of_idx = i else: break if as_of_idx < cand["bar_idx_10"] + 2: continue setup = detect_vwap_pullback_reclaim( bars=bars, start_bar_idx=cand["bar_idx_10"], as_of_bar_idx=as_of_idx, params=ent, prev_close=self._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"] risk_per_share = entry_price - stop_price if risk_per_share <= 0: continue risk_dollars = equity * (rsk.risk_per_trade_pct / 100.0) shares = max(1, int(risk_dollars / risk_per_share)) pos = TGTCPositionRow( session_id=self._session_id, date=date_str, symbol=sym, entry_price=entry_price, stop_price=stop_price, current_stop=stop_price, shares=shares, entered_at=now_et.isoformat(), peak_price=entry_price, is_dry_run=True, ) self._state.save_position(pos) self._state.update_candidate_status(self._session_id, date_str, sym, "filled") self._open_positions[sym] = pos self._log.info("TGTC DRY-RUN ENTRY: %s @ %.2f stop=%.2f shares=%d", sym, entry_price, stop_price, shares) # ── Stop check ───────────────────────────────────────────────────────────── def run_stop_check(self, date_str: str) -> None: """Every 5m: check stops and trend health for open positions.""" try: self._do_stop_check(date_str) except Exception as exc: self._log.error("TGTC stop_check failed: %s", exc, exc_info=True) def _do_stop_check(self, date_str: str) -> None: # V1 trend health signals removed in TGTC V2 transition. raise NotImplementedError( "V1 stop_check uses deleted signals. " "V2 hazard-based exit to be implemented post exit kill-test (Gate 2)." ) from libs.tgtc.gainers_reconstruct import _bar_ts_naive_utc if not self._open_positions: return ex = self._params.exit equity = self._state.get_equity(self._session_id) or self._params.risk.initial_equity now_et = dt.datetime.now(tz=_TZ_ET) date = dt.date.fromisoformat(date_str) cutoff_utc_now = dt.datetime( date.year, date.month, date.day, now_et.hour, now_et.minute ) + dt.timedelta(hours=4) for sym in list(self._open_positions): pos = self._open_positions[sym] bars = self._bars_by_symbol.get(sym, []) if not bars: continue as_of_idx = -1 for i, b in enumerate(bars): if _bar_ts_naive_utc(b) <= cutoff_utc_now: as_of_idx = i else: break if as_of_idx < 0: continue bar = bars[as_of_idx] cur_price = float(bar["close"]) cur_low = float(bar["low"]) # Peak tracking if cur_price > pos.peak_price: pos.peak_price = cur_price self._state.save_position(pos) risk_per_share = pos.entry_price - pos.stop_price # Partial exit at 1R if not pos.partial_taken and risk_per_share > 0: target_1r = pos.entry_price + risk_per_share if cur_price >= target_1r: partial_shares = max(1, int(pos.shares * ex.partial_at_1r)) partial_pnl = (cur_price - pos.entry_price) * partial_shares 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 equity += partial_pnl self._state.save_position(pos) self._log.info("TGTC DRY-RUN PARTIAL: %s +%.2f", sym, partial_pnl) # Stop hit if cur_low <= pos.current_stop: exit_price = min(pos.current_stop, float(bar["open"])) self._close_position(pos, exit_price, "stop_loss", now_et.isoformat(), date_str) continue # Trend health exit trend = compute_trend_health(bars, as_of_idx) if trend <= 1: self._close_position(pos, cur_price, "trend_health_exit", now_et.isoformat(), date_str) continue def _close_position(self, pos: TGTCPositionRow, exit_price: float, reason: str, exit_time: str, date_str: str) -> None: """Mark position closed and write trade record.""" risk_per_share = pos.entry_price - pos.stop_price pnl = round((exit_price - pos.entry_price) * pos.shares, 2) r_mult = round(pnl / (risk_per_share * pos.shares), 2) if risk_per_share * pos.shares > 0 else 0.0 pos.exit_price = exit_price pos.exit_reason = reason pos.exited_at = exit_time pos.pnl = pnl pos.r_multiple = r_mult pos.status = "closed" self._state.save_position(pos) trade = TGTCTradeRow( trade_id=str(uuid.uuid4())[:8], session_id=pos.session_id, date=date_str, symbol=pos.symbol, entry_price=pos.entry_price, exit_price=exit_price, entered_at=pos.entered_at, exited_at=exit_time, shares=pos.shares, pnl=pnl, r_multiple=r_mult, exit_reason=reason, is_dry_run=pos.is_dry_run, ) self._state.save_trade(trade) self._open_positions.pop(pos.symbol, None) self._log.info("TGTC DRY-RUN EXIT: %s @ %.2f reason=%s pnl=%.2f", pos.symbol, exit_price, reason, pnl) # ── EOD exit ────────────────────────────────────────────────────────────── def run_eod_exit(self, date_str: str) -> None: """15:55 ET: close all remaining positions at last price.""" self._state.update_phase(self._session_id, date_str, "eod_exit") self._log.info("TGTC eod_exit: closing %d positions", len(self._open_positions)) now_iso = dt.datetime.now(tz=_TZ_ET).isoformat() from libs.tgtc.gainers_reconstruct import _bar_ts_naive_utc date = dt.date.fromisoformat(date_str) eod_utc = dt.datetime(date.year, date.month, date.day, 19, 55) # 15:55 ET ≈ 19:55 UTC for sym in list(self._open_positions): pos = self._open_positions[sym] bars = self._bars_by_symbol.get(sym, []) exit_price = pos.entry_price if bars: for b in reversed(bars): if _bar_ts_naive_utc(b) <= eod_utc: exit_price = float(b["close"]) break self._close_position(pos, exit_price, "eod_exit", now_iso, date_str) # ── Post close ──────────────────────────────────────────────────────────── def run_post_close(self, date_str: str) -> None: """16:00 ET: write daily snapshot.""" self._state.update_phase(self._session_id, date_str, "done") trades = self._state.get_trades(self._session_id) today_trades = [t for t in trades if t.get("date") == date_str] daily_pnl = sum(t["pnl"] for t in today_trades if t.get("pnl") is not None) all_snap = self._state.get_daily_snapshots(self._session_id, limit=1000) total_pnl = sum(s.get("daily_pnl", 0.0) or 0.0 for s in all_snap) + daily_pnl equity = (self._params.risk.initial_equity + total_pnl) snap = TGTCDailySnapshotRow( session_id=self._session_id, date=date_str, equity=round(equity, 2), daily_pnl=round(daily_pnl, 2), total_pnl=round(total_pnl, 2), trades_taken=len(today_trades), phase="done", ) self._state.save_daily_snapshot(snap) self._log.info("TGTC post_close %s: pnl=%.2f equity=%.2f trades=%d", date_str, daily_pnl, equity, len(today_trades)) # ── Helpers ─────────────────────────────────────────────────────────────────── def _bars_to_enrichment_fmt(bars: dict) -> dict[str, list[dict]]: """Convert AlpacaBroker.get_multi_daily_bars() Bar list to enrich_daily_bars() format.""" result: dict[str, list[dict]] = {} for sym, bar_list in bars.items(): result[sym] = [ { "date": b.date, "open": b.open, "high": b.high, "low": b.low, "close": b.close, "volume": b.volume, } for b in bar_list ] return result def make_tgtc_engine( session_id: str, config_path: str, db_path: str | None = None, ) -> TGTCEngine: """Factory: load config, create state manager, return engine.""" from libs.tgtc.domain import load_tgtc_config cfg = load_tgtc_config(config_path) state = TGTCStateManager(db_path) return TGTCEngine(session_id=session_id, config=cfg, state=state)