一、前言

量化交易最怕的就是错过重要信号。当你不在电脑前时，如何第一时间收到策略的交易信号？企业微信机器人是最简单实用的方案——无需服务器、配置简单、消息实时送达手机。

本文将介绍：

• 企业微信机器人配置
• 消息推送接口封装
• 与TqSdk策略集成
• 多种消息格式应用

二、为什么选择天勤量化（TqSdk）

开发量化交易策略需要功能完整的框架。**天勤量化（TqSdk）**是目前国内最适合个人投资者的期货量化平台：

功能	说明
模拟交易	内置TqSim模拟盘，无需开户即可测试
回测功能	支持历史数据回测
实盘对接	一套代码模拟实盘通用
多账户	支持同时管理多个账户

安装方法：

pip install tqsdk

重要提示：建议先使用模拟盘充分测试策略，确认无误后再考虑实盘交易。

三、企业微信机器人配置

3.1 创建群聊机器人

1. 打开企业微信电脑版
2. 创建一个群聊（或使用现有群）
3. 右键群聊 → 添加群机器人
4. 创建机器人，复制Webhook地址

3.2 Webhook地址格式

https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx

参数	说明
key	机器人唯一标识，创建时自动生成
保密性	不要泄露给他人

四、消息推送封装

4.1 基础推送类

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：企业微信机器人消息推送
说明：本代码仅供学习参考
"""

import requests
import json
from datetime import datetime

class WeChatBot:
    """企业微信机器人"""
    
    def __init__(self, webhook_url):
        """
        初始化机器人
        
        参数：
            webhook_url: 机器人Webhook地址
        """
        self.webhook_url = webhook_url
    
    def send_text(self, content, mentioned_list=None):
        """
        发送文本消息
        
        参数：
            content: 消息内容
            mentioned_list: @的成员列表，如 ["user1", "@all"]
        """
        data = {
            "msgtype": "text",
            "text": {
                "content": content,
            }
        }
        
        if mentioned_list:
            data["text"]["mentioned_list"] = mentioned_list
        
        return self._send(data)
    
    def send_markdown(self, content):
        """
        发送Markdown消息
        
        参数：
            content: Markdown格式内容
        """
        data = {
            "msgtype": "markdown",
            "markdown": {
                "content": content
            }
        }
        return self._send(data)
    
    def send_image(self, image_base64, image_md5):
        """
        发送图片消息
        
        参数：
            image_base64: 图片base64编码
            image_md5: 图片MD5值
        """
        data = {
            "msgtype": "image",
            "image": {
                "base64": image_base64,
                "md5": image_md5
            }
        }
        return self._send(data)
    
    def _send(self, data):
        """发送请求"""
        try:
            response = requests.post(
                self.webhook_url,
                json=data,
                timeout=10
            )
            result = response.json()
            
            if result.get("errcode") == 0:
                return True
            else:
                print(f"发送失败: {result}")
                return False
                
        except Exception as e:
            print(f"发送异常: {e}")
            return False


# ============ 测试 ============
if __name__ == "__main__":
    # 替换为你的Webhook地址
    WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your-key-here"
    
    bot = WeChatBot(WEBHOOK_URL)
    
    # 测试文本消息
    bot.send_text("测试消息：期货监控系统启动成功！")
    
    # 测试Markdown消息
    markdown_content = """
### 📊 交易信号

**合约**: SHFE.rb2510
**方向**: <font color="info">买入开仓</font>
**价格**: 3456
**时间**: 2025-01-09 10:30:00

> 策略: 双均线策略
    """
    bot.send_markdown(markdown_content)

4.2 运行效果

发送成功后，手机企业微信会收到消息推送。

五、与TqSdk策略集成

5.1 交易信号推送

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：双均线策略+微信推送
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosTask
import requests
from datetime import datetime

# ============ 微信推送配置 ============
WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your-key-here"

class SignalNotifier:
    """交易信号通知器"""
    
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url
        self.enabled = bool(webhook_url and "your-key" not in webhook_url)
        
        if not self.enabled:
            print("提示: 微信推送未配置，将只在控制台显示")
    
    def notify_signal(self, symbol, direction, price, strategy_name):
        """推送交易信号"""
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        # 控制台输出
        print(f"\n[{now}] 交易信号: {symbol} {direction} @ {price}")
        
        if not self.enabled:
            return
        
        # 方向颜色
        if "买" in direction or "多" in direction:
            color = "info"     # 绿色
            emoji = "📈"
        else:
            color = "warning"  # 橙色
            emoji = "📉"
        
        content = f"""### {emoji} 交易信号

**合约**: {symbol}
**方向**: <font color="{color}">{direction}</font>
**价格**: {price:.0f}
**时间**: {now}

> 策略: {strategy_name}
"""
        self._send_markdown(content)
    
    def notify_account(self, balance, profit, position_info):
        """推送账户状态"""
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        profit_color = "info" if profit >= 0 else "warning"
        
        content = f"""### 💰 账户状态

**权益**: {balance:.0f}
**盈亏**: <font color="{profit_color}">{profit:+.0f}</font>
**持仓**: {position_info}
**时间**: {now}
"""
        self._send_markdown(content)
    
    def notify_alert(self, title, message, level="INFO"):
        """推送预警消息"""
        emoji_map = {
            "INFO": "ℹ️",
            "WARNING": "⚠️",
            "DANGER": "🚨"
        }
        emoji = emoji_map.get(level, "ℹ️")
        
        content = f"""### {emoji} {title}

{message}

**时间**: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
"""
        self._send_markdown(content)
    
    def _send_markdown(self, content):
        """发送Markdown消息"""
        try:
            data = {
                "msgtype": "markdown",
                "markdown": {"content": content}
            }
            requests.post(self.webhook_url, json=data, timeout=10)
        except Exception as e:
            print(f"推送失败: {e}")


# ============ 策略参数 ============
SYMBOL = "SHFE.rb2510"
FAST_PERIOD = 5
SLOW_PERIOD = 20
LOTS = 2

# ============ 初始化 ============
api = TqApi(TqSim(init_balance=1000000), auth=TqAuth("快期账户", "快期密码"))

klines = api.get_kline_serial(SYMBOL, 300, SLOW_PERIOD + 20)
account = api.get_account()
position = api.get_position(SYMBOL)
target_pos = TargetPosTask(api, SYMBOL)

notifier = SignalNotifier(WEBHOOK_URL)

print("=" * 60)
print("双均线策略 + 微信推送")
print("=" * 60)
print(f"交易合约: {SYMBOL}")
print(f"快线周期: {FAST_PERIOD}")
print(f"慢线周期: {SLOW_PERIOD}")
print("-" * 60)

# 发送启动通知
notifier.notify_alert("策略启动", f"双均线策略已启动\n合约: {SYMBOL}\n参数: MA{FAST_PERIOD}/MA{SLOW_PERIOD}")

last_signal = 0
last_report_hour = -1

while True:
    api.wait_update()
    
    if api.is_changing(klines.iloc[-1], "datetime"):
        close = klines["close"]
        fast_ma = close.rolling(window=FAST_PERIOD).mean().iloc[-1]
        slow_ma = close.rolling(window=SLOW_PERIOD).mean().iloc[-1]
        
        current_price = close.iloc[-1]
        
        # 信号判断
        if fast_ma > slow_ma:
            signal = 1
            signal_text = "买入做多"
        else:
            signal = -1
            signal_text = "卖出做空"
        
        # 信号变化时交易并推送
        if signal != last_signal:
            target_volume = signal * LOTS
            target_pos.set_target_volume(target_volume)
            
            # 推送交易信号
            notifier.notify_signal(
                symbol=SYMBOL,
                direction=signal_text,
                price=current_price,
                strategy_name="双均线策略"
            )
            
            last_signal = signal
        
        # 整点推送账户状态
        current_hour = datetime.now().hour
        if current_hour != last_report_hour and datetime.now().minute == 0:
            pos_info = f"多{position.pos_long}手 空{position.pos_short}手"
            notifier.notify_account(
                balance=account.balance,
                profit=account.float_profit,
                position_info=pos_info
            )
            last_report_hour = current_hour

5.2 消息格式说明

格式	说明	示例
<font color="info">	绿色文字	买入信号
<font color="warning">	橙色文字	卖出信号
<font color="comment">	灰色文字	辅助信息
> 引用	引用样式	备注信息

六、定时报告功能

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：定时推送账户报告
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth, TqSim
import requests
from datetime import datetime
import schedule
import time

WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your-key-here"

def send_daily_report(api):
    """发送每日报告"""
    account = api.get_account()
    
    content = f"""### 📋 每日交易报告

**日期**: {datetime.now().strftime("%Y-%m-%d")}

---

**账户权益**: {account.balance:.0f}
**可用资金**: {account.available:.0f}
**持仓盈亏**: {account.float_profit:+.0f}
**今日盈亏**: {account.close_profit:+.0f}

---

> 报告由量化系统自动生成
"""
    
    data = {
        "msgtype": "markdown",
        "markdown": {"content": content}
    }
    
    try:
        requests.post(WEBHOOK_URL, json=data, timeout=10)
        print(f"[{datetime.now()}] 每日报告已发送")
    except Exception as e:
        print(f"发送失败: {e}")


# 初始化
api = TqApi(TqSim(init_balance=1000000), auth=TqAuth("快期账户", "快期密码"))

# 设置定时任务：每天15:30发送报告（收盘后）
schedule.every().day.at("15:30").do(send_daily_report, api)

print("定时报告服务已启动...")
print("将在每天15:30发送账户报告")

while True:
    api.wait_update()
    schedule.run_pending()

七、常见问题

Q1: 消息发送频率限制？

限制	说明
每分钟	最多20条
每小时	建议不超过100条
解决方案	合并消息、降低频率

Q2: 如何避免消息轰炸？

class RateLimiter:
    """消息频率限制"""
    
    def __init__(self, min_interval=60):
        self.min_interval = min_interval
        self.last_send = {}
    
    def can_send(self, msg_type):
        now = time.time()
        if msg_type in self.last_send:
            if now - self.last_send[msg_type] < self.min_interval:
                return False
        self.last_send[msg_type] = now
        return True

八、总结

要点	内容
推送方式	企业微信群机器人
配置简单	只需Webhook地址
消息格式	支持文本、Markdown、图片
集成方式	在策略中调用推送函数
频率控制	注意限制避免轰炸

---

免责声明：本文仅供学习交流使用，不构成任何投资建议。期货交易有风险，入市需谨慎。

更多资源：

• 天勤量化官网：https://www.shinnytech.com
• GitHub开源地址：https://github.com/shinnytech/tqsdk-python
• 官方文档：https://doc.shinnytech.com/tqsdk/latest