QMT 与手机通信 - 邮件推送实现方法

QMT 邮件推送的重要性

在使用 QMT 量化交易系统时,我们经常需要将交易信号、错误信息或系统状态及时通知到手机,以便及时处理异常情况。邮件推送是一种简单、可靠且免费的通信方式,通过 Python 的 SMTP 模块可以轻松实现。

实现邮件推送的准备工作(发件邮箱)

步骤 1:选择合适的邮箱服务

推荐使用以下邮箱服务,它们都支持 SMTP 协议:

  • Gmail
  • QQ 邮箱
  • 163 邮箱
  • 网易邮箱
  • 企业邮箱

步骤 2:开启 SMTP 服务并获取授权码

以 QQ 邮箱为例:

  1. 登录 QQ 邮箱
  2. 点击 “设置” → “账户”
  3. 找到 “POP3/IMAP/SMTP/Exchange/CardDAV/CalDAV 服务”
  4. 开启 “POP3/SMTP 服务”
  5. 按照提示生成授权码(不是邮箱密码)
  6. 保存好授权码,后续会用到

实现邮件推送的准备工作(收件邮箱)

选择一个收件邮箱还是很重要的,最好是当你邮箱有邮件收到的时候,手机上有提醒,比如你可以在手机上安装对应的app获取实时推送。

我这边用的是QQ邮件,然后微信绑定了QQ邮箱,当我的QQ邮箱收到了一个邮件的时候,微信就会收到一个提醒,说进来一封新邮件。类似微信消息,这个几乎是实时的。这只是我选用的方式,你选择适合自己的方法就好。

Python 实现 SMTP 邮件发送

基本邮件发送代码

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
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header

# 邮件配置
SMTP_SERVER = "smtp.qq.com" # SMTP 服务器地址
SMTP_PORT = 587 # SMTP 端口
SENDER_EMAIL = "[email protected]" # 发件人邮箱
SENDER_PASSWORD = "your_authorization_code" # 授权码(不是密码)
RECEIVER_EMAIL = "[email protected]" # 收件人邮箱(可以是手机邮箱)

def send_email(subject, content):
"""
发送邮件函数
:param subject: 邮件主题
:param content: 邮件内容
:return: 是否发送成功
"""
try:
# 创建邮件对象
message = MIMEMultipart()
message['From'] = Header("QMT 通知", 'utf-8')
message['To'] = Header("用户", 'utf-8')
message['Subject'] = Header(subject, 'utf-8')

# 添加邮件正文
message.attach(MIMEText(content, 'plain', 'utf-8'))

# 连接 SMTP 服务器并发送邮件
server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT)
server.starttls() # 开启 TLS 加密
server.login(SENDER_EMAIL, SENDER_PASSWORD)
server.sendmail(SENDER_EMAIL, RECEIVER_EMAIL, message.as_string())
server.quit()

print("邮件发送成功")
return True
except Exception as e:
print(f"邮件发送失败: {str(e)}")
return False

# 测试邮件发送
if __name__ == "__main__":
send_email("QMT 测试通知", "这是一封测试邮件,用于验证 QMT 邮件推送功能是否正常。")

邮件发送函数封装

为了方便在 QMT 策略中使用,我们可以将邮件发送功能封装成一个独立的模块:

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
# email_utils.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header

class EmailSender:
def __init__(self, smtp_server, smtp_port, sender_email, sender_password, receiver_email):
self.smtp_server = smtp_server
self.smtp_port = smtp_port
self.sender_email = sender_email
self.sender_password = sender_password
self.receiver_email = receiver_email

def send(self, subject, content):
"""
发送邮件
:param subject: 邮件主题
:param content: 邮件内容
:return: 是否发送成功
"""
try:
message = MIMEMultipart()
message['From'] = Header("QMT 通知", 'utf-8')
message['To'] = Header("用户", 'utf-8')
message['Subject'] = Header(subject, 'utf-8')

message.attach(MIMEText(content, 'plain', 'utf-8'))

server = smtplib.SMTP(self.smtp_server, self.smtp_port)
server.starttls()
server.login(self.sender_email, self.sender_password)
server.sendmail(self.sender_email, self.receiver_email, message.as_string())
server.quit()

return True
except Exception as e:
print(f"邮件发送失败: {str(e)}")
return False

# 创建全局邮件发送实例
email_sender = EmailSender(
smtp_server="smtp.qq.com",
smtp_port=587,
sender_email="[email protected]",
sender_password="your_authorization_code",
receiver_email="[email protected]"
)

def send_qmt_email(subject, content):
"""
发送 QMT 相关邮件
:param subject: 邮件主题
:param content: 邮件内容
:return: 是否发送成功
"""
return email_sender.send(subject, content)

封装后的email.utils.py应当与其他的策略的py文件放在同一个文件夹下,通常是在安装目录下的python文件夹。
参考qmt-shared-code文件中详细的解释。

在 QMT 策略中集成邮件推送

示例 1:策略启动通知

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from email_utils import send_qmt_email

def init(ContextInfo):
# 设置账号
ContextInfo.set_account("your_account")

# 发送策略启动通知
subject = "QMT 策略启动通知"
content = "策略已成功启动,开始运行。"
send_qmt_email(subject, content)

def handle_bar(ContextInfo):
# 策略逻辑
pass

示例 2:交易信号通知

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from email_utils import send_qmt_email

def handle_bar(ContextInfo):
# 获取股票数据
code = "600000.SH"
price = ContextInfo.get_market_data([code], "close")[code]

# 交易信号判断
if price > 10.0:
# 发送买入信号通知
subject = "QMT 交易信号通知"
content = f"买入信号:{code} 当前价格:{price}\n建议:买入"
send_qmt_email(subject, content)

# 执行买入操作
# ContextInfo.order(code, 100, "buy")

示例 3:错误信息通知

1
2
3
4
5
6
7
8
9
10
11
12
13
from email_utils import send_qmt_email

def handle_bar(ContextInfo):
try:
# 策略逻辑
code = "600000.SH"
price = ContextInfo.get_market_data([code], "close")[code]
# 其他操作...
except Exception as e:
# 发送错误通知
subject = "QMT 策略错误通知"
content = f"策略运行出错:{str(e)}\n请及时检查。"
send_qmt_email(subject, content)

高级功能:邮件模板和附件

带 HTML 格式的邮件

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
def send_html_email(subject, html_content):
"""
发送 HTML 格式邮件
:param subject: 邮件主题
:param html_content: HTML 内容
:return: 是否发送成功
"""
try:
message = MIMEMultipart()
message['From'] = Header("QMT 通知", 'utf-8')
message['To'] = Header("用户", 'utf-8')
message['Subject'] = Header(subject, 'utf-8')

message.attach(MIMEText(html_content, 'html', 'utf-8'))

# 发送邮件代码...
# ...

return True
except Exception as e:
print(f"邮件发送失败: {str(e)}")
return False

# 使用示例
html_content = """
<html>
<body>
<h2>QMT 交易报告</h2>
<p>今日交易情况:</p>
<table border="1">
<tr>
<th>股票代码</th>
<th>交易类型</th>
<th>价格</th>
<th>数量</th>
</tr>
<tr>
<td>600000.SH</td>
<td>买入</td>
<td>10.5</td>
<td>100</td>
</tr>
</table>
</body>
</html>
"""
send_html_email("QMT 交易报告", html_content)

带附件的邮件

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
def send_email_with_attachment(subject, content, attachment_path):
"""
发送带附件的邮件
:param subject: 邮件主题
:param content: 邮件内容
:param attachment_path: 附件路径
:return: 是否发送成功
"""
try:
message = MIMEMultipart()
message['From'] = Header("QMT 通知", 'utf-8')
message['To'] = Header("用户", 'utf-8')
message['Subject'] = Header(subject, 'utf-8')

# 添加邮件正文
message.attach(MIMEText(content, 'plain', 'utf-8'))

# 添加附件
with open(attachment_path, 'rb') as f:
attachment = MIMEText(f.read(), 'base64', 'utf-8')
attachment['Content-Type'] = 'application/octet-stream'
attachment['Content-Disposition'] = f'attachment; filename="{attachment_path.split("/")[-1]}"'
message.attach(attachment)

# 发送邮件代码...
# ...

return True
except Exception as e:
print(f"邮件发送失败: {str(e)}")
return False

故障排除

常见问题及解决方案

  1. SMTP 服务器连接失败

    • 检查网络连接
    • 确认 SMTP 服务器地址和端口是否正确
    • 检查防火墙设置
  2. 认证失败

    • 确认使用的是授权码而不是邮箱密码
    • 确认授权码是否正确
    • 检查邮箱账号是否被锁定
  3. 邮件被标记为垃圾邮件

    • 优化邮件内容,避免使用垃圾邮件关键词
    • 确保发件人信息完整
    • 定期清理邮件发送记录
  4. 邮件发送延迟

    • 检查网络状况
    • 避免频繁发送邮件
    • 考虑使用邮件队列

最佳实践

  1. 合理设置邮件发送频率:避免频繁发送邮件,以免被邮箱服务提供商限制
  2. 邮件内容简洁明了:重点突出,便于在手机上快速查看
  3. 设置邮件主题前缀:如 “[QMT]”,便于识别和过滤
  4. 定期测试邮件功能:确保在需要时能正常发送
  5. 使用环境变量存储敏感信息:避免在代码中硬编码邮箱密码和授权码

总结

通过 Python 的 SMTP 模块,我们可以轻松实现 QMT 与手机的邮件通信,及时获取交易信号和系统状态。本文介绍了从邮箱设置到代码实现的完整流程,以及在 QMT 策略中的集成方法。

希望这篇文章对您有所帮助!如果您有任何问题或建议,欢迎在评论区留言。


关注我们微信公众号,并在公众号后台发送"qmt","ibkr"或者"invest",第一时间获取最新资讯

如有任何问题,欢迎添加微信,或者在公众号后台联系我们

微信二维码 Share