ib-async: Interactive Brokers API Wrapper

2.1.0 · active · verified Thu Apr 16

ib-async is an asynchronous and synchronous Python framework for the Interactive Brokers API, providing a user-friendly interface to TWS/Gateway for trading, market data, and account management. It is a fork/continuation of the popular ib_insync library. The current version is 2.1.0, with releases occurring semi-regularly, focusing on bug fixes and usability improvements.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to connect to the Interactive Brokers TWS/Gateway, define a stock contract, and retrieve 1-minute historical trade data for the last 10 minutes using the `IB.run()` method to manage the asyncio event loop. Ensure TWS/Gateway is running and API access is enabled.

from ib_insync import IB, Contract, util
import asyncio

async def main():
    ib = IB()
    try:
        # Connect to TWS/Gateway. Default is localhost:7497 for TWS, 4002 for Gateway.
        # Change host/port if TWS/Gateway is running elsewhere.
        await ib.connect('127.0.0.1', 7497, clientId=1)

        print("Connected to IB TWS/Gateway")

        # Define a contract
        contract = Contract(symbol='SPY', secType='STK', exchange='SMART', currency='USD')

        # Request market data
        # Ensure you have subscribed to market data for SPY in TWS/Gateway
        # Data from reqMktData is real-time, consider historical data for backtesting
        # for data in ib.reqMktData(contract):
        #    print(f"Time: {data.time}, Bid: {data.bid}, Ask: {data.ask}")

        # Request historical data for the last 10 minutes
        bars = await ib.reqHistoricalData(
            contract, 
            endDateTime='', 
            durationStr='10 T', 
            barSizeSetting='1 min', 
            whatToShow='TRADES', 
            useRTH=True
        )

        if bars:
            print(f"Retrieved {len(bars)} historical bars for {contract.symbol}")
            for bar in bars:
                print(f"Time: {bar.date}, Open: {bar.open}, Close: {bar.close}, Volume: {bar.volume}")
        else:
            print(f"No historical data found for {contract.symbol}")

    except ConnectionRefusedError:
        print("Connection refused. Is TWS/Gateway running and configured for API access?")
    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        if ib.isConnected():
            ib.disconnect()
            print("Disconnected from IB TWS/Gateway")

# ib-async recommends using IB.run() for managing the event loop for simpler cases
# or asyncio.run() for more complex applications.

# To run the async main function:
if __name__ == '__main__':
    # Use IB.run() for simple script execution without manually managing asyncio loop
    IB.run(main())
    # Alternatively, for more control over the asyncio event loop:
    # asyncio.run(main())

view raw JSON →