量化交易进阶：接入第三方股票API并自动发送Telegram交易通知

量化交易（Quantitative Trading）实时接收交易信号、订单状态和账户变动通知非常重要。Telegram Bot 是一种低成本、高可靠性的方式，可以让你随时随地掌握交易动态。本文将教你如何将 Telegram Bot 接入量化交易系统，并在关键事件（如成交、止损、策略信号）时发送通知

一、选择数据源

iTick 作为国内专业的金融数据服务商，提供股票、期货、期权等多市场的实时行情和历史数据。本文将详细介绍如何通过 API 接入 iTick 数据源，并整合到量化交易系统中

1、iTick 数据源概述

1. 主要数据类型

2. 接入方式

2、准备工作

1. 注册账户：30 秒完成，无需信用卡，第三方账号实现一键登录注册，
2. 获取 API 密钥：控制台一键生成
3. 查阅文档：丰富的代码示例和教程
4. 安装 Python 依赖

pip install requests websocket-client pandas hmac hashlib

3、HTTP REST API 接入示例

1. 获取实时行情

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']}")

2. 获取历史数据

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、性能优化建议

1. 数据缓存：将获取的行情数据缓存到本地文件，避免重复请求数据。
2. 多线程优化：使用多线程或进程池来提高程序性能。

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()

Q3: 数据延迟较高怎么优化？

A3: 免费套餐升级 iTick 付费套餐