-
-
Notifications
You must be signed in to change notification settings - Fork 0
Plugin Design
A plugin is a Python class that subclasses PluginBase. The engine manages its lifecycle, feeds it market data, routes fills and P&L callbacks to it, and executes the trade signals it returns.
plugins/
my_strategy/
__init__.py ← must re-export the class
plugin.py ← implementation
instruments.json ← optional: tradable instruments
__init__.py must export the class:
from .plugin import MyStrategyPlugin
__all__ = ["MyStrategyPlugin"]Load path for ibctl:
./ibctl.py plugin load plugins.my_strategy.pluginfrom typing import List, Dict, Any
from plugins.base import PluginBase, TradeSignal
from ib.contract_builder import ContractBuilder
from ib.data_feed import DataType
class MyStrategyPlugin(PluginBase):
VERSION = "1.0.0"
def __init__(self, base_path=None, portfolio=None,
shared_holdings=None, message_bus=None):
super().__init__(
"my_strategy", # unique snake_case name
base_path, portfolio, shared_holdings, message_bus,
)
@property
def description(self) -> str:
return "One-line description shown in plugin list"
def start(self) -> bool:
state = self.load_state()
self._bar_count = state.get("bar_count", 0)
contract = ContractBuilder.us_stock("SPY")
self.request_stream("SPY", contract,
{DataType.BAR_1MIN}, on_bar=self._on_bar)
return True
def stop(self) -> bool:
self.cancel_stream("SPY")
self.unsubscribe_all()
self.save_state({"bar_count": self._bar_count})
self.save_holdings()
return True
def freeze(self) -> bool:
self.save_state({"bar_count": self._bar_count})
return True
def resume(self) -> bool:
return True
def calculate_signals(self) -> List[TradeSignal]:
return []
def handle_request(self, request_type: str, payload: Dict) -> Dict:
if request_type == "get_status":
return {"success": True, "data": {"bar_count": self._bar_count}}
return {"success": False, "message": f"Unknown: {request_type}"}
def _on_bar(self, bar) -> None:
self._bar_count += 1| Attribute | Type | Description |
|---|---|---|
VERSION |
str |
Stored with saved state (for migration logic) |
IS_SYSTEM_PLUGIN |
bool |
Default False. True prevents plugin unload
|
INSTRUMENT_COMPLIANCE |
bool |
Default False. True blocks signals for symbols not in the plugin's instrument set |
UNLOADED
│ load() ← engine calls: loads instruments + holdings from SQLite
▼
LOADED
│ start() ← your code: subscribe to streams, load state
▼
STARTED ◄──────────────────────────────┐
│ freeze() │ resume()
▼ │
FROZEN ────────────────────────────────┘
│ stop()
▼
STOPPED
│ unload() ← engine calls
▼
UNLOADED
calculate_signals is not called while the plugin is FROZEN or STOPPED.
Called when transitioning LOADED → STARTED. Set up everything.
- Call
load_state()to restore persisted data - Call
request_stream()for each symbol you need - Subscribe to MessageBus channels
- Return
Trueon success;Falseleaves the plugin in ERROR state
Called when transitioning to STOPPED.
- Call
cancel_stream()for each symbol (or let it happen via executive cleanup) - Call
unsubscribe_all() - Call
save_state()andsave_holdings() - Return
True
Pause without tearing down.
- Save state (streams remain active, callbacks still fire)
- Return
True
Return from FROZEN → STARTED.
- Restore any in-memory state you need
- Return
True
Called by the executive on each execution tick (bar boundary or timer).
- Return a list of
TradeSignalobjects (may be empty) - Keep it fast — this is the critical path
- Exceptions are caught and counted toward the circuit breaker (5 consecutive → breaker trips)
Called from the socket thread when plugin request or plugin message is used.
-
plugin request <name> <type> [json]delivers the literal<type>string -
plugin message <name> [json]deliversrequest_type="message" - Must return
{"success": True/False, ...} - Keep fast; no blocking
Optional but strongly recommended. Return a human-readable string documenting
your plugin's custom handle_request commands. Retrieved with:
ibctl plugin help my_strategyDefault implementation returns a generic "no custom commands" string.
State is stored in SQLite (~/.ib_plugin_store.db) and survives restarts.
# Save
self.save_state({
"bar_count": self._bar_count,
"last_price": self._last_price,
"positions": {"SPY": 100}, # any JSON-serializable data
})
# Load (returns {} if nothing saved)
state = self.load_state()
self._bar_count = state.get("bar_count", 0)
# Clear
self.clear_state()Legacy state.json files are automatically migrated to SQLite on the first load_state() call and left in place as backup.
from ib.contract_builder import ContractBuilder
from ib.data_feed import DataType
contract = ContractBuilder.us_stock("SPY", primary_exchange="ARCA")
self.request_stream(
symbol="SPY",
contract=contract,
data_types={DataType.BAR_1MIN, DataType.TICK},
on_bar=self._on_bar,
on_tick=self._on_tick,
)You can request multiple data types in one call. The on_bar callback receives all bar sizes — use bar.timestamp to distinguish them.
| DataType | Description |
|---|---|
TICK |
Throttled price/size updates from reqMktData
|
BAR_5SEC |
5-second OHLCV real-time bars |
BAR_1MIN |
Aggregated from 5-sec bars |
BAR_5MIN |
Aggregated from 5-sec bars |
BAR_15MIN |
Aggregated from 5-sec bars |
BAR_1HOUR |
Aggregated from 5-sec bars |
TICK_BY_TICK_LAST |
Every last-sale print |
TICK_BY_TICK_BIDASK |
Every quote change |
TICK_BY_TICK_MIDPOINT |
Every midpoint change |
MARKET_DEPTH |
L2 order book (10 levels) |
def _on_bar(self, bar) -> None:
# Called on the IB reader thread — return quickly
# bar.symbol str
# bar.timestamp str (ISO)
# bar.open float
# bar.high float
# bar.low float
# bar.close float
# bar.volume int
# bar.wap float
# bar.is_bullish bool (close > open)
# bar.is_bearish bool
# bar.range float (high - low)
# bar.body float (abs(close - open))
# bar.mid float ((high + low) / 2)
self._last_close = bar.closedef _on_tick(self, tick) -> None:
# tick.symbol str
# tick.price float
# tick.tick_type str (LAST, BID, ASK, CLOSE, DELAYED_*, *_SIZE, VOLUME)
# tick.size Optional[int] None for price ticks
# tick.timestamp datetime
if tick.tick_type == "LAST":
self._last_price = tick.priceself.cancel_stream("SPY") # Call in stop()Streams are reference-counted. IB sees only one subscription per symbol regardless of how many plugins subscribe.
feed = self._executive.data_feed
# Last N bars
bars = feed.get_bars("SPY", DataType.BAR_1MIN, count=50)
# Ticks since a time
ticks = feed.get_ticks("SPY", since=some_datetime)
# L2 snapshot
depth = feed.get_depth("SPY")
bids = depth.bids # List[DepthLevel], best first
asks = depth.asksbars = self.get_historical_data(
contract=ContractBuilder.us_stock("AAPL"),
duration_str="5 D", # N S|D|W|M|Y
bar_size_setting="1 hour",
what_to_show="TRADES", # TRADES, MIDPOINT, BID, ASK
use_rth=True,
end_date_time="", # "" = now
timeout=60.0,
)
# Returns List[BarData] or None on timeout
# bar.date, bar.open, bar.high, bar.low, bar.close, bar.volume, bar.wapCall this from start() or a background thread, never from on_bar/on_tick.
from decimal import Decimal
from plugins.base import TradeSignal
def calculate_signals(self) -> List[TradeSignal]:
if self._should_buy():
return [TradeSignal(
symbol="SPY",
action="BUY", # "BUY", "SELL", or "HOLD"
quantity=Decimal("10"), # ALWAYS Decimal, ALWAYS from str
reason="SMA crossover", # logged in execution history
confidence=0.80, # 0.0–1.0 (optional)
urgency="Normal", # Patient | Normal | Urgent (optional)
target_weight=0.20, # optional portfolio weight
)]
return []A signal is actionable when action is "BUY" or "SELL" and quantity > 0. "HOLD" signals are silently ignored.
Always use Decimal(str(your_float)) for quantity — never Decimal(float).
Register an order to receive callbacks:
order_id = self.portfolio.place_order_custom(contract, order)
if order_id:
self.register_order(order_id)Called when an order is completely filled.
def on_order_fill(self, order_record) -> None:
logger.info(f"Filled {order_record.quantity} {order_record.symbol} "
f"@ {order_record.avg_fill_price}")
self._holdings.add_position(
order_record.symbol,
order_record.filled_quantity,
cost_basis=order_record.avg_fill_price,
)Called on every IB status change (submitted, partial fill, cancelled, etc.).
def on_order_status(self, order_record) -> None:
if order_record.is_complete:
# FILLED, CANCELLED, or ERROR
passorder_record fields: order_id, symbol, action, quantity, order_type, status, filled_quantity, avg_fill_price, remaining, is_filled, is_complete, fill_value.
Called when IB delivers a commission report. realized_pnl is 1.7e308 (IB's sentinel) for opening trades.
Real-time P&L updates when subscribed via self.portfolio.request_pnl(account).
Access holdings via self._holdings (populated by _load_holdings() during load()):
# Check available cash
cash = self._holdings.current_cash
# Check a position
pos = self._holdings.get_position("SPY")
if pos:
qty = pos.quantity
basis = pos.cost_basis
value = pos.market_value
# After a buy fill:
self._holdings.add_cash(-cost)
self._holdings.add_position("SPY", qty, cost_basis=price)
# After a sell fill:
self._holdings.remove_position("SPY", qty)
self._holdings.add_cash(proceeds)
# Persist
self.save_holdings() # call in stop() and on important eventsself.publish(
channel="indicators_rsi",
payload={"rsi": 68.3, "symbol": "SPY"},
message_type="data", # data | signal | alert | metric
)def start(self) -> bool:
self.subscribe("indicators_rsi", self._on_rsi)
return True
def stop(self) -> bool:
self.unsubscribe_all() # always clean up in stop()
...
def _on_rsi(self, message) -> None:
rsi = message.payload["rsi"]
source = message.metadata.source_plugin| Pattern | Use |
|---|---|
indicators_<name> |
Computed indicators (RSI, SMA, …) |
<plugin>_signals |
Trade signals from a plugin |
<plugin>_metrics |
Health and performance metrics |
alerts |
System-wide alerts |
synthetic_<name> |
Synthetic spreads or virtual tickers |
# Read-only account data
positions = self.portfolio.positions # Dict[str, Position]
pos = positions.get("SPY")
if pos:
pos.quantity # float
pos.avg_cost # float
pos.market_value # float
pos.unrealized_pnl # float
# Place a direct order (not via signals)
from ib.order_builder import OrderBuilder
order = OrderBuilder.market("BUY", quantity=10)
order_id = self.portfolio.place_order_custom(contract, order)
self.register_order(order_id)
# Cancel an order
self.portfolio.cancel_order(order_id)
# Connection info
self.portfolio.connected # bool
self.portfolio.managed_accounts # List[str]For paper accounts (delayed data), call in start():
self.portfolio.reqMarketDataType(3) # delayed (free)from ib.contract_builder import ContractBuilder
ContractBuilder.us_stock("SPY")
ContractBuilder.us_stock("AAPL", primary_exchange="NASDAQ")
ContractBuilder.etf("QQQ")
ContractBuilder.option("AAPL", expiry="20260117", strike=200.0, right="C")
ContractBuilder.future("ES", expiry="202603", exchange="CME")
ContractBuilder.continuous_future("ES", exchange="CME")
ContractBuilder.forex("EUR", "USD")
ContractBuilder.index("SPX", exchange="CBOE")
ContractBuilder.crypto("BTC", exchange="PAXOS")A plugin can request its own removal (e.g., after a one-shot task):
self.request_unload()The executive removes the plugin asynchronously — safe to call from any callback.
| Do | Don't |
|---|---|
Return quickly from on_bar, on_tick
|
Block in IB reader callbacks |
Use self.publish() from any thread |
Call request_stream from IB reader callbacks |
Do heavy computation in calculate_signals
|
Call save_state frequently on hot paths |
Call save_state in stop() and freeze()
|
Hold locks across await boundaries |
publish() is thread-safe (uses an internal RLock). Everything else should be called from the thread documented in the threading model.
| Item | Convention | Example |
|---|---|---|
| Plugin directory | snake_case |
plugins/sma_crossover/ |
Plugin name (passed to super()) |
snake_case |
"sma_crossover" |
| Class name |
PascalCase + Plugin
|
SMACrossoverPlugin |
| MessageBus channels | snake_case |
"indicators_sma_20" |
handle_request types |
snake_case |
"get_status", "set_period"
|
| State dict keys | snake_case |
"bar_count", "last_signal"
|
Load the same plugin class twice by assigning each a unique slot with the = syntax:
ibctl plugin load plugins/strategy/plugin.py=spy_leg
ibctl plugin load plugins/strategy/plugin.py=qqq_legEach instance has fully independent state, holdings, and CLI address (spy_leg, qqq_leg).
Combine with INSTRUMENT_COMPLIANCE = True to enforce per-instance security restrictions.
Set INSTRUMENT_COMPLIANCE = True on your plugin class to opt in to
instrument-set enforcement. The executive will block any TradeSignal
whose symbol is not in the plugin's registered instrument set and log a
warning. This is a no-op for plugins that only signal on their registered
symbols, but essential when running multiple instances of the same class on
different baskets:
class BasketPlugin(PluginBase):
INSTRUMENT_COMPLIANCE = True
VERSION = "1.0.0"
...ibctl plugin load plugins/basket/plugin.py=growth
ibctl plugin load plugins/basket/plugin.py=techEach instance is silently restricted to its own basket. No signal from
growth can trigger a trade in AAPL.
Instrument lists are stored per slot in SQLite and can be read and modified at any time — even while the plugin is running.
# Add instruments to each slot
ibctl plugin instruments add growth SPY --weight 0.6
ibctl plugin instruments add growth QQQ --weight 0.4
ibctl plugin instruments add tech AAPL --weight 0.5
ibctl plugin instruments add tech MSFT --weight 0.5
# See what's registered
ibctl plugin instruments list growth
# Temporarily disable a symbol without removing it
ibctl plugin instruments disable growth QQQ
# Re-read from SQLite into the running plugin's memory
ibctl plugin instruments reload growth
# Remove one instrument permanently
ibctl plugin instruments remove tech AAPL
# Wipe the entire list for a slot
ibctl plugin instruments clear techadd flags: --name TEXT, --weight FLOAT, --min-weight FLOAT, --max-weight FLOAT, --exchange TEXT, --currency TEXT, --sec-type TEXT, --disabled
Migration: If an instruments.json file exists in the plugin directory when the plugin is first loaded, it is automatically imported into SQLite. The import is idempotent and the source file is left intact as a backup.
The executive wraps each call to calculate_signals in a circuit breaker:
| Threshold | Effect |
|---|---|
| 5 consecutive exceptions | Breaker trips — signals suppressed |
| 5 minutes after trip | Breaker enters half-open — one attempt allowed |
| First successful run | Breaker closes — normal operation |
Tripped plugins appear in plugin status with circuit_breaker: open. The plugin is NOT stopped — use plugin trigger to attempt an early recovery call.
from collections import deque
from decimal import Decimal
from typing import List, Dict
from plugins.base import PluginBase, TradeSignal
from ib.contract_builder import ContractBuilder
from ib.data_feed import DataType
class SMACrossoverPlugin(PluginBase):
VERSION = "1.0.0"
FAST = 10
SLOW = 30
def __init__(self, **kwargs):
super().__init__("sma_crossover", **kwargs)
self._closes: deque = deque(maxlen=self.SLOW + 1)
self._position = 0
@property
def description(self) -> str:
return f"SMA {self.FAST}/{self.SLOW} crossover on SPY"
def start(self) -> bool:
state = self.load_state()
self._position = state.get("position", 0)
contract = ContractBuilder.us_stock("SPY", primary_exchange="ARCA")
self.request_stream("SPY", contract, {DataType.BAR_1MIN},
on_bar=self._on_bar)
return True
def stop(self) -> bool:
self.cancel_stream("SPY")
self.unsubscribe_all()
self.save_state({"position": self._position})
self.save_holdings()
return True
def freeze(self) -> bool:
self.save_state({"position": self._position})
return True
def resume(self) -> bool:
return True
def _on_bar(self, bar) -> None:
self._closes.append(bar.close)
def calculate_signals(self) -> List[TradeSignal]:
closes = list(self._closes)
if len(closes) < self.SLOW:
return []
fast = sum(closes[-self.FAST:]) / self.FAST
slow = sum(closes[-self.SLOW:]) / self.SLOW
if fast > slow and self._position <= 0:
self._position = 10
return [TradeSignal("SPY", "BUY", Decimal("10"),
reason="fast SMA crossed above slow")]
if fast < slow and self._position > 0:
self._position = 0
return [TradeSignal("SPY", "SELL", Decimal("10"),
reason="fast SMA crossed below slow")]
return []
def handle_request(self, request_type: str, payload: Dict) -> Dict:
if request_type == "get_status":
closes = list(self._closes)
return {"success": True, "data": {
"bars": len(closes),
"position": self._position,
}}
return {"success": False, "message": f"Unknown: {request_type}"}
def cli_help(self) -> str:
return (
"sma_crossover commands:\n"
" plugin request sma_crossover get_status {}\n"
" → {bars, position}\n"
)TWS Headless
- Startup sequence
- Market data & streams
- Plugin execution
- Holdings & bookkeeping
- Order lifecycle
- State persistence
- See what's going on
- Fund a plugin
- Transfer assets
- Load and start a plugin
- Stop or pause a plugin
- Place a manual trade
- Send a plugin request
- Manage instrument list
- Reconcile holdings
- Move paper → live
- Shut down
- Full command reference
Plugin Manual ← complete reference
- File layout
- Lifecycle methods
- State persistence
- Market data streams
- Trade signals
- Order callbacks
- Holdings management
- MessageBus
- ContractBuilder
- Instrument compliance
- Multiple instances (slots)
- CLI help & messaging
- Threading rules
- Full example