[[419168]]

Python中文社區(qū) （ID:python-china）

 在本文中，我們將介紹如何為 Interactive Brokers Native Python API 提供的 EClient 和 EWrapper 類派生子類。然后，我們將提供端到端連接測試腳本，以確保我們能夠與 IB 進行對話。

盈透證券（Interactive Brokers）一直是受交易者歡迎的經(jīng)紀商。最初，這可能部分歸因于 IB 提供了一個應用程序編程接口 (API)，允許量化交易者獲取市場數(shù)據(jù)并直接在代碼中進行交易。許多相互競爭的經(jīng)紀商花了一些時間來開發(fā)自己的 API，使 IB 在零售量化交易領域獲得了合理的先發(fā)優(yōu)勢。

雖然最初的 IB API 以接口復雜而著稱，但近年來隨著 IB Native Python API 庫的發(fā)布，情況發(fā)生了變化。

在這個新系列文章中，我們將使用 ibapi 庫，來解釋如何通過“原生 Python”接口與Interactive Brokers API交互。

最終我們將學習如何請求市場數(shù)據(jù)、定義合同和處理訂單。本文將重點介紹接口本身并測試基本連接。

本文假設您有一個可運行的 Python 虛擬環(huán)境（例如 Anaconda 個人版）并已成功將 IB Python API 安裝到該環(huán)境中。安裝說明是特定于操作系統(tǒng)的?？梢栽贗nteractive Brokers API 站點上找到最新的說明。

概述

IB API 通過異步'request-response'模型進行工作。消息通過客戶端類發(fā)送到 IB 服務器（通過 Trader Workstation 或IB Gateway），而響應（稱為“errors”）由包裝類單獨處理。

大部分內部連接處理都通過 Python API 從最終用戶那里抽取出來，允許最少必要的'boilerplate'代碼進行連接。但是，連接到 IB 的原始遺留機制仍然部分地影響了 API 的設計。因此，對于那些不習慣面向對象設計原則的人來說，這可能會令人困惑。

雖然最初似乎不清楚所有組件如何組合在一起，但在對以下類進行編碼后，應該開始了解 API 的構建方式。

為了連接到 IB API，需要對四個主要組件進行編碼。

第一個是 IB API EWrapper 類的派生子類。 EWrapper 用于處理來自 IB 服務器的所有響應（'errors'）。

第二個是IB API EClient 類的派生子類。 EClient 用于將所有消息發(fā)送到 IB 服務器。

第三個是從 EWrapper 和 EClient 派生的子類的多重繼承類，用作 IB API 應用程序的基礎，它將所有通訊聯(lián)系在一起。

最后會有一個 if __name__ == "__main__": 入口點，旨在允許從命令行執(zhí)行腳本。

初始化

第一步是導入腳本中使用的必要庫組件。

我們將需要 IB API EWrapper 和 EClient 類，這將在下面描述。我們還需要分別來自線程 和隊列庫的Thread和Queue標準庫組件。最后，我們將導入 datetime 以將 Unix 時間戳轉換為更易讀的格式： 

# ib_api_connection.py  
import datetime  
import queue  
import threading  
from ibapi.client import EClient  
from ibapi.wrapper import EWrapper 

我們現(xiàn)在可以定義 IBAPIWrapper 類。

IBAPIWrapper 類

EWrapper 類提供了一個接口處理來自 IB 服務器的響應（描述為'errors'）。接口指定可以在派生子類中實現(xiàn)的方法。通過繼承這個類，我們可以覆蓋這些方法以適應我們自己特定的數(shù)據(jù)處理方法。

讓我們首先創(chuàng)建 EWrapper 的 IBAPIWrapper 派生子類并覆蓋一些方法。顯示此組件的完整代碼段,并將依次介紹每種方法： 

# ib_api_connection.py  
class IBAPIWrapper(EWrapper):  
    """  
    A derived subclass of the IB API EWrapper interface  
    that allows more straightforward response processing 
    from the IB Gateway or an instance of TWS.  
    """  
    def init_error(self):  
        """  
        Place all of the error messages from IB into a  
        Python queue, which can be accessed elsewhere.  
        """  
        error_queue = queue.Queue()  
        self._errors = error_queue  
    def is_error(self):  
        """  
        Check the error queue for the presence  
        of errors. 
        Returns  
        -------  
        `boolean`  
            Whether the error queue is not empty  
        """  
        return not self._errors.empty()  
    def get_error(self, timeout=5):  
        """  
        Attempts to retrieve an error from the error queue,  
        otherwise returns None.  
        Parameters  
        ----------  
        timeout : `float`  
            Time-out after this many seconds.  
        Returns  
        -------  
        `str` or None  
            A potential error message from the error queue.  
        """  
        if self.is_error():  
            try:  
                return self._errors.get(timeouttimeout=timeout)  
            except queue.Empty:  
                return None  
        return None  
    def error(self, id, errorCode, errorString):  
        """  
        Format the error message with appropriate codes and  
        place the error string onto the error queue.  
        """  
        error_message = (  
            "IB Error ID (%d), Error Code (%d) with "  
            "response '%s'" % (id, errorCode, errorString)  
        )  
        self._errors.put(error_message)  
    def init_time(self):  
        """  
        Instantiates a new queue to store the server  
        time, assigning it to a 'private' instance  
        variable and also returning it.  
        Returns  
        -------  
        `Queue`  
            The time queue instance.  
        """  
        time_queue = queue.Queue()  
        self._time_queue = time_queue  
        return time_queue  
    def currentTime(self, server_time):  
        """  
        Takes the time received by the server and  
        appends it to the class instance time queue.  
        Parameters  
        ----------  
        server_time : `str`  
            The server time message.  
        """  
        self._time_queue.put(server_time) 

init_error 的任務是創(chuàng)建一個 Python Queue隊列并將其附加一個名為_errors 的“私有”實例變量。該隊列將在整個類中用于存儲 IB 錯誤消息以供以后處理。

is_error 是一個簡單的方法，用于判斷_errors 隊列是否為空。

get_error嘗試從隊列中檢索錯誤消息，規(guī)定的超時時間以秒為單位。如果隊列為空或超時，則該方法不會返回任何內容。

error 將提供的錯誤代碼與錯誤消息一起格式化為適當?shù)淖址袷?，然后將其放?_errors 隊列。此方法用于在針對 API 執(zhí)行代碼時在控制臺上提供更好的調試信息。

這四種方法完成了對來自盈透證券的響應（'errors'）的處理。需要注意的是，ibapi 庫中有很多內部機器執(zhí)行此處理。從我們的派生子類中無法直接看到大部分工作。

其余兩個方法 init_time 和 currentTime 用于執(zhí)行連接“健全性檢查”（'sanity check'）。確定我們是否連接成功的一種簡單方法是檢索 IB 服務器上的本地時間。

這兩種方法只是創(chuàng)建一個新隊列來存儲服務器時間消息，并在請求時將新時間消息放置到這個隊列中。

我們的 EWrapper 簡單子類到此結束。我們現(xiàn)在能夠處理來自 IB 服務器的某些響應。下一個任務是實際向 IB 服務器發(fā)送消息。為此，我們需要覆蓋 EClient 類。

IBAPIClient 類

EClient的 IBAPIClient 派生子類用于向 IB 服務器發(fā)送消息。

需要特別注意的是，我們派生子類的構造函數(shù) __init__方法接受一個包裝參數(shù)，然后將其傳遞給EClient父構造函數(shù)。這意味著在 IBAPIClient類中沒有覆蓋本地 IB API 方法。相反，我們提供了包裝器實例（從 IBAPIWrapper實例化）來處理響應。 

# ib_api_connection.py  
class IBAPIClient(EClient):  
    """  
    Used to send messages to the IB servers via the API. In this  
    simple derived subclass of EClient we provide a method called  
    obtain_server_time to carry out a 'sanity check' for connection  
    testing.  
    Parameters  
    ----------  
    wrapper : `EWrapper` derived subclass  
        Used to handle the responses sent from IB servers  
    """ 
    MAX_WAIT_TIME_SECONDS = 10  
    def __init__(self, wrapper):  
        EClient.__init__(self, wrapper)  
    def obtain_server_time(self):  
        """  
        Requests the current server time from IB then  
        returns it if available.  
        Returns  
        -------  
        `int`  
            The server unix timestamp.  
        """  
        # Instantiate a queue to store the server time  
        time_queue = self.wrapper.init_time()  
        # Ask IB for the server time using the EClient method  
        self.reqCurrentTime()  
        # Try to obtain the latest server time if it exists  
        # in the queue, otherwise issue a warning  
        try:  
            server_time = time_queue.get(  
                timeout=IBAPIClient.MAX_WAIT_TIME_SECONDS  
            )  
        except queue.Empty:  
            print(  
                "Time queue was empty or exceeded maximum timeout of "  
                "%d seconds" % IBAPIClient.MAX_WAIT_TIME_SECONDS  
            )  
            server_time = None   
        # Output all additional errors, if they exist  
        while self.wrapper.is_error():  
            print(self.get_error())  
        return server_time 

在obtain_server_time中，我們首先創(chuàng)建一個隊列來保存來自服務器的時間戳消息。然后我們調用原生 EClient 方法 reqCurrentTime 從服務器獲取時間。

隨后，我們在 try...except 塊中包裝了一個從時間隊列中獲取值的調用。我們提供10秒的超時時間。如果超時或隊列為空，我們將服務器時間設置為None。

我們運行一個 while 循環(huán)來檢查 EWrapper 派生子類中定義的錯誤隊列中的任何其他響應。如果它們存在，它們將打印到控制臺。

最后我們返回服務器時間。

下一階段是創(chuàng)建一種機制來實例化 IBAPIWrapper和 IBAPIClient，以及實際連接到 IB 服務器。

IBAPIApp

本文中要派生的最后一個類是 IBAPIApp 類。

此類利用多重繼承從 IBAPIWrapper 和 IBAPIClient 類繼承。在初始化時，這兩個類也被初始化。但是，請注意 IBAPIClient 類將 wrapper=self 作為初始化關鍵字參數(shù)，因為 IBAPIApp也是從 IBAPIWrapper 派生的。

在初始化兩個父類之后，使用適當?shù)倪B接參數(shù)調用connect原生方法。

下一部分代碼初始化應用程序所需的各種線程?？蛻舳藢嵗幸粋€線程，另一個用于將響應消息添加到各個隊列。

最后調用 init_error 方法開始監(jiān)聽 IB 響應。 

# ib_api_connection.py  
class IBAPIApp(IBAPIWrapper, IBAPIClient):  
    """  
    The IB API application class creates the instances  
    of IBAPIWrapper and IBAPIClient, through a multiple  
    inheritance mechanism.  
    When the class is initialised it connects to the IB  
    server. At this stage multiple threads of execution  
    are generated for the client and wrapper.  
    Parameters  
    ----------  
    ipaddress : `str`  
        The IP address of the TWS client/IB Gateway  
    portid : `int`  
        The port to connect to TWS/IB Gateway with  
    clientid : `int`  
        An (arbitrary) client ID, that must be a positive integer  
    """  
    def __init__(self, ipaddress, portid, clientid):  
        IBAPIWrapper.__init__(self)  
        IBAPIClient.__init__(self, wrapper=self)  
        # Connects to the IB server with the  
        # appropriate connection parameters  
        self.connect(ipaddress, portid, clientid)  
        # Initialise the threads for various components  
        thread = threading.Thread(target=self.run)  
        thread.start()  
        setattr(self, "_thread", thread)  
        # Listen for the IB responses  
        self.init_error() 

現(xiàn)在定義了前三個類，我們就可以創(chuàng)建腳本入口點了。

執(zhí)行代碼

我們首先設置連接參數(shù)，包括主機 IP 地址、連接到 TWS/IB 網(wǎng)關的端口和（任意）正整數(shù)客戶端 ID。

然后我們使用適當?shù)倪B接參數(shù)實例化一個應用程序實例。

我們使用該應用程序從 IB 獲取服務器時間，然后使用 datetime 庫的 utcfromtimestamp 方法將 Unix 時間戳適當?shù)馗袷交癁楦呖勺x性的日期格式。

最后我們斷開與IB服務器的連接并結束程序。 

# ib_api_connection.py  
if __name__ == '__main__':  
    # Application parameters  
    host = '127.0.0.1'  # Localhost, but change if TWS is running elsewhere  
    port = 7497  # Change to the appropriate IB TWS account port number  
    client_id = 1234  
    print("Launching IB API application...")  
    # Instantiate the IB API application  
    app = IBAPIApp(host, port, client_id)  
    print("Successfully launched IB API application...")  
    # Obtain the server time via the IB API app  
    server_time = app.obtain_server_time()  
    server_time_readable = datetime.datetime.utcfromtimestamp(  
        server_time  
    ).strftime('%Y-%m-%d %H:%M:%S')  
    print("Current IB server time: %s" % server_time_readable)  
    # Disconnect from the IB server  
    app.disconnect()  
    print("Disconnected from the IB API application. Finished.") 

在這個階段，我們準備運行 ib_api_connection.py。只需導航到您存儲文件的目錄，確保帶有 ibapi的虛擬環(huán)境處于活動狀態(tài)，并且 TWS（或 IB 網(wǎng)關）已正確加載和配置，然后鍵入以下內容：

python ib_api_connection.py 

您應該會看到類似于以下內容的輸出： 

Launching IB API application...  
Successfully launched IB API application...  
IB Error ID (-1), Error Code (2104) with response 'Market data farm connection is OK:usfarm'  
IB Error ID (-1), Error Code (2106) with response 'HMDS data farm connection is OK:ushmds'  
IB Error ID (-1), Error Code (2158) with response 'Sec-def data farm connection is OK:secdefnj'  
Current IB server time: 2020-07-29 13:27:18  
Disconnected from the IB API application. Finished.  
unhandled exception in EReader thread  
Traceback (most recent call last):  
  File "/home/mhallsmoore/venv/qstrader/lib/python3.6/site-packages/ibapi/reader.py", line 34, in run  
    data = self.conn.recvMsg() 
   File "/home/mhallsmoore/venv/qstrader/lib/python3.6/site-packages/ibapi/connection.py", line 99, in recvMsg  
    buf = self._recvAllMsg()  
  File "/home/mhallsmoore/venv/qstrader/lib/python3.6/site-packages/ibapi/connection.py", line 119, in _recvAllMsg  
    buf = self.socket.recv(4096)  
OSError: [Errno 9] Bad file descriptor 

第一組輸出是帶有代碼 2104、2106 和 2158 的 IB 'errors'。這些實際上是說明與各種服務器的連接正常運行的響應。也就是說，它們不是'errors'！

服務器時間也從 Unix 時間戳正確轉換為更易讀的格式和輸出。在此階段，應用程序斷開連接。

但是請注意，在 EReader 線程中引發(fā)了 OSError 異常。這是 IB API 本身的一個內部問題，目前還沒有一個修復程序。出于本教程的目的，它可以被忽略。

現(xiàn)在完成了連接到 IB Python Native API 的教程。 ib_api_connection.py 的完整代碼請掃描下方二維碼獲取。

我們已經(jīng)成功連接到IB服務器，并通過調用檢查連接，獲取當前服務器時間。后面我們將確定如何從 IB API 檢索股票的市場數(shù)據(jù)。