量化交易進階：接入第三方股票API並自動發送Telegram交易通知

iTick
2025-04-29
AI量化

量化交易（Quantitative Trading）實時接收交易信號、訂單狀態和賬戶變動通知非常重要。Telegram Bot 是一種低成本、高可靠性的方式，可以讓你隨時隨地掌握交易動態。本文將教你如何將 Telegram Bot 接入量化交易系統，並在關鍵事件（如成交、止損、策略信號）時發送通知。

一、選擇數據源

iTick 作為國內專業的金融數據服務商，提供股票、期貨、期權等多市場的實時行情和歷史數據。本文將詳細介紹如何通過 API 接入 iTick 數據源，並整合到量化交易系統中。

1、iTick 數據源概述

主要數據類型

數據類型	更新頻率	適用場景
實時 tick 數據	毫秒級	高頻交易、做市策略
分鐘/日線 K 線	1 分鐘/日	趨勢跟蹤策略

接入方式

接入方式	延遲	適用場景
HTTP REST API	中（300ms+）	低頻數據獲取
WebSocket	低（100ms 內）	實時行情訂閱

2、準備工作

註冊賬戶：30 秒完成，無需信用卡，第三方賬號實現一鍵登錄註冊。
獲取 API 密鑰：控制台一鍵生成。
查閱文檔：豐富的代碼示例和教程。
安裝 Python 依賴：

      pip install requests websocket-client pandas hmac hashlib

3、HTTP REST API 接入示例

獲取實時行情

      import requests

API_KEY = "YOUR_API_KEY"

headers = {
    "accept": "application/json",
    "token": API_KEY
}

def get_realtime_quote(symbolInfo):
    url = "https://api.itick.org/stock/quote"
    params = {
        "region": symbolInfo.region,
        "code":symbolInfo.code
    }

    response = requests.get(url, headers=headers, params=params)
    if response.status_code == 200:
        return response.json()["data"]
    else:
        raise Exception(f"請求失敗: {response.text}")

# 獲取貴州茅台實時行情
params = {
    "region": "SH",
    "code": "600519"
}
quote = get_realtime_quote(params)
print(f"最新價: {quote['ld']}, 成交量: {quote['v']}")

    

獲取歷史數據

      API_KEY = "YOUR_API_KEY"

headers = {
    "accept": "application/json",
    "token": API_KEY
}
def get_historical_kline(symbol, region, period="8", limit=100):
    url = "https://api.itick.org/stock/kline"
    params = {
        "region": region,
        "symbol": symbol,
        "kType": period,  # 支持 週期 1 分鐘、2 五分鐘、3 十分鐘、4 三十分鐘、5 一小時、6 兩小時、7 四小時、8 一天、9 一週、10 一月
        "et": int(time.time() * 1000), # 查詢截止時間
        "limit": limit,
    }

    response = requests.get(url, headers=headers, params=params)
    return pd.DataFrame(response.json()["data"])

# 獲取貴州茅台日線
df = get_historical_kline("600519","SH", "8", 100)
df.to_csv("600519_daily.csv", index=False)

    

4、WebSocket 接入示例

      import websocket
import json
import threading

#  API KEY
API_KEY = "YOUR_API_KEY"

# WebSocket伺服器的地址
ws_url = "wss://api.itick.org/sws"

# 用於鑒權
auth_message = {
  "ac":"auth",
  "params":API_KEY
}

# 訂閱貴州茅台和寧德時代
subscribe_message = {
  "ac":"subscribe",
  "params":"600519$SH,300750$SZ",
  "types":"depth,quote"
}

def on_message(ws, message):
    data = json.loads(message)
    if "data" in data:
        print(f"數據內容: {data['data']}")

def on_error(ws, error):
    print("WebSocket錯誤:", error)

def on_close(ws):
    print("WebSocket連接關閉")

def on_open(ws):

    # 發送鑒權消息
    ws.send(json.dumps(auth_message))

    # 將訂閱消息轉換為JSON格式並發送
    ws.send(json.dumps(subscribe_message))

# 建立WebSocket連接

ws = websocket.WebSocketApp(
    ws_url,
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)

# 在後台運行WebSocket
threading.Thread(target=ws.run_forever).start()

    

5、高級應用：交易信號觸發 Telegram 通知

布林帶策略示例

      import numpy as np

def bollinger_strategy(symbol,region):
    # 獲取歷史數據
    df = get_historical_kline(symbol,region, "3", 100)
    closes = df["c"].astype(float)

    # 計算布林帶
    mean = closes.rolling(20).mean()
    std = closes.rolling(20).std()
    upper = mean + 2 * std
    lower = mean - 2 * std

    # 判斷信號
    last_close = closes.iloc[-1]
    if last_close > upper.iloc[-1]:
        send_telegram_alert(f"🔔 {symbol} 突破布林帶上軌！考慮賣出")
    elif last_close < lower.iloc[-1]:
        send_telegram_alert(f"🔔 {symbol} 跌破布林帶下軌！考慮買入")

def send_telegram_alert(message):
    requests.post(
        f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage",
        json={"chat_id": CHAT_ID, "text": message}
    )

# 每10分鐘運行一次
schedule.every(10).minutes.do(bollinger_strategy, "600519","SH")

    

6、性能優化建議

數據緩存：將獲取的行情數據緩存到本地文件，避免重複請求數據。
多線程優化：使用多線程或進程池來提高程序性能。

7、常見問題解答

Q1: 如何獲取免費的測試數據？

A1: iTick 提供免費 30 天試用，申請後可在測試環境獲取數據（正式環境請慎用）。

Q2: WebSocket 斷連如何處理？

A2: WebSocket 斷連時，可以通過重連機制來處理。

      def run_ws():
    while True:
        try:
            ws.run_forever()
        except Exception as e:
            print(f"連接斷開，5秒後重試... 錯誤: {e}")
            time.sleep(5)

threading.Thread(target=run_ws).start()

    