diff --git a/apps/paper_trader/engine.py b/apps/paper_trader/engine.py index 77222ba..d9e3098 100644 --- a/apps/paper_trader/engine.py +++ b/apps/paper_trader/engine.py @@ -385,29 +385,53 @@ class PaperTradingEngine: ) return report - def _verify_order_fill(self, order_id: str, symbol: str, timeout_sec: float = 15.0) -> Order | None: - """Poll broker to verify order fill. Returns filled Order or None.""" + def _verify_order_fill(self, order_id: str, symbol: str, timeout_sec: float = 15.0) -> tuple[Order | None, str]: + """Poll broker to verify order fill. + + Returns (order, "") on success. + Returns (None, "alpaca_rejected:{status}:{order_id}") if Alpaca explicitly rejects. + Returns (None, "order_timeout:{order_id}") if fill not confirmed within timeout_sec. + + Callers must distinguish the two failure modes: + - alpaca_rejected → record permanently in processed_events (real problem) + - order_timeout → do NOT record (allow retry on next run_next_open) + """ deadline = time.monotonic() + timeout_sec while time.monotonic() < deadline: try: order = self._broker.get_order(order_id) if order.status == "filled" and order.filled_qty > 0: - return order + return order, "" if order.status in ("canceled", "expired", "rejected", "cancelled"): + reason = f"alpaca_rejected:{order.status}:{order_id}" logger.warning( "paper_engine_order_rejected", - symbol=symbol, order_id=order_id, status=order.status, + symbol=symbol, order_id=order_id, alpaca_status=order.status, + hint="Check Alpaca dashboard for rejection details", ) - return None + return None, reason except Exception: pass time.sleep(0.5) - # Timeout — market orders almost always fill instantly + # Timeout — likely submitted outside market hours or Alpaca latency + reason = f"order_timeout:{order_id}" logger.warning( "paper_engine_order_fill_timeout", symbol=symbol, order_id=order_id, timeout_sec=timeout_sec, + hint="Order may have been submitted outside market hours — will retry on next run", ) - return None + return None, reason + + @staticmethod + def _is_market_open() -> bool: + """Return True if US equity market is currently open (9:30–16:00 ET, weekdays).""" + import zoneinfo + now_et = dt.datetime.now(tz=zoneinfo.ZoneInfo("America/New_York")) + if now_et.weekday() >= 5: # Saturday=5, Sunday=6 + return False + market_open = now_et.replace(hour=9, minute=30, second=0, microsecond=0) + market_close = now_et.replace(hour=16, minute=0, second=0, microsecond=0) + return market_open <= now_et < market_close def _check_kill_switch(self, drawdown_pct: float, session_st: Any) -> bool: """Activate kill switch if drawdown exceeds threshold. Returns True if triggered.""" @@ -785,6 +809,19 @@ class PaperTradingEngine: continue # Submit market buy via Alpaca + if not self._is_market_open(): + logger.warning( + "paper_engine_market_closed_skip_entry", + symbol=candidate.symbol, + hint="Market is closed — skipping without recording so retry fires next run", + ) + rejected.append({ + "symbol": candidate.symbol, + "event_type": candidate.event_type, + "score": candidate.score, + "reason": "market_closed", + }) + continue try: order = self._broker.submit_market_buy(candidate.symbol, plan.shares) logger.info( @@ -812,17 +849,25 @@ class PaperTradingEngine: continue # Verify fill - verified = self._verify_order_fill(order.id, candidate.symbol) + verified, fill_fail_reason = self._verify_order_fill(order.id, candidate.symbol) if verified is None: - self._state.record_processed_event( - session_id, candidate.event_id, today.isoformat(), - "rejected", skip_reason="order_not_filled", - ) + if fill_fail_reason.startswith("alpaca_rejected:"): + # Alpaca explicitly rejected — record permanently + self._state.record_processed_event( + session_id, candidate.event_id, today.isoformat(), + "rejected", skip_reason=fill_fail_reason, + ) + else: + # Timeout — likely market closed or transient; do NOT record so retry works + logger.warning( + "paper_engine_fill_timeout_not_recorded", + symbol=candidate.symbol, order_id=order.id, + ) rejected.append({ "symbol": candidate.symbol, "event_type": candidate.event_type, "score": candidate.score, - "reason": "order_not_filled", + "reason": fill_fail_reason, }) continue fill_price = verified.filled_avg_price or plan.entry_price_limit @@ -955,6 +1000,19 @@ class PaperTradingEngine: }) continue + if not self._is_market_open(): + logger.warning( + "paper_engine_market_closed_skip_entry", + symbol=candidate.symbol, + hint="Market is closed — skipping without recording so retry fires next run", + ) + rejected.append({ + "symbol": candidate.symbol, + "event_type": candidate.event_type, + "score": candidate.score, + "reason": "market_closed", + }) + continue try: order = self._broker.submit_market_buy(candidate.symbol, plan.shares) except Exception as exc: @@ -970,17 +1028,23 @@ class PaperTradingEngine: }) continue - verified = self._verify_order_fill(order.id, candidate.symbol) + verified, fill_fail_reason = self._verify_order_fill(order.id, candidate.symbol) if verified is None: - self._state.record_processed_event( - session_id, candidate.event_id, today.isoformat(), - "rejected", skip_reason="order_not_filled", - ) + if fill_fail_reason.startswith("alpaca_rejected:"): + self._state.record_processed_event( + session_id, candidate.event_id, today.isoformat(), + "rejected", skip_reason=fill_fail_reason, + ) + else: + logger.warning( + "paper_engine_fill_timeout_not_recorded", + symbol=candidate.symbol, order_id=order.id, + ) rejected.append({ "symbol": candidate.symbol, "event_type": candidate.event_type, "score": candidate.score, - "reason": "order_not_filled", + "reason": fill_fail_reason, }) continue fill_price = verified.filled_avg_price or plan.entry_price_limit @@ -1512,7 +1576,7 @@ class PaperTradingEngine: shares_to_sell = min(qty, max(1, math.ceil(needed / cur_price))) try: self._broker.close_position(sym, qty=shares_to_sell) - time.sleep(1) + time.sleep(3) new_qty = qty - shares_to_sell if new_qty <= 0: self._state.close_parking_state(session_id) @@ -1920,15 +1984,6 @@ class PaperTradingEngine: except Exception as exc: logger.warning("paper_engine_macro_long_candidate_invalid", error=str(exc)) - # MOMENTUM BREAKOUT: screen for price/volume breakouts without events. - # Only fires when momentum_breakout.enabled=True (default: off). - # Requires walk-forward validation before enabling. - if self._config.momentum_breakout.enabled: - momentum_cands = await self._generate_momentum_breakout_candidates( - today, open_symbols_now - ) - add_on_candidates.extend(momentum_cands) - entries, rejected = await self._process_entries( today, next_open_rows, account, alpaca_positions, strategy_states, session_st, macro_data, self._broker.submit_market_buy, @@ -2446,6 +2501,7 @@ class PaperTradingEngine: macro_data=macro_data, engine_daily_new_risk_used=engine_risk_used if engine_cfg else 0.0, ) + freed = False if plan.skip_reason == "insufficient_cash": # 1) Attempt to free parking cash before giving up needed = plan.shares * float(candidate.entry_price_est) if plan.shares else float(candidate.entry_price_est) @@ -2481,6 +2537,16 @@ class PaperTradingEngine: ) if plan.skip_reason: + # If we just freed parking cash but plan still shows insufficient_cash, + # Alpaca may not have settled the SGOV sell yet → skip without recording + # so the event retries on the next run_next_open. + if freed and plan.skip_reason == "insufficient_cash": + logger.warning( + "paper_engine_parking_freed_cash_not_settled", + symbol=candidate.symbol, + hint="SGOV sold but cash not yet reflected in account — will retry on next run", + ) + continue self._state.record_processed_event( session_id, candidate.event_id, today.isoformat(), "rejected", skip_reason=plan.skip_reason, @@ -2508,6 +2574,17 @@ class PaperTradingEngine: "score": candidate.score, "reason": gap_reason, }) continue + # Market-hours guard: skip (without recording) if market is closed. + # Allows retry on next run_next_open when market is open. + is_moc = (order_fn != self._broker.submit_market_buy) + if not is_moc and not self._is_market_open(): + logger.warning( + "paper_engine_market_closed_skip_entry", + symbol=candidate.symbol, + hint="Market is closed — skipping without recording so retry fires next run", + ) + rejected.append({"symbol": candidate.symbol, "event_type": candidate.event_type, "score": candidate.score, "reason": "market_closed"}) + continue try: order = order_fn(candidate.symbol, plan.shares) logger.info("paper_engine_buy_submitted", symbol=candidate.symbol, qty=plan.shares, order_id=order.id) @@ -2521,15 +2598,22 @@ class PaperTradingEngine: continue # Verify fill (skip for MOC orders — they fill at close) - is_moc = (order_fn != self._broker.submit_market_buy) if not is_moc: - verified = self._verify_order_fill(order.id, candidate.symbol) + verified, fill_fail_reason = self._verify_order_fill(order.id, candidate.symbol) if verified is None: - self._state.record_processed_event( - session_id, candidate.event_id, today.isoformat(), - "rejected", skip_reason="order_not_filled", - ) - rejected.append({"symbol": candidate.symbol, "event_type": candidate.event_type, "score": candidate.score, "reason": "order_not_filled"}) + if fill_fail_reason.startswith("alpaca_rejected:"): + # Alpaca explicitly rejected — record permanently + self._state.record_processed_event( + session_id, candidate.event_id, today.isoformat(), + "rejected", skip_reason=fill_fail_reason, + ) + else: + # Timeout — do NOT record so retry works next run + logger.warning( + "paper_engine_fill_timeout_not_recorded", + symbol=candidate.symbol, order_id=order.id, + ) + rejected.append({"symbol": candidate.symbol, "event_type": candidate.event_type, "score": candidate.score, "reason": fill_fail_reason}) continue fill_price = verified.filled_avg_price or plan.entry_price_limit else: @@ -3047,102 +3131,6 @@ class PaperTradingEngine: # Momentum Breakout Sleeve (Phase 4) # ------------------------------------------------------------------ # - async def _generate_momentum_breakout_candidates( - self, - today: dt.date, - open_symbols: set[str], - ) -> list[Candidate]: - """Screen universe for momentum breakout signals (no event required). - - Uses live Alpaca bar data. Only fires if momentum_breakout.enabled=True. - Returns synthetic Candidates for add_on_candidates in _process_entries. - - WARNING: Validate with walk-forward OOT backtest before using in live trading. - """ - cfg = self._config.momentum_breakout - if not cfg.enabled: - return [] - - from libs.backtest.momentum_screener import MomentumBreakoutScreener - - # Universe: collect symbols from snapshot store or skip if not available - universe_symbols: list[str] = [] - if self._snapshot_store is not None: - # Use the snapshot universe if available - try: - universe_symbols = list(self._snapshot_store.get_universe_symbols()) - except Exception: - pass - if not universe_symbols: - logger.debug("paper_engine_momentum_no_universe") - return [] - - # Fetch bar data for all universe symbols (need 25+ days) - bars_by_sym = await self._fetch_etf_bars(universe_symbols, today, lookback_days=35) - spy_bars = bars_by_sym.get("SPY", {}) - if not spy_bars: - spy_bars_fetched = await self._fetch_etf_bars(["SPY"], today, lookback_days=35) - spy_bars = spy_bars_fetched.get("SPY", {}) - - screener = MomentumBreakoutScreener() - signals = screener.screen( - date=today, - universe_symbols=universe_symbols, - bars_by_symbol=bars_by_sym, - spy_bars=spy_bars, - open_symbols=open_symbols, - min_gap_up_pct=cfg.min_gap_up_pct, - min_volume_ratio=cfg.min_volume_ratio, - min_close_location=cfg.min_close_location, - min_relative_strength_20d=cfg.min_relative_strength_20d, - min_avg_dollar_volume=cfg.min_avg_dollar_volume, - max_results=cfg.max_new_per_day, - ) - - if not signals: - return [] - - candidates: list[Candidate] = [] - for sig in signals: - # Use a synthetic engine_id for position management - cand = Candidate( - event_id=f"synth_momentum_{sig.symbol.lower()}_{today.isoformat()}", - symbol=sig.symbol, - score=sig.score, - sector="MOMENTUM", - event_type="momentum_breakout", - event_timestamp=dt.datetime.combine(today, dt.time(16, 0), tzinfo=dt.timezone.utc), - event_date=today, - filing_time_bucket="after_close", - reaction_date=today, - execution_date=today, - entry_price_est=sig.close_price, - avg_dollar_volume=sig.avg_dollar_volume_20d, - atr_14=sig.atr_14, - score_bucket="medium", - engine_id="idle_momentum_breakout", - entry_timing_policy="next_open", - trade_direction="long", - engine_max_holding_days=cfg.hold_days, - engine_early_failure_no_progress_days=999, - features={ - "momentum_gap_up_pct": sig.gap_up_pct, - "momentum_volume_ratio_20d": sig.volume_ratio_20d, - "momentum_relative_strength_20d": sig.relative_strength_20d, - "momentum_close_location": sig.close_location, - }, - ) - candidates.append(cand) - logger.info( - "paper_engine_momentum_breakout_triggered", - date=today.isoformat(), - symbol=sig.symbol, - score=round(sig.score, 3), - gap_up_pct=round(sig.gap_up_pct * 100, 2), - volume_ratio=round(sig.volume_ratio_20d, 2), - ) - return candidates - # ------------------------------------------------------------------ # # Rotation & Recycle (Phase 1) # ------------------------------------------------------------------ #