Ib integration

qstrader/price_handler/ib_bar.py

@@ -0,0 +1,115 @@
+import os
+import datetime
+import queue
+from .base import AbstractBarPriceHandler
+
+from qstrader.service.ib import IBService
+
+from ibapi.client import EClient
+from ibapi.wrapper import EWrapper
+from ibapi.utils import iswrapper
+from ibapi.contract import *
+from ibapi.common import *
+
+
+
+#types
+from ibapi.utils import (current_fn_name, BadMessage)
+from ibapi.common import *
+from ibapi.order_condition import *
+from ibapi.contract import *
+from ibapi.order import *
+from ibapi.order_state import *
+from ibapi.execution import Execution
+from ibapi.execution import ExecutionFilter
+from ibapi.commission_report import CommissionReport
+from ibapi.scanner import ScannerSubscription
+from ibapi.ticktype import *
+
+from ibapi.account_summary_tags import *
+
+
+class IBBarPriceHandler(AbstractBarPriceHandler):
+    """
+    Designed to feed either live or historic market data bars
+    from an Interactive Brokers connection.
+
+    Uses the IBService to make requests and collect data once responses have returned.
+
+    TODO:
+        * Historic/Live mode to be set by whether QSTrader is in Backtest or Live mode
+        * IBService should be an initialization parameter
+        * Ports, etc, connection strings from config
+    """
+    def __init__(
+        self, events_queue, param_tickers, settings, mode="historic",
+        hist_end_date = datetime.datetime.now() - datetime.timedelta(days=3),
+        hist_duration="5 D", hist_barsize="1 min"
+    ):
+        self.ib_service = IBService()
+        self.ib_service.connect("127.0.0.1",4001,0)
+        self.mode = mode
+        self.continue_backtest = True
+        self.hist_end_date = hist_end_date
+        self.hist_duration = hist_duration
+        self.hist_barsize = hist_barsize
+
+        # The position of a ticker in this dict is used as its IB ID.
+        self.tickers = {} # TODO gross
+        self.ticker_lookup = {}
+
+        for ticker in param_tickers:  # TODO gross param_tickers -- combine above?
+            self._subscribe_ticker(ticker)
+
+        self._wait_for_hist_population()
+
+
+    def _subscribe_ticker(self, ticker):
+        """
+        Request ticker data from IB
+        """
+        if ticker not in self.tickers:
+            # Set up an IB ContractS
+            contract = Contract()
+            contract.exchange = "SMART"
+            contract.symbol = ticker
+            contract.secType = "STK"
+            contract.currency = "AUD"
+
+        if self.mode == "historic":
+            ib_ticker_id = len(self.tickers)
+            end_time = datetime.datetime.strftime(self.hist_end_date, "%Y%m%d 17:00:00")
+            self.ib_service.reqHistoricalData(
+                ib_ticker_id, contract, end_time, self.hist_duration, self.hist_barsize,
+                "TRADES", True, 2, None)
+
+        # TODO gross
+        self.ticker_lookup[len(self.tickers)] = ticker
+        self.tickers[ticker] = {}
+
+
+    def _wait_for_hist_population(self):
+        """
+        # TODO this *needs* to be an infinite loop running inside the service,
+        # with the service launched on a new Thread or Process
+        """
+        while (self.ib_service.conn.isConnected() or not self.ib_service.msg_queue.empty()) and len(self.ib_service.waitingHistoricalData) != 0:
+            try:
+                text = self.ib_service.msg_queue.get(block=True, timeout=0.2)
+                if len(text) > MAX_MSG_LEN:
+                    self.ib_service.wrapper.error(NO_VALID_ID, BAD_LENGTH.code(),
+                        "%s:%d:%s" % (BAD_LENGTH.msg(), len(text), text))
+                    self.ib_service.disconnect()
+                    break
+            except queue.Empty:
+                print("queue.get: empty")
+            else:
+                fields = comm.read_fields(text)
+                print("fields %s", fields)
+                self.ib_service.decoder.interpret(fields)
+
+            print("conn:%d queue.sz:%d",
+                         self.ib_service.conn.isConnected(),
+                         self.ib_service.msg_queue.qsize())
+
+        self.ib_service.disconnect()

qstrader/service/ib.py

@@ -0,0 +1,92 @@
+import sys
+import argparse
+import datetime
+import collections
+import inspect
+import queue
+
+import logging
+import time
+import os.path
+
+from ibapi.wrapper import EWrapper
+from ibapi.client import EClient
+from ibapi.utils import iswrapper
+
+#types
+from ibapi.utils import (current_fn_name, BadMessage)
+from ibapi.common import *
+from ibapi.order_condition import *
+from ibapi.contract import *
+from ibapi.order import *
+from ibapi.order_state import *
+from ibapi.execution import Execution
+from ibapi.execution import ExecutionFilter
+from ibapi.commission_report import CommissionReport
+from ibapi.scanner import ScannerSubscription
+from ibapi.ticktype import *
+
+from ibapi.account_summary_tags import *
+
+
+class IBService(EWrapper, EClient):
+    """
+    The IBService is the primary conduit of data from QStrader to Interactive Brokers.
+    This service provides functions to request data, and allows for
+    callbacks to be triggered, which populates "data queues" with the response.
+
+    All methods of the EClient are available (i.e. API Requests), as are
+    the callbacks for EWrapper (i.e. API responses). It also provides a set of Queues
+    which are populated with the responses from EWrapper. Other components in the
+    system should use these queues collect the API response data.
+
+    Any module or component that wishes to interact with IB should do so by using
+    methods offered in this class. This ensures that the logic required to talk with IB
+    is contained within this class exclusively, with the added benefit that we
+    can easily create mock instances of the IBService for testing.
+    """
+    def __init__(self):
+        EWrapper.__init__(self)
+        EClient.__init__(self, wrapper=self)
+
+        self.historicalDataQueue = queue.Queue()
+        self.waitingHistoricalData = []
+
+
+    def error(self, reqId:TickerId, errorCode:int, errorString:str):
+        super().error(reqId, errorCode, errorString)
+        print("Error. Id: " , reqId, " Code: " , errorCode , " Msg: " , errorString)
+
+
+    """
+    Append `reqId` to waitingHistoricalData, then call the super method.
+    """
+    def reqHistoricalData(self, reqId:TickerId , contract:Contract, endDateTime:str,
+                          durationStr:str, barSizeSetting:str, whatToShow:str,
+                          useRTH:int, formatDate:int, chartOptions:TagValueList):
+        self.waitingHistoricalData.append(reqId)
+        print("REQUESTING HISTORIC, WAITING FOR %s" % len(self.waitingHistoricalData))
+        super().reqHistoricalData( reqId, contract, endDateTime,
+                                  durationStr, barSizeSetting, whatToShow,
+                                  useRTH, formatDate, chartOptions)
+
+
+    """
+    Populate the HistoricalData queue.
+    """
+    def historicalData(self, reqId:TickerId , date:str, open:float, high:float,
+                       low:float, close:float, volume:int, barCount:int,
+                        WAP:float, hasGaps:int):
+        print("HistoricalData. ", reqId, " Date:", date, "Open:", open,
+            "High:", high, "Low:", low, "Close:", close, "Volume:", volume,
+            "Count:", barCount, "WAP:", WAP, "HasGaps:", hasGaps)
+        self.historicalDataQueue.put((reqId, date, open, high, low, close,
+                                        volume, barCount, WAP, hasGaps))
+
+    """
+    Remove `reqId` from waitingHistoricalData
+    TODO: Will it work with multiple historical requests for same symbol?
+    """
+    def historicalDataEnd(self, reqId:int, start:str, end:str):
+        print("FINISHED FOR %s" % reqId)
+        self.waitingHistoricalData.remove(reqId)

tests/test_ib_price_handler.py

@@ -0,0 +1,27 @@
+import unittest
+
+from qstrader.price_handler.ib_bar import IBBarPriceHandler
+from qstrader.compat import queue
+from qstrader import settings
+
+class TestPriceHandlerSimpleCase(unittest.TestCase):
+    def setUp(self):
+        """
+        Set up the PriceHandler object with a small
+        set of initial tickers for a backtest in historic mode.
+        """
+        self.config = settings.TEST
+        fixtures_path = self.config.CSV_DATA_DIR
+        events_queue = queue.Queue()
+        init_tickers = ["CBA"]
+        self.price_handler = IBBarPriceHandler(
+            events_queue, init_tickers, self.config
+        )
+
+    def test(self):
+        self.assertEqual(1,2)
+
+
+
+if __name__ == "__main__":
+    unittest.main()