16.14、異步任務(wù)：

16.14.1、使用協(xié)程任務(wù)：

函數(shù)create_task()用來(lái)創(chuàng)建協(xié)程任務(wù)，并將任務(wù)加入事件循環(huán)以實(shí)現(xiàn)異步并發(fā)。

wait_update()不能用在協(xié)程中，若在協(xié)程中等待業(yè)務(wù)更新，可調(diào)用register_update_notify函數(shù)把業(yè)務(wù)數(shù)據(jù)注冊(cè)到TqChan，當(dāng)業(yè)務(wù)數(shù)據(jù)有更新時(shí)會(huì)通知該TqChan，在協(xié)程里就可以用實(shí)時(shí)更新的業(yè)務(wù)數(shù)據(jù)運(yùn)算。例如：

from tqsdk import TqApi, TqAuth

api = TqApi(auth=TqAuth("信易賬號(hào)", "密碼"))

quote1 = api.get_quote("CFFEX.T2103")

quote2 = api.get_quote("CFFEX.TF2103")

#帶有業(yè)務(wù)更新的協(xié)程

async def demo(quote):

#將quote注冊(cè)到TqChan命名為update_chan

async with api.register_update_notify(quote) as update_chan:

async for _ in update_chan: #遍歷隊(duì)列通知

print('品種：',quote.instrument_name,'最新價(jià)：',quote.last_price)

#無(wú)業(yè)務(wù)更新的協(xié)程

async def func():

return quote1.instrument_name,quote2.instrument_name

# 創(chuàng)建task1、task2，把quote1、quote2注冊(cè)到TqChan

task1=api.create_task(demo(quote1))

task2=api.create_task(demo(quote2))

#把帶有返回值的協(xié)程創(chuàng)建成task3

task3=api.create_task(func())

while True:

api.wait_update()

if task3.done(): #task3結(jié)束后，獲取返回值

print(task3.result())

'''輸出結(jié)果為：('債十2103', '債五2103')品種： 債十2103 最新價(jià)： 97.435('債十2103', '債五2103')品種： 債十2103 最新價(jià)： 97.435('債十2103', '債五2103')品種： 債十2103 最新價(jià)： 97.43('債十2103', '債五2103')品種： 債十2103 最新價(jià)： 97.43'''

16.14.2、使用多線程：

當(dāng)用戶(hù)策略實(shí)例很多，導(dǎo)致網(wǎng)絡(luò)連接數(shù)無(wú)法容納時(shí)，可以使用多線程。首先需要在主線程中創(chuàng)建一個(gè) TqApi 實(shí)例 api_master，并用 TqApi.copy 函數(shù)創(chuàng)建多個(gè)slave副本，把slave副本用在多個(gè)線程中，主線程里的api_master 仍然需要持續(xù)調(diào)用 wait_update。

子線程和主線程其實(shí)是運(yùn)行在同一個(gè)事件循環(huán)里，如果在子線程里調(diào)用api_slave.close()會(huì)引發(fā)主線程事件循環(huán)關(guān)閉的異常，如果主線程里調(diào)用api_master.close()，子線程可能因等待事件循環(huán)響應(yīng)而阻塞，若想讓子線程和主線程一起退出，可設(shè)置子線程為守護(hù)線程。

使用多線程需要自定義一個(gè)線程類(lèi)，并重寫(xiě)run函數(shù)，在run函數(shù)里執(zhí)行策略代碼，例如：

import threading

from tqsdk import TqApi, TqAuth

#自定義線程類(lèi)

class WorkerThread(threading.Thread):

def __init__(self, api, symbol):

threading.Thread.__init__(self)

self.api = api #初始化參數(shù)

self.symbol = symbol #初始化參數(shù)

#重寫(xiě)run函數(shù)，策略代碼寫(xiě)在run函數(shù)中

def run(self):

SHORT = 30 # 短周期

LONG = 60 # 長(zhǎng)周期

data_length = LONG + 2 # k線數(shù)據(jù)長(zhǎng)度

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"): # 產(chǎn)生新k線:重新計(jì)算SMA

short_avg = ma(klines["close"], SHORT) # 短周期

long_avg = ma(klines["close"], LONG) # 長(zhǎng)周期

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__":

#主線程創(chuàng)建TqApi實(shí)例

api_master = TqApi(auth=TqAuth("信易賬號(hào)", "密碼"))

# 實(shí)例化線程類(lèi)，傳入TqApi實(shí)例的副本api_master.copy()

thread1 = WorkerThread(api_master.copy(), "SHFE.cu1901")

thread2 = WorkerThread(api_master.copy(), "SHFE.rb1901")

# 啟動(dòng)線程實(shí)例

thread1.start()

thread2.start()

while True:

api_master.wait_update() #主線程保持對(duì)wait_update()的調(diào)用

當(dāng)線程太多時(shí)，操作系統(tǒng)因調(diào)度線程，可能把主要工作都用在了調(diào)度線程上，而降低了多線程的效率，更宜使用異步協(xié)程實(shí)現(xiàn)多策略。

16.14.3、使用多進(jìn)程：

當(dāng)程序比較耗CPU時(shí)，可以采用多進(jìn)程，比如回測(cè)時(shí)，需要對(duì)大量的數(shù)據(jù)計(jì)算，可以用多個(gè)進(jìn)程同時(shí)回測(cè)多個(gè)品種，注意: 由于服務(wù)器流控限制, 同時(shí)執(zhí)行的回測(cè)任務(wù)請(qǐng)勿超過(guò)10個(gè)，例如：

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("信易賬戶(hù)", "賬戶(hù)密碼"))

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, "最終權(quán)益=", acc.account.balance) # 每次回測(cè)結(jié)束時(shí), 輸出使用的參數(shù)和最終權(quán)益

if __name__ == '__main__':

#提供凍結(jié)以產(chǎn)生 Windows 可執(zhí)行文件的支持，在非 Windows 平臺(tái)上是無(wú)效的

multiprocessing.freeze_support()

p = Pool(4) # 進(jìn)程池, 建議小于cpu數(shù)

for s in range(20, 40):

p.apply_async(MyStrategy, args=(s,)) # 把20個(gè)回測(cè)任務(wù)交給進(jìn)程池執(zhí)行

print('Waiting for all subprocesses done...')

p.close()

p.join()

print('All subprocesses done.')

17、TqSdk部分函數(shù)解讀

17.1、DIFF 協(xié)議：

DIFF (Differential Information Flow for Finance) 是一個(gè)基于websocket和json的應(yīng)用層協(xié)議。websocket是全雙工通信，當(dāng)客戶(hù)端和服務(wù)器端建立連接后，就可以相互發(fā)數(shù)據(jù)，建立連接又稱(chēng)為“握手”，“握手”成功就可以建立通信了，不用在每次需要傳輸信息時(shí)重新建立連接，即不會(huì)“掉線”。json是數(shù)據(jù)存儲(chǔ)格式，json數(shù)據(jù)可以方便的反序列化為Python數(shù)據(jù)。

DIFF協(xié)議可以簡(jiǎn)單的理解為服務(wù)端和客戶(hù)端的通信方式，協(xié)議規(guī)定了數(shù)據(jù)格式，以便于服務(wù)端和客戶(hù)端可以解讀對(duì)方發(fā)來(lái)的數(shù)據(jù)。

DIFF 協(xié)議分為兩部分:數(shù)據(jù)訪問(wèn)和數(shù)據(jù)傳輸。

17.1.1、數(shù)據(jù)傳輸：

DIFF 協(xié)議要求服務(wù)端將業(yè)務(wù)數(shù)據(jù)以JSON Merge Patch的格式推送給客戶(hù)端，JSON Merge Patch的格式形如Python字典，可以在客戶(hù)端反序列化為Python字典(其實(shí)是映射類(lèi)型Entity)。例如：

{

"aid": "rtn_data", # 業(yè)務(wù)信息截面更新

"data": [ # 數(shù)據(jù)更新數(shù)組

{

"balance": 10237421.1, # 賬戶(hù)資金

},

{

"float_profit": 283114.780999997, # 浮動(dòng)盈虧

},

{

"quotes":{

"SHFE.cu1612": {

"datetime": "2016-12-30 14:31:02.000000",

"last_price": 36605.0, # 最新價(jià)

"volume": 25431, # 成交量

"pre_close": 36170.0, # 昨收

}

}

}

]

}aid 字段值即為數(shù)據(jù)包類(lèi)型，"aid":"rtn_data"表示該包的類(lèi)型為業(yè)務(wù)信息截面更新包。整個(gè) data 數(shù)組相當(dāng)于一個(gè)事務(wù)，其中的每一個(gè)元素都是一個(gè) JSON Merge Patch，處理完整個(gè)數(shù)組后業(yè)務(wù)截面即完成了從上一個(gè)時(shí)間截面推進(jìn)到下一個(gè)時(shí)間截面。

DIFF 協(xié)議要求客戶(hù)端發(fā)送 peek_message 數(shù)據(jù)包以獲得業(yè)務(wù)信息截面更新，例如：

{

"aid": "peek_message"

}服務(wù)端在收到 peek_message 數(shù)據(jù)包后應(yīng)檢查是否有數(shù)據(jù)更新，如果有則應(yīng)將更新內(nèi)容立即發(fā)送給客戶(hù)端，如果沒(méi)有則應(yīng)等到有更新發(fā)生時(shí)再回應(yīng)客戶(hù)端。

服務(wù)端發(fā)送 rtn_data 數(shù)據(jù)包后可以等收到下一個(gè) peek_message 后再發(fā)送下一個(gè) rtn_data 數(shù)據(jù)包。

一個(gè)簡(jiǎn)單的客戶(hù)端實(shí)現(xiàn)可以在連接成功后及每收到一個(gè) rtn_data 數(shù)據(jù)包后發(fā)送一個(gè) peek_message 數(shù)據(jù)包，這樣當(dāng)客戶(hù)端帶寬不足時(shí)會(huì)自動(dòng)降低業(yè)務(wù)信息截面的更新頻率以適應(yīng)低帶寬。

當(dāng)數(shù)據(jù)包中的 aid 字段不是 rtn_data 或 peek_message 則表示該包為一個(gè)指令包，具體指令由各業(yè)務(wù)模塊定義，例如 subscribe_quote 表示訂閱行情，insert_order 表示下單。由于客戶(hù)端和服務(wù)端存在網(wǎng)絡(luò)通訊延遲，客戶(hù)端的指令需要過(guò)一段時(shí)間才會(huì)影響到業(yè)務(wù)信息截面中的業(yè)務(wù)數(shù)據(jù)，為了使客戶(hù)端能分辨出服務(wù)端是否處理了該指令，通常服務(wù)端會(huì)將客戶(hù)端的請(qǐng)求以某種方式體現(xiàn)在截面中(具體方式由各業(yè)務(wù)模塊定義)。例如 subscribe_quote 訂閱行情時(shí)服務(wù)端會(huì)將業(yè)務(wù)截面中的 ins_list 字段更新為客戶(hù)端訂閱的合約列表，這樣當(dāng)客戶(hù)端檢查服務(wù)端發(fā)來(lái)的業(yè)務(wù)截面時(shí)如果 ins_list 包含了客戶(hù)端訂閱的某個(gè)合約說(shuō)明服務(wù)端處理了訂閱指令，但若 quotes 沒(méi)有該合約則說(shuō)明該合約不存在訂閱失敗。

服務(wù)端發(fā)送包含"aid":"rtn_data"字段的業(yè)務(wù)數(shù)據(jù)截面更新包，客戶(hù)端發(fā)送包含"aid":"peek_message"字段的數(shù)據(jù)包請(qǐng)求業(yè)務(wù)數(shù)據(jù)截面，或發(fā)送包含"aid":"subscribe_quote "、"aid":"insert_order"等字段的指令包，如此，服務(wù)端和客戶(hù)端相互發(fā)信息，服務(wù)端和客戶(hù)端根據(jù)字段識(shí)別數(shù)據(jù)及處理數(shù)據(jù)。

17.1.2、數(shù)據(jù)訪問(wèn)：

DIFF 協(xié)議要求服務(wù)端維護(hù)一個(gè)業(yè)務(wù)信息截面，例如：

{

"account_id": "41007684", # 賬號(hào)

"static_balance": 9954306.319000003, # 靜態(tài)權(quán)益

"balance": 9963216.550000003, # 賬戶(hù)資金

"available": 9480176.150000002, # 可用資金

"float_profit": 8910.231, # 浮動(dòng)盈虧

"risk_ratio": 0.048482375, # 風(fēng)險(xiǎn)度

"using": 11232.23, # 占用資金

"position_volume": 12, # 持倉(cāng)總手?jǐn)?shù)

"ins_list": "SHFE.cu1609,...." # 行情訂閱的合約列表

"quotes":{ # 所有訂閱的實(shí)時(shí)行情

"SHFE.cu1612": {

"instrument_id": "SHFE.cu1612",

"datetime": "2016-12-30 13:21:32.500000",

"ask_priceN": 36590.0, #賣(mài)N價(jià)

"ask_volumeN": 121, #賣(mài)N量

"bid_priceN": 36580.0, #買(mǎi)N價(jià)

"bid_volumeN": 3, #買(mǎi)N量

"last_price": 36580.0, # 最新價(jià)

"highest": 36580.0, # 最高價(jià)

"lowest": 36580.0, # 最低價(jià)

"amount": 213445312.5, # 成交額

"volume": 23344, # 成交量

"open_interest": 23344, # 持倉(cāng)量

"pre_open_interest": 23344, # 昨持

"pre_close": 36170.0, # 昨收

"open": 36270.0, # 今開(kāi)

"close" : "-", # 收盤(pán)

"lower_limit": 34160.0, #跌停

"upper_limit": 38530.0, #漲停

"average": 36270.1 #均價(jià)

"pre_settlement": 36270.0, # 昨結(jié)

"settlement": "-", # 結(jié)算價(jià)

},

...

}

}

對(duì)應(yīng)的客戶(hù)端也維護(hù)了一個(gè)該截面的鏡像，因此業(yè)務(wù)層可以簡(jiǎn)單同步的訪問(wèn)到全部業(yè)務(wù)數(shù)據(jù)。

TqSdk即是客戶(hù)端，TqSdk把收到的業(yè)務(wù)數(shù)據(jù)截面以上面的格式合并到_data屬性里，_data為多層嵌套的映射類(lèi)型Entity，業(yè)務(wù)數(shù)據(jù)例如“quotes”，也是Entity，其“鍵”是合約代碼，例如“SHFE.cu1612”，其“值”是最終的業(yè)務(wù)數(shù)據(jù)——Quote對(duì)象，業(yè)務(wù)函數(shù)get_quote()便是把_data里的Quote對(duì)象的一個(gè)引用返回給調(diào)用方，調(diào)用方獲得的是Quote對(duì)象的動(dòng)態(tài)引用。

_data是可變映射類(lèi)型，會(huì)接收服務(wù)端發(fā)來(lái)的更新，因此業(yè)務(wù)函數(shù)返回的對(duì)象引用也會(huì)指向隨時(shí)更新的業(yè)務(wù)數(shù)據(jù)。

17.2、業(yè)務(wù)函數(shù)：

以get_quote()為例，上節(jié)已經(jīng)介紹了get_quote()與_data的關(guān)系，現(xiàn)在我們結(jié)合函數(shù)的代碼再看下其執(zhí)行過(guò)程，我們只取代碼的主要部分：

def get_quote(self, symbol: str) -> Quote:

# 從_data屬性中提取Quote

quote = _get_obj(self._data, ["quotes", symbol], self._prototype["quotes"]["#"])

# 若合約symbol是新添加的合約，則向服務(wù)端發(fā)送訂閱該合約的指令包

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

其他的業(yè)務(wù)函數(shù)工作邏輯類(lèi)似。業(yè)務(wù)對(duì)象Quote、Trade、Order、Position、Account等都是Entity的子類(lèi)，可以像類(lèi)一樣獲取其屬性，也可以像字典一樣使用。業(yè)務(wù)對(duì)象在模塊objs中定義。

17.3、insert_order()：

insert_order用來(lái)下單，我們只截取主要代碼看一下執(zhí)行過(guò)程：

def insert_order(...) -> Order:

"""發(fā)送下單指令. **注意: 指令將在下次調(diào)用** :py:meth:`~tqsdk.api.TqApi.wait_update` **時(shí)發(fā)出**"""

if self._loop.is_running(): #事件循環(huán)正在運(yùn)行

# 把下單請(qǐng)求函數(shù)打包成task排入事件循環(huán)

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: #事件循環(huán)還未運(yùn)行

#打包一個(gè)指令包

pack = self._get_insert_order_pack(...)

#發(fā)送指令包

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 #返回委托單

#發(fā)送指令包函數(shù)

def _send_pack(self, pack):

#立即向隊(duì)列發(fā)送指令包

if not self._is_slave:

self._send_chan.send_nowait(pack)

else:

self._master._slave_send_pack(pack)

#下單請(qǐng)求函數(shù)

async def _insert_order_async(...):

#打包一個(gè)指令包

pack = self._get_insert_order_pack(...)

#發(fā)送指令包

self._send_pack(pack)

下單的主要流程為：用協(xié)程任務(wù)打包一個(gè)指令包再發(fā)出去。create_task是無(wú)阻塞的，創(chuàng)建完task立即返回，get_order獲取委托單也是無(wú)阻塞的，因此insert_order執(zhí)行后會(huì)立即返回一個(gè)Order對(duì)象引用——order，不會(huì)等待委托單成交與否。

create_task會(huì)在下單函數(shù)發(fā)送出指令包后(執(zhí)行結(jié)束)停止事件循環(huán)，(主線程在執(zhí)行時(shí)事件循環(huán)可能已經(jīng)是停止?fàn)顟B(tài))，需要在調(diào)用wait_update啟動(dòng)事件循環(huán)時(shí)再?gòu)年?duì)列取出指令包并發(fā)送向服務(wù)端。

17.4、create_task()：

create_task用來(lái)把協(xié)程打包成Task對(duì)象，以便于在事件循環(huán)中并發(fā)執(zhí)行，我們看下函數(shù)的代碼：

def create_task(self, coro: asyncio.coroutine) -> asyncio.Task:

task = self._loop.create_task(coro) #把協(xié)程打包成Task

# 獲取當(dāng)前正在運(yùn)行的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)

# 當(dāng)前Task沒(méi)有正在運(yùn)行，則將剛創(chuàng)建的task添加進(jìn)_tasks

if current_task is None:

self._tasks.add(task)

task.add_done_callback(self._on_task_done) #為task添加結(jié)束時(shí)會(huì)調(diào)用的函數(shù)

return task #返回task

函數(shù)asyncio.current_task(loop=self._loop)用來(lái)返回正在運(yùn)行的Task，如果沒(méi)有正在運(yùn)行的Task則返回None。

_tasks是由api維護(hù)的所有根task，不包含子task，子task由其父task維護(hù)。

add_done_callback()用來(lái)為T(mén)ask添加一個(gè)回調(diào)，回調(diào)將在 Task 對(duì)象完成時(shí)被運(yùn)行。

_on_task_done()函數(shù)用來(lái)將執(zhí)行結(jié)束的task從_tasks里移除，并停止事件循環(huán)，執(zhí)行結(jié)束包括正常結(jié)束和遇到異常結(jié)束。函數(shù)代碼如下：

def _on_task_done(self, task):

"""當(dāng)由 api 維護(hù)的 task 執(zhí)行完成后取出運(yùn)行中遇到的例外并停止 ioloop"""

try:

exception = task.exception()#返回 Task 對(duì)象的異常，如果沒(méi)有異常返回None

if exception:

self._exceptions.append(exception)

except asyncio.CancelledError:

pass

finally:

self._tasks.remove(task)

self._loop.stop()

self._loop.stop()停止事件循環(huán)，以使wait_update()釋放，讓進(jìn)程后續(xù)任務(wù)獲得動(dòng)作機(jī)會(huì)，并等待再次調(diào)用wait_update()。

TqSdk中大量用到了create_task創(chuàng)建Task，而Task執(zhí)行結(jié)束后會(huì)調(diào)用回調(diào)函數(shù)_on_task_done()停止事件循環(huán)，而且主線程在執(zhí)行時(shí)(取得了控制權(quán))事件循環(huán)可能已經(jīng)是停止?fàn)顟B(tài)，因此需要循環(huán)調(diào)用wait_update()再次開(kāi)啟事件循環(huán)以執(zhí)行Task。

17.5、TqChan：

TqChan定義在模塊channel中，TqChan是異步隊(duì)列asyncio.Queue的子類(lèi)，TqSdk中大量用到了TqChan，TqSdk各組件間通過(guò)TqChan傳遞數(shù)據(jù)，一個(gè)組件向TqChan放入數(shù)據(jù)，另一個(gè)組件從TqChan里取出數(shù)據(jù)。

TqChan里定義了發(fā)送數(shù)據(jù)和接收數(shù)據(jù)的函數(shù)，因此用TqChan可以連接收、發(fā)組件，使組件間建立通信。

數(shù)據(jù)在組件間單向傳遞，由TqChan連接的組件構(gòu)成了生產(chǎn)者、消費(fèi)者模型。

我們看下TqChan的主要代碼，代碼各部分的含義注釋的很清楚了：

class TqChan(asyncio.Queue):

"""用于協(xié)程間通訊的channel"""

_chan_id: int = 0

def __init__(self, api: 'TqApi', last_only: bool = False, logger = None,

chan_name: str = "") -> None:

"""創(chuàng)建channel實(shí)例Args:api (tqsdk.api.TqApi): TqApi 實(shí)例last_only (bool): 為T(mén)rue時(shí)只存儲(chǔ)最后一個(gè)發(fā)送到channel的對(duì)象"""

TqChan._chan_id += 1

asyncio.Queue.__init__(self, loop=api._loop)

self._last_only = last_only

self._closed = False

# 關(guān)閉函數(shù)

async def close(self) -> None:

"""關(guān)閉channel,并向隊(duì)列放入一個(gè)None值關(guān)閉后send將不起作用，因此recv在收完剩余數(shù)據(jù)后會(huì)立即返回None"""

if not self._closed:

self._closed = True

await asyncio.Queue.put(self, None)

#發(fā)送數(shù)據(jù)的函數(shù)

async def send(self, item: Any) -> None:

"""異步發(fā)送數(shù)據(jù)到channel中Args:item (any): 待發(fā)送的對(duì)象"""

if not self._closed:

if self._last_only: #只存儲(chǔ)最新數(shù)據(jù)

while not self.empty():

asyncio.Queue.get_nowait(self)#取出全部歷史數(shù)據(jù)再放入最新數(shù)據(jù)

await asyncio.Queue.put(self, item) #放入新數(shù)據(jù)，如果隊(duì)列已滿則阻塞等待

#發(fā)送數(shù)據(jù)的函數(shù)

def send_nowait(self, item: Any) -> None:

"""類(lèi)似send函數(shù)，但是立即發(fā)送數(shù)據(jù)到channel中Args:item (any): 待發(fā)送的對(duì)象Raises:asyncio.QueueFull: 如果channel已滿則會(huì)拋出 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) #立即向隊(duì)列中放入數(shù)據(jù)

#接收數(shù)據(jù)的函數(shù)

async def recv(self) -> Any:

"""異步接收channel中的數(shù)據(jù)，如果channel中沒(méi)有數(shù)據(jù)則一直等待Returns:any: 收到的數(shù)據(jù)，如果channel已被關(guān)閉則會(huì)立即收到None"""

if self._closed and self.empty(): #channel已關(guān)閉且已空

return None #返回None值

item = await asyncio.Queue.get(self) #取出channel里的數(shù)據(jù)，若無(wú)則阻塞等待

return item #返回取到的值

#接收數(shù)據(jù)的函數(shù)

def recv_nowait(self) -> Any:

"""類(lèi)似recv，但是立即接收channel中的數(shù)據(jù)Returns:any: 收到的數(shù)據(jù)，如果channel已被關(guān)閉則會(huì)立即收到NoneRaises:asyncio.QueueFull: 如果channel中沒(méi)有數(shù)據(jù)則會(huì)拋出 asyncio.QueueEmpty"""

if self._closed and self.empty(): #channel已關(guān)閉且已空

return None #返回None值

item = asyncio.Queue.get_nowait(self) #立即取出隊(duì)列中的數(shù)據(jù)

return item #返回取出的數(shù)據(jù)

#接收最新數(shù)據(jù)的函數(shù)

def recv_latest(self, latest: Any) -> Any:

"""嘗試立即接收channel中的最后一個(gè)數(shù)據(jù)Args:latest (any): 如果當(dāng)前channel中沒(méi)有數(shù)據(jù)或已關(guān)閉則返回該對(duì)象Returns:any: channel中的最后一個(gè)數(shù)據(jù)"""

while (self._closed and self.qsize() > 1) or (not self._closed and not self.empty()):

latest = asyncio.Queue.get_nowait(self)

return latest

#重寫(xiě)的__iter__方法，返回自身的異步迭代器

def __aiter__(self):

return self

#重寫(xiě)的__next__方法，返回異步迭代器下一個(gè)元素

async def __anext__(self):

value = await asyncio.Queue.get(self) #如果隊(duì)列無(wú)元素，則阻塞直到有數(shù)據(jù)

if self._closed and self.empty():

raise StopAsyncIteration

return value

#重寫(xiě)的 __enter__方法，使channel可用在上下文管理語(yǔ)句async with中開(kāi)啟自身

async def __aenter__(self):

return self

##重寫(xiě)的__exit__方法，使channel可用在上下文管理語(yǔ)句async with中以退出自身

async def __aexit__(self, exc_type, exc, tb):

await self.close()

TqSdk中大量用到了TqChan在組件間收發(fā)數(shù)據(jù)，當(dāng)事件循環(huán)被stop停止時(shí)，收數(shù)據(jù)一端執(zhí)行item = await asyncio.Queue.get(self)時(shí)會(huì)掛起自身并交出控制權(quán)給事件循環(huán)的調(diào)用方，調(diào)用方再次啟動(dòng)事件循環(huán)時(shí)，事件循環(huán)繼續(xù)輪詢(xún)執(zhí)行task。

17.6、register_update_notify()：

register_update_notify()函數(shù)用于把業(yè)務(wù)數(shù)據(jù)注冊(cè)到TqChan，實(shí)際上是把TqChan添加到業(yè)務(wù)對(duì)象的_listener屬性里，當(dāng)業(yè)務(wù)對(duì)象更新時(shí)會(huì)向TqChan添加一個(gè)True值，當(dāng)TqChan為空時(shí)則等待業(yè)務(wù)對(duì)象更新。

我們先看一個(gè)以TqChan實(shí)例在協(xié)程中接收數(shù)據(jù)更新的例子：

from tqsdk import TqApi, TqAuth

api = TqApi(auth=TqAuth("信易賬號(hào)", "密碼"))

quote = api.get_quote("CFFEX.T2103") #訂閱盤(pán)口行情

#定義一個(gè)協(xié)程

async def func():

from tqsdk.channel import TqChan #導(dǎo)入TqChan

chan = TqChan(api,last_only=True) #實(shí)例化TqChan，接收數(shù)據(jù)更新

quote["_listener"].add(chan) #把chan添加進(jìn)quote的_listener屬性

async for p in chan: #若quote有更新會(huì)執(zhí)行循環(huán)體，如無(wú)更新則阻塞等待

print(p)

print(quote.datetime,quote.last_price) #打印盤(pán)口時(shí)間和最新價(jià)

break

await chan.close() #chan使用完關(guān)閉

return quote.instrument_name,quote.instrument_name #返回值

task=api.create_task(func()) #把協(xié)程打包成Task

while True:

api.wait_update()

if task.done(): #Task結(jié)束后獲取協(xié)程返回值

print(task.result())

'''輸出結(jié)果為：True2021-02-05 13:11:02.300000 97.3('債十2103', 1615532400.0)('債十2103', 1615532400.0)('債十2103', 1615532400.0)'''

register_update_notify()函數(shù)是對(duì)上述代碼的簡(jiǎn)化，再用with語(yǔ)句管理上下文，例如：

from tqsdk import TqApi, TqAuth

api = TqApi(auth=TqAuth("信易賬號(hào)", "密碼"))

quote = api.get_quote("CFFEX.T2103") #訂閱盤(pán)口行情

#定義一個(gè)協(xié)程

async def func():

async with api.register_update_notify(quote) as chan: #把quote注冊(cè)到chan

async for p in chan: #若quote有更新會(huì)執(zhí)行循環(huán)體，如無(wú)更新則阻塞等待

print(p)

print(quote.datetime,quote.last_price) #打印盤(pán)口時(shí)間和最新價(jià)

break

return quote.instrument_name,quote.instrument_name #返回值

task=api.create_task(func()) #把協(xié)程打包成Task

while True:

api.wait_update()

if task.done(): #Task結(jié)束后獲取協(xié)程返回值

print(task.result())

'''輸出結(jié)果為：True2021-02-05 13:48:53.800000 97.26('債十2103', '債十2103')('債十2103', '債十2103')('債十2103', '債十2103')'''

若async for p in chan循環(huán)不用break跳出，則會(huì)隨quote更新循環(huán)執(zhí)行，若quote無(wú)更新，比如停盤(pán)，異步迭代函數(shù)__anext__()里將阻塞，循環(huán)也跟著阻塞，等待再次收到quote更新。

17.7、wait_update():

wait_update用于等待業(yè)務(wù)更新，我們結(jié)合其代碼分析下其執(zhí)行機(jī)制：

def wait_update(self, deadline: Optional[float] = None) -> None:

if self._loop.is_running(): #wait_update被放入了事件循環(huán)里

raise Exception("不能在協(xié)程中調(diào)用 wait_update, 如需在協(xié)程中等待業(yè)務(wù)數(shù)據(jù)更新請(qǐng)使用 register_update_notify")

elif asyncio._get_running_loop():

raise Exception(

"TqSdk 使用了 python3 的原生協(xié)程和異步通訊庫(kù) asyncio，您所使用的 IDE 不支持 asyncio, 請(qǐng)使用 pycharm 或其它支持 asyncio 的 IDE")

self._wait_timeout = False #是否觸發(fā)超時(shí)

# 先嘗試執(zhí)行各個(gè)task,再請(qǐng)求下個(gè)業(yè)務(wù)數(shù)據(jù)

self._run_until_idle()

# 總會(huì)發(fā)送 serial_extra_array 數(shù)據(jù)，由 TqWebHelper 處理

for _, serial in self._serials.items():

self._process_serial_extra_array(serial)

# 上句發(fā)送數(shù)據(jù)創(chuàng)建的有task，先嘗試執(zhí)行各個(gè)task,再請(qǐng)求下個(gè)業(yè)務(wù)數(shù)據(jù)

self._run_until_idle()

#非api副本，且已收到了上次返回的更新數(shù)據(jù)，再次請(qǐng)求新數(shù)據(jù)

if not self._is_slave and self._diffs:

self._send_chan.send_nowait({

"aid": "peek_message"

})

# 先收取數(shù)據(jù)再判斷 deadline, 避免當(dāng)超時(shí)立即觸發(fā)時(shí)無(wú)法接收數(shù)據(jù)

update_task = self.create_task(self._fetch_msg()) #從服務(wù)端收取數(shù)據(jù)

#超時(shí)后重置self._wait_timeout為T(mén)rue，并停止事件循環(huán)

deadline_handle = None if deadline is None else self._loop.call_later(max(0, deadline - time.time()),

self._set_wait_timeout)

try: #未觸發(fā)超時(shí)且無(wú)待處理的新數(shù)據(jù)，啟動(dòng)事件循環(huán)執(zhí)行全部Task

while not self._wait_timeout and not self._pending_diffs:

self._run_once() #未設(shè)置超時(shí)也未收到新數(shù)據(jù)，將在此阻塞

return len(self._pending_diffs) != 0 #True:還有待處理數(shù)據(jù)，False:數(shù)據(jù)已處理完或超時(shí)未收到數(shù)據(jù)

finally: #處理待處理的數(shù)據(jù)，將數(shù)據(jù)合并到self._data

self._diffs = self._pending_diffs

self._pending_diffs = []

# 清空K線更新范圍，避免在 wait_update 未更新K線時(shí)仍通過(guò) is_changing 的判斷

self._klines_update_range = {}

for d in self._diffs:

# 判斷賬戶(hù)類(lèi)別, 對(duì)股票和期貨的 trade 數(shù)據(jù)分別進(jìn)行處理

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)

# 非交易數(shù)據(jù)均按照期貨處理邏輯

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的更新與原始數(shù)據(jù)、left_id、right_id、more_data、last_id相關(guān)，其中任何一個(gè)發(fā)生改變都應(yīng)重新計(jì)算df

# 注：訂閱某K線后再訂閱合約代碼、周期相同但長(zhǎng)度更短的K線時(shí), 服務(wù)器不會(huì)再發(fā)送已有數(shù)據(jù)到客戶(hù)端，即chart發(fā)生改變但內(nèi)存中原始數(shù)據(jù)未改變。

# 檢測(cè)到K線數(shù)據(jù)或chart的任何字段發(fā)生改變則更新serial的數(shù)據(jù)

if self.is_changing(serial["df"]) or self.is_changing(serial["chart"]):

if len(serial["root"]) == 1: # 訂閱單個(gè)合約

self._update_serial_single(serial)

else: # 訂閱多個(gè)合約

self._update_serial_multi(serial)

if deadline_handle: #取消超時(shí)回調(diào)

deadline_handle.cancel()

update_task.cancel() #取消收取新業(yè)務(wù)task

# 最后處理 raise Exception，保證不會(huì)因?yàn)閽佸e(cuò)導(dǎo)致后面的代碼沒(méi)有執(zhí)行

for d in self._diffs:

for query_id, query_result in d.get("symbols", {}).items():

if query_result.get("error", None):

raise Exception(f"查詢(xún)合約服務(wù)報(bào)錯(cuò) {query_result['error']}")

從wait_update的代碼可知，wait_update的工作可分成四大塊：1、先執(zhí)行事件循環(huán)中存在的task

2、向服務(wù)端請(qǐng)求新數(shù)據(jù)

3、事件循環(huán)輪詢(xún)執(zhí)行未完成的task，若未設(shè)置超時(shí)也未收到新數(shù)據(jù)，將阻塞

4、收到了新數(shù)據(jù)，停止事件循環(huán)，用新數(shù)據(jù)更新_data，等待下次調(diào)用wait_update

wait_update其實(shí)是事件循環(huán)的調(diào)用方(執(zhí)行self._loop.run_forever())，因此，wait_update的核心工作是開(kāi)啟事件循環(huán)。

開(kāi)啟事件循環(huán)的函數(shù)：

def _run_once(self):

"""執(zhí)行 ioloop 直到 ioloop.stop 被調(diào)用"""

if not self._exceptions:

self._loop.run_forever()

if self._exceptions:

raise self._exceptions.pop(0)

def _run_until_idle(self):

"""執(zhí)行 ioloop 直到?jīng)]有待執(zhí)行任務(wù)"""

while self._check_rev != self._event_rev:

#用來(lái)追蹤是否有任務(wù)未完成并等待執(zhí)行

check_handle = self._loop.call_soon(self._check_event, self._event_rev + 1)

try:

self._run_once()

finally:

check_handle.cancel()

函數(shù)_run_until_idle中調(diào)用_run_once，核心工作就是執(zhí)行self._loop.run_forever()來(lái)開(kāi)啟事件循環(huán)。

事件循環(huán)里有各種task，比如交易策略、業(yè)務(wù)處理任務(wù)等，事件循環(huán)會(huì)輪詢(xún)執(zhí)行各個(gè)task，當(dāng)task執(zhí)行結(jié)束或收到新數(shù)據(jù)時(shí)，事件循環(huán)會(huì)被stop停止，事件循環(huán)被停止才可以將控制權(quán)交給調(diào)用方wait_update(比如task執(zhí)行await asyncio.Queue.get()時(shí)讓出控制權(quán))，執(zhí)行wait_update后續(xù)代碼，用新數(shù)據(jù)更新業(yè)務(wù)字段，wait_update執(zhí)行完后之后，主程序會(huì)再次調(diào)用wait_update再次開(kāi)啟事件循環(huán)(在主程序while循環(huán)中)，事件循環(huán)接著上次停止的上下文狀態(tài)繼續(xù)執(zhí)行未完成的task。

即：task執(zhí)行結(jié)束或收到新數(shù)據(jù)時(shí)，會(huì)停止事件循環(huán)并讓出控制權(quán)給調(diào)用方wait_update使wait_update執(zhí)行結(jié)束。主程序調(diào)用wait_update時(shí)則開(kāi)啟事件循環(huán)。

wait_update是事件循環(huán)的調(diào)用方，因此，wait_update不能用在事件循環(huán)中，函數(shù)代碼開(kāi)頭部分會(huì)先檢查wait_update是否被放入了事件循環(huán)。

事件循環(huán)每次只運(yùn)行一個(gè)task，task執(zhí)行結(jié)束或收到業(yè)務(wù)更新使事件循環(huán)停止，才能讓出控制權(quán)給wait_update使后續(xù)任務(wù)得到執(zhí)行，否則事件循環(huán)保持運(yùn)行，主程序?qū)⒆枞趙ait_update，停止后的事件循環(huán)還需要重新開(kāi)啟以恢復(fù)執(zhí)行未完成的task及繼續(xù)收取新數(shù)據(jù)，因此，應(yīng)在主程序中將wait_update放在while True循環(huán)中循環(huán)調(diào)用，即可隨著業(yè)務(wù)更新對(duì)事件循環(huán)啟、停操作。

數(shù)據(jù)流通過(guò)隊(duì)列TqChan傳遞，隊(duì)列中有數(shù)據(jù)才能get出，否則將阻塞，因此task阻塞實(shí)際發(fā)生在get阻塞時(shí)，若事先沒(méi)有訂閱數(shù)據(jù)或已停盤(pán)，隊(duì)列無(wú)法get出數(shù)據(jù)，事件循環(huán)也沒(méi)有被stop而保持運(yùn)行等待get，則事件循環(huán)無(wú)法讓出控制權(quán)，主程序?qū)⒆枞趙ait_update。

若是設(shè)置了超時(shí)，則超時(shí)后會(huì)停止事件循環(huán)，超時(shí)語(yǔ)句為：

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)后被調(diào)用。

因此事件循環(huán)超時(shí)后執(zhí)行了函數(shù)self._set_wait_timeout，代碼為：

def _set_wait_timeout(self):

self._wait_timeout = True #重置超時(shí)變量為T(mén)rue

self._loop.stop() #停止事件循環(huán)

即超時(shí)后也會(huì)主動(dòng)停止事件循環(huán)以讓出控制權(quán)給wait_update。

總結(jié)

以上是生活随笔為你收集整理的python应用于期货_Python期货量化交易基础教程（17）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。