聚宽对接 QMT：如何通过飞书 API 中转量化交易信号？

在量化交易中，许多交易者喜欢在**聚宽（JoinQuant）云端进行策略回测与模拟运行，但由于聚宽目前不直接提供实盘交易通道，需要将交易信号传输到本地的QMT（Quantitative Multi-market Trading System）**终端执行。

本文将教你如何利用飞书机器人 API作为中转站，实现“聚宽触发信号 -> 飞书群消息中转 -> 本地 QMT 轮询解析 -> 自动下单”的极速实盘对接方案。

一、 整体架构设计

1. 信号发送端（聚宽云端）：策略触发买卖点时，通过 Python 的 requests 库向飞书群机器人的 Webhook 发送格式化的 JSON 交易信号。
2. 信号中转站（飞书云端）：飞书群接收并保存交易信号消息。
3. 信号接收与执行端（本地 QMT）：本地 Python 脚本（运行在 QMT 的 Python 环境或 XtQuant 中）定时轮询飞书群消息，获取最新未处理的信号，解析后调用 QMT 的交易接口（如 passorder）向券商柜台报单。

二、 聚宽端：发送交易信号

在聚宽策略中，我们可以利用 Python 标准库 requests 向飞书机器人发送自定义消息。首先，你需要在飞书群聊中添加一个“自定义机器人”，并获取其 Webhook 地址。

聚宽策略信号发送源码：

import requests
import json

def send_feishu_signal(security, action, amount, price=None):
    # 替换为你的飞书机器人 Webhook 地址
    webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
    
    # 构造标准化的交易信号协议（方便QMT端解析）
    signal_data = {
        "source": "JoinQuant",
        "security": security,  # 例如 '000001.XSHE'
        "action": action,      # 'BUY' 或 'SELL'
        "amount": amount,      # 股数
        "price": price         # 价格（可选，限价单使用）
    }
    
    headers = {
        "Content-Type": "application/json"
    }
    
    # 飞书自定义机器人富文本或文本消息格式
    payload = {
        "msg_type": "text",
        "content": {
            "text": json.dumps(signal_data)  # 将信号转为字符串发送
        }
    }
    
    try:
        response = requests.post(webhook_url, data=json.dumps(payload), headers=headers, timeout=5)
        if response.status_code == 200:
            print("飞书信号发送成功: ", signal_data)
        else:
            print("飞书信号发送失败: ", response.text)
    except Exception as e:
        print("发送异常: ", str(e))

# 聚宽策略调用示例
def handle_data(context, data):
    # 假设触发买入信号
    # send_feishu_signal('000001.XSHE', 'BUY', 100)
    pass

三、 QMT 端：轮询飞书消息并执行交易

在本地 QMT 极客版（或使用 XtQuant 独立运行的 Python 脚本）中，我们需要通过飞书开放平台的 获取群消息 API 轮询群聊内容。

注意：为了能够读取群消息，你需要创建一个飞书“企业自建应用”，开启机器人权限，并将其加入到接收信号的群聊中。

QMT 端轮询与执行源码：

import time
import json
import requests

# 飞书应用凭证（在飞书开放平台获取）
APP_ID = "cli_xxxxxxxxxxxxxxxx"
APP_SECRET = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
CHAT_ID = "oc_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" # 信号群的chat_id

last_msg_id = None  # 记录最后一条处理过的消息ID，防止重复下单

def get_tenant_access_token():
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = {"app_id": APP_ID, "app_secret": APP_SECRET}
    res = requests.post(url, json=payload)
    return res.json().get("tenant_access_token")

def poll_feishu_signals():
    global last_msg_id
    token = get_tenant_access_token()
    if not token:
        return None
        
    # 获取群消息列表接口
    url = f"https://open.feishu.cn/open-apis/im/v1/messages?container_id_type=chat&container_id={CHAT_ID}&sort_type=ByCreateTimeDesc&page_size=1"
    headers = {"Authorization": f"Bearer {token}"}
    
    try:
        res = requests.get(url, headers=headers)
        items = res.json().get("data", {}).get("items", [])
        if not items:
            return None
            
        latest_msg = items[0]
        msg_id = latest_msg.get("message_id")
        
        # 如果是新消息，则进行解析
        if msg_id != last_msg_id:
            last_msg_id = msg_id
            content_str = latest_msg.get("body", {}).get("content", "")
            # 飞书返回的content是转义后的JSON字符串
            content_data = json.loads(content_str)
            text = content_data.get("text", "")
            
            # 解析交易信号
            signal = json.loads(text)
            if signal.get("source") == "JoinQuant":
                return signal
    except Exception as e:
        print("轮询飞书消息异常:", e)
    return None

# QMT 实盘下单逻辑
def execute_trade(signal):
    # 转换聚宽代码格式为QMT格式 (例如 '000001.XSHE' -> '000001.SZ')
    jq_code = signal["security"]
    code, market = jq_code.split('.')
    qmt_market = 'SZ' if market == 'XSHE' else 'SH'
    qmt_code = f"{code}.{qmt_market}"
    
    action = signal["action"]
    amount = signal["amount"]
    
    print(f"收到信号，准备下单: 标的={qmt_code}, 动作={action}, 数量={amount}")
    
    # ---------------- QMT 下单 API 示例 ----------------
    # 实际运行在QMT终端时，需使用QMT内置的交易对象，如 xtconstant
    # if action == 'BUY':
    #     passorder(23, 1101, 'YOUR_ACCOUNT', qmt_code, 11, -1, amount, 2, 1, context)
    # elif action == 'SELL':
    #     passorder(24, 1101, 'YOUR_ACCOUNT', qmt_code, 11, -1, amount, 2, 1, context)
    # --------------------------------------------------

# 主循环
if __name__ == "__main__":
    print("开始监听飞书交易信号...")
    while True:
        signal = poll_feishu_signals()
        if signal:
            execute_trade(signal)
        time.sleep(1)  # 每秒轮询一次

四、 核心注意事项与优化建议

1. 网络延迟与滑点：通过飞书 API 中转会有 200ms - 1s 左右的延迟。建议在聚宽端使用限价单或在 QMT 端执行时加入滑点保护（如 QMT 下单时采用最新价加减几分钱）。
2. 消息去重与防漏：网络波动可能导致轮询失败或重复获取。QMT 端必须严格校验 message_id，并建立本地信号日志，确保同一信号绝不重复执行。
3. 风控与异常处理：
   • 聚宽端发送信号前，需使用 get_current_data() 检查标的是否停牌。
   • QMT 端下单前，需调用 QMT 资金接口检查可用资金或可用持仓是否足够，避免废单。