Python IB API 接駁教學 – 2. API 函數

文章提供: 核心引擎

開始連接 IB API

以下是一個簡單的程式碼,用於測試與 IB API 的連接。如果需要,請更改函數中的第2個參數 7497 (port number)。而第3個參數 123 是用於向 API 識別程式的用戶端 ID。你可以把它設為任何唯一的正整數。

from ibapi.client import EClient
from ibapi.wrapper import EWrapper  

class IBapi(EWrapper, EClient):
     def __init__(self):
         EClient.__init__(self, self) 

app = IBapi()
app.connect('127.0.0.1', 7497, 123)
app.run()

'''
#Uncomment this section if unable to connect
#and to prevent errors on a reconnect
import time
time.sleep(2)
app.disconnect()
'''

您的輸出應如下所示:

如果您已嘗試運行腳本幾次但未看到輸出,請將上面提到的第3個參數用戶端ID更改為唯一的ID。

另一個原因也可能是因為腳本在建立連接之前已結束。在這種情況下,請嘗試在代碼段末尾使用睡眠計時器將腳本暫停幾秒鐘。

EClient 和 EWrapper 的重要性

IB Python API 用戶端資料夾中有幾個原始程式碼檔。兩個最重要的檔是 EClient 和 EWrapper。

在大多數情況下,EClient 處理所有傳出請求,而 EWrapper 處理傳入消息。正如其名稱一樣,EWrapper的作用類似於傳入消息的包裝器,在大多數情況下,需要在腳本中覆蓋其中的函數,以將輸出重定向到您希望它的位置。

值得流覽一些原始程式碼檔以熟悉API。請記住,您可以隨時在Python終端中鍵入help(EClient)或help(EWrapper),以獲取有關其中包含的函數的更多資訊。

此處提供的所有範例都從基本腳本開始。在其中,首先導入 EClient 和 Ewrapper 類。然後創建一個新的自定義類,並將 EClient 和 Ewrapper 類傳遞到其中。

此時,我們使用範例中的變數實例化類,並調用命令來指定創建連接所需的參數。該命令執行啟動通信,同時在腳本末尾用於結束工作階段並關閉連接。appapp.connect()app.run()app.disconnect()

我們將向基本腳本添加線程。API 連接將在其自己的線程中運行,以確保與伺服器之間的通信不會被腳本主塊中的其他命令阻止。

如何檢索蘋果股票(AAPL)的現價

為了獲得股票的最新現價,我們創建了一個定義股票參數的合約物件。然後,我們調用 EClient 中的一個函數,讓 API 知道我們需要數據。

EWrapper 中的函數需要被覆蓋才能將回應列印到螢幕上。

或者,可以將回應保存到檔或變數。在生產環境中,您可能會將其保存到變數中。

代碼如下:

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract

import threading
import time


class IBapi(EWrapper, EClient):
	def __init__(self):
		EClient.__init__(self, self)
	def tickPrice(self, reqId, tickType, price, attrib):
		if tickType == 2 and reqId == 1:
			print('The current ask price is: ', price)

def run_loop():
	app.run()

app = IBapi()
app.connect('127.0.0.1', 7497, 123)

#Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True)
api_thread.start()

time.sleep(1) #Sleep interval to allow time for connection to server

#Create contract object
apple_contract = Contract()
apple_contract.symbol = 'AAPL'
apple_contract.secType = 'STK'
apple_contract.exchange = 'SMART'
apple_contract.currency = 'USD'

#Request Market Data
app.reqMktData(1, apple_contract, '', False, False, [])

time.sleep(10) #Sleep interval to allow time for incoming price data
app.disconnect()

此代碼將調用以請求 AAPL 的價格數據流,並在更新時在螢幕上列印最新價格。讓我們看一下所需的參數reqMktData

The is a unique positive integer you assign to your request which will be included in the response. This way, if you make several market data requests at the same time, you’ll know which returned data belongs to which asset. ReqId

The , left empty in this example, allows you to specify what kind of data you’re looking for. Since the ask price is part of the default dataset returned, we don’t need to specify a . You can run the code snippet below to get a full list of all the tickTypes available. tickTypetickType

from ibapi.ticktype import TickTypeEnum

for i in range(91):
	print(TickTypeEnum.to_str(i), i)

賣出價的數值為 2,因此我們腳本中函數中的 if 語句僅過濾掉賣出價。tickPrice

下面的第四個參數是,如果您希望獲得沒有訂閱的資產的快照數據。如果您有市場數據訂閱,或者不需要訂閱,請將其設置為 False。reqMktData

第五項是獲取快照而不是流數據。這適用於已訂閱的資產,或者不需要訂閱。

檢索其他資產的市場數據 – 歐元/美元,比特幣和黃金

如果您想提取其他市場的最新要價,只需根據需要更改合約物件即可。腳本的其餘部分保持不變。

要獲取合約物件所需的詳細資訊,請右鍵按兩下TWS觀察清單中需要數據的資產,然後選擇描述。將出現一個彈出框,其中包含您需要的資訊。它看起來像這樣:

現在我們有了歐元/美元所需的數據,讓我們為它創建一個合約物件。

eurusd_contract = Contract()
eurusd_contract.symbol = 'EUR'
eurusd_contract.secType = 'CASH'
eurusd_contract.exchange = 'IDEALPRO'
eurusd_contract.currency = 'USD'

你有它。我們現在有一個新的合約對象,我們可以使用與上一個示例相同的語法為其發出市場數據請求。我們只需要交換合約物件。

有興趣交易比特幣期貨嗎?下面是用於接收市場數據的合約物件的範例:

BTC_futures__contract = Contract()
BTC_futures__contract.symbol = 'BRR'
BTC_futures__contract.secType = 'FUT'
BTC_futures__contract.exchange = 'CMECRYPTO'
BTC_futures__contract.lastTradeDateOrContractMonth  = '202003'

上面的代碼片段中有一些更改。首先,期貨合約通常不需要合約貨幣。其次,需要添加合同到期時間。

» 如果您熱衷於期貨交易,請查看我們的”5種期貨交易策略指南“。

最後,如果您是大宗商品交易者,請查看如何創建現貨黃金合約:

XAUUSD_contract = Contract() 
XAUUSD_contract.symbol = 'XAUUSD' 
XAUUSD_contract.secType = 'CMDTY' 
XAUUSD_contract.exchange = 'SMART' 
XAUUSD_contract.currency = 'USD'

提示:如果您發現自己對同一資產類別中的工具發出大量請求,則創建一個基於預定義參數創建合約物件的函數可能會更容易。

使用 Python IB API 取得最近10小時 K 線

獲取歷史數據與檢索最新賣出價非常相似。不同之處在於,它被稱為 而不是 。我們可以覆蓋函數來處理回應。確保傳入包含所有數據的物件。reqHistoricalDatareqMktDatahistoricalDatabar

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract

import threading
import time

class IBapi(EWrapper, EClient):
	def __init__(self):
		EClient.__init__(self, self)
	def historicalData(self, reqId, bar):
		print(f'Time: {bar.date} Close: {bar.close}')
		
def run_loop():
	app.run()

app = IBapi()
app.connect('127.0.0.1', 7497, 123)

#Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True)
api_thread.start()

time.sleep(1) #Sleep interval to allow time for connection to server

#Create contract object
eurusd_contract = Contract()
eurusd_contract.symbol = 'EUR'
eurusd_contract.secType = 'CASH'
eurusd_contract.exchange = 'IDEALPRO'
eurusd_contract.currency = 'USD'

#Request historical candles
app.reqHistoricalData(1, eurusd_contract, '', '2 D', '1 hour', 'BID', 0, 2, False, [])

time.sleep(5) #sleep to allow enough time for data to be returned
app.disconnect()

reqHistoricalData需要更多的參數,這是一個細分。

由於我們正在尋找最近的10根蠟燭,因此我們可以將結束日期留空。

對於間隔,我們選擇了”2 D”,它代表兩天。間隔是從前一天的收盤價開始計算的,因此,如果您選擇”1 D”,則根據一天中的時間,您可能會得到少於10根蠟燭。

時間段很簡單,我們將其設置為”1小時”,因為我們正在尋找每小時的蠟燭。

數據類型通常為 BID、ASK 或 MIDPOINT。在大多數圖表平臺上,使用買入價。

有關可用資料類型、時間段和間隔的完整清單,請查看 – https://interactivebrokers.github.io/tws-api/historical_bars.html

RTH代表常規交易時間,主要用於股票。如果您要查找盤前數據,請將其設置為 1。

時間格式有兩個選項。如果希望響應數據包含可讀時間,請將其設置為 1,對於 Epcoh (Unix) 時間,請將其設置為 2。第二個選項使轉換為Python DateTime物件變得更加容易。

最後,如果 Streaming 設置為True,它將每五秒不斷更新價格柱(即使蠟燭尚未關閉)。

注意:IB將發送最新的蠟燭,即使它尚未關閉。在大多數情況下,不完整的蠟燭是沒有用的,應該被丟棄。

存儲歷史數據以供日後使用的最佳方式是什麼?

存儲數據的一種簡單方法是將其另存為 CSV 檔。這可以使用Python中的標準寫入檔方法完成,也可以使用Pandas庫中的內置方法完成。

熊貓圖書館由交易者設計,用於交易。至少在最初,它後來被修改為附帶更多的功能。該庫允許輕鬆操作數據和存儲。

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract

import threading
import time

class IBapi(EWrapper, EClient):
	def __init__(self):
		EClient.__init__(self, self)
		self.data = [] #Initialize variable to store candle

	def historicalData(self, reqId, bar):
		print(f'Time: {bar.date} Close: {bar.close}')
		self.data.append([bar.date, bar.close])
		
def run_loop():
	app.run()

app = IBapi()
app.connect('127.0.0.1', 7497, 123)

#Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True)
api_thread.start()

time.sleep(1) #Sleep interval to allow time for connection to server

#Create contract object
eurusd_contract = Contract()
eurusd_contract.symbol = 'EUR'
eurusd_contract.secType = 'CASH'
eurusd_contract.exchange = 'IDEALPRO'
eurusd_contract.currency = 'USD'

#Request historical candles
app.reqHistoricalData(1, eurusd_contract, '', '2 D', '1 hour', 'BID', 0, 2, False, [])

time.sleep(5) #sleep to allow enough time for data to be returned

#Working with Pandas DataFrames
import pandas

df = pandas.DataFrame(app.data, columns=['DateTime', 'Close'])
df['DateTime'] = pandas.to_datetime(df['DateTime'],unit='s') 
df.to_csv('EURUSD_Hourly.csv')  

print(df)


app.disconnect()

上面的代碼片段構建自上一個示例,我們在該示例中檢索了歐元/美元的 10 個最後一小時蠟燭。為將其另存為 CSV 檔案所做的更改如下所示:

首先,我們創建了一個空變數,調用並指示函數在蠟燭圖傳入時向其追加。app.datahistoricalData

然後,為了使用 Pandas 匯出數據,我們創建了一個數據幀。調用該函數是為了將傳入的數據轉換為 DateTime 物件,以便以後更容易操作。這是將數據保存到文件的簡單方法。若要稍後檢索它,只需通過運行並保存對變數的回應來調用該檔。pandas.to_datetime.to_csvpandas.read_csv(filename)

計算 20 天 SMA 的 3 種方法

有幾種方法可以計算 20 天移動平均線的值。我們將討論三個:使用熊貓,手動計算和利用第三方庫。Python Pandas 只需一行即可實現。

df['20SMA'] = df['Close'].rolling(20).mean()
print(df.tail(10))

這就是它所需要的一切。您的輸出應如下所示:

或者,如果要手動計算移動平均線,請使用以下代碼段:

total = 0
for i in app.data[-20:]:  
    total += float(i[1])

print('20SMA =', round(total/20, 5)) 

上面的代碼將最後 20 根蠟燭收盤的總和除以 20,得出 20 SMA。

最後一種方法涉及使用名為TA-Lib的第三方庫。一些經紀人在他們的自定義圖表軟體中使用此庫,它非常受歡迎。雖然原始庫在 Python 中不可用,但可以使用包裝器來允許 Python 使用者訪問。

如何使用IB Python原生API進行交易?

要觸發訂單,我們只需創建一個包含資產詳細資訊的合約物件和一個包含訂單詳細資訊的訂單物件。然後致電提交訂單。app.placeOrder

IB API 需要與所有訂單關聯的訂單 ID,並且它必須是唯一的正整數。它還需要大於使用的最後一個訂單ID。幸運的是,有一個內置函數,可以告訴您下一個可用的訂單ID。

我們還可以使用這個內置函數來確認連接,因為一旦建立連接,這個訂單ID就會被發送出去。

出於這個原因,我們啟用了一些錯誤檢查,告訴腳本在腳本的早期等待訂單ID,以確保我們實際上已連接。

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
from ibapi.order import *

import threading
import time

class IBapi(EWrapper, EClient):
	def __init__(self):
		EClient.__init__(self, self)

	def nextValidId(self, orderId: int):
		super().nextValidId(orderId)
		self.nextorderId = orderId
		print('The next valid order id is: ', self.nextorderId)

	def orderStatus(self, orderId, status, filled, remaining, avgFullPrice, permId, parentId, lastFillPrice, clientId, whyHeld, mktCapPrice):
		print('orderStatus - orderid:', orderId, 'status:', status, 'filled', filled, 'remaining', remaining, 'lastFillPrice', lastFillPrice)
	
	def openOrder(self, orderId, contract, order, orderState):
		print('openOrder id:', orderId, contract.symbol, contract.secType, '@', contract.exchange, ':', order.action, order.orderType, order.totalQuantity, orderState.status)

	def execDetails(self, reqId, contract, execution):
		print('Order Executed: ', reqId, contract.symbol, contract.secType, contract.currency, execution.execId, execution.orderId, execution.shares, execution.lastLiquidity)


def run_loop():
	app.run()

#Function to create FX Order contract
def FX_order(symbol):
	contract = Contract()
	contract.symbol = symbol[:3]
	contract.secType = 'CASH'
	contract.exchange = 'IDEALPRO'
	contract.currency = symbol[3:]
	return contract

app = IBapi()
app.connect('127.0.0.1', 7497, 123)

app.nextorderId = None

#Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True)
api_thread.start()

#Check if the API is connected via orderid
while True:
	if isinstance(app.nextorderId, int):
		print('connected')
		break
	else:
		print('waiting for connection')
		time.sleep(1)

#Create order object
order = Order()
order.action = 'BUY'
order.totalQuantity = 100000
order.orderType = 'LMT'
order.lmtPrice = '1.10'

#Place order
app.placeOrder(app.nextorderId, FX_order('EURUSD'), order)
#app.nextorderId += 1

time.sleep(3)

#Cancel order 
print('cancelling order')
app.cancelOrder(app.nextorderId)

time.sleep(3)
app.disconnect()

為了確認連接已建立,我們正在等待 API 通過 發送,並使用睡眠計時器將腳本保存在迴圈中,直到收到為止。nextorderid

除此之外,我們還創建了一個函數來創建特定於外匯的合約。這簡化了合約的創建,因為大多數參數都是相似的。

為了下訂單,我們創建了一個訂單物件,指定您是要買入還是賣出。訂單大小和限價也在此處設置。如果您想創建市價單,請設置為「MKT」並註釋掉.order.orderTypeorderlmtPrice

下訂單後,請記得增加您的訂單。nextorderId

您還會注意到在腳本頂部附近定義的幾個附加函數。這些是EWrapper返回的所有與下訂單相關的消息。我們將此輸出定向到螢幕,但與之前類似,您可能希望將其中一些保存到變數中以供以後使用。

以下是執行上述文稿後輸出的外觀:

API 將許多專案視為錯誤,即使它們不是。例如,即使沒有問題,訂單取消也會顯示為錯誤。這可以通過覆蓋錯誤消息的 EWrapper 函數來更改。下面是一個示例:

def error(self, reqId, errorCode, errorString):
	if errorCode == 202:
		print('order canceled') 

可以在此處找到 API 代碼(包括錯誤代碼)的完整清單 – https://interactivebrokers.github.io/tws-api/message_codes.html

最好使用與市場數據連接關聯的代碼來確保您具有有效的數據連接,並在提交訂單時實施錯誤檢查,以確保連接處於活動狀態並且價格數據是最新的。

如何使用IB Python原生API實現止損或止盈?

止損本質上是一旦達到特定價格就要執行的訂單。最好將止損訂單與您的原始訂單「分組」。。這樣,如果您決定刪除原始訂單,您的止損單將自動刪除。

IB將訂單分組稱為括號訂單。主訂單被視為「父訂單」,止損或止盈被視為「子」訂單。

通過將父訂單的訂單號分配為子訂單中的 a,將兩個訂單綁定在一起。parentId

雖然止損和止盈訂單組合在一起形成一個括號訂單,但請注意,兩個訂單都需要一個單獨的訂單。因此,請記住增加併為您的止損和止盈訂單分配一個。orderIdorderId

要記住的另一件重要事情是 父訂單具有行 。這是為了確保在傳輸其餘括號訂單之前不會處理第一個訂單。通過發送的最後一個訂單應處理整個支架訂單。order.transmit = FalseplaceOrderorder.transmit = True

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
from ibapi.order import *

import threading
import time

class IBapi(EWrapper, EClient):
	
	def __init__(self):
		EClient.__init__(self, self)
	
	def nextValidId(self, orderId: int):
		super().nextValidId(orderId)
		self.nextorderId = orderId
		print('The next valid order id is: ', self.nextorderId)

	def orderStatus(self, orderId, status, filled, remaining, avgFullPrice, permId, parentId, lastFillPrice, clientId, whyHeld, mktCapPrice):
		print('orderStatus - orderid:', orderId, 'status:', status, 'filled', filled, 'remaining', remaining, 'lastFillPrice', lastFillPrice)
	
	def openOrder(self, orderId, contract, order, orderState):
		print('openOrder id:', orderId, contract.symbol, contract.secType, '@', contract.exchange, ':', order.action, order.orderType, order.totalQuantity, orderState.status)

	def execDetails(self, reqId, contract, execution):
		print('Order Executed: ', reqId, contract.symbol, contract.secType, contract.currency, execution.execId, execution.orderId, execution.shares, execution.lastLiquidity)


def run_loop():
	app.run()

def FX_order(symbol):
	contract = Contract()
	contract.symbol = symbol[:3]
	contract.secType = 'CASH'
	contract.exchange = 'IDEALPRO'
	contract.currency = symbol[3:]
	return contract

app = IBapi()
app.connect('127.0.0.1', 7497, 123)

app.nextorderId = None

#Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True)
api_thread.start()

#Check if the API is connected via orderid
while True:
	if isinstance(app.nextorderId, int):
		print('connected')
		print()
		break
	else:
		print('waiting for connection')
		time.sleep(1)

#Create order object
order = Order()
order.action = 'BUY'
order.totalQuantity = 100000
order.orderType = 'LMT'
order.lmtPrice = '1.10'
order.orderId = app.nextorderId
app.nextorderId += 1
order.transmit = False

#Create stop loss order object
stop_order = Order()
stop_order.action = 'SELL'
stop_order.totalQuantity = 100000
stop_order.orderType = 'STP'
stop_order.auxPrice = '1.09'
stop_order.orderId = app.nextorderId
app.nextorderId += 1
stop_order.parentId = order.orderId
order.transmit = True

#Place orders
app.placeOrder(order.orderId, FX_order('EURUSD'), order)
app.placeOrder(stop_order.orderId, FX_order('EURUSD'), stop_order)

app.disconnect()

可以通過創建類似於我們創建上述止損訂單的物件來添加止盈。Order()

止盈中的價格變數可能看起來像這樣,因為止盈是限價單。因此,請使用它而不是.請記住,無論最後發送哪個訂單,都應該有,而其餘的應該有。take_profit.lmtPricestop_order.auxPricetransmit=Truetransmit=False

當谷歌達到一定價格時,如何為蘋果下訂單?

盈透證券的一大優勢是它支援高級訂單類型。它甚至有幾個大多數其他經紀人不支援。

我們將在下一個示例中重點介紹高級訂單類型,我們將展示一旦Google(GOOG)越過某個價格點,如何在Apple(AAPL)中執行交易。

對於其他經紀人,您可能需要手動跟蹤Google的股票價格,並在滿足條件后發送訂單。但是,有一個更乾淨的解決方案,允許我們發送訂單,讓IB的伺服器在滿足條件時進行跟蹤,以便可以執行交易。

由於我們使用的是特殊訂單功能,因此需要從 中導入兩個類。ibapi.order_condition

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
from ibapi.order_condition import Create, OrderCondition
from ibapi.order import *

import threading
import time

除此之外,我們還提供了一些在前面的示例中使用的相同導入,用於創建合約和訂單物件。

class IBapi(EWrapper, EClient):
	def __init__(self):
		EClient.__init__(self, self)
		self.contract_details = {} #Contract details will be stored here using reqId as a dictionary key

	def nextValidId(self, orderId: int):
		super().nextValidId(orderId)
		self.nextorderId = orderId
		print('The next valid order id is: ', self.nextorderId)

	def orderStatus(self, orderId, status, filled, remaining, avgFullPrice, permId, parentId, lastFillPrice, clientId, whyHeld, mktCapPrice):
		print('orderStatus - orderid:', orderId, 'status:', status, 'filled', filled, 'remaining', remaining, 'lastFillPrice', lastFillPrice)
	
	def openOrder(self, orderId, contract, order, orderState):
		print('openOrder id:', orderId, contract.symbol, contract.secType, '@', contract.exchange, ':', order.action, order.orderType, order.totalQuantity, orderState.status)

	def execDetails(self, reqId, contract, execution):
		print('Order Executed: ', reqId, contract.symbol, contract.secType, contract.currency, execution.execId, execution.orderId, execution.shares, execution.lastLiquidity)

接下來,我們覆蓋了更多函數,這些函數將在訂單發送后和執行訂單時返回數據。我們所做的只是指示 API 將此資訊列印到主控台,只是為了說明它們的工作原理。

下一個代碼片段與我們嘗試實現的目標更相關。為了創造價格條件,我們需要我們嘗試交易的資產的合約ID或ConID。

您可以通過搜索IB合約和交易品種資料庫來獲取此ID。但更簡單的方法是使用 API 的功能。它將返回已填寫ConID的合同。reqContractDetails

	def contractDetails(self, reqId: int, contractDetails):
		self.contract_details[reqId] = contractDetails

	def get_contract_details(self, reqId, contract):
		self.contract_details[reqId] = None
		self.reqContractDetails(reqId, contract)
		#Error checking loop - breaks from loop once contract details are obtained
		for err_check in range(50):
			if not self.contract_details[reqId]:
				time.sleep(0.1)
			else:
				break
		#Raise if error checking loop count maxed out (contract details not obtained)
		if err_check == 49:
			raise Exception('error getting contract details')
		#Return contract details otherwise
		return app.contract_details[reqId].contract

有兩個函數可用於獲取包含 ConID 的更新合同。第一個是 哪個是 的函數。當我們請求合同詳細資訊時,它將在此處返回。contractDetailsEWrapper

我們將把這裡返回的任何內容存儲在字典檔中。我們用於發出請求的請求ID或將用作字典的鍵值。reqId

接下來,我們創建了一個用於請求合約詳細資訊的自定義函數。要訪問它,我們必須通過reqId和我們請求詳細信息的合約。

此自定義函數涉及許多內容。它發出資料請求,並創建存儲數據的變數。此外,它還進行了一些錯誤檢查,以確保數據確實已返回並且沒有問題。最後,它等待數據,因此在數據傳入之前不會執行其他命令。讓我們更詳細地介紹一下此函數。

首先,我們創建一個變數來存儲傳入的數據。我們將此項設置為”無”。這樣,我們可以稍後檢查變數是否具有值以確認我們的數據已到達。

接下來,該函數會將請求發送到 API。

此時,我們將檢查我們的數據是否已到達。迴圈已設置為運行 50 次。在每次反覆運算中,它會檢查我們的合約詳細資訊是否已返回,如果返回,迴圈將中斷。

如果迴圈運行了整整 50 次,這意味著它沒有成功突破,則 的值將為 49。在這種情況下,我們將提出一個異常,以提醒我們在獲取合同詳細資訊時出現問題。err_check

現在我們已經完成了類函數,讓我們繼續討論主腳本。在這裡,我們創建了兩個函數。

def run_loop():
	app.run()

def Stock_contract(symbol, secType='STK', exchange='SMART', currency='USD'):
	''' custom function to create stock contract '''
	contract = Contract()
	contract.symbol = symbol
	contract.secType = secType
	contract.exchange = exchange
	contract.currency = currency
	return contract

第一個只是一個函數,我們稍後將調用該函數以在線程中運行我們的應用程式,類似於前面的示例。任何需要聲明或在此線程啟動時運行的內容都可以添加到函數中。run_loop

第二個功能是簡化合同的創建。我們已經傳遞了一些預設值,因為大多數股票將屬於同一類別。

app = IBapi()
app.connect('127.0.0.1', 7496, 123)

app.nextorderId = None

#Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True)
api_thread.start()

#Check if the API is connected via orderid
while True:
	if isinstance(app.nextorderId, int):
		print('connected')
		break
	else:
		print('waiting for connection')
		time.sleep(1)

上面的代碼類似於前面的範例。我們已連接到 API,啟動了一個線程,並檢查是否存在以確認連接。nextorderid

#Create contracts
apple_contract = Stock_contract('AAPL')
google_contract = Stock_contract('GOOG')

我們的下一步是創建兩個合同,一個用於GOOG,一個用於AAPL。由於我們只使用基於 GOOG 價格的價格條件函數,因此只需要該合約的 ConID。

#Update contract ID
google_contract = app.get_contract_details(101, google_contract)

因此,我們使用自定義功能來更新Google合約,而不是Apple合約。get_contract_details

我們終於準備好創造我們的價格條件。第一步是創建訂單條件物件。

priceCondition = Create(OrderCondition.Price)

讓我們分解一下上面的代碼。 只是將存儲我們條件的變數的名稱。你可以隨心所欲地命名它。priceCondition

Create是 API 中找到的order_condition.py檔中的函數。正如它的名字一樣,它用於創建一個物件,或者更確切地說,根據我們的需求實例化正確的類。

在這種情況下,我們需要類,所以這就是進來的地方。PriceConditionOrderCondition.price

您可以創建幾種其他類型的條件,這是您聲明要處理的條件的位置。

總共有六種不同類型的訂單條件 – 、 、 、 、 和 。PriceTimeMarginExecutionVolumePercentChange

因此,例如,如果要根據當天的百分比變化創建條件,則可以改用。priceCondition = Create(OrderCondition.PercentChange)

接下來,我們通過我們設置條件的資產的合約ID及其交易的交易所。

priceCondition.conId = google_contract.conId
priceCondition.exchange = google_contract.exchange

此資訊已在合約物件中,因此我們只需將其指向合約的相應屬性即可。

我們準備設定一些條件。

#create conditions
priceCondition.isMore = True
priceCondition.triggerMethod = priceCondition.TriggerMethodEnum.Last
priceCondition.price = 1400.00

我們希望谷歌的價格高於1400美元才能執行這筆交易。因此,我們將該屬性設置為 True,並向該屬性添加了浮點值 1400.00。.isMore.price

我們要使用的觸發方法是GOOG交易的最後價格

API 要求將觸發器方法作為整數輸入,但有一個調用的函數會將值 Last 轉換為整數,這就是我們在這裡所做的。TriggerMethodEnum

我們的價格條件是完整的,隨時可以去。現在剩下的就是將條件添加到訂單中並提交它。

#Create order object
order = Order()
order.action = 'BUY'
order.totalQuantity = 100
order.orderType = 'MKT'
#order.lmtPrice = '300' - optional - you can add a buy limit here

在上面的代碼中,我們以與之前的示例相同的方式創建了一個訂單。請注意,我們可以在此處創建限價訂單。假設我們設定了300美元的限額。

在這種情況下,一旦GOOG突破1400美元,訂單將被觸發,但訂單將被發送以300美元的價格購買AAPL。

無論GOOG接下來做什麼,該買入訂單都將保持活躍,但除非AAPL回落至300美元,否則不會觸發。如果AAPL當時已經以300美元或更低的價格交易,它將立即被觸發。

我們將使用市價單,但如果您決定限價單,請確保將更改為”LMT”。orderType

我們之前創建的價格條件仍然需要添加到訂單中。操作方法如下:

order.conditions.append(priceCondition)

並且不要忘記設置順序.傳輸到 True。

order.transmit = True

最後一步是提交訂單。

app.placeOrder(app.nextorderId, apple_contract, order)

此時,訂單位於IB的伺服器上,它將從那裡進行管理。即使我們關閉腳本,該訂單仍將保持活動狀態,IB將在滿足條件時執行該訂單。

當谷歌在過去5分鐘內移動超過5%時,如何為Apple下訂單?

此策略與上一個策略有一些相似之處,儘管我們需要採用完全不同的方法並手動編碼。

價格條件函數確實允許我們根據價格變化的百分比提交訂單,但是,它從一天開始就計算此變化。

我們追求的是過去5分鐘內發生的價格變化。

我們將訂閱即時報價數據並將其存儲在熊貓的數據幀中。這將使我們能夠檢查5%的變化,此時我們可以提交訂單。

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
from ibapi.order import *

import pandas as pd
import threading
import time

我們從進口開始,這裡唯一的新東西就是我們進口了熊貓。

class IBapi(EWrapper, EClient):
	def __init__(self):
		EClient.__init__(self, self)
		self.bardata = {} #Initialize dictionary to store bar data
	
	def nextValidId(self, orderId: int):
		super().nextValidId(orderId)
		self.nextorderId = orderId
		print('The next valid order id is: ', self.nextorderId)

到目前為止,類函數看起來也很熟悉。這裡唯一不同的是,我們創建了一個名為 的字典檔。我們稍後將使用它來存儲我們的價格數據幀。bardata

	def tick_df(self, reqId, contract):
		''' custom function to init DataFrame and request Tick Data '''
		self.bardata[reqId] = pd.DataFrame(columns=['time', 'price'])
		self.bardata[reqId].set_index('time', inplace=True)
		self.reqTickByTickData(reqId, contract, "Last", 0, True)
		return self.bardata[reqId]

在這裡,我們創建了一個自定義函數。它將創建一個空的數據幀,並將索引設置為時間列。這樣,我們將有一個時間序列索引數據幀,當我們必須將數據縮小到5分鐘的視窗時,它將簡化以後的事情。

同時,我們使用了來自的函數來啟動數據流。reqTickByTickDataEClient

將其設置為自定義函數的原因是可以啟動多個數據饋送,每個饋送都有自己單獨的數據幀。同樣,reqId 將用作鍵,因此可以從我們之前在函數中聲明的變數 bardata 訪問所有數據。__init__

Next, we will overwrite the function of the .tickByTickAllLastEWrapper

	def tickByTickAllLast(self, reqId, tickType, time, price, size, tickAtrribLast, exchange, specialConditions):
		if tickType == 1:
			self.bardata[reqId].loc[pd.to_datetime(time, unit='s')] = price

This function will return the last price. The tick type for that is 1. The function should not return any other type of data, but we are checking to make sure the tick type is in fact 1 before adding to our DataFrame, just to be sure.

Let’s break down the next line of code. is the bardata dictionary file with the reqId as the key. In other words, this is our pandas DataFrame.self.bardata[reqId]

該函數來自pandas,它允許我們指定要插入數據的行和列。.loc

我們正在創建一個新行,使用時間作為索引。在該行中,我們在價格列下插入最後一個價格。

Panda通常會識別時間戳何時通過,並自動將其轉換為日期時間值。在這種情況下,它沒有。這就是為什麼我們使用Pandas的內置函數將時間值轉換為DateTime值的原因。pd.to_datetime(time, unit='s')

	def Stock_contract(self, symbol, secType='STK', exchange='SMART', currency='USD'):
		''' custom function to create contract '''
		contract = Contract()
		contract.symbol = symbol
		contract.secType = secType
		contract.exchange = exchange
		contract.currency = currency
		return contract

我們做的最後一件事是創建一個自定義函數,以便更輕鬆地創建股票合約。

請注意,它是在類中創建的,在上一個示例中,我們在類外部創建了它。

這兩種方法都有效,並將提供相同的最終結果。如果您計劃創建多個腳本,並認為您將在每個腳本中使用特定的函數,那麼在類中編寫它是有意義的。

這樣,您就可以將類導入到另一個腳本中,而不必重寫相同的函數。

我們在類之外創建了一些函數。

def run_loop():
	app.run()

def submit_order(contract, direction, qty=100, ordertype='MKT', transmit=True):
	#Create order object
	order = Order()
	order.action = direction
	order.totalQuantity = qty
	order.orderType = ordertype
	order.transmit = transmit
	#submit order
	app.placeOrder(app.nextorderId, contract, order)
	app.nextorderId += 1

第二個功能簡化了提交訂單的過程。這是一個很好的例子,說明可以包含在類中的內容。

但是,我們已經介紹了幾種不同的訂單類型,例如包含止損水準或止盈水準的支架訂單,以及價格條件訂單。由於訂單處理的複雜性,不將其包含在類中更有意義。

接下來,我們有我們的策略功能。這就是我們是否應該執行交易的決策發生的地方。

def check_for_trade(df, contract):
	start_time = df.index[-1] - pd.Timedelta(minutes=5)
	min_value = df[start_time:].price.min()
	max_value = df[start_time:].price.max()

	if df.price.iloc[-1] < max_value * 0.95:
		submit_order(contract, 'SELL')
		return True

	elif df.price.iloc[-1] > min_value * 1.05:
		submit_order(contract, 'BUY')
		return True

第一行是獲取 DataFrame 中的最後一個索引值,即我們收到的最後一個數據的時間值。我們使用 Pandas 內置的方法從該時間值中減去 5 分鐘。Timedelta

現在我們知道通過使用來查看有多遠。在下面的代碼行中,我們使用了它,它返回了從 5 分鐘前到現在的所有數據。start_timedf[start_time:]

然後,我們可以使用 Pandas 中的 and 函數來確定過去五分鐘內的最高價和最低價。min()max()

有了這些值,我們可以檢查當前價格(數據框中的最後一個價格值)是否大於或小於最小值或最大值 5%。

如果滿足條件,我們將提交訂單。該函數還將傳回布林值 。通過這種方式,我們知道訂單已提交。True

此時,我們可以轉到主腳本。

#Main
app = IBapi()
app.nextorderId = None
app.connect('127.0.0.1', 7496, 123)

#Start the socket in a thread
api_thread = threading.Thread(target=run_loop)
api_thread.start()

#Check if the API is connected via orderid
while True:
	if isinstance(app.nextorderId, int):
		print('connected')
		break
	else:
		print('waiting for connection')
		time.sleep(1)

#Create contract object
google_contract = app.Stock_contract('GOOG')
apple_contract = app.Stock_contract('AAPL')

上面的腳本與前面的示例相同。它連接到 API,啟動線程,並通過檢查下一個有效的訂單 ID 來確保建立連接。我們還創建了兩個股票合約。

#Request tick data for google using custom function
df = app.tick_df(401, google_contract)

在這裡,我們開始了GOOG的數據流。回想一下,我們在類中為此創建了一個函數。我們只需要傳遞一個reqId,它可以是任何唯一的整數,以及合約。

#Verify data stream
time.sleep(10)
for i in range(100):
	if len(df) > 0:
		break
	time.sleep(0.3)
	
if i == 99:
	app.disconnect()
	raise Exception ('Error with Tick data stream')

接下來,我們只想驗證數據是否從流進入我們的 DataFrame。我們給它一些時間,但如果它失敗了,就會引發異常。我們通過檢查以確保數據幀的長度大於0來實現此目的。

此時,我們知道數據流正在工作,並且我們正在數據幀中捕獲它。但我們仍然需要五分鐘的數據才能開始執行交易。

#Check if there is enough data
data_length = df.index[-1] - df.index[0]
if data_length.seconds < 300:
	time.sleep(300 - data_length.seconds)

在上面的代碼中,我們通過將 DataFrame 中的最後一個時間值減去第一個時間值來檢查已經過去了多少秒。

我們至少需要5分鐘或300秒的數據。因此,我們將讓腳本休眠 300 秒,減去已經經過的時間。

現在一切都準備好了,我們準備開始尋找交易了。

#Main loop
while True:
	if check_for_trade(df, apple_contract): break
	time.sleep(0.1)

app.disconnect()

上面的代碼是一個無限迴圈,它調用函數以查看是否發生了5%的偏差,如果發生了偏差,則執行交易。check_for_trade

回想一下,如果執行交易,該函數返回布爾值?如果發生這種情況,腳本將脫離無限循環並結束。True

如果要使腳本保持連續運行,可以從上面的代碼段中刪除 和。如果您走這條路,如果交易被執行,最好實施5分鐘的睡眠。if: break

否則,一旦滿足條件,腳本將發送多個連續訂單,因為它在無限迴圈中運行。

最後,我們添加了0.1秒的睡眠,以便在每次檢查后非常短暫地暫停腳本。這是為了避免我們的CPU在執行無限循環時進入過載狀態。

有幾種不同的方法可以使用 API 流式傳輸數據。作為此示例中使用的逐筆報價數據的替代方法,我們可以使用該函數。reqMktData

這兩種方法都有其注意事項。該函數每 250 毫秒發送一次價格變動數據(股票和期貨)。因此,數據不如 準確。reqMktDatareqTickByTickData

有趣的是,它不返回交易發生的時間,這是此示例中未使用它的主要原因。reqMktData

更準確,但會返回最後的價格或買入價和賣出價。並且,單獨的功能用於管理這些功能。reqTickByTickDataEWrapper

使用 時,可能會有幾筆交易使用相同的時間戳快速進入。這可能會導致數據丟失,因為我們根據時間值存儲數據。reqTickByTickData

也許IB開發人員會在未來的版本中考慮這些不一致之處。現在,可能值得檢查這兩個端點,以確定哪一個最適合您的系統。

如何購買看漲期權和看跌期權?

下達期權訂單類似於為任何其他資產下訂單。我們可以重用前面部分的大部分代碼,我們在該部分中經歷了一個觸發訂單的範例

但是我們需要更改一些合約參數。因此,讓我們從創建一個合約對象開始。

contract = Contract()
contract.symbol = 'TSLA'
contract.secType = 'OPT'
contract.exchange = 'SMART'

上面的代碼應該看起來很熟悉。與前面的示例唯一有點不同的是,我們對安全類型(secType)使用了”OPT”來區分它作為一個選項。

我們需要填充其他一些欄位才能正確定義期權合約。

contract.lastTradeDateOrContractMonth = '20201002'
contract.strike = 424
contract.right = 'C'
contract.multiplier = '100'

在這裡,我們指定了2020年10月2日的期權到期日,執行價格為424美元。

我們將購買一個看漲期權,該期權由.如果您想交易看跌期權,只需將其換成”P”即可。最後,合約乘數為 100。在大多數情況下,合約乘數將為 100。這可以通過TWS確認,或者還有一種方法可以通過API進行檢查。contract.right

總而言之,與股票合約相比,我們需要聲明額外的四個參數,並且我們需要使用”OPT”作為secType。

現在我們可以創建一個訂單物件。

order = Order()
order.action = 'BUY'
order.totalQuantity = 1
order.orderType = 'MKT'

這裡唯一需要注意的是,總數量是1。由於此股票期權的乘數為100,因此數量為1類似於交易100股TSLA。

最後,我們可以提交訂單。

app.placeOrder(app.nextorderId, contract, order)

我們的螢幕確認訂單已發送並執行。

如果您按照此代碼示例進行操作,則在 2020 年 10 月 2 日之後閱讀本文時,必須更改選項到期時間。

您可以在期權交易工具下或在TWS中查找有效的期權到期日和行使價,也可以右鍵按兩下關注清單中的資產並按下期權圖示以拉動鏈條。




Python API

Python IB API 接駁教學 – 3. 運用 Telegram Bot 自動發送通知

如何運用 Telegram Bot 自動發送通知給你? 現在,您已經能夠獲取市場數據並創建訂單,您可能希望實現通知系統。例如當訂單被觸發時,或者達到某個你自訂的條件。 Telegram 有一種非常簡單的方法來創建實時通知訊息。 設定 Telegram Bot 步驟: 從電報中打開與 "BotFather" 的聊天室。鍵入命令 /newbot它將提示你輸入機器人名稱,並向你發送訪問權杖 (Token)。與剛創建的機器人打開聊天室。輸入任意的聊天訊息以啟用聊天室。轉到以下...

Python IB API 接駁教學 – 1. IB API 介紹

Python IB API (Native) 介紹 它是一項允許您通過Python代碼自動交易的功能。用更專業的術語來說,它是一種通信協議,允許與盈透證券(IB)伺服器和自定義軟體應用程式交換資訊。作為橋樑,該API允許從自定義軟體或腳本發送訂單,接收即時或歷史數據以及其他幾個有用的應用程式。 IB平臺上的圖表示例。它顯示特斯拉期權上漲30,000%,並在幾天內崩潰 - 大約在2020年 Python IB API (Native) 的好處...

Python 富途 API 函數表

筆者整理了一個富途 API 函數表,方便同學深入認識 Futu API 可以拿到的數據類型。 富途 API 可分為2大類 透過行情接口下載各種市場產品的即時數據 (分鐘級、秒級)。透過交易接口以程式即時進行下單交易、帳戶管理等功能。 行情接口 以下是行情接口函數表,看有沒有你相要的數據:...

Python 富途 API 接駁方法

很多同學都不知道原來富途牛牛有免費開放一套 API 給大家拿取實時市場數據,這篇教學會向大家富途 API 接駁方法。 富途 API 軟件架構 Python 富途 API 由 Python futu-api 函式庫及 Futu OpenD 軟件組成。我們只需在電腦安裝運行 Futu OpenD 軟件,然後在 Python 調用 futu-api 函數,就可拎到實時市場數據。 安裝 3 步曲 開立富途牛牛帳戶下載並安裝 Futu OpenD 軟件下載並安裝 Python futu-api 第3方函式庫 第1步: 開立富途牛牛帳戶...

PYTHON 自動化數據擷取講座
富途 OPEN API 程式交易講座