群益API行情串接(四)


像股票這樣的金融商品百百種,加上市場的變化性與多樣的市場規則,使得這類的分析總是相當困難。但相關數據取得與蒐集的管道也相當稀少,甚至難以整合與上手。國內幾家券商提供之API服務便顯得相當地重要。

前言

這個系列至第三篇為止,已經把當日Tick的回補功能做了基本的實現。本篇要另外來提一下即時的資料獲取功能。不同時間、頻率的資料,能夠做到的應用不同。頻率越高,產生的資料意義會不同,甚至也可能因為更貼近原始數據,結構也會不同。因此,如果能夠獲取即時的資料,我們是能夠以更為微觀的角度去進行分析,甚至若真的能夠做到"即時",我們還可以做到高頻交易(不過,這樣的過程,一般散戶是很難去實現,常常是研究階段,並發表成果,或是不足以去實施這樣的方案)。

理所當然,越深入的探討,就像是去尋找真理與基本元素的組合方式一樣,肯定是不容易的,畢竟我們並不是身處在那樣的環境。又或者,你今天接觸了一個高度抽象化的程式,事實上,自己本身可以不知道底層是如何運作與如何實現、計算,一般是知道這段程式需要什麼樣的輸入、怎樣的輸出結果,並且如何應用,這樣便已足夠去使用。反過來要去深入探討,解碼後,完全就是一個不同的世界啊~。

話說遠了,進入本篇主軸,就像是開頭所說,本篇主要是講述即時的資料獲取,除了即時報價之外,也注重在上下最佳五檔

程式環境

程式語言: Python 3.6.8 x86

作業系統: Windows 10 64位元

API版本: 2.13.16 x86版本

程式碼回顧

首先先來快速回顧過去已經寫過的程式,中間一些細部會省略,主要是提個大概的基本元素。比較著急或已經非常熟悉的朋友可以跳著看。

先前我們主要使用到了兩個套件,分別是pythoncom以及comtypes,前者用來讀取Message Queue中的資料,後者用來調用COM元件。

import pythoncom
import comtypes.client

接著,我們指定了API的元件路徑,並且設定了一些基礎變數,包含帳號密碼以及市場別編號的Mapping等等。

api_path = r'C:/SKCOM/2.13.16/SKCOM.dll'

account = 'your account'
password = 'your password'

market_number_dict = {
0: '上市',
1: '上櫃',
2: '期貨',
3: '選擇權',
4: '興櫃'
}

然後,我們撰寫了第一個函式,用於API的初始化:

def capital_api_init(api_path_):
# link api module.
comtypes.client.GetModule(api_path_)
import comtypes.gen.SKCOMLib as sk

# Login.
skC = comtypes.client.CreateObject(sk.SKCenterLib, interface=sk.ISKCenterLib)

# Domestic Quote.
skQ = comtypes.client.CreateObject(sk.SKQuoteLib, interface=sk.ISKQuoteLib)

return skC, skQ

再來是登入功能:

def capital_Login(acc_, pass_):
m_n_code = skC.SKCenterLib_Login(acc_, pass_)
print("Login [{}]".format(skC.SKCenterLib_GetReturnCodeMessage(m_n_code)))

連線報價伺服器的部分,我們也有寫了連接與結束連線功能:

def EnterMonitor():
m_nCode = skQ.SKQuoteLib_EnterMonitor()
print("Quote enter monitor [{}]".format(skC.SKCenterLib_GetReturnCodeMessage(m_nCode)))


def LeaveMonitor():
m_nCode = skQ.SKQuoteLib_LeaveMonitor()
print("Quote leave monitor [{}]".format(skC.SKCenterLib_GetReturnCodeMessage(m_nCode)))

處理報價事件的部分,我們寫了一個Class,用來封裝相關的事件處理函式等等。其中,主要包含三個部分,分別是初始化、Set與Get函式和事件函式。

class SKQuoteLibEvents:
def __init__(self):
...

"""
set、get fuction.
"""

def set_is_ready(self, status: bool):
...

...

def get_is_ready(self):
...

def print(self, message: str):
...

"""
define events.
"""

def OnConnection(self, nKind, nCode):
...

def OnNotifyHistoryTicks(self, sMarketNo: int, sStockIdx: int, nPtr: int, lDate: int, lTimehms: int, lTimemillismicros: int, nBid: int, nAsk: int, nClose: int, nQty: int, nSimulate: int):
...

def OnNotifyTicks(self, sMarketNo: int, sStockIdx: int, nPtr: int, lDate: int, lTimehms: int, lTimemillismicros: int, nBid: int, nAsk: int, nClose: int, nQty: int, nSimulate: int):
...

def OnNotifyStockList(self, sMarketNo: int, bstrStockData: str):
...

最後是負責統一處理從Message Queue中取出訊息的函式:

def run_callback_sync(QuoteEvent: SKQuoteLibEvents):
...
while not QuoteEvent.get_is_ready():
pythoncom.PumpWaitingMessages()

...

事實上,整體來說,程式寫起來並不難,通常API串接的主要問題在於環境要求,也就是環境設定的部分,以及如何去調用,包含其規格是什麼。


實現即時資料獲取功能

關於即時報價的部分,我們先前其實已經有使用到,就是OnNotifyTicks()函式。為了方便講解,我們稍微修改一下這邊的函式內部的實作,注意到,我們這邊把原本會將資料序列儲存在dict結構的變數部分給移除掉了:


def OnNotifyTicks(self, sMarketNo: int, sStockIdx: int, nPtr: int, lDate: int, lTimehms: int, lTimemillismicros: int, nBid: int, nAsk: int, nClose: int, nQty: int, nSimulate: int):

self.__is_ready = True
self.__is_data_get = True

print(sMarketNo, sStockIdx, '[OnNotifyTicks]', str(nPtr), str(lDate), str(lTimehms), str(lTimemillismicros), str(nBid), str(nAsk), str(nClose), str(nQty), str(nSimulate))

並且,我們還會想要看到上下五檔的變動資料,會用到的是OnNotifyBest5()函式,我們在原先SKQuoteLibEvents的Class中添加以下程式碼:


def OnNotifyBest5(self, sMarketNo: int, sStockidx: int, nBestBid1: int, nBestBidQty1: int, nBestBid2: int, nBestBidQty2: int, nBestBid3: int, nBestBidQty3: int, nBestBid4: int, nBestBidQty4: int, nBestBid5: int, nBestBidQty5: int, nExtendBid: int, nExtendBidQty: int, nBestAsk1: int, nBestAskQty1: int, nBestAsk2: int, nBestAskQty2: int, nBestAsk3: int, nBestAskQty3: int, nBestAsk4: int, nBestAskQty4: int, nBestAsk5: int, nBestAskQty5: int, nExtendAsk: int, nExtendAskQty: int, nSimulate: int):

self.__is_ready = True
self.__is_data_get = True

print(sMarketNo, sStockidx, '[OnNotifyBest5]', nBestBid1, nBestBidQty1, nBestBid2, nBestBidQty2, nBestBid3, nBestBidQty3, nBestBid4, nBestBidQty4, nBestBid5, nBestBidQty5, nExtendBid, nExtendBidQty, nBestAsk1, nBestAskQty1, nBestAsk2, nBestAskQty2, nBestAsk3, nBestAskQty3, nBestAsk4, nBestAskQty4, nBestAsk5, nBestAskQty5, nExtendAsk, nExtendAskQty, nSimulate)

註:這邊有一個另外值得一提的事,原先在Class中的自定義函式裡面,有一個set_Stock_id()函式,這個函式負責設定當下所存入dict結構的變數的key值,但這個用來識別資料所屬的key值其實在事件函式中也能得到,只不過需要做一點mapping,因為透過事件函式所取得的商品編號,會是經過該系統特殊編碼後的代碼,與實際商品交易代碼會不一樣。由於這只不過是一個小小的處理,本系列基本上把介紹聚焦在如何使用,就不特別進一步提及了。


剛才提及的兩個事件函式,OnNotifyTicks()與OnNotifyBest5(),與國內報價中的skQ.SKQuoteLib_RequestTicks()函式相連動,我們這裡透過此函式來進行商品資料請求。

實際呼叫的部分,同樣為了方便講解,我們將固定幾個參數,包含這邊只會索取兩個個股即時資料,包含上下五檔。另外,這邊設定一個只會偵測一個固定次數的迴圈:

sk, skC, skQ = capital_api_init(api_path)
capital_Login(account, password)

# event registration.
SKQuoteEvent = SKQuoteLibEvents()
SKQuoteLibEventHandler = comtypes.client.GetEvents(skQ, SKQuoteEvent)

EnterMonitor()

print('start...')
run_callback_sync(SKQuoteEvent)

stock_code_list = ['1101', '2330']
iter_ = 0
for stock_code_index, stock_code in enumerate(stock_code_list):
# SKQuoteEvent.set_Stock_id(stock_code)
skQ.SKQuoteLib_RequestTicks(stock_code_index, stock_code)

print("RequestTicks: {}\t{}\t{}".format(stock_code, '上市', stock_code_index))
while iter_ <= 1000:
run_callback_sync(SKQuoteEvent)
iter_ += 1


LeaveMonitor()

SKQuoteLibEventHandler.disconnect()

註:這邊要提一下,由於是請求商品明細資料,而目前其規定一個page只能索取一檔商品,因此,各位要注意一下不要不小心覆蓋掉了先前的請求。

實際執行結果如下(中間有多輸出一些抓取各市場商品編號的部分,可忽略):

...


結論

本篇另外介紹了一個能夠即時獲取股價與上下五檔的方法,我想這對於一個交易模擬系統,不管是用在預測模型的測試或是單純的系統模擬,應該挺有幫助。雖然平常我們使用歷史資料進行測試時,都是用假裝看不見的方式去做,透過不給模型或機器看過那些用來測試的資料,來衡量演算法或模型的可行性等等。但若能實際上即時獲取資料,才能體現出其真正的價值,不然只不過是紙上談兵,做出了一個無法實現的效益績優的模型。

2 則留言:

  1. 首先感謝J大分享這麼好的文章 請問股票代號 怎麼沒辦法顯示

    回覆刪除
    回覆
    1. 需要進一步的描述,不然我得通靈一下。

      刪除