绘图 - 即时 tick 资料

以下内容，都是 shioaji 的官网文件的内容，只是加了一些我自己的理解，感谢永丰提供这麽完整的 sample code

绘图会用到 numpy, pandas, bqplot...等套件，如果之前没有安装，可以先行安装

pip install numpy pandas bqplot

• 引用套件

import json
import datetime as dt
import numpy as np
import pandas as pd
import bqplot as bq
import shioaji as sj

• 登入 shioaji

api = sj.Shioaji()
api.login(person_id="帐号", passwd="密码")

• 建立表格的资料集

last_code = 10200

tick_plot_length = 150
x_tick_index = np.arange(tick_plot_length)
y_price = np.zeros((tick_plot_length,)) + last_code
y_vol = np.zeros((tick_plot_length,))
updown_color = np.zeros((tick_plot_length,)) + 1
askbid_color_data = np.zeros((tick_plot_length,))


new_index = 0
new_deal_price = 1
new_vol = 1
new_updown_color = 1
new_askbid_color_data = 1
ask_price = 0
bid_price = 0

• 建立绘图的方法

def init_tickchar():
    global line_chart, scatter_chart
    global x_tick_index, y_price, y_vol, updown_color, askbid_color_data

    x_sc = bq.LinearScale()
    y_sc = bq.LinearScale()

    sc_size = bq.LinearScale()
    color_line = bq.ColorScale(colors=['red', 'gray', 'green'])
    c_ord = bq.ColorScale(colors=['gray', 'HotPink', 'SeaGreen'])

    line_chart = bq.FlexLine(
        x = x_tick_index,
        y = y_price,
        color = updown_color,
        scales = {'x': x_sc, 'y': y_sc, 'color': color_line},
        display_legend = True,
        labels = ["台积电"]
    )
    
    def_tt = bq.Tooltip(
        fields = ['x', 'y', 'size'],
        formats = ['', '.0f', '.0f'],
        labels = ['tick', 'price', 'vol']
    )

    scatter_chart = bq.Scatter(
        x = x_tick_index,
        y = y_price,
        size = y_vol,
        color = askbid_color_data,
        scales = {'x': x_sc, 'y': y_sc, 'size': sc_size, 'color': c_ord},
        tooltip = def_tt, 
        labels = ['tick', 'price', 'vol'],
        selected_style = {'opacity': 1.0},
        unselected_style = {'opacity': 0.7}
    )

    ax_x = bq.Axis(scale=x_sc)
    ax_y = bq.Axis(scale=y_sc, orientation='vertical', tick_format='0.0f')
    ax_color = bq.ColorAxis(label='updown', scale=color_line, tick_format='.0f')

    fig = bq.Figure(marks=[line_chart, scatter_chart], axes=[ax_x, ax_y, ax_color],)
    
    return fig

• 建立资料处理方法

def proc_tick_chartdata(topic, quote_msg):
    global new_index, new_deal_price, new_vol, new_updown_color, new_askbid_color_data
    global x_tick_index, y_price, y_vol, updown_color, askbid_color_data
    new_index += 1
    new_deal_price = quote_msg.get('close', [0,])[-1]
    new_vol = quote_msg.get('Volume', [0,])[-1]
    ask_volsum = quote_msg.get('TradeAskVolSum', 0)
    bid_volsum = quote_msg.get('TradeBidVolSum', 0)
    new_askbid_color_data = quote_msg.get('TickType', [0,])[-1] # 0: -, 1: ask, 2: bid
    
    if new_deal_price > y_price[-1]:
        new_updown_color = 0
    elif new_deal_price < y_price[-1]:
        new_updown_color = 2
    else:
        new_updown_color = 1

    x_tick_index[:-1] = x_tick_index[1:]
    x_tick_index[-1] = new_index
    y_price[:-1] = y_price[1:]
    y_price[:-1] = new_deal_price
    y_vol[:-1] = y_vol[1:]
    y_vol[:-1] = new_vol
    updown_color[:-1] = updown_color[1:]
    updown_color[-2] = new_updown_color
    askbid_color_data[:-1] = askbid_color_data[1:]
    askbid_color_data[-1] = new_askbid_color_data
    

• 建立图表更新方法

def update_tick_chart(update_freq):
    global line_chart, scatter_chart

    if (new_index % update_freq == 0) and new_index >= tick_plot_length:
        with line_chart.hold_sync() and scatter_chart.hold_sync():
            line_chart.x = x_tick_index.copy()
            line_chart.y = y_price.copy()
            line_chart.color = updown_color.copy()
            scatter_chart.x = x_tick_index.copy()
            scatter_chart.y = y_price.copy()
            scatter_chart.size = y_vol.copy() * 10
            scatter_chart.color = askbid_color_data.copy()

• 建立取得价格後的回呼方法

@sj.on_quote
def update_tick_chart(update_freq):
    global line_chart, scatter_chart

    if (new_index % update_freq == 0) and new_index >= tick_plot_length:
        with line_chart.hold_sync() and scatter_chart.hold_sync():
            line_chart.x = x_tick_index.copy()
            line_chart.y = y_price.copy()
            line_chart.color = updown_color.copy()
            scatter_chart.x = x_tick_index.copy()
            scatter_chart.y = y_price.copy()
            scatter_chart.size = y_vol.copy() * 10
            scatter_chart.color = askbid_color_data.copy()

• 建立 solace 事件回呼方法

@sj.on_event
def event_callback(resp_code, event_code, event):
    print("Response Code: {} | Event Code: {} | Event: {}".format(resp_code, event_code, event))

• 设定回呼方法

api.quote.set_callback(quote_callback)
api.quote.set_event_callback(event_callback)

• 绘图

init_tickchar()

• 订阅报价

stock2330 = api.Contracts.Stocks["2330"]
api.quote.subscribe(stock2330, quote_type="bidask")

因为我是在盘後才跑这个程序，所以没有办法取得即资料，所以暂时不确定以上的程序能不能正确运行，之後如果有机会在交易时间跑的话，会再验证，如果有问题的话，会再修改。