Browse documentation

Extending BrokerBridge with plugins

This page is for developers writing Python to add a data source. If you just want to trade, see Help instead. AI models are not pluggable today. Pick your model and pay with AI credits, as described in AI Providers. Retail accounts connect Interactive Brokers for market data; linking your own third-party data subscription (Unusual Whales, EODHD, or similar) is not available today.

Add a custom data source by implementing one abstract base class. The plugin can supply ranked signals to the research pipeline. It does not gain model selection, approval, or broker authority.

Architecture overview

DataSourcePlugin

Provides trading signals. The pipeline calls fetch_signals() during each scan interval to get ranked signals.

Examples: IBKR market data, custom scanner

DataSourcePlugin supports three auth modes: api_key, oauth, and connection. These credentials are for your own third-party data source, never for an AI provider.

DataSourcePlugin interface

python
class DataSourcePlugin(ABC):
    name: str                    # Unique plugin identifier
    display_name: str            # Human-readable name
    auth_type: AuthType          # "api_key" | "oauth" | "connection"
    required: bool = False       # System won't start without it if True

    async def validate_credentials(self) -> bool: ...
    async def fetch_signals(self, symbols: list[str]) -> list[Signal]: ...
    async def health_check(self) -> PluginStatus: ...

Tutorial: custom data source

my_scanner_plugin.py
from datetime import datetime, timezone
import httpx
from brokerbridge.plugins.base import DataSourcePlugin
from brokerbridge.plugins.types import AuthType, PluginStatus, Signal


class MyScannerPlugin(DataSourcePlugin):
    name = "my_scanner"
    display_name = "My Custom Scanner"
    auth_type: AuthType = "api_key"

    def __init__(self, api_key: str):
        self._api_key = api_key

    async def validate_credentials(self) -> bool:
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                "https://api.example.com/ping",
                headers={"Authorization": f"Bearer {self._api_key}"},
            )
            return resp.status_code == 200

    async def fetch_signals(self, symbols: list[str]) -> list[Signal]:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://api.example.com/scan",
                headers={"Authorization": f"Bearer {self._api_key}"},
                json={"symbols": symbols},
            )
            resp.raise_for_status()
            data = resp.json()

        signals = []
        for item in data.get("signals", []):
            signals.append(Signal(
                symbol=item["symbol"],
                direction=item["direction"],
                score=item["score"],
                timestamp=datetime.now(timezone.utc),
                source=self.name,
            ))
        return sorted(signals, key=lambda s: s.score, reverse=True)

    async def health_check(self) -> PluginStatus:
        try:
            valid = await self.validate_credentials()
            return PluginStatus(connected=valid)
        except Exception as e:
            return PluginStatus(connected=False, message=str(e))

Available plugins

PluginTypeStatus
IBKR Market DataDataSourceAvailable
← API ReferenceSecurity & Privacy →