[[419168]]

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

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

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

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

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

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

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

概述

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

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

雖然最初似乎不清楚所有組件如何組合在一起，但在對以下類進(jìn)行編碼后，應(yīng)該開始了解 API 的構(gòu)建方式。

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

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

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

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

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

初始化

第一步是導(dǎo)入腳本中使用的必要庫組件。

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

# 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 服務(wù)器的響應(yīng)（描述為'errors'）。接口指定可以在派生子類中實(shí)現(xiàn)的方法。通過繼承這個類，我們可以覆蓋這些方法以適應(yīng)我們自己特定的數(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 的任務(wù)是創(chuàng)建一個 Python Queue隊(duì)列并將其附加一個名為_errors 的“私有”實(shí)例變量。該隊(duì)列將在整個類中用于存儲 IB 錯誤消息以供以后處理。

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

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

error 將提供的錯誤代碼與錯誤消息一起格式化為適當(dāng)?shù)淖址袷剑缓髮⑵浞湃?_errors 隊(duì)列。此方法用于在針對 API 執(zhí)行代碼時在控制臺上提供更好的調(diào)試信息。

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

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

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

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

IBAPIClient 類

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

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

# 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)建一個隊(duì)列來保存來自服務(wù)器的時間戳消息。然后我們調(diào)用原生 EClient 方法 reqCurrentTime 從服務(wù)器獲取時間。

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

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

最后我們返回服務(wù)器時間。

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

IBAPIApp

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

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

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

下一部分代碼初始化應(yīng)用程序所需的各種線程?？蛻舳藢?shí)例有一個線程，另一個用于將響應(yīng)消息添加到各個隊(duì)列。

最后調(diào)用 init_error 方法開始監(jiān)聽 IB 響應(yīng)。 

# 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)建腳本入口點(diǎn)了。

執(zhí)行代碼

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

然后我們使用適當(dāng)?shù)倪B接參數(shù)實(shí)例化一個應(yīng)用程序?qū)嵗?/p>

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

最后我們斷開與IB服務(wù)器的連接并結(jié)束程序。 

# 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.") 

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

python ib_api_connection.py 

您應(yīng)該會看到類似于以下內(nèi)容的輸出： 

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'。這些實(shí)際上是說明與各種服務(wù)器的連接正常運(yùn)行的響應(yīng)。也就是說，它們不是'errors'！

服務(wù)器時間也從 Unix 時間戳正確轉(zhuǎn)換為更易讀的格式和輸出。在此階段，應(yīng)用程序斷開連接。

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

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

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