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

CLI display: STARTED is displayed as running in plugin list and plugin status output.

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 and survives restarts. The registry path is per-account (~/.ib_plugin_store_<account_id>.db), selected automatically at engine start based on the connected IB account. Plugin state files (state.json, holdings.json, instruments.json) are likewise isolated under plugins/<slot>/<account_id>/.

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

Cached historical data (get_bars_cached)

Prefer get_bars_cached over get_historical_data for warm-up fetches — it routes through the shared historical/bars.db BarStore and only calls IB for uncached ranges:

from datetime import datetime, timedelta, timezone

def start(self) -> bool:
    end_dt   = datetime.now(timezone.utc)
    start_dt = end_dt - timedelta(days=5)
    bars = self.get_bars_cached(
        contract=ContractBuilder.etf("SPY"),
        start_dt=start_dt,
        end_dt=end_dt,
        bar_size_setting="5 mins",
        what_to_show="MIDPOINT",
        use_rth=True,
    )
    return True

Live bar subscription with auto-caching (subscribe_live_bars)

Streams bars (backfill + live updates) and automatically caches every bar to the shared BarStore — no manual insert_bar calls needed:

def start(self) -> bool:
    self._req = self.subscribe_live_bars(
        contract=ContractBuilder.etf("GLD"),
        on_bar=self._on_bar,
        duration_str="2 D",
        bar_size_setting="5 mins",
        what_to_show="MIDPOINT",
        use_rth=True,
    )
    return True

def stop(self) -> bool:
    if self._req:
        self.portfolio.cancel_historical_data(self._req)
    return True

Bars cached this way are immediately available to get_bars_cached on the next restart.

---

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

ibctl plugin load plugins/basket/plugin.py=growth
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.

Managing the instrument list from the CLI

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 tech

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

---

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