Plugin Design

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.

---

File Layout

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.plugin

---

Minimal Skeleton

from 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

---

Class Attributes

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

---

Lifecycle

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.

---

The Six Required Methods

start(self) -> bool

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 True on success; False leaves the plugin in ERROR state

stop(self) -> bool

Called when transitioning to STOPPED.

• Call cancel_stream() for each symbol (or let it happen via executive cleanup)
• Call unsubscribe_all()
• Call save_state() and save_holdings()
• Return True

freeze(self) -> bool

Pause without tearing down.

• Save state (streams remain active, callbacks still fire)
• Return True

resume(self) -> bool

Return from FROZEN → STARTED.

• Restore any in-memory state you need
• Return True

calculate_signals(self) -> List[TradeSignal]

Called by the executive on each execution tick (bar boundary or timer).

• Return a list of TradeSignal objects (may be empty)
• Keep it fast — this is the critical path
• Exceptions are caught and counted toward the circuit breaker (5 consecutive → breaker trips)

handle_request(self, request_type, payload) -> Dict

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] delivers request_type="message"
• Must return {"success": True/False, ...}
• Keep fast; no blocking

cli_help(self) -> str

Optional but strongly recommended. Return a human-readable string documenting your plugin's custom handle_request commands. Retrieved with:

ibctl plugin help my_strategy

Default implementation returns a generic "no custom commands" string.

---

State Persistence

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.

---

Market Data

Requesting a stream

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.

Available data types

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)

Bar callback

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.close

Tick callback

def _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.price

Cancelling streams

self.cancel_stream("SPY")   # Call in stop()

Streams are reference-counted. IB sees only one subscription per symbol regardless of how many plugins subscribe.

Accessing buffered data

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.asks

Historical data (blocking)

bars = 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.wap

Call this from start() or a background thread, never from on_bar/on_tick.

---

Trade Signals

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).

---

Order Callbacks

Register an order to receive callbacks:

order_id = self.portfolio.place_order_custom(contract, order)
if order_id:
    self.register_order(order_id)

on_order_fill(self, order_record)

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,
    )

on_order_status(self, order_record)

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
        pass

order_record fields: order_id, symbol, action, quantity, order_type, status, filled_quantity, avg_fill_price, remaining, is_filled, is_complete, fill_value.

on_commission(self, exec_id, commission, realized_pnl, currency)

Called when IB delivers a commission report. realized_pnl is 1.7e308 (IB's sentinel) for opening trades.

on_pnl(self, pnl_data)

Real-time P&L updates when subscribed via self.portfolio.request_pnl(account).

---

Holdings Management

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 events

---

MessageBus

Publishing

self.publish(
    channel="indicators_rsi",
    payload={"rsi": 68.3, "symbol": "SPY"},
    message_type="data",    # data | signal | alert | metric
)

Subscribing

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

Channel naming conventions

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

---

Portfolio Access

# 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)

---

ContractBuilder

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")

---

Self-Unload

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.

---

Threading Rules

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.

---

Naming Conventions

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"

Multiple instances

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_leg

Each instance has fully independent state, holdings, and CLI address (spy_leg, qqq_leg). Combine with INSTRUMENT_COMPLIANCE = True to enforce per-instance security restrictions.

---

Instrument Compliance

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"
    ...

# Instance A: instruments.json contains SPY, QQQ
ibctl plugin load plugins/basket/plugin.py=growth

# Instance B: instruments.json contains AAPL, MSFT
ibctl plugin load plugins/basket/plugin.py=tech

Each instance is silently restricted to its own basket. No signal from growth can trigger a trade in AAPL.

---

Circuit Breaker

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.

---

Example: Simple SMA Crossover

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"
        )