qmt

讯投QMT使用小技巧: QMT推送发送消息到微信的几种方案

说明：本文包含 AI 创作内容，请自行判断是否适用。文中包含的代码片段，切勿直接使用，需要根据实际情况修改。

概述

策略上线后，最朴素的诉求就是”出事的时候手机能响一下”。邮件方案在 邮件推送 里讲过了，但邮件有个问题：到达手机的即时性依赖邮箱 App 的推送策略，有时会延迟几分钟甚至几十分钟。而微信几乎是国人打开频率最高的 App，把告警直接送到微信，体验会好很多。

本文整理 QMT 策略中可用的几条微信推送通道——Server酱、企业微信群机器人、WxPusher、PushPlus，给出统一封装的工具类和 QMT 集成示例，并和 运行状态监控与心跳报警 配合，构成完整的告警链路。

一、方案选型

方案 触达位置 注册门槛 频率限制 适合场景
Server酱 个人微信「Server酱Turbo」应用 扫码登录拿 SendKey 免费版 5 条/天，Turbo 200 条/天 个人单策略、低频告警
企业微信群机器人 企业微信群 建群+添加机器人 20 条/分钟 多策略集中告警、团队协作
WxPusher 个人微信「WxPusher」公众号 注册+关注公众号 免费版较宽松 一对多推送、带 UI 管理
PushPlus 个人微信「PushPlus推送加」公众号 注册+关注公众号 免费版 200 条/天 个人推送、支持模板

选型建议：个人单策略选 Server酱或 PushPlus，多策略/团队选企业微信群机器人。企业微信群机器人不需要个人开通企业微信会员，建个只有自己的群也能用。

二、Server酱：最简单的方案

Server酱（sct.ftqq.com）走的是「一个 HTTP 请求 = 一条微信消息」的模式，整个接入只需要一个 SendKey。

步骤

  1. 访问 sct.ftqq.com，微信扫码登录
  2. 在「SendKey」页面拿到 SCT**** 开头的 key
  3. 在「微信推送」页面扫码绑定要接收消息的微信号
  4. 调用接口

代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# wechat_utils.py
# -*- coding: utf-8 -*-
import requests

SERVERCHAN_KEY = "SCT****your_send_key****"

def send_serverchan(title, content=""):
    """
    通过 Server酱 推送到微信
    :param title: 消息标题（必填，最长 32）
    :param content: 消息内容，支持 Markdown
    :return: 是否发送成功
    """
    url = f"https://sctapi.ftqq.com/{SERVERCHAN_KEY}.send"
    try:
        r = requests.post(url, data={"title": title, "desp": content}, timeout=10)
        return r.json().get("code") == 0
    except Exception as e:
        print(f"Server酱推送失败: {e}")
        return False

Server酱免费版每天 5 条，对监控告警来说偏少。要么升级 Turbo 版，要么配合下面的节流策略只用它发关键告警。

三、企业微信群机器人：多策略群告警

企业微信群机器人不需要企业认证，建一个群、加个机器人就能拿到 Webhook URL，最适合把多个策略的告警集中到一个群里。

步骤

  1. 在企业微信里建一个群（自己一个人的群也行）
  2. 群设置 → 群机器人 → 添加机器人 → 起个名字
  3. 复制 Webhook 地址，形如 https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx

文本消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import requests

WECOM_WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key"

def send_wecom_text(content, mentioned_list=None):
    """
    发送文本消息
    :param content: 文本内容
    :param mentioned_list: 需要@的用户ID列表，['@all'] 表示@所有人
    """
    data = {
        "msgtype": "text",
        "text": {
            "content": content,
            "mentioned_list": mentioned_list or [],
            "mentioned_mobile_list": []
        }
    }
    try:
        r = requests.post(WECOM_WEBHOOK, json=data, timeout=10)
        return r.json().get("errcode") == 0
    except Exception as e:
        print(f"企业微信推送失败: {e}")
        return False

Markdown 消息

告警内容用 Markdown 排版会清晰很多：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def send_wecom_markdown(content):
    """
    发送 Markdown 消息（企业微信群机器人支持有限 Markdown 语法）
    """
    data = {
        "msgtype": "markdown",
        "markdown": {"content": content}
    }
    try:
        r = requests.post(WECOM_WEBHOOK, json=data, timeout=10)
        return r.json().get("errcode") == 0
    except Exception as e:
        print(f"企业微信推送失败: {e}")
        return False

# 使用示例
msg = """## QMT 告警
> **策略**: 双均线
> **账号**: 600000
> **时间**: 2026-08-04 14:30

**事件**: 账号掉线
<font color="warning">请及时检查</font>"""
send_wecom_markdown(msg)

企业微信群机器人支持 <font color="info|comment|warning"> 给文字上色，warning 是橙红色，比较醒目。

四、WxPusher / PushPlus

两者都是「关注公众号 → 注册拿 token → HTTP 调接口」的模式，用法和 Server酱 类似，只是接口参数不同。

WxPusher

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
WXPUSHER_TOKEN = "your_app_token"

def send_wxpusher(title, content):
    url = "https://wxpusher.????.com/api/send/message"
    data = {
        "appToken": WXPUSHER_TOKEN,
        "content": content,
        "summary": title,             # 消息摘要（必填，否则推送列表里显示空白）
        "contentType": 1,             # 1=文本, 2=html, 3=markdown
        "topicIds": [],               # 主题ID，可留空
        "uids": ["your_uid"]          # 目标用户UID，在公众号「我的」里查
    }
    try:
        r = requests.post(url, json=data, timeout=10)
        return r.json().get("success")
    except Exception as e:
        print(f"WxPusher 推送失败: {e}")
        return False

PushPlus

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
PUSHPLUS_TOKEN = "your_token"

def send_pushplus(title, content, template="html"):
    url = "http://www.pushplus.plus/send"
    data = {
        "token": PUSHPLUS_TOKEN,
        "title": title,
        "content": content,
        "template": template           # html / json / markdown
    }
    try:
        r = requests.post(url, json=data, timeout=10)
        return r.json().get("code") == 200
    except Exception as e:
        print(f"PushPlus 推送失败: {e}")
        return False

五、统一封装：notify.py

不同渠道接口不同，但策略代码只想关心「发条消息」。封装一个统一的 notify，支持渠道切换、节流、容错：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
# notify.py
# -*- coding: utf-8 -*-
import time
import threading

class Notifier:
    def __init__(self, default_channel="serverchan"):
        self.default_channel = default_channel
        self._last = {}                # (channel, key) -> ts
        self._lock = threading.Lock()

    def _throttle(self, channel, key, cooldown=600):
        """同一类消息 cooldown 秒内只发一次，避免刷屏"""
        k = (channel, key)
        now = time.time()
        with self._lock:
            if now - self._last.get(k, 0) < cooldown:
                return False
            self._last[k] = now
            return True

    def send(self, title, content="", channel=None, key=None, cooldown=600):
        """
        发送消息
        :param title: 标题
        :param content: 内容
        :param channel: 渠道，默认 self.default_channel
        :param key: 节流键，默认用 title；同一 key 在 cooldown 内只发一次
        :param cooldown: 节流秒数，默认 600 秒
        """
        channel = channel or self.default_channel
        key = key or title
        if not self._throttle(channel, key, cooldown):
            print(f"[notify] 节流命中，跳过: {title}")
            return False

        try:
            if channel == "serverchan":
                return send_serverchan(title, content)
            elif channel == "wecom":
                return send_wecom_text(f"{title}\n{content}")
            elif channel == "wecom_md":
                return send_wecom_markdown(f"## {title}\n{content}")
            elif channel == "wxpusher":
                return send_wxpusher(title, content)
            elif channel == "pushplus":
                return send_pushplus(title, content)
            else:
                print(f"[notify] 未知渠道: {channel}")
                return False
        except Exception as e:
            # 告警本身失败不能影响策略
            print(f"[notify] 发送异常: {e}")
            return False

notifier = Notifier(default_channel="serverchan")

def notify(title, content="", **kwargs):
    """供策略调用的统一入口"""
    return notifier.send(title, content, **kwargs)

把 notify.py 放到 QMT 安装目录的 python 文件夹下，所有策略都能引用，路径配置见 公共代码的编写与引用。

六、在 QMT 策略中集成

示例 1：策略启动通知

1
2
3
4
5
6
from notify import notify

def init(C):
    C.set_account(account)
    # 开盘前推送一条，确认策略已加载
    notify("QMT 策略启动", f"账号 {account} 策略已启动")

示例 2：成交回调通知

成交是发通知的最佳时机——不是”下单了”就发，而是”真成交了”才发，避免被废单打扰。成交回调用法详见 成交回调：

1
2
3
4
5
6
7
8
9
10
11
from notify import notify

def deal_callback(C, dealInfo):
    code = dealInfo.m_strInstrumentID
    if code.startswith('6'):
        code += '.SH'
    else:
        code += '.SZ'
    side = '买入' if dealInfo.m_nOffset == 48 else '卖出'
    msg = f"{side} {code}\n价格 {dealInfo.m_dPrice}\n数量 {dealInfo.m_nVolume}"
    notify("成交回报", msg)

示例 3：异常巡检告警

和 心跳报警 的巡检结合，发现异常立刻推微信：

1
2
3
4
5
6
def patrol(C):
    problems = []
    # ... 巡检逻辑，发现异常追加到 problems ...
    if problems:
        notify("QMT巡检异常", "\n".join(problems),
               channel="wecom_md", key="patrol", cooldown=600)

这里 key="patrol" 把所有巡检异常归并到同一个节流桶，10 分钟内不会因为同一个巡检任务刷屏。不同问题想分开告警，就给不同的 key。

示例 4：收盘日报

1
2
3
4
5
6
7
8
9
10
11
12
13
def init(C):
    # 每个交易日 15:05 推送日报
    C.run_time('daily_report', '1nDay', '20260101 15:05:00')

def daily_report(C):
    acc = get_trade_detail_data(account, accountType, 'account')[0]
    holdings = get_trade_detail_data(account, accountType, 'position')
    md = f"## 收盘报告\n> 账号 {account}\n\n"
    md += f"- 可用资金: {acc.m_dAvailable:.2f}\n"
    md += f"- 持仓数: {len(holdings)}\n\n"
    for h in holdings:
        md += f"- {h.m_strInstrumentID}: {h.m_nVolume}股 浮盈{h.m_dProfit:.2f}\n"
    notify("收盘报告", md, channel="wecom_md", cooldown=0)  # 日报不节流

run_time 定时任务的写法见 每天代码重置与定时任务。

七、常见坑

  1. QMT 内置 Python 缺 requests：QMT 自带的 Python 环境通常已包含 requests，但版本可能较旧。若提示 No module named 'requests'，按 安装第三方库 装一下，或用标准库 urllib 改写。
  2. 超时阻塞策略：HTTP 请求默认会阻塞，务必传 timeout=10，并放在 run_time 任务或回调里发，别放 handlebar 高频路径。
  3. 频率超限：Server酱免费版 5 条/天，企业微信 20 条/分钟。监控类告警务必节流，否则一天的额度几分钟就刷光。
  4. 中文编码：JSON body 用 json= 传参会自动处理；用 data= 传 form 时注意 requests 会按 utf-8 编码，一般没问题，但日志里打印 content 别用 gbk 控制台，会报错。
  5. 告警本身失败：所有推送函数都要 try/except 兜底，告警挂了不能拖垮策略。notify 已经做了这层保护。
  6. 关键告警走多通道：账号掉线、重复下单风险这种关键告警，建议同时发 Server酱（个人微信）和企业微信群，避免单通道故障漏报。

八、和邮件方案的关系

邮件（邮件推送）和微信不是二选一，而是互补：

  • 微信：即时性强、阅读率高，适合异常告警、成交回报、关键事件。
  • 邮件：留档性好、可带附件、可发长报告，适合日报、周报、详细日志。

推荐组合：实时事件走微信，日报兜底走邮件。notify 封装里再加一个 email 渠道，调用 email_utils.send_qmt_email 即可统一入口。

总结

把消息推到微信，QMT 策略就有了”会叫的watchdog”：

  • 个人单策略用 Server酱，多策略/团队用 企业微信群机器人，需要一对多用 WxPusher/PushPlus。
  • 统一封装到 notify.py，支持渠道切换、节流、容错。
  • 集成到 init（启动通知）、deal_callback（成交回报）、patrol（巡检告警）、daily_report（收盘日报）四个时机。
  • 关键告警走多通道，实时走微信、留档走邮件。

配合 运行状态监控与心跳报警 的心跳/巡检/watchdog体系，整套 QMT 运维闭环就齐了——出事手机能响，没事每天一报，安心睡觉。

说明：本文包含 AI 创作内容，请自行判断是否适用。文中包含的代码片段，切勿直接使用，需要根据实际情况修改。

相关文章

QMT 与手机通信 - 邮件推送实现方法 开始阅读

讯投QMT使用小技巧: 运行状态监控与心跳报警 开始阅读

QMT 委托/成交回调机制详解 开始阅读

讯投QMT使用小技巧: 公共代码的编写与引用 开始阅读

讯投QMT使用小技巧: 每天代码重置与定时任务 开始阅读

QMT 安装第三方 Python 库 开始阅读

所有与QMT相关的文章查看目录