用永豐 API 抓取台股即時盤中內外盤成交資訊

BY PJ. －2025 年 5 月 2 日
(最後更新於: 2025 年 5 月 10 日)

繼上次透過永豐證券 API 找出短線交易潛力股，我們獲得了一組股票清單。接下來，我們將介紹如何使用永豐 API 取得盤中即時交易資訊，並進一步分析內外盤成交來識別即時的買賣訊號。

內外盤成交分析

外盤成交：代表買方提高買價，成交在較高價位，顯示出買方較強。
內盤成交：代表賣方降低賣價，成交在較低價位，顯示出賣方較強。

內外盤成交明細

透過這些資訊，我們可以判斷股票的買賣盤力道，從而預測未來的價格走勢。

永豐(Shioaji)即時行情API說明

登入永豐API

import shioaji as sj

api = sj.Shioaji(simulation=True)  # 使用模擬環境
api.login(
    api_key="your_api_key", 
    secret_key="your_secret_key", 
    receive_window=60000
)

訂閱股票

訂閱股票後，當盤中有成交訊息時，永豐將自動回傳資料。我們不需要透過 requests 查詢，只需接收並擷取回傳的資料即可。

訂閱數量限制：每個帳號最多可訂閱200檔股票，並且根據交易金額的不同，流量限制會有所不同。

近 30 日使用 API 成交金額	每日流量限制
0	500MB
1 - 1億	2GB
>1億	10GB

詳細流量限制參考：永豐API限制說明

訂閱 tick 資訊與 bidask 資訊

你可以選擇訂閱 tick 或 bidask 來獲取不同的市場資訊：

Tick：
即時成交量與價格資訊。
Bidask：
委買賣五檔價格資訊。

以訂閱股票 2330 為例：

api.quote.subscribe(
    api.Contracts.Stocks["2330"], 
    quote_type = sj.constant.QuoteType.Tick,
    version = sj.constant.QuoteVersion.v1
)

* 使用 callback 函數擷取回傳資訊

當永豐傳回資料後，我們可以使用 callback 函數來擷取資料並進行後續處理。

from shioaji import TickSTKv1, Exchange

def quote_callback(exchange: Exchange, tick: TickSTKv1):
    print(f"Exchange: {exchange}, Tick: {tick}")

api.quote.set_on_tick_stk_v1_callback(quote_callback) 這段程式碼會把回傳的交易資料打印出來，接下來我們將演示如何將這些資訊儲存至資料庫，便於後續分析與應用。

實作：資料庫儲存與回傳資訊分析

我們將盤前分析的潛力股票清單 (stock.xlsx)，透過永豐 API 訂閱並將回傳資訊儲存至 SQLite 資料庫，方便後續進行即時下單應用。
在stock.xlsx同層資料夾，我們分別建立四個 Python 檔案：main.py、tick.py、sql.py、type.py。

main.py - 永豐 API 登入與資料處理

import shioaji as sj
import tick

class RunMachine:
    def __init__(self):

        # 登入永豐模擬環境
        self.api = sj.Shioaji(simulation=True)
        self.api.login(
            api_key="your_api_key", 
            secret_key="your_secret_key", 
            receive_window=60000
        )


        # 執行 tick 串接 API 程式
        tick_func = tick.update_tick_df(self.api)
        tick_func.save_loop() #不定數迴圈執行程式

RunMachine()

tick.py - 訂閱股票與回傳資料儲存

import shioaji as sj
import time , datetime
import pandas as pd
import time
from sql import use_sql
from type import chg_type
from shioaji import BidAskSTKv1, Exchange, TickSTKv1

class update_tick_df:
    def __init__(self,api):
        self.api = api
        self.df = pd.read_excel('stock.xlsx',dtype=object)
        self.df.index = self.df.name
        self.df = self.df.drop('name',axis=1)


        #使用sqlite
        self.ldb = use_sql()
        self.ldb.delete_table() #清空原資料

        #訂閱股票
        self.subscribe_stock(self.df.symbol)
        self.add_column()
        self.tick()
        # self.bidask()

    #訂閱
    def subscribe_stock(self,stock_list):
        for symbol in stock_list:
            self.api.quote.subscribe(
                self.api.Contracts.Stocks[symbol], 
                quote_type = sj.constant.QuoteType.Tick,
                version = sj.constant.QuoteVersion.v1)
            print('subscribing tick to %s' % symbol)
            self.api.quote.subscribe(
                self.api.Contracts.Stocks[symbol], 
                quote_type = sj.constant.QuoteType.BidAsk,
                version = sj.constant.QuoteVersion.v1)
            print('subscribing bidask to %s' % symbol)
    #取消訂閱


    def add_column(self):

        self.df['high_price'] = 0.0
        self.df['low_price'] = 9999.0
        self.df['price'] = 0.0
        self.df['bid_side_total_vol'] =\
        self.df['ask_side_total_vol'] = 0
        self.df['tick_time'] = ""
        self.df['bid_continuous_times'] = self.df['ask_continuous_times'] = self.df['bid_continuous_volumn'] = self.df['ask_continuous_volumn'] = 0
        self.df['origin_ask_times'] = self.df['origin_bid_times'] = 0
        self.df['origin_ask_volumn'] = self.df['origin_bid_volumn'] = 0
    def tick(self):
        @self.api.on_tick_stk_v1()
        def quote_callback(exchange: Exchange, tick:TickSTKv1):
            try:
                self.df.loc[self.df.symbol == tick.code,'bid_side_total_vol']=tick.bid_side_total_vol
                self.df.loc[self.df.symbol == tick.code,'ask_side_total_vol']=tick.ask_side_total_vol
                self.df.loc[self.df.symbol == tick.code,'price']=tick.close
                self.df.loc[self.df.symbol == tick.code,'tick_time']=tick.datetime.strftime('%Y-%m-%d %H:%M:%S')
                if tick.tick_type == 1:
                    if (self.df.loc[self.df.symbol == tick.code,'ask_continuous_times'] != 0).any():
                        self.df.loc[self.df.symbol == tick.code,'origin_ask_times'] = self.df.loc[self.df.symbol == tick.code,'ask_continuous_times']
                        self.df.loc[self.df.symbol == tick.code,'origin_ask_volumn'] = self.df.loc[self.df.symbol == tick.code,'ask_continuous_volumn']
                    self.df.loc[self.df.symbol == tick.code,'origin_bid_times'] = 0
                    self.df.loc[self.df.symbol == tick.code,'origin_bid_volumn'] = 0
                    # 原內盤10次以上，出現外盤連續2次~5次 > 賣方竭點
                    self.df.loc[self.df.symbol == tick.code,'bid_continuous_times'] += 1
                    self.df.loc[self.df.symbol == tick.code,'ask_continuous_times'] = 0
                    self.df.loc[self.df.symbol == tick.code,'bid_continuous_volumn'] += tick.volume
                    self.df.loc[self.df.symbol == tick.code,'ask_continuous_volumn'] = 0
                elif tick.tick_type == 2:
                    if (self.df.loc[self.df.symbol == tick.code,'bid_continuous_times'] != 0).any():
                        self.df.loc[self.df.symbol == tick.code,'origin_bid_times'] = self.df.loc[self.df.symbol == tick.code,'bid_continuous_times']
                        self.df.loc[self.df.symbol == tick.code,'origin_bid_volumn'] = self.df.loc[self.df.symbol == tick.code,'bid_continuous_volumn']
                    self.df.loc[self.df.symbol == tick.code,'origin_ask_times'] = 0
                    self.df.loc[self.df.symbol == tick.code,'origin_ask_volumn'] = 0
                    # 原外盤10次以上，出現內盤連續2次~5次 > 買方竭點
                    self.df.loc[self.df.symbol == tick.code,'ask_continuous_times'] += 1
                    self.df.loc[self.df.symbol == tick.code,'bid_continuous_times'] = 0
                    self.df.loc[self.df.symbol == tick.code,'ask_continuous_volumn'] += tick.volume
                    self.df.loc[self.df.symbol == tick.code,'bid_continuous_volumn'] = 0
                if (self.df.loc[self.df.symbol == tick.code,'high_price'] < tick.close).any():
                    self.df.loc[self.df.symbol == tick.code,'high_price'] = tick.close
                if (self.df.loc[self.df.symbol == tick.code,'low_price'] > tick.close).any():
                    self.df.loc[self.df.symbol == tick.code,'low_price'] = tick.close
            except:
                print('tick data Error symbol: %s\n' % tick.code)
        time.sleep(3)


    def save_df(self):
        #讀取excel調整dtype格式
        df = chg_type(self.df)
        #try:
        if len(df) > 0:
           self.ldb.update_table(df)
        else:
            print("tick None")

    def save_loop(self):
        while True:
            if datetime.datetime.now().strftime("%H:%M:%S") <= '13:30:00'\
                and datetime.datetime.now().strftime("%H:%M:%S") >= '09:00:00':
                time.sleep(0.5)
                self.save_df()
            else:
                break

sql.py - 資料庫操作

import sqlite3
from pandas.io import sql
import pandas as pd
import time



class use_sql:
    def __init__(self):
        self.con = sqlite3.connect('stock_database.db', check_same_thread=False)
        self.cursorObj = self.con.cursor()

        self.cursorObj.execute("""create table if not exists stock_data(name varchar, symbol varchar, 
                               avg_volume int, high float,low float, open float, close float, price_change float,
                               high_price float,low_price float , price float,
                               bid_side_total_vol int, ask_side_total_vol int, 
                               tick_time varchar, 
                               bid_continuous_times int, ask_continuous_times int, bid_continuous_volumn int, ask_continuous_volumn int,
                               origin_ask_times int , origin_bid_times int,
                               origin_ask_volumn int , origin_bid_volumn int)
                               """)
        self.con.commit()
    def update_table(self, df):

        self.cursorObj.execute("DELETE FROM stock_data")
        self.con.commit()
        sql.to_sql(df, name='stock_data', con = self.con,if_exists='append')
    def delete_all(self):

        self.cursorObj.execute("delete from stock_data")
        self.con.commit()

type.py - 資料型態轉換

def chg_type(df):
    def convert_decimal_to_float(value):
        return float(value)
    df = df.applymap(str)
    df.symbol = df.symbol.astype('str')
    df.avg_volume = df.avg_volume.apply(convert_decimal_to_float)
    df.high = df.high.apply(convert_decimal_to_float)
    df.low = df.low.apply(convert_decimal_to_float)
    df.open = df.open.apply(convert_decimal_to_float)
    df.close = df.close.apply(convert_decimal_to_float)
    df.price_change = df.price_change.apply(convert_decimal_to_float)
    df.high_price = df.high_price.apply(convert_decimal_to_float)
    df.low_price = df.low_price.apply(convert_decimal_to_float)
    df.price = df.price.apply(convert_decimal_to_float)
    df.bid_side_total_vol = df.bid_side_total_vol.astype('int')
    df.ask_side_total_vol = df.ask_side_total_vol.astype('int')
    df.tick_time = df.tick_time.astype('str')
    df.bid_continuous_times = df.bid_continuous_times.astype('int')
    df.ask_continuous_times = df.ask_continuous_times.astype('int')
    df.bid_continuous_volumn = df.bid_continuous_volumn.astype('int')
    df.ask_continuous_volumn = df.ask_continuous_volumn.astype('int')
    df.origin_ask_times = df.origin_ask_times.astype('int')
    df.origin_bid_times = df.origin_bid_times.astype('int')
    df.origin_ask_volumn = df.origin_ask_volumn.astype('int')
    df.origin_bid_volumn = df.origin_bid_volumn.astype('int')
    return df

此檔案負責將 DataFrame 中的各欄位資料轉換為適合存儲至資料庫的資料型態。在處理股市資料時，某些欄位需要轉換為浮點數或整數，以確保資料庫中存儲的資料能夠正確地進行查詢和計算。

產生結果

螢幕擷取畫面 2025-05-02 154707

註：sqlite檔案可以用db browser for sqlite開啟

使用資料庫分析與判斷買賣訊號

寫入的table: stock_data，我將擷取的資料做了一些運算，可以用來查看各股票買賣盤力道：

欄位名稱	說明
bid_continuous_times	連續外盤成交量（若出現內盤成交則從 0 開啟計算）
ask_continuous_times	連續內盤成交量（若出現外盤成交則從 0 開啟計算）
bid_continuous_volumn	連續外盤成交筆數（若出現內盤成交則從 0 開啟計算）
ask_continuous_volumn	連續內盤成交筆數（若出現外盤成交則從 0 開啟計算）
origin_ask_volumn	外盤成交之前，連續內盤成交量
origin_bid_volumn	內盤成交之前，連續外盤成交量
origin_ask_times	外盤成交之前，連續內盤成交筆數
origin_bid_times	內盤成交之前，連續外盤成交筆數
bid_side_total_vol	外盤成交總量
ask_side_total_vol	內盤成交總量
high_price	程式開始執行後的最高股價
low_price	程式開始執行後的最低股價

這些資料可用來判斷股市的即時買賣訊號。舉例來說，當股價接近今日低點並且出現大量買盤，可能是反彈的訊號，這部分就留給大家自行運用。

結語

這篇文章介紹了如何使用永豐證券 API，透過盤中即時交易資訊來進行短線交易分析。我們透過訂閱tick api，在新的成交紀錄回傳即時更新SQLite 資料庫，用來分析買賣盤力道，從而判斷即時的交易訊號。

若想進一步實現自動下單，可串接永豐下單function，並透過多執行緒同時更新資料庫並使用；或可考慮串接 Telegram API，實現即時推播功能，可參考【Telegram API】使用 Python 打造 Telegram Notify 新手教學。

#永豐證券API #台股分析 #程式交易 #即時交易訊號 #Python金融分析 #股市盤中分析 #短線交易 #API應用 #台股即時行情 #Shioaji #台股內外盤

💬 留言區

PJ 說：

Hi，我是在開盤前先執行歷史資料抓清單
開盤後用清單的股票代碼，訂閱即時資料，把資料寫入SQLite
再做後續應用

2025-05-13 17:02

小正說：

Hi PJ你好,
想請問你的SQLite 資料庫內容是如何規劃的,如何將歷史資料和目前及時跑的資料互相結合呢?
謝謝

2025-05-13 16:41

文章分類: