ibkr-trade-ai 0.1.3

IBKR Trade AI - Lightweight asyncio based trading library for Interactive Brokers.

IBKR Trade AI

Disclaimer

This software is for educational purposes only. Do not risk money which you are afraid to lose. USE THE SOFTWARE AT YOUR OWN RISK. THE AUTHORS AND ALL AFFILIATES ASSUME NO RESPONSIBILITY FOR YOUR TRADING RESULTS.

Trading stocks and other financial instruments carries a high level of risk, and may not be suitable for all investors. Before deciding to trade, you should carefully consider your investment objectives, level of experience, and risk appetite.

A lightweight, asyncio-based trading library for Interactive Brokers (IBKR) with built-in trade tracking, order management, and persistent caching.

Features

• Async-first design: Built on asyncio for high-performance concurrent operations
• Persistent trade tracking: Uses diskcache for reliable trade state persistence
• Type-safe: Fully typed with comprehensive type hints for better IDE support
• Thread-safe: Built-in thread validation decorators for multi-threaded environments
• Order management: Support for market and limit orders with automatic order ID generation
• Real-time updates: Event-driven architecture for order status and execution updates
• Overnight trading: Built-in support for regular and overnight trading hours
• Flexible callbacks: Register async callbacks for trade events

Installation

From PyPI (when published)

pip install ibkr-trade-ai

From Source

git clone https://github.com/osjayaprakash/ibkr-trade-ai.git
cd ibkr-trade-ai
pip install -e .

Development Installation

pip install -e ".[dev]"

Quick Start

Basic Usage

import asyncio
from diskcache import Cache
from bolt2 import TradeDataManager

async def main():
    # Initialize cache for trade persistence
    cache = Cache('./trade_cache')

    # Create trade manager
    manager = TradeDataManager(cache)

    # Place a market order
    await manager.placeOrderAsync(
        sym='AAPL',
        lmtPrice=None,  # None for market orders
        qty=10,
        action='BUY',
        slot=None
    )

    # Run the event loop
    await manager.runLoop()

if __name__ == '__main__':
    asyncio.run(main())

Placing Limit Orders

# Place a limit order
await manager.placeOrderAsync(
    sym='TSLA',
    lmtPrice=250.50,  # Limit price
    qty=5,
    action='SELL',
    slot='portfolio_1'
)

Registering Callbacks

async def on_trade_update(trade):
    print(f"Trade updated: {trade.symbol} - {trade.status}")
    # Your custom logic here

# Register the callback
manager.callbacks.append(on_trade_update)

Architecture

Core Components

TradeDataManager

The main orchestrator that handles:

• Order placement and validation
• Event loop management
• Trade state updates
• Callback execution

Trade

Represents a single trade with:

• Order details (symbol, quantity, price)
• Execution information
• State tracking (Submitted, Filled, Cancelled, etc.)
• Persistent caching

TradeWrapper

IBKR API wrapper that:

• Manages connection to Interactive Brokers TWS/Gateway
• Handles API callbacks
• Queues events for processing

API Reference

TradeDataManager

__init__(cache: Cache, timeout: int, clientId: int)

Initialize the trade manager with a cache instance, timeout, and client ID.

Parameters:

• cache (Cache): diskcache.Cache instance for trade persistence

async placeOrderAsync(sym: str, lmtPrice: Optional[float], qty: int, action: str, slot: Optional[str]) -> None

Place an order asynchronously.

Parameters:

• sym (str): Stock symbol
• lmtPrice (Optional[float]): Limit price, or None for market orders
• qty (int): Quantity to trade (must be >= 1)
• action (str): 'BUY' or 'SELL'
• slot (Optional[str]): Optional identifier for grouping trades

Raises:

• ValueError: If action is not 'BUY' or 'SELL', qty < 1, or invalid limit price
• RuntimeError: If unable to get next valid order ID

Example:

# Market order
await manager.placeOrderAsync('AAPL', None, 100, 'BUY', None)

# Limit order
await manager.placeOrderAsync('MSFT', 380.50, 50, 'SELL', 'tech_portfolio')

async runLoop() -> None

Run the main event processing loop. Processes events from the queue and executes callbacks.

Example:

# Run in background task
asyncio.create_task(manager.runLoop())

callbacks: list[Callable[[Trade], Awaitable[Any]]]

List of async callbacks to execute on trade updates.

Example:

async def log_trade(trade):
    print(f"Trade update: {trade}")

manager.callbacks.append(log_trade)

Trade States

The library uses the following trade states:

State	Description
PendingSubmit	Order is being prepared for submission
PendingCancel	Order cancellation is pending
PreSubmitted	Order is pre-submitted to IBKR
Submitted	Order is active on the exchange
Filled	Order has been completely filled
Cancelled	Order has been cancelled
Inactive	Order is inactive
CustomSubmitInitiated	Custom state for initial submission
CustomCancelInitiated	Custom state for cancel initiation
DirtyCancelled	Order was cancelled but no confirmation received

Trading Hours

Regular Trading Hours

9:30 AM - 4:00 PM EST

from bolt2 import isRegularTradingHourMin

if isRegularTradingHourMin():
    # Place regular hours trades
    pass

Overnight Trading Hours

8:00 PM - 3:50 AM EST

from bolt2 import isOvernightTradingHourMin

if isOvernightTradingHourMin():
    # Place overnight trades (requires limit orders)
    pass

Configuration

Cache Setup

The library uses diskcache for persistent storage:

from diskcache import Cache

# Basic cache
cache = Cache('./trade_cache')

# Cache with custom settings
cache = Cache(
    directory='./trade_cache',
    size_limit=500 * 1024 * 1024,  # 500 MB
    eviction_policy='least-recently-used'
)

Logging

Configure logging for debugging:

from bolt2.utils import setupRootLogger

# Setup with default log file
setupRootLogger()

# Custom log file
setupRootLogger('./logs/trading.log')

Thread Safety

The library includes thread validation decorators:

from bolt2.utils import setUserThread, setApiThread, inUserThread, inApiThread
import threading

# Set thread IDs
setUserThread(threading.get_ident())

# Decorate functions that must run in specific threads
@inUserThread
def user_thread_function():
    pass

Best Practices

1. Always Use Async Context

async def trade_workflow():
    cache = Cache('./cache')
    manager = TradeDataManager(cache, timeout=30, clientId=1)

    try:
        await manager.placeOrderAsync('AAPL', None, 10, 'BUY', None)
        await manager.runDataManagerLoop()
    except Exception as e:
        logger.error(f"Trading error: {e}")

2. Handle Errors Gracefully

try:
    await manager.placeOrderAsync('INVALID', None, -5, 'BUY', None)
except ValueError as e:
    print(f"Invalid order parameters: {e}")
except RuntimeError as e:
    print(f"Runtime error: {e}")

3. Use Callbacks for Trade Monitoring

async def monitor_trade(trade):
    if trade.status == TradeState.Filled.value:
        await send_notification(f"Order filled: {trade.symbol}")
    elif trade.status == TradeState.Cancelled.value:
        await log_cancelled_order(trade)

manager.callbacks.append(monitor_trade)

4. Validate Trading Hours

from bolt2 import isRegularTradingHourMin, isOvernightTradingHourMin

if isRegularTradingHourMin():
    # Market orders OK
    await manager.placeOrderAsync('AAPL', None, 10, 'BUY', None)
elif isOvernightTradingHourMin():
    # Must use limit orders
    await manager.placeOrderAsync('AAPL', 180.50, 10, 'BUY', None)
else:
    print("Market is closed")

Development

Setup Development Environment

# Clone repository
git clone https://github.com/osjayaprakash/ibkr-trade-ai.git
cd ibkr-trade-ai

# Install with dev dependencies
pip install -e ".[dev]"

# Run type checking
mypy bolt2

# Run tests
pytest

# Format code
black bolt2
ruff check bolt2

Project Structure

ibkr-trade-ai/
├── bolt2/                  # Main package
│   ├── __init__.py        # Public API exports
│   ├── manager.py         # TradeDataManager class
│   ├── trade.py           # Trade class and utilities
│   ├── utils.py           # Utility functions and enums
│   ├── wrapper.py         # IBKR API wrapper
│   └── py.typed           # PEP 561 type marker
├── .github/
│   └── workflows/
│       └── publish-to-test-pypi.yml  # CI/CD pipeline
├── tests/                 # Test suite (if available)
├── pyproject.toml         # Project configuration
├── README.md             # This file
└── LICENSE               # MIT License

Publishing & CI/CD Setup

This project uses GitHub Actions for automated testing and publishing to PyPI.

Automated Workflow Features

The CI/CD pipeline automatically:

• ✅ Tests on Python 3.9, 3.10, 3.11, 3.12, 3.13
• ✅ Runs type checking (mypy)
• ✅ Runs linting (ruff)
• ✅ Checks code formatting (black)
• ✅ Validates version consistency across files
• ✅ Builds and verifies package distributions
• ✅ Tests package installation
• ✅ Publishes to TestPyPI for release candidates (v*-rc* tags)
• ✅ Publishes to PyPI for stable releases (v* tags)
• ✅ Creates GitHub releases with auto-generated notes

Initial Setup (One-Time)

1. Create GitHub Environments

Go to Settings → Environments and create:

• pypi - Production PyPI environment

  • Optional: Add 5-minute wait timer (gives review time)
  • Optional: Set deployment branch to main only

• testpypi - Testing PyPI environment

  • No restrictions needed

2. Configure PyPI Trusted Publishing (Recommended)

This is more secure than using API tokens.

For PyPI (Production):

1. Go to https://pypi.org/manage/account/publishing/
2. Click "Add a new pending publisher"
3. Fill in:
   • PyPI Project Name: ibkr-trade-ai
   • Owner: osjayaprakash (your GitHub username)
   • Repository name: ibkr-trade-ai
   • Workflow name: publish-to-test-pypi.yml
   • Environment name: pypi
4. Save

For TestPyPI (Testing):

1. Go to https://test.pypi.org/manage/account/publishing/
2. Repeat the same process with environment name: testpypi

Alternative: Using API Tokens

If you prefer API tokens instead of trusted publishing:

1. Generate tokens at https://pypi.org/manage/account/token/
2. Add to GitHub: Settings → Secrets and variables → Actions
   • PYPI_API_TOKEN - PyPI token
   • TEST_PYPI_API_TOKEN - TestPyPI token
3. Update workflow to use password: ${{ secrets.PYPI_API_TOKEN }}

3. (Optional) Branch Protection

Protect the main branch:

Settings → Branches → Add rule for main:

• ✅ Require status checks to pass:
  • test (Python 3.9)
  • test (Python 3.10)
  • test (Python 3.11)
  • test (Python 3.12)
  • test (Python 3.13)
  • build
• ✅ Require branches to be up to date

Release Process

For Release Candidates (Testing on TestPyPI):

# 1. Update version to include 'rc' suffix in both:
#    - pyproject.toml: version = "0.1.0-rc1"
#    - bolt2/__init__.py: __version__ = "0.1.0-rc1"

# 2. Commit and tag
git add .
git commit -m "Prepare release 0.1.0-rc1"
git tag v0.1.0-rc1
git push origin main --tags

# 3. Workflow automatically:
#    - Runs all tests
#    - Builds package
#    - Publishes to TestPyPI (because tag contains 'rc')

# 4. Test installation from TestPyPI
pip install -i https://test.pypi.org/simple/ ibkr-trade-ai==0.1.0rc1

# 5. Test your package
python -c "from bolt2 import TradeDataManager; print('Success!')"

For Stable Releases (Publishing to PyPI):

# 1. Update version (no 'rc' suffix) in both:
#    - pyproject.toml: version = "0.1.0"
#    - bolt2/__init__.py: __version__ = "0.1.0"

# 2. Commit and tag
git add .
git commit -m "Release 0.1.0"
git tag v0.1.0
git push origin main --tags

# 3. Workflow automatically:
#    - Runs all tests
#    - Validates version consistency
#    - Builds package
#    - Publishes to PyPI (no 'rc' in tag)
#    - Creates GitHub Release with artifacts

# 4. Install from PyPI
pip install ibkr-trade-ai

# 5. Verify installation
python -c "from bolt2 import TradeDataManager; print('Success!')"

Version Management

Important: Keep versions synchronized across:

1. pyproject.toml - Line 7: version = "0.1.0"
2. bolt2/__init__.py - Line 28: __version__ = "0.1.0"

The CI pipeline will fail if versions don't match the git tag.

Version Format:

• Stable releases: v0.1.0, v1.0.0, v2.1.3
• Release candidates: v0.1.0-rc1, v1.0.0-rc2
• Alpha/Beta: v0.1.0-alpha1, v0.1.0-beta1

Workflow Triggers

The workflow runs on:

• ✅ Pushes to main branch (tests only)
• ✅ Pull requests to main (tests only)
• ✅ Version tags v* (tests + publish)
• ✅ Manual trigger via GitHub Actions UI

Troubleshooting

"Version mismatch" error:

• Check that tag, pyproject.toml, and init.py all have the same version
• Tag should be v0.1.0 for version 0.1.0

"Trusted publishing failed":

• Verify you created the pending publisher on PyPI/TestPyPI
• Check that repository name, owner, and workflow name match exactly
• Wait a few minutes after creating the publisher

Tests failing:

• Check the Actions tab for detailed error messages
• Run tests locally: pytest tests/
• Run type checking: mypy bolt2

Package not found on PyPI:

• Check the Actions tab to see if publish step succeeded
• Wait a few minutes for PyPI to index the package
• Verify you're using the correct package name: ibkr-trade-ai

Requirements

• Python 3.9+
• diskcache
• ibapi-latest (Interactive Brokers API)
• pytz

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Guidelines

1. Follow the existing code style (black, mypy)
2. Add tests for new features
3. Update documentation as needed
4. Ensure all tests pass before submitting

License

This project is licensed under the MIT License - see the LICENSE file for details.

Acknowledgments

• Built on top of Interactive Brokers API
• Inspired by the need for a modern, async-first trading library
• Thanks to the Python async community

Support

• Issues: GitHub Issues
• Discussions: GitHub Discussions

Roadmap

• Add comprehensive test suite
• Support for options trading
• Portfolio management features
• Real-time market data integration
• Backtesting framework
• Web dashboard
• Enhanced order types (stop-loss, trailing stop, etc.)

---

Made with ❤️ by Jayaprakash Sundararaj