import logging import time from abc import ABC, abstractmethod from datetime import date, datetime, time as dtime, timedelta from typing import Optional import pandas as pd from ib_insync import IB, Contract, StopOrder, Trade from bars import BarManager from config import config from orders import has_open_order, trade_commission, wait_trade_done from state import PositionTracker logger = logging.getLogger(__name__) class BaseStrategy(ABC): def __init__(self, ib: IB, bar_manager: BarManager, tracker: PositionTracker): self.ib = ib self.bar_manager = bar_manager self.tracker = tracker self.name = self.__class__.__name__ self._stop_trades: dict[str, Trade] = {} self._stop_consumed: set = set() # uids of done stop orders already processed self._stop_cooldown: dict[str, datetime] = {} # key -> last stop-out time self._peak_high: dict[str, float] = {} # key -> highest high since entry (trailing) @abstractmethod async def on_tick(self): pass @abstractmethod async def on_bar(self): pass async def on_start(self): pass async def on_stop(self): pass def __repr__(self) -> str: return f"<{self.name}>" def _order_quantity(self, key: str, price: float) -> int: """Share count targeting the (per-symbol) dollar value per trade (min 1).""" value = self._trade_value_usd(key) if price and price > 0: return max(1, int(value // price)) return 1 def _trade_value_usd(self, key: str) -> float: """Per-trade dollar value for a symbol; global per-symbol override wins.""" override = config.symbol_trade_value_usd.get(key) return override if override else self.cfg.trade_value_usd def _can_open_position(self, key: str, order_value: float, price: float = 0.0) -> tuple[bool, str]: """Global entry gates shared by all strategies. Returns (allowed, reason_if_blocked): - sell_only mode -> no new positions at all - daily realized loss hit max_daily_loss -> no new positions today - concurrent lots hit max_positions -> no new positions until some close - symbol's combined position value would exceed max_symbol_value_usd - symbol sold by ANY strategy within sell_cooldown_minutes -> no re-entry - beyond the cooldown (within sell_improvement_window_minutes), a re-buy is only allowed if its price is >= sell_improvement_pct below the last sell price (prevents same-price flip-flops across strategies) """ rec = self.tracker.recent_sell(key) if rec is not None: age_min = (time.time() - rec["ts"]) / 60 if age_min < config.sell_cooldown_minutes: return False, (f"{key} sold {age_min:.0f} min ago by another strategy - " f"global sell cooldown {config.sell_cooldown_minutes} min") if price > 0 and age_min < config.sell_improvement_window_minutes: need = rec["price"] * (1 - config.sell_improvement_pct / 100) if price >= need: return False, (f"{key} re-buy price {price:.2f} not >= " f"{config.sell_improvement_pct:.1f}% below last sell " f"{rec['price']:.2f} (need < {need:.2f})") if config.sell_only: return False, "sell-only mode (liquidating)" day_pnl = self.tracker.day_realized_pnl() if day_pnl <= -config.max_daily_loss: return False, f"daily loss limit hit (realized {day_pnl:+.2f} <= -{config.max_daily_loss:.0f})" if self.tracker.total_positions() >= config.max_positions: return False, f"global cap of {config.max_positions} positions reached" symbol_value = self.tracker.symbol_value(key) if symbol_value + order_value > config.max_symbol_value_usd: return False, (f"{key} value cap: ${symbol_value + order_value:,.0f} would exceed " f"${config.max_symbol_value_usd:,.0f}") return True, "" # ---------- stop-out / order-failure cooldown (per strategy+symbol) ---------- @staticmethod def _next_day_start() -> datetime: """End-of-day boundary: next local midnight (rest of today = no re-entry).""" return datetime.combine(date.today() + timedelta(days=1), dtime.min) def _mark_cooldown(self, key: str, reason: str, until: Optional[datetime] = None): expiry = until or (datetime.now() + timedelta(minutes=self.cfg.stop_cooldown_minutes)) self._stop_cooldown[key] = expiry desc = "until EOD" if until else f"{self.cfg.stop_cooldown_minutes} min" logger.info("%s: %s cooldown for %s (%s, %s)", self.name, key, key, desc, reason) def _mark_stop_cooldown(self, key: str): self._mark_cooldown(key, "stop-out") def _mark_hard_stop_cooldown(self, key: str): """After a hard-stop fill, block re-entry on this symbol for the WHOLE day (prevents same-price re-buys after an overnight-gap stop-out).""" self._mark_cooldown(key, "hard stop-out", until=self._next_day_start()) def _in_stop_cooldown(self, key: str) -> bool: until = self._stop_cooldown.get(key) if until is None: return False if datetime.now() >= until: del self._stop_cooldown[key] return False return True # ---------- exchange-side hard stop (GTC STP order) management ---------- def _use_hard_stop(self) -> bool: return getattr(self.cfg, "use_hard_stop", False) def _stop_ref(self, key: str) -> str: return f"{self.name}:{key}:STP" def _has_active_stop(self, key: str) -> bool: t = self._stop_trades.get(key) return t is not None and not t.isDone() # ---------- trailing stop: raise the stop as price makes new highs ---------- def _use_trailing_stop(self) -> bool: return bool(getattr(self.cfg, "use_trailing_stop", False)) def _target_stop_price(self, key: str, entry_price: float, peak_high: float | None) -> float: """Desired STP price for a position. Fixed stop = entry - stop_loss_pct (always the floor). Once the peak high since entry is >= min_profit above entry, switch to a trailing stop = peak_high - trailing_stop_pct, but never below the fixed stop. """ fixed = round(entry_price * (1 - self.cfg.stop_loss_pct / 100), 2) if not self._use_trailing_stop() or not peak_high: return fixed peak_profit_pct = (peak_high - entry_price) / entry_price * 100 if peak_profit_pct < self.cfg.min_profit_pct: return fixed trailing = round(peak_high * (1 - self.cfg.trailing_stop_pct / 100), 2) return max(trailing, fixed) def _track_peak_high(self, key: str, high: float): """Remember the highest high seen for a symbol (call with each completed bar).""" current = self._peak_high.get(key) if current is None or high > current: self._peak_high[key] = high def _current_peak_high(self, key: str) -> float | None: return self._peak_high.get(key) @staticmethod def _trade_uid(trade: Trade): """Stable unique id for a trade (permId once assigned by IB).""" return trade.order.permId or trade.order.orderId or id(trade) async def _sync_stop_orders(self): """Reconcile in-memory/exchange stop orders with tracked positions. Runs at the top of every on_bar cycle: - STP filled -> record the sell in the tracker (position is gone) - STP dead (cancelled/inactive) while position owned -> re-place next cycle - STP open -> adopt it (e.g. after a bot restart); raise it to the trailing target if it has moved up - owned but no STP anywhere -> place a new one """ if not self._use_hard_stop(): return by_ref = {} for t in self.ib.trades(): ref = getattr(t.order, "orderRef", "") or "" if ref: by_ref[ref] = t for key, contract in self.contracts.items(): owned = self.tracker.get(self.name, key) trade = self._stop_trades.get(key) if trade is None: candidate = by_ref.get(self._stop_ref(key)) if candidate is not None: if not candidate.isDone(): trade = candidate # adopt open order (e.g. after restart) elif self._trade_uid(candidate) not in self._stop_consumed: trade = candidate # consume terminal state exactly once if trade is not None and trade.isDone(): self._stop_consumed.add(self._trade_uid(trade)) status = trade.orderStatus.status filled = trade.orderStatus.filled or 0.0 if status == "Filled" and filled > 0: if owned: logger.info( "%s HARD STOP filled: %s x%g @ %.2f", self.name, key, filled, trade.orderStatus.avgFillPrice, ) self.tracker.record_sell(self.name, key, filled, trade.orderStatus.avgFillPrice, reason="HardStop", commission=trade_commission(trade)) self._mark_hard_stop_cooldown(key) elif owned: logger.warning( "%s stop order for %s ended with status=%s - re-placing now", self.name, key, status, ) self._stop_trades.pop(key, None) if owned and not (status == "Filled" and filled > 0): # re-place immediately (same cycle), don't wait for the next one target = await self._current_stop_target(key, contract, owned) await self._place_stop(key, contract, owned["quantity"], target) continue if trade is not None: self._stop_trades[key] = trade # adopt existing open order if owned: await self._raise_stop_if_needed(key, contract, owned) continue if owned: target = await self._current_stop_target(key, contract, owned) await self._place_stop(key, contract, owned["quantity"], target) async def _current_stop_target(self, key: str, contract: Contract, owned: dict) -> float: """Desired STP price for the position right now (fixed or trailing).""" entry = owned["entry_price"] peak = await self._peak_high_since_entry(key, contract, owned) return self._target_stop_price(key, entry, peak) async def _peak_high_since_entry(self, key: str, contract: Contract, owned: dict) -> float | None: """Highest high of completed bars since the position was opened, if any.""" try: df = await self._get_df(contract) except Exception: return None if df is None or df.empty: return None cutoff = None entry_ts = owned.get("entry_ts") if entry_ts: try: cutoff = pd.Timestamp(entry_ts, tz="UTC") except Exception: cutoff = None if cutoff is None: ed = owned.get("entry_date") if ed: try: cutoff = pd.Timestamp(ed, tz="America/New_York").tz_convert("UTC") except Exception: cutoff = None if cutoff is None: return None since = df.loc[df["date"] >= cutoff, "high"] return float(since.max()) if not since.empty else None async def _raise_stop_if_needed(self, key: str, contract: Contract, owned: dict): """Cancel and re-place the stop if its price no longer matches the target. Handles both upward trailing moves and stale/zombie orders whose auxPrice differs from the desired stop (e.g. a PreSubmitted order that IB never accepted; cancel+re-place recovers it instead of silently keeping it). """ trade = self._stop_trades.get(key) if trade is None or trade.isDone(): return target = await self._current_stop_target(key, contract, owned) current = getattr(trade.order, "auxPrice", 0) or 0 if current and abs(target - current) < 0.01: return logger.info( "%s raising stop %s %.2f -> %.2f (trailing)", self.name, key, current, target, ) stop_filled, stop_price, confirmed = await self._cancel_stop(key) if stop_filled > 0: logger.info("%s: stop filled %g during trailing raise", key, stop_filled) self.tracker.record_sell(self.name, key, stop_filled, stop_price, reason="HardStopDuringTrail") remaining = owned["quantity"] - stop_filled if remaining <= 0: return if not confirmed: logger.warning( "%s: trailing raise aborted for %s - stop cancel not confirmed", self.name, key, ) return await self._place_stop(key, contract, remaining, target) async def _place_stop(self, key: str, contract: Contract, quantity: float, stop_price: float): """Place a GTC stop-loss sell order at the exchange.""" stop_price = round(stop_price, 2) ref = self._stop_ref(key) if has_open_order(self.ib, ref): _matching = [ f"{t.order.orderRef}|{t.orderStatus.status}|client{t.order.clientId}|done={t.isDone()}" for t in self.ib.openTrades() if t.order.orderRef == ref ] logger.info( "%s skipping stop place for %s: matching open order exists %s (debug: owned=%s stop_trades=%s)", self.name, key, _matching, self.tracker.get(self.name, key), self._stop_trades.get(key), ) return order = StopOrder("SELL", quantity, stop_price) if config.ib.account: order.account = config.ib.account order.orderRef = ref order.tif = "GTC" order.outsideRth = True # allow triggering in extended hours / overnight gaps try: trade = self.ib.placeOrder(contract, order) self._stop_trades[key] = trade logger.info( "%s hard stop placed: %s x%g STP @ %.2f (GTC, outsideRth)", self.name, key, quantity, stop_price, ) except Exception as e: logger.error("%s failed to place stop for %s: %s", self.name, key, e) async def _cancel_stop(self, key: str) -> tuple[float, float, bool]: """Cancel the stop order for key. Returns (quantity, avg_price) filled in the cancel race, and whether the cancellation was confirmed before giving up. If not confirmed, the caller must NOT proceed with a market sell - the STP may still be live and both orders could fill (double sell). Leave the position for next cycle. """ trade = self._stop_trades.pop(key, None) if trade is None: return 0.0, 0.0, True if not trade.isDone(): self.ib.cancelOrder(trade.order) if not await wait_trade_done(trade, 5.0): logger.warning( "%s: cancel of stop for %s not confirmed after 5s - " "assuming it may still be active", self.name, key, ) return trade.orderStatus.filled or 0.0, trade.orderStatus.avgFillPrice or 0.0, False return trade.orderStatus.filled or 0.0, trade.orderStatus.avgFillPrice or 0.0, True