"""Lightweight IB client wrapper using ibapi. This wraps a minimal subset of EClient/EWrapper to subscribe to market data and track positions in-memory. It's designed to be used by a small Flask backend that serves a browser dashboard. Notes: - Keep behavior simple: subscribe to market data for provided symbols. - Store latest quotes in `self.quotes` and positions in `self.positions`. """ import logging import threading import time from typing import Dict, List from ibapi.client import EClient from ibapi.contract import Contract from ibapi.wrapper import EWrapper class IBClient(EWrapper, EClient): def __init__(self, host: str = "127.0.0.1", port: int = 7496, client_id: int = 1): EClient.__init__(self, self) # normalize/validate connection params if not host: host = "127.0.0.1" if not isinstance(port, int): try: port = int(port) except Exception: port = 7496 self.host = host self.port = port self.client_id = client_id self._thread = None self._next_req_id = 1 # connection sync self._connected_event = threading.Event() # subscriptions requested before connection self._pending_subscriptions: Dict[str, Dict] = {} # in-memory stores self.quotes: Dict[str, Dict] = {} self.positions: List[Dict] = [] # store executions (append as execDetails callbacks arrive) self.executions: List[Dict] = [] self._execution_ids = set() self.historical_closes: Dict[str, float] = {} # tracking for subscriptions: reqId -> symbol self._req_map: Dict[int, str] = {} self._historical_req_map: Dict[int, str] = {} self._requested_symbols = set() self._requested_accounts = set() self._requested_historical = set() # basic logging logging.basicConfig(level=logging.INFO) def _normalize_expiry(self, expiry): expiry = str(expiry or "").strip() if len(expiry) == 8 and expiry.isdigit(): return f"{expiry[0:4]}-{expiry[4:6]}-{expiry[6:8]}" if len(expiry) == 6 and expiry.isdigit(): return f"{expiry[0:4]}-{expiry[4:6]}" return expiry def _contract_key(self, contract) -> str: if not contract: return "" con_id = getattr(contract, "conId", None) if con_id: return f"CONID:{con_id}" sec_type = (getattr(contract, "secType", None) or "").upper() symbol = getattr(contract, "symbol", None) or "" if sec_type == "OPT": expiry = self._normalize_expiry(getattr(contract, "lastTradeDateOrContractMonth", None)) strike = getattr(contract, "strike", None) right = getattr(contract, "right", None) or "" multiplier = getattr(contract, "multiplier", None) or "" return f"OPT:{symbol}:{expiry}:{strike}:{right}:{multiplier}" return f"{sec_type or 'STK'}:{symbol}" def _contract_to_record(self, contract) -> Dict: if not contract: return {} sec_type = (getattr(contract, "secType", None) or "STK").upper() symbol = getattr(contract, "symbol", None) or "" if sec_type == "OPT": expiry = self._normalize_expiry(getattr(contract, "lastTradeDateOrContractMonth", None)) strike = getattr(contract, "strike", None) right = (getattr(contract, "right", None) or "").upper() multiplier = getattr(contract, "multiplier", None) else: expiry = None strike = None right = None multiplier = None local_symbol = getattr(contract, "localSymbol", None) rec = { "instrument_key": self._contract_key(contract), "con_id": getattr(contract, "conId", None) or None, "symbol": symbol, "security_type": sec_type, "underlying_symbol": symbol, "expiry": expiry or None, "strike": float(strike) if strike not in (None, "") else None, "option_type": right or None, "multiplier": str(multiplier) if multiplier not in (None, "") else None, "local_symbol": local_symbol or None, "exchange": getattr(contract, "exchange", None) or None, "primary_exchange": getattr(contract, "primaryExchange", None) or None, "currency": getattr(contract, "currency", None) or None, "trading_class": getattr(contract, "tradingClass", None) or None, } rec["display_symbol"] = self._format_display_symbol(rec) return rec def _format_display_symbol(self, rec: Dict) -> str: sec_type = (rec.get("security_type") or "STK").upper() symbol = rec.get("underlying_symbol") or rec.get("symbol") or "" if sec_type == "OPT": expiry = rec.get("expiry") or "?" strike = rec.get("strike") if isinstance(strike, float) and strike.is_integer(): strike_str = str(int(strike)) else: strike_str = str(strike) if strike is not None else "?" option_type = rec.get("option_type") or "?" return f"{symbol} {expiry} {strike_str} {option_type}" return symbol def _price_multiplier(self, rec: Dict) -> float: sec_type = (rec.get("security_type") or "STK").upper() if sec_type != "OPT": return 1.0 multiplier = rec.get("multiplier") try: return float(multiplier) if multiplier not in (None, "", 0, "0") else 100.0 except Exception: return 100.0 def _make_contract_from_record(self, rec: Dict) -> Contract: sec_type = (rec.get("security_type") or "STK").upper() con_id = rec.get("con_id") if con_id: c = Contract() c.conId = int(con_id) c.secType = sec_type c.exchange = rec.get("exchange") or "SMART" currency = rec.get("currency") if currency: c.currency = currency return c if sec_type == "OPT": c = Contract() c.symbol = rec.get("underlying_symbol") or rec.get("symbol") c.secType = "OPT" c.exchange = rec.get("exchange") or "SMART" c.currency = rec.get("currency") or "USD" expiry = rec.get("expiry") or "" c.lastTradeDateOrContractMonth = expiry.replace("-", "") strike = rec.get("strike") if strike is not None: c.strike = float(strike) option_type = rec.get("option_type") if option_type: c.right = option_type multiplier = rec.get("multiplier") if multiplier: c.multiplier = str(multiplier) local_symbol = rec.get("local_symbol") if local_symbol: c.localSymbol = local_symbol primary_exchange = rec.get("primary_exchange") if primary_exchange: c.primaryExchange = primary_exchange trading_class = rec.get("trading_class") if trading_class: c.tradingClass = trading_class return c return self._make_stock_contract( rec.get("symbol") or rec.get("underlying_symbol"), exchange=rec.get("exchange") or "SMART", currency=rec.get("currency") or "USD", ) def connect_and_start(self) -> None: if not self.isConnected(): logging.info("Connecting to IB gateway %s:%s", self.host, self.port) self.connect(self.host, self.port, self.client_id) self._thread = threading.Thread(target=self.run, daemon=True) self._thread.start() # don't block here; nextValidId will mark connection ready def disconnect_and_stop(self) -> None: try: if self.isConnected(): self.disconnect() except Exception: pass # ---- EWrapper overrides (minimal) ---- def nextValidId(self, orderId: int): super().nextValidId(orderId) # ensure we start with a reasonably large req id to avoid collisions if orderId and orderId > self._next_req_id: self._next_req_id = orderId # mark connection as ready try: self._connected_event.set() except Exception: pass # process any pending subscriptions if self._pending_subscriptions: pending = list(self._pending_subscriptions.values()) self._pending_subscriptions.clear() for s in pending: try: self.subscribe_market_data(s) except Exception: logging.exception("Failed to process pending subscription %s", s) def error(self, reqId, errorCode, errorString): logging.error("IB error (req=%s code=%s): %s", reqId, errorCode, errorString) def tickPrice(self, reqId, tickType, price, attrib): # store last or closing price when available if price is None or price == -1: return instrument_key = self._req_map.get(reqId) if not instrument_key: return data = self.quotes.setdefault(instrument_key, {}) if tickType in (4, 68): data["last"] = price elif tickType in (9, 75): data["close"] = price elif tickType in (1, 66): data["bid"] = price elif tickType in (2, 67): data["ask"] = price elif tickType == 37: data["mark"] = price def tickSize(self, reqId, tickType, size): instrument_key = self._req_map.get(reqId) if not instrument_key: return data = self.quotes.setdefault(instrument_key, {}) data["size"] = size def updatePortfolio(self, contract, position, marketPrice, marketValue, averageCost, unrealizedPNL, realizedPNL, accountName): # simple representation of the position if not contract or not contract.symbol: return contract_rec = self._contract_to_record(contract) pos = { **contract_rec, "position": position, "marketPrice": marketPrice, "marketValue": marketValue, "averageCost": averageCost, "account": accountName, } # replace existing entry for the symbol for i, p in enumerate(self.positions): if p.get("instrument_key") == pos["instrument_key"] and p.get("account") == accountName: self.positions[i] = pos break else: self.positions.append(pos) def positionEnd(self): # called when a batch of position updates is finished logging.info("Position update batch complete. %d positions stored", len(self.positions)) def position(self, account, contract, position, avgCost): """Handle positions returned by `reqPositions()`. Signature from EWrapper: position(self, account, contract, position, avgCost) """ try: if not contract or not getattr(contract, 'symbol', None): return contract_rec = self._contract_to_record(contract) pos = { **contract_rec, "position": position, "averageCost": avgCost, "account": account, } # replace existing entry for the symbol for i, p in enumerate(self.positions): if p.get("instrument_key") == pos["instrument_key"] and p.get("account") == account: self.positions[i] = pos break else: self.positions.append(pos) except Exception: logging.exception("Error handling position callback") def execDetails(self, reqId, contract, execution): """Capture execution details for building lots. `execution` is an `Execution` object from ibapi which contains .shares, .price, .side, .execId, .time (may be empty) """ try: symbol = getattr(contract, "symbol", None) if not symbol: return contract_rec = self._contract_to_record(contract) entry = { "execId": getattr(execution, "execId", None), **contract_rec, "shares": float(getattr(execution, "shares", 0)), "price": float(getattr(execution, "price", 0.0)), "side": getattr(execution, "side", ""), "time": getattr(execution, "time", None), "permId": getattr(execution, "permId", None), } exec_id = entry.get("execId") if exec_id and exec_id in self._execution_ids: return if exec_id: self._execution_ids.add(exec_id) self.executions.append(entry) except Exception: logging.exception("Failed to record execution detail") def execDetailsEnd(self, reqId): logging.info("Finished receiving executions (req=%s). %d execs total", reqId, len(self.executions)) # ---- helpers ---- def _new_req_id(self) -> int: rid = self._next_req_id self._next_req_id += 1 return rid def _make_stock_contract(self, symbol: str, exchange: str = "SMART", currency: str = "USD") -> Contract: c = Contract() c.symbol = symbol c.secType = "STK" c.exchange = exchange c.currency = currency return c def subscribe_market_data(self, contract_or_record) -> None: """Subscribe to basic market data for a contract or simple symbol string. Repeated calls for the same symbol are ignored. """ if isinstance(contract_or_record, str): contract = self._make_stock_contract(contract_or_record) queued_record = {"symbol": contract_or_record, "security_type": "STK"} instrument_key = self._contract_key(contract) else: contract = self._make_contract_from_record(contract_or_record) queued_record = dict(contract_or_record) instrument_key = self._contract_key(contract) if instrument_key in self._requested_symbols: return # if not connected yet, queue the subscription if not self._connected_event.is_set(): logging.info("Not connected yet; queuing subscription for %s", instrument_key) self._pending_subscriptions[instrument_key] = queued_record # ensure we start connection try: self.connect_and_start() except Exception: logging.exception("Failed to start connection while queuing subscription") return req_id = self._new_req_id() self._req_map[req_id] = instrument_key # prefer delayed data when live market data is not available try: self.reqMarketDataType(3) self.reqMktData(req_id, contract, "", False, False, []) self._requested_symbols.add(instrument_key) logging.info("Subscribed to market data for %s (req=%s)", instrument_key, req_id) except Exception as e: logging.exception("Failed to subscribe to %s: %s", instrument_key, e) def request_positions(self) -> None: """Ask the IB API to send current positions (triggers updatePortfolio callbacks).""" if not self._connected_event.is_set(): logging.info("Not connected yet; will request positions only if connection parameters are valid") # validate host/port before trying to connect if not isinstance(self.host, str) or not isinstance(self.port, int): logging.error("Invalid connection parameters: host=%r port=%r", self.host, self.port) return try: self.connect_and_start() except Exception: logging.exception("Failed to start connection for positions request") # schedule a delayed reqPositions after connection is established def _delayed(): if self._connected_event.wait(5): try: self.reqPositions() except Exception: logging.exception("Failed to request positions after connect") threading.Thread(target=_delayed, daemon=True).start() return try: self.reqPositions() except Exception: logging.exception("Failed to request positions") def request_account_updates(self, account: str) -> None: """Subscribe to portfolio/account updates for a specific account.""" if not account or account in self._requested_accounts: return if not self._connected_event.is_set(): try: self.connect_and_start() except Exception: logging.exception("Failed to start connection for account updates") return try: self.reqAccountUpdates(True, account) self._requested_accounts.add(account) except Exception: logging.exception("Failed to request account updates for %s", account) def request_executions(self, start_time: str | None = None) -> None: """Request executions from IB (will trigger execDetails callbacks). Uses an empty ExecutionFilter to fetch recent executions. The results arrive via `execDetails` callbacks. """ try: # import here to avoid module-level dependency issues from ibapi.execution import ExecutionFilter filt = ExecutionFilter() if start_time: filt.time = start_time rid = self._new_req_id() self.reqExecutions(rid, filt) except Exception: logging.exception("Failed to request executions") def historicalData(self, reqId, bar): try: instrument_key = self._historical_req_map.get(reqId) if not instrument_key: return close = getattr(bar, "close", None) if close not in (None, -1): self.historical_closes[instrument_key] = float(close) except Exception: logging.exception("Failed to process historical data") def historicalDataEnd(self, reqId, start, end): self._historical_req_map.pop(reqId, None) def request_historical_close(self, contract_or_record) -> None: if isinstance(contract_or_record, str): contract = self._make_stock_contract(contract_or_record) else: contract = self._make_contract_from_record(contract_or_record) instrument_key = self._contract_key(contract) if not instrument_key or instrument_key in self._requested_historical: return if not self._connected_event.is_set(): try: self.connect_and_start() except Exception: logging.exception("Failed to start connection for historical close") return try: req_id = self._new_req_id() self._historical_req_map[req_id] = instrument_key self._requested_historical.add(instrument_key) self.reqHistoricalData(req_id, contract, "", "2 D", "1 day", "TRADES", 0, 1, False, []) except Exception: logging.exception("Failed to request historical close for %s", instrument_key) def get_quotes(self): return dict(self.quotes) def get_positions(self): return list(self.positions) def compute_open_lots(self, include_position_fallback: bool = False): """Compute open lots using FIFO matching from recorded executions. Returns a list of lots with: symbol, qty, cost_basis, avg_price, market_price, market_value, gain_dollars, gain_percent, cagr_percent, buy_time """ # group executions by symbol from collections import defaultdict from datetime import datetime by_symbol = defaultdict(list) for e in self.executions: by_symbol[e["instrument_key"]].append(e) # include symbols from positions as well so we can fallback position_symbols = set() position_by_symbol = {} for p in self.positions: instrument_key = p.get("instrument_key") if not instrument_key: continue position_symbols.add(instrument_key) if instrument_key not in position_by_symbol: position_by_symbol[instrument_key] = p lots = [] all_symbols = set(by_symbol.keys()) | position_symbols for instrument_key in all_symbols: execs = by_symbol.get(instrument_key, []) position = position_by_symbol.get(instrument_key) or {} meta = execs[0] if execs else position # sort by time if available, else keep order def _time_key(x): t = x.get("time") if not t: return 0 # try parsing common formats for fmt in ("%Y%m%d %H:%M:%S", "%Y%m%d %H:%M:%S", "%Y%m%d-%H:%M:%S"): try: return datetime.strptime(t, fmt).timestamp() except Exception: continue try: return float(t) except Exception: return 0 execs_sorted = sorted(execs, key=_time_key) if execs else [] # build open lots using FIFO for both long and short positions open_longs = [] open_shorts = [] for e in execs_sorted: side = (e.get("side") or "").upper() shares = abs(float(e.get("shares", 0) or 0)) price = float(e.get("price", 0.0) or 0.0) time_str = e.get("time") if side == "BOT" or side == "BUY": qty_to_buy = shares while qty_to_buy > 0 and open_shorts: s = open_shorts[0] matched = min(qty_to_buy, abs(s["qty"])) s["qty"] += matched qty_to_buy -= matched if abs(s["qty"]) < 1e-9: open_shorts.pop(0) if qty_to_buy > 0: open_longs.append({"qty": qty_to_buy, "price": price, "time": time_str}) else: qty_to_sell = shares while qty_to_sell > 0 and open_longs: b = open_longs[0] matched = min(qty_to_sell, b["qty"]) b["qty"] -= matched qty_to_sell -= matched if b["qty"] < 1e-9: open_longs.pop(0) if qty_to_sell > 0: open_shorts.append({"qty": -qty_to_sell, "price": price, "time": time_str}) # remaining buys are open lots # ensure market data available for symbol if instrument_key not in self.quotes: try: self.subscribe_market_data(meta) except Exception: pass market_price = None q = self.quotes.get(instrument_key) if q: market_price = q.get("last") if market_price is None: market_price = q.get("close") if market_price is None: market_price = q.get("mark") if market_price is None: market_price = self.historical_closes.get(instrument_key) if market_price is None: market_price = position.get("marketPrice") try: position_qty = float(position.get("position") or 0.0) except Exception: position_qty = 0.0 try: position_market_value = float(position.get("marketValue")) if position.get("marketValue") is not None else None except Exception: position_market_value = None known_market_value_remaining = position_market_value first_exec_time = execs_sorted[0].get("time") if execs_sorted else None price_multiplier = self._price_multiplier(meta or position) if execs_sorted: open_lots = open_longs + open_shorts for b in open_lots: qty = float(b["qty"]) avg_price = float(b["price"]) * price_multiplier cost_basis = qty * avg_price effective_market_price = None if market_price is None else market_price * price_multiplier market_value = None if effective_market_price is None else qty * effective_market_price if market_value is None and known_market_value_remaining is not None and position_qty: market_value = known_market_value_remaining * (qty / position_qty) if known_market_value_remaining is not None and market_value is not None: known_market_value_remaining -= market_value gain_dollars = None if market_value is None else market_value - cost_basis gain_percent = None cagr_percent = None base_cost = abs(cost_basis) if base_cost and market_value is not None: try: gain_percent = (gain_dollars / base_cost) * 100 except Exception: gain_percent = None # estimate CAGR using buy time if available buy_time = b.get("time") if buy_time and market_value is not None and base_cost > 0: try: # parse time similar to above for fmt in ("%Y%m%d %H:%M:%S", "%Y%m%d %H:%M:%S", "%Y%m%d-%H:%M:%S"): try: dt = datetime.strptime(buy_time, fmt) break except Exception: dt = None if dt is None: # try numeric timestamp try: dt = datetime.fromtimestamp(float(buy_time)) except Exception: dt = None if dt is not None: years = max((datetime.utcnow() - dt).days / 365.0, 1e-6) try: ending_value = base_cost + gain_dollars if ending_value > 0: cagr = (ending_value / base_cost) ** (1.0 / years) - 1.0 cagr_percent = cagr * 100.0 else: cagr_percent = None except Exception: cagr_percent = None except Exception: cagr_percent = None lots.append({ "symbol": meta.get("symbol"), "display_symbol": meta.get("display_symbol") or self._format_display_symbol(meta), "security_type": meta.get("security_type"), "underlying_symbol": meta.get("underlying_symbol"), "expiry": meta.get("expiry"), "strike": meta.get("strike"), "option_type": meta.get("option_type"), "instrument_key": instrument_key, "qty": qty, "avg_price": avg_price, "cost_basis": cost_basis, "market_price": effective_market_price, "market_value": market_value, "gain_dollars": gain_dollars, "gain_percent": gain_percent, "cagr_percent": cagr_percent, "buy_time": buy_time, }) elif include_position_fallback: # fallback: use positions data to create a single lot per symbol for p in self.positions: if p.get("instrument_key") != instrument_key: continue qty = p.get("position") or 0 avg_price = p.get("averageCost") or p.get("average_cost") or 0.0 try: qtyf = float(qty) except Exception: qtyf = 0.0 try: avg_pricef = float(avg_price) except Exception: avg_pricef = 0.0 cost_basis = qtyf * avg_pricef effective_market_price = None if market_price is None else market_price * self._price_multiplier(p) if effective_market_price is not None: market_value = qtyf * effective_market_price else: market_value = position_market_value gain_dollars = None if market_value is None else market_value - cost_basis gain_percent = None base_cost = abs(cost_basis) if base_cost and market_value is not None: try: gain_percent = (gain_dollars / base_cost) * 100 except Exception: gain_percent = None lots.append({ "symbol": p.get("symbol"), "display_symbol": p.get("display_symbol") or self._format_display_symbol(p), "security_type": p.get("security_type"), "underlying_symbol": p.get("underlying_symbol"), "expiry": p.get("expiry"), "strike": p.get("strike"), "option_type": p.get("option_type"), "instrument_key": instrument_key, "qty": qtyf, "avg_price": avg_pricef, "cost_basis": cost_basis, "market_price": effective_market_price, "market_value": market_value, "gain_dollars": gain_dollars, "gain_percent": gain_percent, "cagr_percent": None, "buy_time": first_exec_time, }) return lots