☰
Python实现A股MA5上穿MA10金叉实时筛选:数据源到定时任务全解析
2026/10/6 14:35:12 网站建设 项目流程

先说结论:MA5上穿MA10是A股短线交易里最经典、也被验证过无数次的金叉信号,但真正能把它“实时”跑起来、每天自动扫一遍全市场的代码,远没有网上那些张口就来的版本那么简单。这篇我就把整套筛选逻辑、数据源选型、代码实现和踩坑记录完整拆开,从零讲到一个可以直接丢到定时任务里用的版本。

先说清楚这个东西是什么、能做什么:所谓“实时筛选”,本质是每根K线收盘后,快速扫描全市场A股,找出当天MA5从下方穿过MA10的个股。MA5是5日均线,代表最近5天收盘价的平均值;MA10是10日均线,代表近10天的平均成本。MA5上穿MA10,意味着短线资金开始以更高的价格持续入场,把短周期成本推高到了中短周期成本之上,这是短线趋势转强的技术确认。

适合谁看:想用Python做A股量化选股、又不想只停留在“看代码”阶段的人;以及已经会用通达信、同花顺写公式选股,但觉得内置筛选速度慢、没法自定义复杂逻辑的朋友。这篇博文会从均线金的本质、数据源对比、代码实现到常见报错排查,一次性讲透,保证你看完能直接落地一套自己的实时筛选脚本。

1. 内容整体设计与思路拆解

1.1 为什么MA5上穿MA10这个信号值得做实时筛选

先说一个扎心的事实:均线金叉这玩意儿,几乎所有行情软件都能一键选出来,通达信里一行公式就搞定。但如果你真的拿它做了一段时间实盘,会发现两个问题——信号滞后和信号泛滥。

滞后是因为均线本身就是滞后指标。MA5、MA10都是对历史价格的平滑,股价先动、均线后动,这是数学性质决定的,改不了。但我们能做的是“实时”筛选,也就是在信号刚形成的那一天、甚至在盘中接近形成时就把票找出来,而不是等第二天复盘才看到,那样黄花菜都凉了。

泛滥是因为全市场5000多只股票,哪怕只用MA5上穿MA10这一个条件,每天符合的票可能也有几十上百只。这时候就需要叠加成交量、MACD、板块位置、是否ST、是否次新股等过滤条件。钝化过滤条件这件事,就是纯代码的活,也是这篇博文要重点展开的部分。

先说清楚我的总体设计思路:整个系统拆成四层——数据获取层、指标计算层、信号筛选层、结果输出层。数据获取层负责拉取全市场日线行情和实时行情;指标计算层负责算MA5、MA10以及补充条件;信号筛选层负责判断上穿并输出候选池;结果输出层负责把结果写成CSV或推送到本地文件,方便你每天打开看一眼。这个分层的核心目的是让每一层都能独立替换,比如今天用A数据源,明天想换B数据源,只需要改第一层,后面三层完全不用动。

1.2 实时筛选和收盘后筛选的本质区别

很多人对“实时”有误解,以为盘中每秒扫一遍才叫实时。其实对于MA5上穿MA10这种日线级别的信号,“实时”只需要做到“每天收盘后自动跑一遍”就够了。如果非要在盘中做,反而会有个麻烦——盘中价格还在变动,MA5和MA10的数值每时每刻都在变,早盘的一根大阳线可能让MA5瞬间上穿MA10,尾盘杀跌又跌回去,一天之内信号反复横跳,让你根本没法操作。

所以我的建议是:这套代码定位为“日线级实时筛选器”,每天14:50或者收盘后15:05自动运行一次。14:50跑的好处是能赶在尾盘前看结果,如果真想操作还有时间;收盘后跑的好处是数据稳定,绝对不会有盘中那种“一秒钟一个信号”的尴尬。实际使用中我推荐15:00收盘后跑,然后把结果导出来,晚上慢慢复盘,第二天再决定是否介入,这样心态会稳很多。

既然定了收盘后跑,那数据源必须满足两个条件:一是当天数据完整,二是拉取速度快。这直接决定了下面工具选型的走向。

2. 工具选型与数据源对比

2.1 三种常见方案的优劣对比

做A股实时筛选,市面上主流方案无非三种:通达信等行情软件自带的公式选股、tushare pro、akshare。我分别聊一下实际用下来的感受。

先说通达信自带选股。优点是真的快,本地跑,几秒钟能扫完整个市场,而且数据准确、不用考虑积分和限流。缺点是逻辑表达能力有限,复杂的过滤条件写起来很别扭,比如我想在MA5上穿MA10的同时,要求MACD在零轴上方且连续三天放量,通达信的公式编辑器写这种条件会比较绕。另外通达信的实时选股需要一直开着软件,适合手动操作,不适合做成自动化定时任务。

tushare pro是我最推荐的方案。它是目前国内量化圈用最多的数据接口,日线数据、实时行情都有,而且返回的是干净的DataFrame格式,配合pandas做指标计算非常顺手。缺点是有积分门槛,有些接口需要攒够一定积分才能调用,但这属于一次性投入,注册、签到、写点文档评论就能攒够,不算门槛太高。

akshare是完全免费开源的方案,直接爬取公开网页数据,不需要token。优点是零成本,缺点是数据格式不稳定,网站改版可能导致接口失效。我试过一段时间,日线数据倒是没啥问题,但实时分钟数据的字段会偶尔变动,需要不定期维护。对于想跑一个长期稳定的定时任务,我更倾向于tushare pro,下面是两者的核心参数对比。

对比项tushare proakshare
是否需要token是,需注册获取否,直接安装调用
日线数据稳定性高,格式固定中,偶尔受网站改版影响
实时/分钟数据分钟接口需积分免费但字段变动风险高
文档完善度较完善,社区案例多一般,文档略散
适合场景长期稳定的自动化策略学习、临时拉取数据

实际项目里我用tushare pro做主力数据源,akshare作为备份。一旦tushare当天接口抽风,可以临时切换到akshare顶上,不至于让整个定时任务挂掉。

2.2 tushare pro积分和接口权限的坑

这里必须啰嗦一句,因为十个人有八个会卡在这一步。tushare pro的很多接口都有积分门槛,比如基础的daily日线接口需要120分,实时分钟数据接口要更高的积分。新注册用户初始积分只有几十,得靠签到、完善资料、写使用文档来攒。

我建议的最低配置是:先攒到120分,能调用daily接口就够了。因为前面说了,这套实时筛选本质是“日线级”,只需要每天拉一次日线数据,不需要分钟线。当然,如果你对盘中筛选有执念,那得继续攒分到2000以上才能开通分钟接口,但那种场景通常已经属于高阶玩法,不是这篇讨论的范围。

注册之后记得设置自己的token,代码里通过ts.set_token('你的token')配置,然后优先用pro.daily()接口。这个接口一次最多拉6000条记录,做全市场日线数据时要按股票代码循环拉取,后面代码部分我会给出完整实现。

3. 核心代码实现与实操拆解

3.1 环境准备与依赖安装

先说明运行环境:Windows或Linux都行,Python 3.8以上即可,不需要装额外的数据库,数据全部放内存里算,跑完全市场也就几十秒。需要安装的库只有三个:pandas、tushare、akshare。如果你只用tushare,装前两个就够了。

pip install pandas tushare # 可选,作为备份数据源 pip install akshare

pandas是必须的,因为后面的MA计算、上穿判断全部靠它。tushare负责拉数据。akshare作为备选方案,不装也不影响主流程。另外提醒一下,Windows上如果装tushare遇到依赖问题,优先升级pip再装,我遇到过一次是numpy版本不兼容导致的,升级pip、重新安装numpy就解决了。

3.2 完整实现:全市场MA5上穿MA10实时筛选

直接上核心代码,我从一个可以跑通的最小版本讲起,再逐步叠加过滤条件。

先定义一个获取单只股票日线数据的函数。tushare的daily接口传入股票代码和起止日期,返回日期从新到旧排列的DataFrame,字段包括trade_date、open、high、low、close、vol等。这里的核心坑是:tushare返回的日期是字符串格式,比如'20250110',必须转成pandas的datetime类型,否则排序和切片都会出现问题。

import pandas as pd import tushare as ts ts.set_token('你的token') pro = ts.pro_api() def get_daily(code, start_date='20241001', end_date='20250110'): df = pro.daily(ts_code=code, start_date=start_date, end_date=end_date) if df is None or df.empty: return None df['trade_date'] = pd.to_datetime(df['trade_date'], format='%Y%m%d') df = df.sort_values('trade_date').reset_index(drop=True) return df

然后定义均线上穿判断函数。这里有个关键逻辑:判断“上穿”不是看今天的MA5比MA10大就行,而是要求昨天MA5小于等于MA10、今天MA5大于MA10。这就是金叉的定义,也是把“形成信号”和“维持上涨”区分开来的关键。

def check_cross(df): if df is None or len(df) < 20: return False df['ma5'] = df['close'].rolling(5).mean() df['ma10'] = df['close'].rolling(10).mean() # 用前一天的MA关系判断是否从下方向上穿过 prev_ma5 = df['ma5'].iloc[-2] prev_ma10 = df['ma10'].iloc[-2] curr_ma5 = df['ma5'].iloc[-1] curr_ma10 = df['ma10'].iloc[-1] if prev_ma5 <= prev_ma10 and curr_ma5 > curr_ma10: return True return False

然后是获取全市场股票列表。用stock_basic接口可以拿到所有上市股票的代码、名称、行业、上市日期等信息。这里顺便解决了一个关键问题:过滤ST和次新股。ST股不碰,这是做短线最起码的纪律;上市不足60个交易日的次新股均线还没走稳,MA5和MA10参考意义不大,也要排除。

def get_all_stocks(): stock_list = pro.stock_basic(exchange='', list_status='L', fields='ts_code,name,industry,list_date') # 过滤ST stock_list = stock_list[~stock_list['name'].str.contains('ST', na=False)] # 过滤上市不足60个交易日的次新股 stock_list['list_date'] = pd.to_datetime(stock_list['list_date'], format='%Y%m%d') recent_cutoff = pd.Timestamp.now() - pd.Timedelta(days=120) stock_list = stock_list[stock_list['list_date'] < recent_cutoff] return stock_list

主程序部分,遍历全市场,拉数据、判断信号、收集结果。这里要控制一下节奏,不要一次并发拉太多,否则容易触发tushare的限流。实测下来用time.sleep(0.1)就够稳了,全市场扫一遍大约需要8到10分钟,放在定时任务里完全不叫事。

import time def realtime_scan(): stocks = get_all_stocks() results = [] for _, row in stocks.iterrows(): code = row['ts_code'] try: df = get_daily(code) if check_cross(df): # 补充:今天的涨跌幅、收盘价,方便复盘 latest_close = df['close'].iloc[-1] latest_pct = df['pct_chg'].iloc[-1] results.append({ 'code': code, 'name': row['name'], 'industry': row['industry'], 'close': latest_close, 'pct_chg': latest_pct }) except Exception as e: print(f'处理{code}出错: {e}') time.sleep(0.1) result_df = pd.DataFrame(results) result_df.to_csv('ma5_cross_ma10_results.csv', index=False, encoding='utf-8-sig') print(f'筛选完成,共找到{len(result_df)}只股票') return result_df if __name__ == '__main__': realtime_scan()

这个最小版本已经能跑通。但我要直接说结论:这么裸跑出来的结果,你实际用起来会很难受。因为全市场符合“单纯金叉”的股票太多了,你需要叠加更多过滤条件,把候选池压到每天几只到十几只的合理范围。

3.3 叠加实战过滤条件:从信号池到候选池

我实际用的过滤条件有这么几个,按重要性排序:

第一,量能确认。金叉当天成交量必须高于最近5日均量。没有放量的金叉质量不高,很可能是一日游行情。实现上就是拿当天的vol和过去5天vol的均值做对比,要求vol > mean(vol_5d) * 1.2,即放量20%以上。

第二,MACD零轴位置。要求MACD的DIF线在零轴上方。MACD在零轴上方的金叉,代表中期趋势已经走多,这种金叉的成功率远高于零轴下方的反弹金叉。实现上可以直接用talib库,也可以手写12日和26日EMA来算。手写其实很简单,十几行代码的事,我用的是手写版本,免去额外装库的麻烦。

第三,市值过滤。流通市值低于30亿的票波动太大,散户资金就能拉起来,也容易被庄家控盘,一般不建议做。高于300亿的大盘股,MA5上穿MA10的弹性又不够,涨也涨不了几个点。每个人风格不同,我习惯盯30亿到200亿之间的中盘股。

第四,排除近两日已经大涨过的。如果金叉当天已经涨停或者涨幅超过7%,第二天追进去的风险很大,很容易高位站岗。我一般把当天涨幅上限设在5%到7%之间,具体数值看你自己的风险偏好。

把过滤条件整合进主程序的方式很简单,在check_cross返回True之后,再依次检查量能、MACD位置、市值,全部通过才append到结果里。

def advanced_check(df, row): if not check_cross(df): return False # 条件1:当日放量 latest_vol = df['vol'].iloc[-1] avg_vol5 = df['vol'].iloc[-5:-1].mean() if latest_vol <= avg_vol5 * 1.2: return False # 条件2:MACD快线在零轴上方 dif, dea, macd = calc_macd(df) if dif.iloc[-1] <= 0: return False # 条件3:当日涨幅不过高 if df['pct_chg'].iloc[-1] > 7: return False return True

这里的calc_macd是一个手写的MACD计算函数,按常规参数12、26、9来算。作为素材补充,我贴一个简化实现,有需要的可以直接拿去用:

def calc_macd(df, fast=12, slow=26, signal=9): close = df['close'] ema_fast = close.ewm(span=fast, adjust=False).mean() ema_slow = close.ewm(span=slow, adjust=False).mean() dif = ema_fast - ema_slow dea = dif.ewm(span=signal, adjust=False).mean() macd = (dif - dea) * 2 return dif, dea, macd

加完这些条件,你再看一眼筛选结果,会明显感觉到质量不一样。裸筛选可能有50只,叠加后可能只剩10只,而这10只就是第二天真正值得盯盘的票。

3.4 定时任务:让筛选自动化运行

代码写完只是第一步,真正的“实时”筛选必须让脚本在每天收盘后自动跑。Windows上用任务计划程序,Linux上用crontab。

Windows的做法:打开任务计划程序,新建任务,触发器设为每天15:05,操作设为运行Python脚本。这里有个坑,任务计划程序默认的“起始于”目录不是脚本所在目录,所以脚本里所有相对路径都可能找不到。解决办法很简单:在脚本开头加两行,把工作目录切换到脚本所在目录。

import os, sys os.chdir(os.path.dirname(os.path.abspath(__file__)))

Linux的crontab配置更简洁,一行就够。

5 15 * * 1-5 /usr/bin/python3 /home/user/stock_ma_cross.py

注意这里限定周一至周五跑,周末A股不开盘。另外日志输出建议重定向到文件里,不然crontab的日志会丢,出了问题没处查。我习惯这样写:

5 15 * * 1-5 /usr/bin/python3 /home/user/stock_ma_cross.py >> /home/user/stock_cron.log 2>&1

这样每次运行的结果和报错都会追加到log文件里,第二天打开看log就知道昨晚跑没跑成功。

4. 常见问题与排查技巧实录

4.1 数据源报错与限流问题

现象一:提示抱歉,您没有访问该接口的权限。这是tushare积分不够导致的。解决办法就是继续攒积分,或者等积分够了下个交易日再跑。没有其他捷径。注意不要用多个账号绕过积分限制,tushare的风控会检测到异常调用,可能直接封号。

现象二:全市场扫描中途报错卡死。大概率是拉数据太频繁被限流了。我实测tushare的日线接口,保持在每秒10次以内基本没事,保险起见用了time.sleep(0.1),也就是每秒最多10只。如果报错特别频繁,可以把sleep提高到了0.2。跑一次脚本多花几分钟而已,不亏。

现象三:返回数据是空的。先检查是不是新股没有历史数据,再检查日期范围是否正确。注意tushare的起止日期格式是字符串YYYYMMDD,中间不能有横杠。这个格式问题我至少看到十个人踩过坑,查半天报错才发现是格式不对。

4.2 信号重复和漏选问题

重复问题:如果你把脚本放在盘中跑,一只股票可能连续两天都被选中,因为前天金叉、今天均线还没死叉,但“上穿”条件已经不满足了。我的建议是:严格使用“昨天小于等于、今天大于”的穿越定义,而不是“今天大于”的状态定义。这一行差之毫厘,结果天壤之别。

漏选问题:有些股票当天停牌,没有当天数据。这时候df的最后一行其实是前一个交易日的数据,程序会误判成当天金叉。解决办法是在get_daily时检查最新日期是否为最近交易日,如果不是就跳过。这个逻辑加上之后,整个筛选结果的准确率会提升一个档次。

def is_latest_trade_date(df, expected_date): latest = df['trade_date'].iloc[-1] return latest == expected_date

在主程序里,先获取今天的日期(要排除周末),在扫描前先判断一下最新交易日,只处理那些最新数据为当日的股票。

4.3 从代码到实盘:一个盯着执行的提醒

最后分享一个我个人的经验:代码跑通了,不代表策略就能赚钱。MA5上穿MA10这个信号,它的本质是“跟随趋势”,胜率是靠多次小止损+少数大盈利积累出来的。实盘时我给自己定的纪律是:

第一,金叉第二天,如果高开超过3%,不追;如果平开或低开,观察前30分钟能否翻红。第二,单只股票仓位控制在总资金的10%以内,因为即使过滤条件再严格,单票胜率也就五五开。第三,买入后第三天不涨就走,不要把短线做成长线。均线金叉策略最忌讳的就是“被套就死扛”,因为这是趋势跟踪策略,不是价值投资策略,逻辑搞混了,纪律就容易崩。

这套系统我从开始写到稳定运行,前后改了三版。第一版瞎跑,信号几百个;第二版加了量能和MACD过滤,日子好过多了;第三版加入自动定时和停牌检查,才算真正“实时”起来。现在每天收盘后我就不用管了,早上起来扫一眼结果文件,心里大概有数今天该盯哪几只。这就是“实时筛选”真正的价值——不是替你决策,而是把5000只股票缩减成5只,让你把精力花在真正值得研究的对象上。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询