在本文中，我们将介绍如何派生InteractiveBrokersNativePythonAPI提供的EClient和EWrapper类的子类。然后我们将提供端到端连接测试脚本以确保我们可以与IB通信。盈透证券一直是交易者中受欢迎的经纪商。最初，这可能部分是由于IB提供了一个应用程序编程接口(API)，允许量化交易者访问市场数据并直接在代码中进行交易。许多竞争经纪商花时间开发自己的API，使IB在散户量化交易领域具有合理的先发优势。虽然最初的IBAPI以其复杂的接口而闻名，但近年来随着IBNativePythonAPI库的发布，这种情况发生了变化。在这个新的系列文章中，我们将使用ibapi库来解释如何通过“本机Python”界面与InteractiveBrokersAPI进行交互。最后，我们将学习如何请求市场数据、定义合同和处理订单。本文将重点介绍接口本身并测试基本连接性。本文假设您有一个工作的Python虚拟环境（例如Anaconda个人版）并且已经成功地在其中安装了IBPythonAPI。安装说明是特定于操作系统的。最新的说明可以在盈透证券API网站上找到。概述IBAPI通过异步“请求-响应”模型工作。消息通过客户端类发送到IB服务器（通过交易者工作站或IB网关），而响应（称为“错误”）由包装类单独处理。大多数内部连接处理是通过PythonAPI从最终用户那里抽象出来的，允许最少的必要“样板”代码来执行连接。然而，连接IB的原始遗留机制仍然部分影响了API的设计。所以对于那些不习惯面向对象设计原则的人来说可能会感到困惑。虽然最初似乎不清楚所有组件是如何组合在一起的，但在编写以下类之后，您应该开始了解API是如何构建的。为了连接到IBAPI，需要对四个主要组件进行编码。第一个是IBAPIEWrapper类的派生子类。EWrapper用于处理来自IB服务器的所有响应（“错误”）。第二个是IBAPIEClient类的派生子类。EClient用于将所有消息发送到IB服务器。第三个是多重继承类，其子类派生自EWrapper和EClient，用作IBAPI应用程序的基础，它将所有通信联系在一起。末尾会有一个if__name__=="__main__":入口点，旨在允许从命令行执行脚本。初始化的第一步是导入脚本中使用的必要库组件。我们将需要IBAPIEWrapper和EClient类，如下所述。我们还需要分别来自Thread和Queue库的Thread和Queue标准库组件。最后，我们将导入日期时间以将Unix时间戳转换为更易读的格式：#ib_api_connection.pyimportdatetimeimportqueueimportthreadingfromibapi.clientimportEClientfromibapi.wrapperimportEWrapper我们现在可以定义IBAPIWrapper类。IBAPIWrapper类EWrapper类提供了一个接口来处理来自IB服务器的响应（描述为“错误”）。接口指定可以在派生子类中实现的方法。通过继承这个类，我们可以重写这些方法来适应我们自己特定的数据处理方式。让我们首先创建一个EWrapper的IBAPIWrapper派生子类并覆盖一些方法。显示此组件的完整代码段,并将依次介绍每种方法：#ib_api_connection.pyclassIBAPIWrapper(EWrapper):"""AderivedsubclassoftheIBAPIEWrapperinterfacethatallowsmorestraightforwardresponseprocessingfromtheIBGatewayoraninstanceofTWS."""definit_error(self):"""PlacealloftheerrormessagesfromIBintoaPythonqueue,whichcanbeaccessedelsewhere."""error_queue=queue.Queue()self._errors=error_queuedefis_error(self):"""Checktheerrorqueueforthepresenceoferrors.Returns-------`boolean`Whethertheerrorqueueisnotempty"""returnnotself._errors.empty()defget_error(self,timeout=5):"""Attemptstoretrieanerrorfromtheerrorqueue,otherwisereturnsNone.returnself._errors.get(timeouttimeout=timeout)exceptqueue.Empty:returnNonereturnNonedeferror(self,id,errorCode,errorString):"""用适当的代码格式化错误消息并将错误字符串放入错误队列。"""error_message=("IBErrorID(%d),ErrorCode(%d)with""response'%s'"%(id,errorCode,errorString))self._errors.put(error_message)definit_time(self):"""实例化一个新队列来存储服务器时间，将其分配给一个'私有'实例变量并返回它。返回------`Queue`Thetimequeueinstance.""""time_queue=queue.Queue()self._time_queue=time_queuereturntime_queuedefcurrentTime(self,server_time):"""取服务器接收到的时间追加到类实例时间队列。参数-----------server_time:`str`服务器时间消息。"""self._time_queue.put(server_time)init_error任务是创建一个Python队列并向其附加一个名为_errors的“私有”实例变量.这个队列会在整个类中用来存放IB的错误信息，以便后面的处理。is_error是一个简单的方法，用来判断_errors队列是否为空。get_error尝试从队列中检索错误消息，并指定超时时间（以秒为单位）。如果队列为空或超时，则该方法不返回任何内容。error将提供的错误代码与错误消息一起格式化为适当的字符串格式，并将其放入_errors队列中。此方法用于在针对API执行代码时在控制台上提供更好的调试信息。这四种方法完成了对来自盈透证券的响应（“错误”）的处理。请注意，ibapi库中有很多内部机制可以执行此处理。大多数工作不能直接从我们的派生子类中看到。剩下的两个方法，init_time和currentTime，用于执行连接的“健全性检查”（'sanitycheck'）。确定我们是否连接成功的一种简单方法是检索IB服务器上的本地时间。这两个方法只是创建一个新的队列来存储服务器时间消息，并在请求时将新的时间消息放入这个队列中。EWrapper的简单子类到此结束。我们现在能够处理来自IB服务器的一些响应。下一个任务是将消息实际发送到IB服务器。为此，我们需要覆盖EClient类。IBAPIClient类EClient的IBAPIClient派生子类用于向IB服务器发送消息。重要的是要注意，派生子类的构造函数__init__方法接受一个包装器参数，然后将其传递给EClient父构造函数。这意味着本地IBAPI方法不会在IBAPIClient类中被覆盖。相反，我们提供了一个包装器实例（从IBAPIWrapper实例化）来处理响应。#ib_api_connection.pyclassIBAPIClient(EClient):"""UsedtosendmessagestotheIBserversviatheAPI.InthissimplederivedsubclassofEClientweprovideamethodcalledobtain_server_timetocarryouta'sanitycheck'forconnectiontesting.Parameters----------wrapper:`EWrapper`derivedsubclassUsedtohandletheresponsessentfromIBservers"""MAX_WAIT_TIME_SECONDS=10def__init__(self,wrapper):EClient.__init__(self,wrapper)defobtain_server_time(self):"""从IB请求当前服务器时间，然后返回它是否可用。返回------`int`服务器unix时间戳。"""#Instantiateaqueuetostoretheservertimetime_queue=self.wrapper.init_time()#AskIBfortheservertimeusingtheEClientmethodself.reqCurrentTime()#strytimetoifitexistslate#inthequeue,otherwiseissueawarningtry:server_time=time_queue.get(timeout=IBAPIClient.MAX_WAIT_TIME_SECONDS)exceptqueue.Empty:print("Timequeuewasemptyorexceededmaximumtimeoutof""%dseconds"%IBAPIClient.MAX_WAIT_TIME_SECONDS)server_time=None#Outputalladditionalerrors,iftheyexistwhileself.wrapper.is_error():print(self.get_error())returnserver_time在obtain_server_time中，我们首先创建一个队列来保存来自服务器的带时间戳的消息，然后我们调用本机EClient方法reqCurrentTime从服务器获取时间。然后我们在try...except块中包装一个调用以从时间队列获取值。我们提供10秒的超时。如果超时或队列为空，我们将服务器发送的时间设置为None。我们运行一个while循环来检查EWrapper派生子类中定义的错误队列中是否有任何其他响应。如果存在，则将它们打印到控制台。最后我们返回服务器下一个阶段是创建一个A机制来实例化IBAPIWrapper和IBAPIClient，并真正连接到IB服务器IBAPIApp本文最后要派生的类是IBAPIApp类，该类继承自IBAPIWrapper和IBAPIClient类使用多重继承。在初始化的时候，这两个类也被初始化。但是，请注意，IBAPIClient类使用wrapper=self作为初始化关键字参数，因为IBAPIApp也是从IBAPIWrapper派生的。初始化两个父类后，使用适当的连接参数调用connect本机方法。代码的下一部分初始化应用程序所需的各种线程。一个线程用于客户端实例，另一个线程用于向各种队列添加响应消息。最后，调用init_error方法开始监听IB响应。#ib_api_connection.pyclassIBAPIApp(IBAPIWrapper,IBAPIClient):"""TheIBAPIapplicationclasscreatestheinstancesofIBAPIWrapperandIBAPIClient,throughamultipleinheritancemechanism.WhentheclassisinitialiseditconnectstotheIBserver.Atthisstagemultiplethreadsofexecutionaregeneratedfortheclientandwrapper.Parameters----------ipaddress:`str`TheIPaddressoftheTWSclient/IBGatewayportid:`int`TheporttoconnecttoTWS/IBGatewaywithclientid:`int`（任意）clientID，必须是正整数"""def__init__（self，ipaddress，portid，clientid）：IBAPIWrapper.__init__（self）IBAPIClient.__init__（self，wrapper=self）#ConnectstotheIBserverwiththe#appropriateconnectionparametersself.connect（ipaddress，portid，clientid)#Initialisethethreadsforvariouscomponentsthread=threading.Thread(target=self.run)thread.start()setattr(self,"_thread",thread)#ListenfortheIBresponsesself.init_error()现在确定了前面三个类，我们就可以创建脚入口点了。执行代码，我们首先设置连接参数，包括主机IP地址、连接到TWS/IB网关的端口和（任意）正整数客户端ID。然后，我们使用适当的连接参数实例化一个应用程序实例。我们使用应用程序从IB获取服务器时间，然后使用datetime库的utcfromtimestamp方法将Unix时间戳正确格式化为更易读的日期格式。最后我们断开与IB服务器的连接并结束程序。#ib_api_connection.pyif__name__=='__main__':#Applicationparametershost='127.0.0.1'#Localhost,butchangeifTWSisrunningelsewhereport=7497#ChangetotheappropriateIBTWSaccountportnumberclient_id=1234print("LaunchingIBAPIapplication...")#InstantiatetheIBAPIapplication...成功启动IBAPIapplication...")#通过IBAPI获取服务器时间appserver_time=app.obtain_server_time()server_time_readable=datetime.datetime.utcfromtimestamp(server_time).strftime('%Y-%m-%d%H:%M:%S')print("CurrentIBservertime:%s"%server_time_readable)#DisconnectfromtheIBserverapp.disconnect()print("DisconnectedfromtheIBAPIapplication.Finished.")在此阶段，我们已准备好运行ib_api_connection.py。只需导航到您存储文件的目录，确保带有ibapi的虚拟环境处于活动状态，并且TWS（或IB网关）已正确加载和配置，然后键入以下内容：pythonib_api_connection.py您应该会看到类似内容输出的内容：正在启动IBAPI应用程序...已成功启动IBAPI应用程序...IBErrorID（-1），错误代码（2104）与响应'MarketdatafarmconnectionisOK：usfarm'IBErrorID（-1），错误代码（2106）与响应'HMDSdatafarmconnectionisOK：ushmds'IBErrorID（-1），错误代码(2158)响应'Sec-defdatafarmconnectionisOK:secdefnj'CurrentIBservertime:2020-07-2913:27:18DisconnectedfromtheIBAPIapplication.Finished.unhandledexceptioninEReaderthreadTraceback(mostrecentcallast):File"/home/mhallsmoore6/venvite/python/qpackages/ibapi/reader.py",line34,inrundata=self.conn.recvMsg()File"/home/mhallsmoore/venv/qstrader/lib/python3.6/site-packages/ibapi/connection.py",line99,inrecvMsgbuf=self._recvAllMsg()文件“/home/mhallsmoore/venv/qstrader/lib/python3.6/site-packages/ibapi/connection.py”，第119行，in_recvAllMsgbuf=self.socket.recv(4096)OSError:[Errno9]Badfiledescriptor第一组输出是IB“错误”，代码为2104、2106和2158这些实际上是响应，表明与各种服务器的连接工作正常。也就是说，它们不是“错误”！服务器时间也从Unix时间戳正确转换为更易读的格式和输出。在此阶段，应用程序断开连接。但是请注意，在EReader线程中会引发OSError异常。这是IBAPI本身的内部问题，目前没有修复。出于本教程的目的，可以忽略它。关于连接到IBPythonNativeAPI的教程到此结束。ib_api_connection.py完整代码请扫描下方二维码。我们已经成功连接到IB服务器并通过调用获取当前服务器时间来检查连接。稍后我们将确定如何从IBAPI检索股票的市场数据。
