群益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只能索取一檔商品，因此，各位要注意一下不要不小心覆蓋掉了先前的請求。

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

...

---

結論

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