356 lines
16 KiB
Python
356 lines
16 KiB
Python
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
|