python应用于期货_Python期货量化交易基础教程（17）
16.14、異步任務：
16.14.1、使用協程任務：
函數create_task()用來創建協程任務，并將任務加入事件循環以實現異步并發。
wait_update()不能用在協程中，若在協程中等待業務更新，可調用register_update_notify函數把業務數據注冊到TqChan，當業務數據有更新時會通知該TqChan，在協程里就可以用實時更新的業務數據運算。例如：
from tqsdk import TqApi, TqAuth
api = TqApi(auth=TqAuth("信易賬號", "密碼"))
quote1 = api.get_quote("CFFEX.T2103")
quote2 = api.get_quote("CFFEX.TF2103")
#帶有業務更新的協程
async def demo(quote):
#將quote注冊到TqChan命名為update_chan
async with api.register_update_notify(quote) as update_chan:
async for _ in update_chan: #遍歷隊列通知
print('品種：',quote.instrument_name,'最新價：',quote.last_price)
#無業務更新的協程
async def func():
return quote1.instrument_name,quote2.instrument_name
# 創建task1、task2，把quote1、quote2注冊到TqChan
task1=api.create_task(demo(quote1))
task2=api.create_task(demo(quote2))
#把帶有返回值的協程創建成task3
task3=api.create_task(func())
while True:
api.wait_update()
if task3.done(): #task3結束后，獲取返回值
print(task3.result())
'''輸出結果為：('債十2103', '債五2103')品種： 債十2103 最新價： 97.435('債十2103', '債五2103')品種： 債十2103 最新價： 97.435('債十2103', '債五2103')品種： 債十2103 最新價： 97.43('債十2103', '債五2103')品種： 債十2103 最新價： 97.43'''
16.14.2、使用多線程：
當用戶策略實例很多，導致網絡連接數無法容納時，可以使用多線程。首先需要在主線程中創建一個 TqApi 實例 api_master，并用 TqApi.copy 函數創建多個slave副本，把slave副本用在多個線程中，主線程里的api_master 仍然需要持續調用 wait_update。
子線程和主線程其實是運行在同一個事件循環里，如果在子線程里調用api_slave.close()會引發主線程事件循環關閉的異常，如果主線程里調用api_master.close()，子線程可能因等待事件循環響應而阻塞，若想讓子線程和主線程一起退出，可設置子線程為守護線程。
使用多線程需要自定義一個線程類，并重寫run函數，在run函數里執行策略代碼，例如：
import threading
from tqsdk import TqApi, TqAuth
#自定義線程類
class WorkerThread(threading.Thread):
def __init__(self, api, symbol):
threading.Thread.__init__(self)
self.api = api #初始化參數
self.symbol = symbol #初始化參數
#重寫run函數，策略代碼寫在run函數中
def run(self):
SHORT = 30 # 短周期
LONG = 60 # 長周期
data_length = LONG + 2 # k線數據長度
klines = self.api.get_kline_serial(self.symbol, duration_seconds=60, data_length=data_length)
target_pos = TargetPosTask(self.api, self.symbol)
while True:
self.api.wait_update()
if self.api.is_changing(klines.iloc[-1], "datetime"): # 產生新k線:重新計算SMA
short_avg = ma(klines["close"], SHORT) # 短周期
long_avg = ma(klines["close"], LONG) # 長周期
if long_avg.iloc[-2] < short_avg.iloc[-2] and long_avg.iloc[-1] > short_avg.iloc[-1]:
target_pos.set_target_volume(-3)
print("均線下穿，做空")
if short_avg.iloc[-2] < long_avg.iloc[-2] and short_avg.iloc[-1] > long_avg.iloc[-1]:
target_pos.set_target_volume(3)
print("均線上穿，做多")
if __name__ == "__main__":
#主線程創建TqApi實例
api_master = TqApi(auth=TqAuth("信易賬號", "密碼"))
# 實例化線程類，傳入TqApi實例的副本api_master.copy()
thread1 = WorkerThread(api_master.copy(), "SHFE.cu1901")
thread2 = WorkerThread(api_master.copy(), "SHFE.rb1901")
# 啟動線程實例
thread1.start()
thread2.start()
while True:
api_master.wait_update() #主線程保持對wait_update()的調用
當線程太多時，操作系統因調度線程，可能把主要工作都用在了調度線程上，而降低了多線程的效率，更宜使用異步協程實現多策略。
16.14.3、使用多進程：
當程序比較耗CPU時，可以采用多進程，比如回測時，需要對大量的數據計算，可以用多個進程同時回測多個品種，注意: 由于服務器流控限制, 同時執行的回測任務請勿超過10個，例如：
from tqsdk import TqApi, TqAuth, TqSim, TargetPosTask, BacktestFinished, TqBacktest
from tqsdk.tafunc import ma
from datetime import date
import multiprocessing
from multiprocessing import Pool
def MyStrategy(SHORT):
LONG = 60
SYMBOL = "SHFE.cu1907"
acc = TqSim()
try:
api = TqApi(acc, backtest=TqBacktest(start_dt=date(2019, 5, 6), end_dt=date(2019, 5, 10)), auth=TqAuth("信易賬戶", "賬戶密碼"))
data_length = LONG + 2
klines = api.get_kline_serial(SYMBOL, duration_seconds=60, data_length=data_length)
target_pos = TargetPosTask(api, SYMBOL)
while True:
api.wait_update()
if api.is_changing(klines.iloc[-1], "datetime"):
short_avg = ma(klines.close, SHORT)
long_avg = ma(klines.close, LONG)
if long_avg.iloc[-2] < short_avg.iloc[-2] and long_avg.iloc[-1] > short_avg.iloc[-1]:
target_pos.set_target_volume(-3)
if short_avg.iloc[-2] < long_avg.iloc[-2] and short_avg.iloc[-1] > long_avg.iloc[-1]:
target_pos.set_target_volume(3)
except BacktestFinished:
api.close()
print("SHORT=", SHORT, "最終權益=", acc.account.balance) # 每次回測結束時, 輸出使用的參數和最終權益
if __name__ == '__main__':
#提供凍結以產生 Windows 可執行文件的支持，在非 Windows 平臺上是無效的
multiprocessing.freeze_support()
p = Pool(4) # 進程池, 建議小于cpu數
for s in range(20, 40):
p.apply_async(MyStrategy, args=(s,)) # 把20個回測任務交給進程池執行
print('Waiting for all subprocesses done...')
p.close()
p.join()
print('All subprocesses done.')
17、TqSdk部分函數解讀
17.1、DIFF 協議：
DIFF (Differential Information Flow for Finance) 是一個基于websocket和json的應用層協議。websocket是全雙工通信，當客戶端和服務器端建立連接后，就可以相互發數據，建立連接又稱為“握手”，“握手”成功就可以建立通信了，不用在每次需要傳輸信息時重新建立連接，即不會“掉線”。json是數據存儲格式，json數據可以方便的反序列化為Python數據。
DIFF協議可以簡單的理解為服務端和客戶端的通信方式，協議規定了數據格式，以便于服務端和客戶端可以解讀對方發來的數據。
DIFF 協議分為兩部分:數據訪問和數據傳輸。
17.1.1、數據傳輸：
DIFF 協議要求服務端將業務數據以JSON Merge Patch的格式推送給客戶端，JSON Merge Patch的格式形如Python字典，可以在客戶端反序列化為Python字典(其實是映射類型Entity)。例如：
{
"aid": "rtn_data", # 業務信息截面更新
"data": [ # 數據更新數組
{
"balance": 10237421.1, # 賬戶資金
},
{
"float_profit": 283114.780999997, # 浮動盈虧
},
{
"quotes":{
"SHFE.cu1612": {
"datetime": "2016-12-30 14:31:02.000000",
"last_price": 36605.0, # 最新價
"volume": 25431, # 成交量
"pre_close": 36170.0, # 昨收
}
}
}
]
}aid 字段值即為數據包類型，"aid":"rtn_data"表示該包的類型為業務信息截面更新包。整個 data 數組相當于一個事務，其中的每一個元素都是一個 JSON Merge Patch，處理完整個數組后業務截面即完成了從上一個時間截面推進到下一個時間截面。
DIFF 協議要求客戶端發送 peek_message 數據包以獲得業務信息截面更新，例如：
{
"aid": "peek_message"
}服務端在收到 peek_message 數據包后應檢查是否有數據更新，如果有則應將更新內容立即發送給客戶端，如果沒有則應等到有更新發生時再回應客戶端。
服務端發送 rtn_data 數據包后可以等收到下一個 peek_message 后再發送下一個 rtn_data 數據包。
一個簡單的客戶端實現可以在連接成功后及每收到一個 rtn_data 數據包后發送一個 peek_message 數據包，這樣當客戶端帶寬不足時會自動降低業務信息截面的更新頻率以適應低帶寬。
當數據包中的 aid 字段不是 rtn_data 或 peek_message 則表示該包為一個指令包，具體指令由各業務模塊定義，例如 subscribe_quote 表示訂閱行情，insert_order 表示下單。由于客戶端和服務端存在網絡通訊延遲，客戶端的指令需要過一段時間才會影響到業務信息截面中的業務數據，為了使客戶端能分辨出服務端是否處理了該指令，通常服務端會將客戶端的請求以某種方式體現在截面中(具體方式由各業務模塊定義)。例如 subscribe_quote 訂閱行情時服務端會將業務截面中的 ins_list 字段更新為客戶端訂閱的合約列表，這樣當客戶端檢查服務端發來的業務截面時如果 ins_list 包含了客戶端訂閱的某個合約說明服務端處理了訂閱指令，但若 quotes 沒有該合約則說明該合約不存在訂閱失敗。
服務端發送包含"aid":"rtn_data"字段的業務數據截面更新包，客戶端發送包含"aid":"peek_message"字段的數據包請求業務數據截面，或發送包含"aid":"subscribe_quote "、"aid":"insert_order"等字段的指令包，如此，服務端和客戶端相互發信息，服務端和客戶端根據字段識別數據及處理數據。
17.1.2、數據訪問：
DIFF 協議要求服務端維護一個業務信息截面，例如：
{
"account_id": "41007684", # 賬號
"static_balance": 9954306.319000003, # 靜態權益
"balance": 9963216.550000003, # 賬戶資金
"available": 9480176.150000002, # 可用資金
"float_profit": 8910.231, # 浮動盈虧
"risk_ratio": 0.048482375, # 風險度
"using": 11232.23, # 占用資金
"position_volume": 12, # 持倉總手數
"ins_list": "SHFE.cu1609,...." # 行情訂閱的合約列表
"quotes":{ # 所有訂閱的實時行情
"SHFE.cu1612": {
"instrument_id": "SHFE.cu1612",
"datetime": "2016-12-30 13:21:32.500000",
"ask_priceN": 36590.0, #賣N價
"ask_volumeN": 121, #賣N量
"bid_priceN": 36580.0, #買N價
"bid_volumeN": 3, #買N量
"last_price": 36580.0, # 最新價
"highest": 36580.0, # 最高價
"lowest": 36580.0, # 最低價
"amount": 213445312.5, # 成交額
"volume": 23344, # 成交量
"open_interest": 23344, # 持倉量
"pre_open_interest": 23344, # 昨持
"pre_close": 36170.0, # 昨收
"open": 36270.0, # 今開
"close" : "-", # 收盤
"lower_limit": 34160.0, #跌停
"upper_limit": 38530.0, #漲停
"average": 36270.1 #均價
"pre_settlement": 36270.0, # 昨結
"settlement": "-", # 結算價
},
...
}
}
對應的客戶端也維護了一個該截面的鏡像，因此業務層可以簡單同步的訪問到全部業務數據。
TqSdk即是客戶端，TqSdk把收到的業務數據截面以上面的格式合并到_data屬性里，_data為多層嵌套的映射類型Entity，業務數據例如“quotes”，也是Entity，其“鍵”是合約代碼，例如“SHFE.cu1612”，其“值”是最終的業務數據——Quote對象，業務函數get_quote()便是把_data里的Quote對象的一個引用返回給調用方，調用方獲得的是Quote對象的動態引用。
_data是可變映射類型，會接收服務端發來的更新，因此業務函數返回的對象引用也會指向隨時更新的業務數據。
17.2、業務函數：
以get_quote()為例，上節已經介紹了get_quote()與_data的關系，現在我們結合函數的代碼再看下其執行過程，我們只取代碼的主要部分：
def get_quote(self, symbol: str) -> Quote:
# 從_data屬性中提取Quote
quote = _get_obj(self._data, ["quotes", symbol], self._prototype["quotes"]["#"])
# 若合約symbol是新添加的合約，則向服務端發送訂閱該合約的指令包
if symbol not in self._requests["quotes"]:
self._requests["quotes"].add(symbol)
self._send_pack({
"aid": "subscribe_quote",
"ins_list": ",".join(self._requests["quotes"]),
})
#返回quote，其指向的是_data中的Quote
return quote
其他的業務函數工作邏輯類似。業務對象Quote、Trade、Order、Position、Account等都是Entity的子類，可以像類一樣獲取其屬性，也可以像字典一樣使用。業務對象在模塊objs中定義。
17.3、insert_order()：
insert_order用來下單，我們只截取主要代碼看一下執行過程：
def insert_order(...) -> Order:
"""發送下單指令. **注意: 指令將在下次調用** :py:meth:`~tqsdk.api.TqApi.wait_update` **時發出**"""
if self._loop.is_running(): #事件循環正在運行
# 把下單請求函數打包成task排入事件循環
self.create_task(self._insert_order_async(...))
#下單后獲取委托單order
order = self.get_order(order_id, account=account)
#更新委托單字段
order.update({"order_id": order_id,"exchange_id": exchange_id,...})
return order #返回委托單
else: #事件循環還未運行
#打包一個指令包
pack = self._get_insert_order_pack(...)
#發送指令包
self._send_pack(pack)
##下單后獲取委托單order
order = self.get_order(order_id, account=account)
#更新委托單字段
order.update({ "order_id": order_id,"exchange_id": exchange_id,...})
return order #返回委托單
#發送指令包函數
def _send_pack(self, pack):
#立即向隊列發送指令包
if not self._is_slave:
self._send_chan.send_nowait(pack)
else:
self._master._slave_send_pack(pack)
#下單請求函數
async def _insert_order_async(...):
#打包一個指令包
pack = self._get_insert_order_pack(...)
#發送指令包
self._send_pack(pack)
下單的主要流程為：用協程任務打包一個指令包再發出去。create_task是無阻塞的，創建完task立即返回，get_order獲取委托單也是無阻塞的，因此insert_order執行后會立即返回一個Order對象引用——order，不會等待委托單成交與否。
create_task會在下單函數發送出指令包后(執行結束)停止事件循環，(主線程在執行時事件循環可能已經是停止狀態)，需要在調用wait_update啟動事件循環時再從隊列取出指令包并發送向服務端。
17.4、create_task()：
create_task用來把協程打包成Task對象，以便于在事件循環中并發執行，我們看下函數的代碼：
def create_task(self, coro: asyncio.coroutine) -> asyncio.Task:
task = self._loop.create_task(coro) #把協程打包成Task
# 獲取當前正在運行的Task
current_task = asyncio.Task.current_task(loop=self._loop)\
if (sys.version_info[0] == 3 and sys.version_info[1] < 7) else asyncio.current_task(loop=self._loop)
# 當前Task沒有正在運行，則將剛創建的task添加進_tasks
if current_task is None:
self._tasks.add(task)
task.add_done_callback(self._on_task_done) #為task添加結束時會調用的函數
return task #返回task
函數asyncio.current_task(loop=self._loop)用來返回正在運行的Task，如果沒有正在運行的Task則返回None。
_tasks是由api維護的所有根task，不包含子task，子task由其父task維護。
add_done_callback()用來為Task添加一個回調，回調將在 Task 對象完成時被運行。
_on_task_done()函數用來將執行結束的task從_tasks里移除，并停止事件循環，執行結束包括正常結束和遇到異常結束。函數代碼如下：
def _on_task_done(self, task):
"""當由 api 維護的 task 執行完成后取出運行中遇到的例外并停止 ioloop"""
try:
exception = task.exception()#返回 Task 對象的異常，如果沒有異常返回None
if exception:
self._exceptions.append(exception)
except asyncio.CancelledError:
pass
finally:
self._tasks.remove(task)
self._loop.stop()
self._loop.stop()停止事件循環，以使wait_update()釋放，讓進程后續任務獲得動作機會，并等待再次調用wait_update()。
TqSdk中大量用到了create_task創建Task，而Task執行結束后會調用回調函數_on_task_done()停止事件循環，而且主線程在執行時(取得了控制權)事件循環可能已經是停止狀態，因此需要循環調用wait_update()再次開啟事件循環以執行Task。
17.5、TqChan：
TqChan定義在模塊channel中，TqChan是異步隊列asyncio.Queue的子類，TqSdk中大量用到了TqChan，TqSdk各組件間通過TqChan傳遞數據，一個組件向TqChan放入數據，另一個組件從TqChan里取出數據。
TqChan里定義了發送數據和接收數據的函數，因此用TqChan可以連接收、發組件，使組件間建立通信。
數據在組件間單向傳遞，由TqChan連接的組件構成了生產者、消費者模型。
我們看下TqChan的主要代碼，代碼各部分的含義注釋的很清楚了：
class TqChan(asyncio.Queue):
"""用于協程間通訊的channel"""
_chan_id: int = 0
def __init__(self, api: 'TqApi', last_only: bool = False, logger = None,
chan_name: str = "") -> None:
"""創建channel實例Args:api (tqsdk.api.TqApi): TqApi 實例last_only (bool): 為True時只存儲最后一個發送到channel的對象"""
TqChan._chan_id += 1
asyncio.Queue.__init__(self, loop=api._loop)
self._last_only = last_only
self._closed = False
# 關閉函數
async def close(self) -> None:
"""關閉channel,并向隊列放入一個None值關閉后send將不起作用，因此recv在收完剩余數據后會立即返回None"""
if not self._closed:
self._closed = True
await asyncio.Queue.put(self, None)
#發送數據的函數
async def send(self, item: Any) -> None:
"""異步發送數據到channel中Args:item (any): 待發送的對象"""
if not self._closed:
if self._last_only: #只存儲最新數據
while not self.empty():
asyncio.Queue.get_nowait(self)#取出全部歷史數據再放入最新數據
await asyncio.Queue.put(self, item) #放入新數據，如果隊列已滿則阻塞等待
#發送數據的函數
def send_nowait(self, item: Any) -> None:
"""類似send函數，但是立即發送數據到channel中Args:item (any): 待發送的對象Raises:asyncio.QueueFull: 如果channel已滿則會拋出 asyncio.QueueFull"""
if not self._closed:
if self._last_only:
while not self.empty():
asyncio.Queue.get_nowait(self)
asyncio.Queue.put_nowait(self, item) #立即向隊列中放入數據
#接收數據的函數
async def recv(self) -> Any:
"""異步接收channel中的數據，如果channel中沒有數據則一直等待Returns:any: 收到的數據，如果channel已被關閉則會立即收到None"""
if self._closed and self.empty(): #channel已關閉且已空
return None #返回None值
item = await asyncio.Queue.get(self) #取出channel里的數據，若無則阻塞等待
return item #返回取到的值
#接收數據的函數
def recv_nowait(self) -> Any:
"""類似recv，但是立即接收channel中的數據Returns:any: 收到的數據，如果channel已被關閉則會立即收到NoneRaises:asyncio.QueueFull: 如果channel中沒有數據則會拋出 asyncio.QueueEmpty"""
if self._closed and self.empty(): #channel已關閉且已空
return None #返回None值
item = asyncio.Queue.get_nowait(self) #立即取出隊列中的數據
return item #返回取出的數據
#接收最新數據的函數
def recv_latest(self, latest: Any) -> Any:
"""嘗試立即接收channel中的最后一個數據Args:latest (any): 如果當前channel中沒有數據或已關閉則返回該對象Returns:any: channel中的最后一個數據"""
while (self._closed and self.qsize() > 1) or (not self._closed and not self.empty()):
latest = asyncio.Queue.get_nowait(self)
return latest
#重寫的__iter__方法，返回自身的異步迭代器
def __aiter__(self):
return self
#重寫的__next__方法，返回異步迭代器下一個元素
async def __anext__(self):
value = await asyncio.Queue.get(self) #如果隊列無元素，則阻塞直到有數據
if self._closed and self.empty():
raise StopAsyncIteration
return value
#重寫的 __enter__方法，使channel可用在上下文管理語句async with中開啟自身
async def __aenter__(self):
return self
##重寫的__exit__方法，使channel可用在上下文管理語句async with中以退出自身
async def __aexit__(self, exc_type, exc, tb):
await self.close()
TqSdk中大量用到了TqChan在組件間收發數據，當事件循環被stop停止時，收數據一端執行item = await asyncio.Queue.get(self)時會掛起自身并交出控制權給事件循環的調用方，調用方再次啟動事件循環時，事件循環繼續輪詢執行task。
17.6、register_update_notify()：
register_update_notify()函數用于把業務數據注冊到TqChan，實際上是把TqChan添加到業務對象的_listener屬性里，當業務對象更新時會向TqChan添加一個True值，當TqChan為空時則等待業務對象更新。
我們先看一個以TqChan實例在協程中接收數據更新的例子：
from tqsdk import TqApi, TqAuth
api = TqApi(auth=TqAuth("信易賬號", "密碼"))
quote = api.get_quote("CFFEX.T2103") #訂閱盤口行情
#定義一個協程
async def func():
from tqsdk.channel import TqChan #導入TqChan
chan = TqChan(api,last_only=True) #實例化TqChan，接收數據更新
quote["_listener"].add(chan) #把chan添加進quote的_listener屬性
async for p in chan: #若quote有更新會執行循環體，如無更新則阻塞等待
print(p)
print(quote.datetime,quote.last_price) #打印盤口時間和最新價
break
await chan.close() #chan使用完關閉
return quote.instrument_name,quote.instrument_name #返回值
task=api.create_task(func()) #把協程打包成Task
while True:
api.wait_update()
if task.done(): #Task結束后獲取協程返回值
print(task.result())
'''輸出結果為：True2021-02-05 13:11:02.300000 97.3('債十2103', 1615532400.0)('債十2103', 1615532400.0)('債十2103', 1615532400.0)'''
register_update_notify()函數是對上述代碼的簡化，再用with語句管理上下文，例如：
from tqsdk import TqApi, TqAuth
api = TqApi(auth=TqAuth("信易賬號", "密碼"))
quote = api.get_quote("CFFEX.T2103") #訂閱盤口行情
#定義一個協程
async def func():
async with api.register_update_notify(quote) as chan: #把quote注冊到chan
async for p in chan: #若quote有更新會執行循環體，如無更新則阻塞等待
print(p)
print(quote.datetime,quote.last_price) #打印盤口時間和最新價
break
return quote.instrument_name,quote.instrument_name #返回值
task=api.create_task(func()) #把協程打包成Task
while True:
api.wait_update()
if task.done(): #Task結束后獲取協程返回值
print(task.result())
'''輸出結果為：True2021-02-05 13:48:53.800000 97.26('債十2103', '債十2103')('債十2103', '債十2103')('債十2103', '債十2103')'''
若async for p in chan循環不用break跳出，則會隨quote更新循環執行，若quote無更新，比如停盤，異步迭代函數__anext__()里將阻塞，循環也跟著阻塞，等待再次收到quote更新。
17.7、wait_update():
wait_update用于等待業務更新，我們結合其代碼分析下其執行機制：
def wait_update(self, deadline: Optional[float] = None) -> None:
if self._loop.is_running(): #wait_update被放入了事件循環里
raise Exception("不能在協程中調用 wait_update, 如需在協程中等待業務數據更新請使用 register_update_notify")
elif asyncio._get_running_loop():
raise Exception(
"TqSdk 使用了 python3 的原生協程和異步通訊庫 asyncio，您所使用的 IDE 不支持 asyncio, 請使用 pycharm 或其它支持 asyncio 的 IDE")
self._wait_timeout = False #是否觸發超時
# 先嘗試執行各個task,再請求下個業務數據
self._run_until_idle()
# 總會發送 serial_extra_array 數據，由 TqWebHelper 處理
for _, serial in self._serials.items():
self._process_serial_extra_array(serial)
# 上句發送數據創建的有task，先嘗試執行各個task,再請求下個業務數據
self._run_until_idle()
#非api副本，且已收到了上次返回的更新數據，再次請求新數據
if not self._is_slave and self._diffs:
self._send_chan.send_nowait({
"aid": "peek_message"
})
# 先收取數據再判斷 deadline, 避免當超時立即觸發時無法接收數據
update_task = self.create_task(self._fetch_msg()) #從服務端收取數據
#超時后重置self._wait_timeout為True，并停止事件循環
deadline_handle = None if deadline is None else self._loop.call_later(max(0, deadline - time.time()),
self._set_wait_timeout)
try: #未觸發超時且無待處理的新數據，啟動事件循環執行全部Task
while not self._wait_timeout and not self._pending_diffs:
self._run_once() #未設置超時也未收到新數據，將在此阻塞
return len(self._pending_diffs) != 0 #True:還有待處理數據，False:數據已處理完或超時未收到數據
finally: #處理待處理的數據，將數據合并到self._data
self._diffs = self._pending_diffs
self._pending_diffs = []
# 清空K線更新范圍，避免在 wait_update 未更新K線時仍通過 is_changing 的判斷
self._klines_update_range = {}
for d in self._diffs:
# 判斷賬戶類別, 對股票和期貨的 trade 數據分別進行處理
if "trade" in d:
for k, v in d.get('trade').items():
prototype = self._security_prototype if self._account._is_stock_type(k) else self._prototype
_merge_diff(self._data, {'trade': {k: v} }, prototype, False)
# 非交易數據均按照期貨處理邏輯
diff_without_trade = {k : v for k, v in d.items() if k != "trade"}
if diff_without_trade:
_merge_diff(self._data, diff_without_trade, self._prototype, False)
for query_id, query_result in d.get("symbols", {}).items():
if query_id.startswith("PYSDK_quote") and query_result.get("error", None) is None:
quotes = _symbols_to_quotes(query_result)
_merge_diff(self._data, {"quotes": quotes}, self._prototype, False)
for _, serial in self._serials.items():
# K線df的更新與原始數據、left_id、right_id、more_data、last_id相關，其中任何一個發生改變都應重新計算df
# 注：訂閱某K線后再訂閱合約代碼、周期相同但長度更短的K線時, 服務器不會再發送已有數據到客戶端，即chart發生改變但內存中原始數據未改變。
# 檢測到K線數據或chart的任何字段發生改變則更新serial的數據
if self.is_changing(serial["df"]) or self.is_changing(serial["chart"]):
if len(serial["root"]) == 1: # 訂閱單個合約
self._update_serial_single(serial)
else: # 訂閱多個合約
self._update_serial_multi(serial)
if deadline_handle: #取消超時回調
deadline_handle.cancel()
update_task.cancel() #取消收取新業務task
# 最后處理 raise Exception，保證不會因為拋錯導致后面的代碼沒有執行
for d in self._diffs:
for query_id, query_result in d.get("symbols", {}).items():
if query_result.get("error", None):
raise Exception(f"查詢合約服務報錯 {query_result['error']}")
從wait_update的代碼可知，wait_update的工作可分成四大塊：1、先執行事件循環中存在的task
2、向服務端請求新數據
3、事件循環輪詢執行未完成的task，若未設置超時也未收到新數據，將阻塞
4、收到了新數據，停止事件循環，用新數據更新_data，等待下次調用wait_update
wait_update其實是事件循環的調用方(執行self._loop.run_forever())，因此，wait_update的核心工作是開啟事件循環。
開啟事件循環的函數：
def _run_once(self):
"""執行 ioloop 直到 ioloop.stop 被調用"""
if not self._exceptions:
self._loop.run_forever()
if self._exceptions:
raise self._exceptions.pop(0)
def _run_until_idle(self):
"""執行 ioloop 直到沒有待執行任務"""
while self._check_rev != self._event_rev:
#用來追蹤是否有任務未完成并等待執行
check_handle = self._loop.call_soon(self._check_event, self._event_rev + 1)
try:
self._run_once()
finally:
check_handle.cancel()
函數_run_until_idle中調用_run_once，核心工作就是執行self._loop.run_forever()來開啟事件循環。
事件循環里有各種task，比如交易策略、業務處理任務等，事件循環會輪詢執行各個task，當task執行結束或收到新數據時，事件循環會被stop停止，事件循環被停止才可以將控制權交給調用方wait_update(比如task執行await asyncio.Queue.get()時讓出控制權)，執行wait_update后續代碼，用新數據更新業務字段，wait_update執行完后之后，主程序會再次調用wait_update再次開啟事件循環(在主程序while循環中)，事件循環接著上次停止的上下文狀態繼續執行未完成的task。
即：task執行結束或收到新數據時，會停止事件循環并讓出控制權給調用方wait_update使wait_update執行結束。主程序調用wait_update時則開啟事件循環。
wait_update是事件循環的調用方，因此，wait_update不能用在事件循環中，函數代碼開頭部分會先檢查wait_update是否被放入了事件循環。
事件循環每次只運行一個task，task執行結束或收到業務更新使事件循環停止，才能讓出控制權給wait_update使后續任務得到執行，否則事件循環保持運行，主程序將阻塞在wait_update，停止后的事件循環還需要重新開啟以恢復執行未完成的task及繼續收取新數據，因此，應在主程序中將wait_update放在while True循環中循環調用，即可隨著業務更新對事件循環啟、停操作。
數據流通過隊列TqChan傳遞，隊列中有數據才能get出，否則將阻塞，因此task阻塞實際發生在get阻塞時，若事先沒有訂閱數據或已停盤，隊列無法get出數據，事件循環也沒有被stop而保持運行等待get，則事件循環無法讓出控制權，主程序將阻塞在wait_update。
若是設置了超時，則超時后會停止事件循環，超時語句為：
self._loop.call_later(max(0, deadline - time.time()),self._set_wait_timeout)
loop.call_later(delay, callback, *args, context=None)：安排 callback 在給定的 delay 秒(可以是 int 或者 float)后被調用。
因此事件循環超時后執行了函數self._set_wait_timeout，代碼為：
def _set_wait_timeout(self):
self._wait_timeout = True #重置超時變量為True
self._loop.stop() #停止事件循環
即超時后也會主動停止事件循環以讓出控制權給wait_update。
總結
以上是生活随笔為你收集整理的python应用于期货_Python期货量化交易基础教程（17）的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 内网集群 无法通信_记一次集群内无可用h
- 下一篇： python删除空白没有显示_删除Pyt