Add HistoricInteractiveBrokersClient

examples/live/interactive_brokers/historic_download.py

@@ -13,78 +13,65 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 # -------------------------------------------------------------------------------------------------
+import asyncio
+import datetime
 
-import pandas as pd
+from nautilus_trader.adapters.interactive_brokers.common import IBContract
+from nautilus_trader.adapters.interactive_brokers.historic import HistoricInteractiveBrokersClient
+from nautilus_trader.persistence.catalog import ParquetDataCatalog
 
-# fmt: off
-from nautilus_trader.adapters.interactive_brokers.config import InteractiveBrokersDataClientConfig
-from nautilus_trader.adapters.interactive_brokers.factories import InteractiveBrokersLiveDataClientFactory
-from nautilus_trader.adapters.interactive_brokers.factories import InteractiveBrokersLiveExecClientFactory
-from nautilus_trader.adapters.interactive_brokers.historic.bar_data import BarDataDownloader
-from nautilus_trader.adapters.interactive_brokers.historic.bar_data import BarDataDownloaderConfig
-from nautilus_trader.config import LoggingConfig
-from nautilus_trader.config import TradingNodeConfig
-from nautilus_trader.live.node import TradingNode
-from nautilus_trader.model.data import Bar
 
+async def main():
+    contract = IBContract(
+        secType="STK",
+        symbol="AAPL",
+        exchange="SMART",
+        primaryExchange="NASDAQ",
+    )
+    instrument_id = "TSLA.NASDAQ"
 
-# fmt: on
+    client = HistoricInteractiveBrokersClient(port=4002, client_id=5)
+    await client._connect()
+    await asyncio.sleep(2)
 
-# *** MAKE SURE YOU HAVE REQUIRED DATA SUBSCRIPTION FOR THIS WORK WORK AS INTENDED. ***
+    instruments = await client.request_instruments(
+        contracts=[contract],
+        instrument_ids=[instrument_id],
+    )
 
-df = pd.DataFrame()
+    bars = await client.request_bars(
+        bar_specifications=["1-HOUR-LAST", "30-MINUTE-MID"],
+        end_date_time=datetime.datetime(2023, 11, 6, 16, 30),
+        tz_name="America/New_York",
+        duration="1 D",
+        contracts=[contract],
+        instrument_ids=[instrument_id],
+    )
 
+    trade_ticks = await client.request_ticks(
+        "TRADES",
+        start_date_time=datetime.datetime(2023, 11, 6, 10, 0),
+        end_date_time=datetime.datetime(2023, 11, 6, 10, 1),
+        tz_name="America/New_York",
+        contracts=[contract],
+        instrument_ids=[instrument_id],
+    )
 
-# Data Handler for BarDataDownloader
-def do_something_with_bars(bars: list):
-    global df
-    bars_dict = [Bar.to_dict(bar) for bar in bars]
-    df = pd.concat([df, pd.DataFrame(bars_dict)])
-    df = df.sort_values(by="ts_init")
+    quote_ticks = await client.request_ticks(
+        "BID_ASK",
+        start_date_time=datetime.datetime(2023, 11, 6, 10, 0),
+        end_date_time=datetime.datetime(2023, 11, 6, 10, 1),
+        tz_name="America/New_York",
+        contracts=[contract],
+        instrument_ids=[instrument_id],
+    )
 
+    catalog = ParquetDataCatalog("./catalog")
+    catalog.write_data(instruments)
+    catalog.write_data(bars)
+    catalog.write_data(trade_ticks)
+    catalog.write_data(quote_ticks)
 
-# Configure the trading node
-config_node = TradingNodeConfig(
-    trader_id="TESTER-001",
-    logging=LoggingConfig(log_level="INFO"),
-    data_clients={
-        "InteractiveBrokers": InteractiveBrokersDataClientConfig(
-            ibg_host="127.0.0.1",
-            ibg_port=7497,
-            ibg_client_id=1,
-        ),
-    },
-    timeout_connection=90.0,
-)
 
-# Instantiate the node with a configuration
-node = TradingNode(config=config_node)
-
-# Configure your strategy
-downloader_config = BarDataDownloaderConfig(
-    start_iso_ts="2023-09-01T00:00:00+00:00",
-    end_iso_ts="2023-09-30T00:00:00+00:00",
-    bar_types=[
-        "AAPL.NASDAQ-1-MINUTE-BID-EXTERNAL",
-        "AAPL.NASDAQ-1-MINUTE-ASK-EXTERNAL",
-        "AAPL.NASDAQ-1-MINUTE-LAST-EXTERNAL",
-    ],
-    handler=do_something_with_bars,
-    freq="1W",
-)
-
-# Instantiate the downloader and add into node
-downloader = BarDataDownloader(config=downloader_config)
-node.trader.add_actor(downloader)
-
-# Register your client factories with the node (can take user defined factories)
-node.add_data_client_factory("InteractiveBrokers", InteractiveBrokersLiveDataClientFactory)
-node.add_exec_client_factory("InteractiveBrokers", InteractiveBrokersLiveExecClientFactory)
-node.build()
-
-# Stop and dispose of the node with SIGINT/CTRL+C
 if __name__ == "__main__":
-    try:
-        node.run()
-    finally:
-        node.dispose()
+    asyncio.run(main())

nautilus_trader/adapters/interactive_brokers/client/client.py

@@ -1089,6 +1089,7 @@ async def get_historical_bars(
         use_rth: bool,
         end_date_time: str,
         duration: str,
+        timeout: int = 60,
     ):
         name = str(bar_type)
         if not (request := self.requests.get(name=name)):
@@ -1114,7 +1115,7 @@ async def get_historical_bars(
             )
             self._log.debug(f"reqHistoricalData: {request.req_id=}, {contract=}")
             request.handle()
-            return await self._await_request(request, 20)
+            return await self._await_request(request, timeout)
         else:
             self._log.info(f"Request already exist for {request}")
 
@@ -1294,9 +1295,15 @@ async def get_historical_ticks(
         self,
         contract: IBContract,
         tick_type: str,
-        end_date_time: pd.Timestamp,
-        use_rth: bool,
+        start_date_time: pd.Timestamp | str = "",
+        end_date_time: pd.Timestamp | str = "",
+        use_rth: bool = True,
     ):
+        if isinstance(start_date_time, pd.Timestamp):
+            start_date_time = start_date_time.strftime("%Y%m%d %H:%M:%S %Z")
+        if isinstance(end_date_time, pd.Timestamp):
+            end_date_time = end_date_time.strftime("%Y%m%d %H:%M:%S %Z")
+
         name = (str(ib_contract_to_instrument_id(contract)), tick_type)
         if not (request := self.requests.get(name=name)):
             req_id = self._next_req_id()
@@ -1307,8 +1314,8 @@ async def get_historical_ticks(
                     self._client.reqHistoricalTicks,
                     reqId=req_id,
                     contract=contract,
-                    startDateTime="",
-                    endDateTime=end_date_time.strftime("%Y%m%d %H:%M:%S %Z"),
+                    startDateTime=start_date_time,
+                    endDateTime=end_date_time,
                     numberOfTicks=1000,
                     whatToShow=tick_type,
                     useRth=use_rth,
@@ -1318,13 +1325,19 @@ async def get_historical_ticks(
                 cancel=functools.partial(self._client.cancelHistoricalData, reqId=req_id),
             )
             request.handle()
-            return await self._await_request(request, 20)
+            return await self._await_request(request, 60)
         else:
             self._log.info(f"Request already exist for {request}")
 
-    def historicalTicksBidAsk(self, req_id: int, ticks: list, done: bool):
+    def historicalTicksBidAsk(
+        self,
+        req_id: int,
+        ticks: list,
+        done: bool,
+    ):  # : Override the EWrapper
         self.logAnswer(current_fn_name(), vars())
-
+        if not done:
+            return
         if request := self.requests.get(req_id=req_id):
             instrument_id = InstrumentId.from_str(request.name[0])
             instrument = self._cache.instrument(instrument_id)
@@ -1335,21 +1348,25 @@ def historicalTicksBidAsk(self, req_id: int, ticks: list, done: bool):
                     instrument_id=instrument_id,
                     bid_price=instrument.make_price(tick.priceBid),
                     ask_price=instrument.make_price(tick.priceAsk),
-                    bid_size=instrument.make_price(tick.sizeBid),
-                    ask_size=instrument.make_price(tick.sizeAsk),
+                    bid_size=instrument.make_qty(tick.sizeBid),
+                    ask_size=instrument.make_qty(tick.sizeAsk),
                     ts_event=ts_event,
                     ts_init=ts_event,
                 )
                 request.result.append(quote_tick)
 
             self._end_request(req_id)
 
-    def historicalTicksLast(self, req_id: int, ticks: list, done: bool):
+    def historicalTicksLast(self, req_id: int, ticks: list, done: bool):  # : Override the EWrapper
         self.logAnswer(current_fn_name(), vars())
+        if not done:
+            return
         self._process_trade_ticks(req_id, ticks)
 
-    def historicalTicks(self, req_id: int, ticks: list, done: bool):
+    def historicalTicks(self, req_id: int, ticks: list, done: bool):  # : Override the EWrapper
         self.logAnswer(current_fn_name(), vars())
+        if not done:
+            return
         self._process_trade_ticks(req_id, ticks)
 
     def _process_trade_ticks(self, req_id: int, ticks: list):

nautilus_trader/adapters/interactive_brokers/common.py

@@ -85,7 +85,7 @@ class IBContract(NautilusConfig, frozen=True, repr_omit_defaults=True):
         Exchange where security is traded. Will be SMART for Stocks.
     primaryExchange: str
         Exchange where security is registered. Applies to Stocks.
-    localSymbol: str
+    symbol: str
         Unique Symbol registered in Exchange.
     build_options_chain: bool (default: None)
         Search for full option chain

nautilus_trader/adapters/interactive_brokers/data.py

@@ -376,8 +376,8 @@ async def _handle_ticks_request(
             ticks_part = await self._client.get_historical_ticks(
                 contract,
                 tick_type,
-                end,
-                self._use_regular_trading_hours,
+                end_date_time=end,
+                use_rth=self._use_regular_trading_hours,
             )
             if not ticks_part:
                 break

nautilus_trader/adapters/interactive_brokers/gateway.py

@@ -45,14 +45,14 @@ class InteractiveBrokersGateway:
     A class to manage starting an Interactive Brokers Gateway docker container.
     """
 
-    IMAGE: ClassVar[str] = "ghcr.io/unusualalpha/ib-gateway:stable"
+    IMAGE: ClassVar[str] = "ghcr.io/unusualalpha/ib-gateway:10.19"
     CONTAINER_NAME: ClassVar[str] = "nautilus-ib-gateway"
     PORTS: ClassVar[dict[str, int]] = {"paper": 4002, "live": 4001}
 
     def __init__(
         self,
-        username: str,
-        password: str,
+        username: str | None = None,
+        password: str | None = None,
         host: str | None = "localhost",
         port: int | None = None,
         trading_mode: str | None = "paper",

nautilus_trader/adapters/interactive_brokers/historic/__init__.py

@@ -12,3 +12,12 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 # -------------------------------------------------------------------------------------------------
+# fmt: off
+from nautilus_trader.adapters.interactive_brokers.historic.client import HistoricInteractiveBrokersClient
+
+
+# fmt: on
+
+__all__ = [
+    "HistoricInteractiveBrokersClient",
+]