- 申請key, (api & secret) from https://www.sinotrade.com.tw/newweb/SinoTrade_login/
- install python 3.8 以上
- pip install shioaji & pip install -U shioaji
- hello world
- edit .env
API_KEY=your_api_key
SECRET_KEY=your_secret_key
- edit python
import shioaji as sj
import pandas as pd
import threading
import time
import schedule
from datetime import datetime
import os
from dotenv import load_dotenv
api = sj.Shioaji()
load_dotenv()
API_KEY = os.getenv("API_KEY")
SECRET_KEY = os.getenv("SECRET_KEY")
api.login(api_key=API_KEY, secret_key=SECRET_KEY)
api.logout()
# 创建空的 DataFrame 和缓存columns = ['timestamp', 'open', 'avg_price', 'close', 'high', 'low', 'volume', 'total_volume', 'price_chg', 'pct_chg']
quote_df = pd.DataFrame(columns=columns)
quote_data = []
# 回调函数
def on_quote(exchange, quote):
global quote_data
new_row = {
'timestamp': pd.to_datetime(quote.datetime),
# 'open': float(quote.open), #開盤
'avg_price': float(quote.avg_price),
'close': float(quote.close),
'high': float(quote.high),
'low': float(quote.low),
# 'amount': float(quote.amount),
# 'total_amount': float(quote.total_amount),
'volume': quote.volume, #單量
'total_volume': quote.total_volume,
'price_chg': float(quote.price_chg), #價差
'pct_chg': float(quote.pct_chg) #百分比
}
quote_data.append(new_row) #加入data[]
# 定期将数据从缓存写入 DataFrame
if len(quote_data) >= 100: #超過100筆 寫入data frame
update_dataframe()
def update_dataframe():
global quote_df, quote_data
if quote_data:
quote_df = pd.concat([quote_df, pd.DataFrame(quote_data)], ignore_index=True)
quote_data = [] # 清空缓存
print(quote_df.tail()) #印出來 如果超過100筆
# 定时任务 每分鐘
def print_quote():
global quote_df
if not quote_df.empty:
print("Latest quote:")
print(quote_df.tail()) # tail印多少?
else:
print("No quote data available yet.")
# 设置行情订阅 api = sj.Shioaji()
contract = api.Contracts.Stocks["2408"]
api.quote.subscribe(contract, quote_type=sj.constant.QuoteType.Quote, version=sj.constant.QuoteVersion.v1)
api.quote.set_on_quote_stk_v1_callback(on_quote) #callback
# 定时输出
schedule.every(1).minutes.do(print_quote)
# 主线程运行
try:
while True:
schedule.run_pending()
time.sleep(1)
except KeyboardInterrupt:
print("Stopping program...")
# AttributeError: 'SolaceAPI' object has no attribute 'unsubscribe_all'
api.quote.unsubscribe_all()
quote_df.to_csv('quote_data.csv', index=False)
print("Data saved to 'quote_data.csv'")
