
上一篇我们通过批量回测筛选出了策略“偏爱”的股票,但不少新手反馈:“每天手动跑代码太麻烦”“错过金叉信号就白筛选了”。今天就彻底解决这个痛点——教你用Python打造一套“全自动量化系统”:每天收盘后自动获取最新数据、执行策略,一旦优选股出现金叉信号,立刻通过微信把“股票名称+买入提示”推送给你,真正实现“不用盯盘也能抓机会”。
一、核心工具:3分钟搞定自动化必备组件
自动化的核心是“数据自动获取+信号自动推送”,我们需要两个简单工具,新手跟着步骤走,3分钟就能配齐:
1.1 微信推送工具:Server酱(免费无门槛)
Server酱是专门用于“程序发消息到微信”的工具,不用写复杂接口,复制一个Key就能用,步骤如下:
- 打开官网:https://sct.ftqq.com/,用微信扫码登录(仅需微信授权,安全无广告);
- 登录后点击左侧“发送消息”,找到“SCKEY”(一串字母+数字的字符串),复制保存——这是“微信消息通行证”,后面代码要用到;
- 测试有效性:在官网“发送消息”栏输入任意内容,点击发送,手机微信会立刻收到“Server酱”的推送,说明配置成功。
提示:Server酱免费版足够满足个人需求,每天可发100条消息,完全覆盖量化信号提醒场景。
1.2 自动化核心:复用策略函数+定时任务
我们不需要重新写策略逻辑,直接复用第三篇的“成交量过滤均线策略”,新增三个核心功能:
- 数据实时化:获取“前一个交易日”的最新数据(不用手动改日期);
- 信号判断:只监控第三篇筛选出的“优选股”,避免无效信号;
- 定时执行:用Python的“schedule”库设置每天16:30自动跑代码(A股15:00收盘,16:30数据已更新)。
先安装需要的新库,打开命令提示符(Windows按Win+R输cmd),输入以下命令,回车等待安装完成:
pip install requests schedule # requests用于发微信消息,schedule用于定时
二、完整代码:打造“自动跑策略+微信提醒”系统
整个系统分为“优选股列表→策略函数→微信推送函数→定时任务”四部分,代码全注释,新手替换“SCKEY”和“优选股代码”就能用。
2.1 第一步:配置核心参数(新手必改3处)
# 1. 配置核心参数(新手只改这3处!)
# ① 替换成你的Server酱SCKEY(微信推送用)
SCKEY = "你的Server酱SCKEY"
# ② 替换成第三篇筛选出的优选股(格式:股票代码:股票名称)
GOOD_STOCKS = {
"600519.SH": "贵州茅台",
"300750.SZ": "宁德时代",
"002594.SZ": "比亚迪" # 可根据自己的优选股列表增减
}
# ③ 替换成你的Tushare Token
TUSHARE_TOKEN = "你的Tushare Token"
# 2. 导入所有需要的库
import tushare as ts
import pandas as pd
import numpy as np
import requests
import schedule
import time
from datetime import datetime, timedelta
2.2 第二步:封装微信推送函数(1行代码发消息)
把微信推送逻辑打包成函数,后面只要调用“send_wechat_msg(消息内容)”,就能立刻推送到微信:
# 微信推送函数:传入消息内容,自动推送到微信
def send_wechat_msg(content):
# 构造Server酱的请求链接
url = f"https://sctapi.ftqq.com/{SCKEY}.send"
# 消息参数(title是标题,desp是详细内容)
params = {
"title": "【股票量化信号提醒】",
"desp": content # 支持换行,用
分隔
}
# 发送请求
response = requests.get(url, params=params)
# 验证是否发送成功
if response.json()["code"] == 0:
print("微信消息推送成功!")
else:
print(f"推送失败,缘由:{response.json()['message']}")
# 测试函数:运行后微信收到消息说明没问题
# send_wechat_msg("测试消息:量化系统已启动!")
2.3 第三步:升级策略函数(支持实时数据)
复用第三篇的成交量过滤策略,新增“自动获取最新交易日”功能,不用再手动修改start_date和end_date:
# 初始化Tushare接口
pro = ts.pro_api(TUSHARE_TOKEN)
# 自动获取最新交易日(避免手动改日期)
def get_last_trade_date():
# 获取最近30天的交易日数据,取最后一个
trade_cal = pro.trade_cal(exchange='SSE', start_date='20240101', end_date=datetime.now().strftime('%Y%m%d'))
last_trade_date = trade_cal[trade_cal['is_open']==1]['cal_date'].iloc[-1]
return last_trade_date
# 升级后的策略函数:输入股票代码,返回是否出现金叉信号
def check_golden_cross(ts_code, stock_name):
last_date = get_last_trade_date()
# 获取最近60天的数据(足够计算20日均线+判断信号)
start_date = (datetime.strptime(last_date, '%Y%m%d') - timedelta(days=60)).strftime('%Y%m%d')
# 获取股票数据(和之前一致)
df = pro.daily(ts_code=ts_code, start_date=start_date, end_date=last_date, adj='qfq')
if len(df) < 20:
return False, f"{stock_name}({ts_code})数据不足,无法判断信号"
# 策略逻辑(成交量过滤+均线交叉,和第三篇一致)
df = df.sort_values('trade_date')
df['trade_date'] = pd.to_datetime(df['trade_date'])
df.set_index('trade_date', inplace=True)
df['ma5'] = df['close'].rolling(window=5).mean()
df['ma20'] = df['close'].rolling(window=20).mean()
df['vol_5avg'] = df['vol'].rolling(window=5).mean().shift(1)
df['vol_enlarge'] = np.where(df['vol'] > df['vol_5avg']*1.3, 1, 0)
df['ma5_prev'] = df['ma5'].shift(1)
df['ma20_prev'] = df['ma20'].shift(1)
# 只判断最新一个交易日是否出现金叉
latest_data = df.iloc[-1]
prev_data = df.iloc[-2]
# 金叉条件:前一日ma5<ma20,当日ma5>ma20,且成交量放大
if (prev_data['ma5'] < prev_data['ma20']) & (latest_data['ma5'] > latest_data['ma20']) & (latest_data['vol_enlarge'] == 1):
return True, f"{stock_name}({ts_code})
出现金叉买入信号!
日期:{latest_data.name.strftime('%Y-%m-%d')}
收盘价:{latest_data['close']:.2f}元
成交量:{latest_data['vol']}手(较前5日放大30%+)"
else:
return False, f"{stock_name}({ts_code})未出现金叉信号"
# 测试策略函数:检查贵州茅台是否有信号
has_signal, msg = check_golden_cross("600519.SH", "贵州茅台")
print(msg)
2.4 第四步:封装主函数(批量检查+信号推送)
主函数负责循环检查所有优选股,一旦发现金叉信号,立刻调用微信推送函数,把消息发给你:
# 主函数:批量检查优选股+推送信号
def check_all_stocks():
print(f"===== {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 开始检查优选股信号 =====")
# 收集所有信号结果
results = []
# 循环检查每只优选股
for ts_code, stock_name in GOOD_STOCKS.items():
has_signal, msg = check_golden_cross(ts_code, stock_name)
results.append(msg)
print(msg)
# 有金叉信号就立刻推送微信
if has_signal:
send_wechat_msg(msg)
# 汇总所有结果,推送一次完整报告
summary = "
".join(results)
send_wechat_msg(f" 优选股信号汇总({datetime.now().strftime('%Y-%m-%d')})
{summary}")
print("===== 检查完成,结果已推送至微信 =====")
# 测试主函数:手动触发一次检查
# check_all_stocks()
2.5 第五步:设置定时任务(每天自动跑)
用schedule库设置每天16:30自动执行主函数,只要电脑开着,就会按时跑代码,不用手动操作:
# 设置定时任务:每天16:30自动执行
schedule.every().day.at("16:30").do(check_all_stocks)
# 让程序一直运行,等待定时任务触发
print("量化自动提醒系统已启动!")
print(f"每天16:30自动检查优选股信号,结果将推送至微信")
print("如需停止程序,按Ctrl+C即可")
while True:
schedule.run_pending() # 检查是否有定时任务需要执行
time.sleep(60) # 每隔60秒检查一次,减少电脑资源占用
三、实战测试:3步确认系统能正常工作
代码写好后别直接挂后台,先按以下步骤测试,确保每个环节都没问题:
3.1 第一步:测试微信推送
找到“send_wechat_msg”函数下面的测试代码,去掉注释(删掉#),运行代码:
# 测试微信推送(去掉注释运行)
send_wechat_msg("测试消息:量化系统已启动!")
如果手机微信收到“Server酱”的推送消息,说明微信配置成功;如果没收到,检查SCKEY是否复制正确,或重新登录Server酱刷新SCKEY。
3.2 第二步:测试策略信号判断
运行“check_golden_cross”的测试代码,查看控制台输出:
# 测试策略函数(去掉注释运行)
has_signal, msg = check_golden_cross("600519.SH", "贵州茅台")
print(msg)
如果输出“贵州茅台(600519.SH)未出现金叉信号”或具体的金叉信息,说明策略函数正常;如果报错“数据不足”,检查Tushare Token是否有效。
3.3 第三步:测试定时任务
为了避免等太久,先把定时时间改成“当前时间+1分钟”,列如目前15:20,就设置15:21执行:
# 临时测试定时任务:设置1分钟后执行(测试完改回16:30)
schedule.every().day.at("15:21").do(check_all_stocks)
运行代码后,等待到设定时间,看控制台是否打印“开始检查优选股信号”,并收到微信推送的汇总报告,确认定时任务生效。
四、进阶优化:新手必学的实用技巧
基础版本能满足需求,再教你3个进阶技巧,让系统更稳定、更实用:
4.1 技巧1:加入日志记录(方便排查问题)
新增日志功能,把每天的检查结果存到文件里,就算程序报错,也能找到问题缘由:
import logging
# 配置日志:存到quant_log.txt文件里
logging.basicConfig(
filename='quant_log.txt',
level=logging.INFO,
format='%(asctime)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# 在主函数里添加日志记录
def check_all_stocks():
log_msg = f"===== 开始检查优选股信号 ====="
print(log_msg)
logging.info(log_msg)
results = []
for ts_code, stock_name in GOOD_STOCKS.items():
has_signal, msg = check_golden_cross(ts_code, stock_name)
results.append(msg)
print(msg)
logging.info(msg) # 记录每只股票的结果
if has_signal:
send_wechat_msg(msg)
summary = "
".join(results)
send_wechat_msg(f" 优选股信号汇总({datetime.now().strftime('%Y-%m-%d')})
{summary}")
log_msg = "===== 检查完成 ====="
print(log_msg)
logging.info(log_msg)
4.2 技巧2:电脑关机也能跑(用云服务器)
如果不想一直开着电脑,可以用阿里云、腾讯云的“轻量应用服务器”,学生机每月只要9.9元,步骤如下:
- 购买服务器:选“Windows Server 2019”系统(和电脑操作一样,新手易上手);
- 远程连接:电脑按Win+R输“mstsc”,输入服务器的“公网IP+用户名+密码”,远程控制服务器;
- 配置环境:在服务器上安装Python和需要的库,把代码复制过去,运行后最小化即可(服务器会24小时运行)。
4.3 技巧3:新增死叉提醒(及时止盈)
在策略函数里新增死叉判断,持仓股票出现死叉时立刻提醒卖出,形成“买入-卖出”完整闭环:
# 升级策略函数:同时判断金叉和死叉
def check_cross_signal(ts_code, stock_name, hold_status=False):
# 前面代码和check_golden_cross一致...
# 新增死叉判断(持仓时才需要关注)
if hold_status:
if (prev_data['ma5'] > prev_data['ma20']) & (latest_data['ma5'] < latest_data['ma20']):
return "death", f"{stock_name}({ts_code})
出现死叉卖出信号!
日期:{latest_data.name.strftime('%Y-%m-%d')}
收盘价:{latest_data['close']:.2f}元"
# 金叉判断
if (prev_data['ma5'] < prev_data['ma20']) & (latest_data['ma5'] > latest_data['ma20']) & (latest_data['vol_enlarge'] == 1):
return "golden", f"{金叉消息内容}"
return "none", f"{无信号消息}"
# 调用时指定是否持仓
signal_type, msg = check_cross_signal("600519.SH", "贵州茅台", hold_status=True)
if signal_type in ["golden", "death"]:
send_wechat_msg(msg)
五、本期避坑指南:自动化系统的4个关键注意点
- 避坑1:Tushare限流问题——免费用户每秒最多调用2次接口,批量检查30只股票时,在循环里加“time.sleep(0.5)”,避免被限流;
- 避坑2:交易日判断——周末和节假日A股不开盘,系统会自动跳过(由于get_last_trade_date会获取最近的交易日),不用手动暂停;
- 避坑3:SCKEY保密——别把SCKEY分享给别人,否则可能收到垃圾消息,如需更换,在Server酱官网重新生成即可;
- 避坑4:不依赖单一信号——系统提醒只是“辅助决策”,出现金叉时先看大盘是否稳定(列如沪指是否跌超1%),避免盲目跟风。
六、系列总结与下期预告:策略优化天花板
到这篇为止,我们已经完成了“0基础到自动化”的完整量化闭环:从Tushare获取数据→搭建均线策略→成交量过滤假信号→多股批量回测→全自动微信提醒,新手跟着系列文章操作,已经能搭建属于自己的量化工具。
下期我们将冲击“策略优化天花板”——引入“参数寻优”和“多因子策略”:用Python自动找出每只股票最适合的均线参数(列如有的股票适合5/20日,有的适合10/30日),再结合“MACD+RSI”指标构建多因子策略,进一步提升胜率。
目前就把你的优选股列表填入代码,启动自动化系统吧!如果遇到服务器配置、信号推送等问题,直接在评论区留言,我会逐一帮你解决。



Tushare现在好像是限流每60秒一次了
有视频教程吗
分享给大家
牛叉叉
继续加油💪
收藏了,感谢分享