盘中盯盘最累人的不是看涨跌而是盯均线。你打开行情软件把几十只自选股翻来翻去就为了等MA5上穿MA10的那一下等你喝口水回来行情一变可能又错过了。我后来直接把这件事实时化处理写了一个Python脚本每隔几分钟自动扫一遍沪深A股凡是在当天形成MA5上穿MA10的股票直接打印出来。这篇文章就是把完整思路、核心代码和我在实际使用中踩过的坑整理出来内容包括数据源怎么选、金叉怎么算才不误判、全市场扫描怎么做并发提速以及回测验证和风控细节。适合有Python基础、想做实时选股或者量化入门的读者参考。先说清楚这套东西做的是“筛选器”不是“印钞机”。它帮你把全市场几千只股票里当天刚刚形成均线金叉的标的捞出来省去手动翻盘的时间至于捞出来之后能不能买那是另一套交易逻辑的问题。下面我从信号逻辑讲到完整代码再讲回测和实战过滤把整个过程掰开揉碎。1. 金叉信号到底是什么MA5上穿MA10的选股逻辑与适用边界1.1 均线金叉的数学定义与趋势含义MA5是最近5个交易日收盘价的简单平均值MA10是最近10个交易日的平均值。所谓MA5上穿MA10说的是昨天MA5还在MA10下方或者贴着MA10今天MA5跑到了MA10上方。这个事件发生在K线图上就是两条均线出现了一个向上的交叉点。用大白话解释一下。你可以把MA5理解成“最近5天买入的人平均成本”MA10理解成“最近10天买入的人平均成本”。当MA5从下方突破MA10说明短期买入者的平均成本开始超过中期买入者的平均成本意味着最近的资金更愿意出高价进场短期动能压过了中期趋势。这种信号在波段交易里经常被当成趋势可能转强的早期提示。再打一个生活化的比方假设你开了一家奶茶店最近5天的日均营业额超过了最近10天的日均营业额说明这段时间生意明显变好了。均线金叉就是类似的意思——用平均成本来过滤单日涨跌的随机噪音观察一段时间的动能变化。1.2 这策略适合谁、不适合谁MA5上穿MA10这种双均线金叉本质上是中短线趋势跟踪信号适合以下场景做波段、持股周期在几天到几周的交易者。愿意接受假信号用止损或仓位控制来管理风险的人。不想整天盯盘但希望用程序自动监控全市场的用户。不适合谁呢不适合把它当成“买入就涨”的圣杯。震荡行情里股价上下反复均线会频繁金叉死叉信号多到让人麻木胜率也会明显下降。趋势行情里效果会好一些但趋势行情什么时候来没人能提前预测。所以我的态度一直是金叉负责“筛选”过滤器负责“优化”交易纪律负责“保命”。2. 行情数据源选型免费接口怎么做到不败坏“实时性”2.1 三套主流免费数据源对比A股行情数据源不少但对个人开发者来说真正能免费拿到、且更新足够快的其实就那几个。我列一个自己长期用的对比表方便你选型。数据源接入方式实时性限制与费用适合场景aksharepip install akshare无需token直接调用快照秒级刷新历史K线盘中包含当日未收盘bar免费但接口参数随上游调整需要留意版本更新个人全市场扫描、原型验证tushare注册token后调用ts.pro_bar日线/分钟线积分越高权限越细免费档有积分和调用频率限制历史回测、工程化项目东方财富push2接口直接HTTP请求行情接口秒级有风控并发过高容易封IP有经验的人做小批量自选监控我的建议是刚上手直接用akshare。原因很简单不需要注册token不需要配置环境变量pip装完之后几行代码就能拉到股票列表和历史K线把精力集中在信号逻辑本身。tushare更适合需要字段稳定、接口可控的工程化场景。至于东方财富的原始HTTP接口虽然理论上实时性最好但你要自己处理headers、参数签名、频率控制踩坑成本高新手阶段完全没必要。2.2 为什么A股的“盘中日K”是个特例这里有一个特别容易忽略的点。你在盘中运行脚本时akshare的历史K线接口返回的最后一根bar并不是真正收盘的日K而是“用当前最新价拼出来的未收盘K线”。比如你下午2点跑一次它把2点的最新价当作“今日收盘价”算进MA5下午3点收盘后再跑一次今日收盘价才是真收盘价。这就带来一个现象盘中出现的金叉尾盘可能会消失。比如上午股价拉升MA5瞬间穿过MA10指标亮了下午股价回落最新价把MA5又拉回MA10下方金叉没了。反过来尾盘突然拉升的股票可能在最后半小时才形成金叉。所以“实时筛选MA5上穿MA10”这句话严格一点说是“用最新价模拟收盘价实时预判日线金叉是否会出现”。明白这个原理之后你就知道为什么我建议每天下午14:50左右跑一次全市场扫描——那时候离收盘只剩10分钟当日K线基本定型信号稳定性远高于上午盘中。3. 核心算法计算MA5/MA10并准确识别“当日金叉”3.1 最小数据量的严谨推导先回答一个基础问题判断一只股票今天是否出现MA5上穿MA10至少需要多少根日K线计算今天的MA10需要今天往前数10个交易日的收盘价计算昨天的MA10需要昨天往前数10个交易日。这两组数据叠加起来至少需要11根K线。如果你只取5根、6根今天的MA10都算不出来更谈不上比较。实际代码里我会取最近60个自然日的数据大约能覆盖40多个交易日。这样做的原因有三个一是给前复权数据留缓冲避免复权因子在早期样本上产生误差二是如果想顺手看一下20日均线或者60日均线数据也够用三是对接口的请求压力也不大单只股票一次拉几十行数据响应非常快。3.2 用pandas向量化计算金叉序列核心计算用pandas的rolling函数就可以完成不需要自己写for循环。import pandas as pd def compute_ma_cross(df): df[日期] pd.to_datetime(df[日期]) df df.sort_values(日期).reset_index(dropTrue) df[MA5] df[收盘].rolling(5).mean() df[MA10] df[收盘].rolling(10).mean() ma5_yesterday df[MA5].iloc[-2] ma10_yesterday df[MA10].iloc[-2] ma5_today df[MA5].iloc[-1] ma10_today df[MA10].iloc[-1] if any(pd.isna(x) for x in [ma5_yesterday, ma10_yesterday, ma5_today, ma10_today]): return False return ma5_yesterday ma10_yesterday and ma5_today ma10_today第一件必须做的事是排序。akshare的日线接口默认按日期升序返回但为了防御上游接口调整我会显式按“日期”排序再reset_index保证iloc[-1]是今天、iloc[-2]是昨天。识别“当天金叉”最核心的一点是必须比较两天的状态。很多人第一次写这个需求会直接写成df[MA5] df[MA10]然后筛出所有MA5在MA10上方的股票。这样选出来的都是已经金叉了好几天、甚至金叉后涨了一大波的标的不是“当天刚刚发生上穿”。真正的当日金叉判断是昨天MA5还在MA10下方或相等今天MA5站上了MA10。判断出“MA5 MA10”和“MA5 MA10”之后不要急着用df[MA5].diff()这类方法去看差值变化。直接用两天的布尔值组合最直观也最容易调试。3.3 前复权的重要性这是一个新老股民都很容易踩的坑算均线必须用前复权价。如果不用复权股票除权除息当天价格会突然出现一个很大的跳空。比如一只股票10转10除权日股价从20元变成10元5日均线和10日均线都会被强行拉低接下来几天可能就会形成一个“假金叉”。从图形上看确实交叉了但这个交叉是除权造成的数字游戏跟买卖力量没有任何关系。用前复权价格相当于把历史价格统一到当前股本视角除权产生的价格断层被修正均线才能真实反映趋势。在akshare里面取数时stock_zh_a_hist有一个adjust参数传qfq就是前复权。这个参数不能省省了就是在给策略埋雷。4. 完整实现全市场实时筛选MA5上穿MA10的Python脚本4.1 依赖安装与常见环境问题先安装依赖主要是akshare和pandaspip install -U akshare pandasakshare版本更新比较频繁上游接口一旦调整旧版本就可能报参数错误。所以我习惯每次跑脚本之前先看一眼版本或者直接在计划任务里加一步pip install -U akshare保证接口跟得上。如果你是Windows机器装完pandas后运行脚本时如果提示“由于找不到msvcp140.dll无法继续执行代码”说明系统缺VC运行库。去微软官网下载“Microsoft Visual C 2015-2022 Redistributable (x64)”安装一遍重启终端就好了。这个问题我在新电脑上遇到过不止一次不是代码问题纯粹是运行环境缺组件。4.2 股票池生成与K线抓取直接给出完整主脚本我把它命名为ma_cross_monitor.py。逻辑分三步拉全市场股票列表逐只拉最近K线并判断是否金叉最后输出结果并按成交额排序。import akshare as ak import pandas as pd from datetime import datetime, timedelta from concurrent.futures import ThreadPoolExecutor, as_completed import time import os import json HIST_DAYS 60 # 拉取最近60个自然日约40多个交易日 THREAD_NUM 8 # 并发线程数 INTERVAL 0.1 # 每个请求完成后的间隔时间单位秒 TOP_N 30 # 只输出成交额前30的信号 def get_pool(): 获取沪深A股股票池排除ST和北交所 df ak.stock_info_a_code_name() df[code] df[code].astype(str).str.zfill(6) df df[~df[name].str.contains(ST|退, naFalse)] df df[df[code].str.startswith((60, 00, 30, 68))] return df.reset_index(dropTrue) def check_one(code, name): 拉取单只股票近期日K判断今天是否出现MA5上穿MA10 try: end datetime.now().strftime(%Y%m%d) start (datetime.now() - timedelta(daysHIST_DAYS)).strftime(%Y%m%d) df ak.stock_zh_a_hist( symbolcode, perioddaily, start_datestart, end_dateend, adjustqfq ) if df is None or len(df) 15: return None df[日期] pd.to_datetime(df[日期]) df df.sort_values(日期).reset_index(dropTrue) df[MA5] df[收盘].rolling(5).mean() df[MA10] df[收盘].rolling(10).mean() if pd.isna(df[MA5].iloc[-2]) or pd.isna(df[MA10].iloc[-2]): return None cross_up ( df[MA5].iloc[-2] df[MA10].iloc[-2] and df[MA5].iloc[-1] df[MA10].iloc[-1] ) if not cross_up: return None return { code: code, name: name, close: round(df[收盘].iloc[-1], 2), pct: df[涨跌幅].iloc[-1], ma5: round(df[MA5].iloc[-1], 2), ma10: round(df[MA10].iloc[-1], 2), amount: df[成交额].iloc[-1], } except Exception as exc: print(f[warn] {code} {name} 获取失败: {exc}) return None def scan(): pool get_pool() print(f[info] 待扫描股票数: {len(pool)}) rows [] with ThreadPoolExecutor(max_workersTHREAD_NUM) as executor: future_map { executor.submit(check_one, row[code], row[name]): row for _, row in pool.iterrows() } for future in as_completed(future_map): result future.result() if result: rows.append(result) time.sleep(INTERVAL) if not rows: print([info] 当前无符合条件股票) return out pd.DataFrame(rows).sort_values(amount, ascendingFalse) out out.head(TOP_N) print(out.to_string(indexFalse)) push_notify(out) def push_notify(out): 增量去重当天已经推送过的信号不重复推送 pushed_path pushed.json pushed set() if os.path.exists(pushed_path): with open(pushed_path, r, encodingutf-8) as f: pushed set(json.load(f)) today datetime.now().strftime(%Y-%m-%d) for _, row in out.iterrows(): key f{row[code]}_{today} if key not in pushed: print( f[新信号] {row[name]}({row[code]}) f现价{row[close]} MA5{row[ma5]} MA10{row[ma10]} f涨幅{row[pct]}% 成交额{row[amount]} ) pushed.add(key) with open(pushed_path, w, encodingutf-8) as f: json.dump(list(pushed), f, ensure_asciiFalse, indent2) if __name__ __main__: scan()这里有几个细节说明一下。get_pool里面我用startswith((60, 00, 30, 68))来过滤对应的分别是沪市主板、深市主板、创业板、科创板。北交所代码以8和4开头默认排除如果你需要自己把对应前缀加进去即可。ST和退市整理期的股票也直接排除避免垃圾信号干扰。check_one里面为什么取len(df) 15作为判断一般来说一只股票上市满60个自然日交易日大约是40多天足够计算MA5和MA10。如果数据量太少说明可能次新股或者停牌太久直接跳过。4.3 并发优化与请求节奏控制全市场有5000多只股票如果用for循环一只一只拉每只请求等0.5秒一轮下来四五十分钟完全谈不上“实时筛选”。所以必须用并发。我使用ThreadPoolExecutor开8个线程。每个请求本身是网络IO操作线程在等待响应时并不会占用多少CPU多线程同时等整体耗时会从“总数量x单次耗时”缩短到约“总数量/线程数x单次耗时”。实测下来全市场一轮扫描大约2到3分钟基本可以接受。但并发不是越大越好。我最初把线程开到32跑了不到五分钟接口就开始返回空DataFrame再往后甚至直接超时。这不是代码的问题是上游接口扛不住这种请求密度触发了风控。后来我把线程数降到8并且每完成一个请求在宿主线程里time.sleep(0.1)让请求间隔均匀一些全市场扫描就稳定多了。个人经验是线程数控制在6到12之间间隔0.1到0.3秒速度和安全性能达到一个平衡。如果你不想等两三分钟只想快速盯盘可以把股票池缩小到自选股或某个板块。比如只扫沪深300成分股一轮几十秒只扫自选的30只几秒就出结果。实时监控场景里个人自选股优先级远高于全市场。4.4 增量去重不要让同一个信号刷屏脚本如果每隔几分钟跑一次同一只股票在14:00出现金叉14:05再跑仍然满足条件如果不做去重终端就会被同一个股票刷屏。解决办法很简单用“股票代码日期”作为唯一标识当天已经输出过的信号后面几轮扫描不再提醒。这里用pushed.json文件存储历史推送记录文件里是一个字符串数组每次扫描前加载到内存Set里。因为key里带了日期第二天这个key自然失效不需要手动清理。脚本重启后也会自动读取历史记录不会因为重启导致上午推过的信号下午再推一遍。5. 事后来看用历史数据验证一下这个信号靠不靠谱5.1 一个不需要额外框架的快速回测函数写实时脚本之前我建议你先用历史数据做一个快速验证确认这个信号在统计意义上有没有参考价值。最粗粒度的验证方式是拉一只股票过去几年的日线找出所有满足MA5上穿MA10的日期统计信号出现后若干日的收益然后和“随便找个交易日持有同样天数”的收益做对比。def quick_backtest(df, hold5): 对单只股票做简单回测返回信号日收益与全样本收益序列 df df.copy() df[日期] pd.to_datetime(df[日期]) df df.sort_values(日期).reset_index(dropTrue) df[MA5] df[收盘].rolling(5).mean() df[MA10] df[收盘].rolling(10).mean() df[cross] (df[MA5] df[MA10]) (df[MA5].shift(1) df[MA10].shift(1)) df[future_close] df[收盘].shift(-hold) df[future_ret] df[future_close] / df[收盘] - 1 signal_ret df.loc[df[cross], future_ret].dropna() all_ret df[future_ret].dropna() return signal_ret, all_ret然后把多只股票的结果汇总起来看信号日的平均收益是否明显优于随机持有收益再看胜率。这个方法虽然粗糙但足以判断一个信号有没有继续优化的价值。5.2 用backtrader做多股回测的思路如果要把策略做得更完整比如加入资金管理、止损止盈、多股票组合回测我会用backtrader。它是Python里比较流行的回测框架双均线金叉策略写起来非常简洁import backtrader as bt class MaCross(bt.Strategy): params dict(fast5, slow10) def __init__(self): self.ma_fast bt.ind.SMA(self.data.close, periodself.p.fast) self.ma_slow bt.ind.SMA(self.data.close, periodself.p.slow) self.cross bt.ind.CrossOver(self.ma_fast, self.ma_slow) def next(self): if self.cross 0: self.buy() elif self.cross 0: self.close()CrossOver会返回三种状态1表示上穿-1表示下穿0表示无交叉。策略里cross 0买入cross 0平仓。多股回测时我不建议在一个Cerebro里塞入大量不同股票然后让同一份策略实例处理多数据逻辑容易混乱。更清晰的做法是循环单只股票、单独创建Cerebro回测最后把每只股票的收益率、最大回撤、胜率汇总成一个总表。这样每一只股票是独立交易不会出现资金串仓的问题统计结果也直观。5.3 回测结果里最容易误读的两个陷阱看回测结果时有两点我建议特别留意。第一不要忽视交易成本。A股有佣金、印花税、滑点一笔完整的买入卖出综合成本可能在0.1%到1%之间。如果策略的期望收益就0.2%扣掉成本可能就变成负期望了。回测脚本里至少应该预留万分之几的手续费参数否则结果会偏乐观。第二一字涨停和停牌是回测的盲区。信号出现当天如果是一字涨停你根本买不进去但回测模型默认按收盘价成交会把这种不可成交的信号也算成盈利。所以我在实时脚本里才要加一字板过滤回测时也应该把信号日开盘涨幅超过9.5%的样本剔除否则统计结果会失真。另外任何固定参数都有过拟合风险。MA5和MA10在某个时间段表现好不代表未来永远好。我见过很多人在回测曲线里看到一条漂亮的上扬线就激动但换个年份再测曲线就面目全非。策略验证的重点应该放在多个年份、多种市场环境的稳定性上。6. 实战会用到的几个过滤器从“有信号”到“值得关注”6.1 排除ST、新股、停牌与一字板拿到金叉信号之后不能直接无脑关注。我会在实战中叠加几个过滤条件把明显不适合参与的标的一开始就筛掉。ST和退市整理期涨跌停幅度和普通股票不同资金结构也不同均线金叉的参考意义很弱。上市不足60个交易日的新股MA10都还没完全走顺金叉信号可信度低。停牌股当天没有实时成交历史数据里出现的信息是延迟的判断会失真。一字板或开盘即涨停的股票就算出现了金叉以散户的通道也很难买到信号再漂亮也没有操作性。一字板判断在实时脚本里可以这样加检查当天K线的开盘价、最高价、最低价、收盘价是否全部相等并且涨幅大于9.5%。如果满足说明从开盘到当前一直是封板状态直接跳过。如果只想排除接近涨停的票可以放宽为“开盘涨幅 9.5%”。A股主板涨停幅度是10%创业板和科创板是20%可以按板块设定阈值这里只是抛砖引玉。6.2 叠加量能、量比与板块热度金叉信号加上量能确认是我个人觉得最实用的一层优化。简单来说同样的金叉有成交额配合和没有成交额配合含义完全不同。一只股票平时日成交额只有几千万某天突然放大到五六个亿并出现MA5上穿MA10说明资金关注度显著提升信号的有效性更强。反之如果金叉发生在极度缩量的状态下可能只是买卖力量短暂失衡持续性存疑。在输出排序时我默认按成交额降序本质上就是在做量能过滤。如果你想再严格一点可以在check_one里加一个最低成交额条件比如要求当日成交额大于2亿元。这样做的代价是会漏掉一些小盘股的机会但换来的是更高的信号质量。具体阈值可以根据你自己的资金体量和风险偏好来调。板块热度是个加分项但不建议写进第一版代码。你可以把每天筛选出的信号列表贴到行情软件里观察它们属于哪些板块如果多个信号集中在同一个板块说明板块效应在起作用可以高看一眼。程序自动去做板块归类也能实现但需要额外的板块成分数据工作量更大适合第二期再优化。6.3 盘中、尾盘、收盘后三个时段的玩法同一个金叉信号在不同时间点运行软件意义是不同的。盘中运行适合“打提前量”。你想在信号确认的第一时间知道然后自己去判断要不要介入。缺点是当日K线还没定型信号可能在尾盘消失因此盘中扫描建议频率高一点但不要直接用扫描结果下单。尾盘14:50左右运行是我个人最推荐的。这个时间点离收盘很近当日K线基本定型金叉信号的真实性大幅提高。看到信号后可以放到第二天结合开盘情况再决定是否参与正好也符合A股T1的规则。收盘后再运行适合做交易计划。把当天所有金叉股票整理出来晚上慢慢复盘、看基本面、看板块然后制定第二天的操作预案。这时候没有时间压力胜在从容。我把这三套用途区分清楚之后就不会被盘中那些“虚晃一枪”的信号牵着走了。7. 行情接口的风控细节与自保策略别把自己的IP玩进去7.1 反爬到底长什么样很多人在写这类选股工具时前半天代码都跑得好好的后半天突然发现接口返回空DataFrame再跑几次直接报网络错误。这不是网络故障而是你的请求频率太高被上游接口限流了。东方财富系的产品线包括行情中心、股吧论坛等整体上对异常请求的容忍度都不高。我自己遇到过的情况有几种返回数据为空、响应时间越来越长、HTTP 403、偶尔弹出验证码。这里重点说一个经验行情接口还好如果你顺手去爬股吧论坛的帖子会发现反爬强度明显高一个级别请求频率稍高就断。所以不要以为行情接口免费不限量就使劲薅。“东方股吧反爬”这个说法在不少技术群里都出现过本质上不是什么神秘的东西就是服务端设置了一系列风控策略识别并拦截非浏览器行为。对个人量化研究来说正确做法不是去研究怎么绕过而是严格约束自己的请求节奏避免触发风控。7.2 给自己加三道“礼貌层”我在写实时扫描脚本时给自己定了三条规矩你也可以直接照抄控制线程数和请求频率。线程不超过12个每个请求之间至少留0.1秒间隔全市场一轮控制在两到三分钟。设置兼容的请求头。如果你直接请求东方财富的HTTP接口建议带上常见浏览器的User-Agent和Referer模仿正常浏览器的访问行为。用akshare封装接口时库内部一般已经处理了不需要额外操心。失败时指数退避重试。第一次失败等1秒第二次等2秒第三次等4秒最多重试三次。不要失败后立刻无脑重试那只会让风控判定更严厉。另外个人研究用途的脚本不要部署成高频率任务。每隔5到10分钟跑一次全市场就够了没必要做成每秒刷新的实时推送。A股本来也不是每秒都能产生日线金叉信号过度监控只会增加风控风险和数据噪音。7.3 崩溃恢复与状态持久化实时扫描脚本通常会挂很长时间运行过程中可能因为网络抖动、接口超时、甚至是电脑休眠导致进程中断。我建议做两件小事能让脚本稳定很多。第一把推送记录持久化到本地文件。我上面代码里已经写了pushed.json即使脚本中途崩溃重启后也不会把之前推送过的信号再推一遍。这个细节看着不起眼但能避免很多重复提醒带来的烦躁。第二把日志输出到文件而不是只打到终端。全市场扫描结果如果只靠print终端里翻几屏就找不到了。可以用一个简单的日志文件每次运行把时间戳、信号数量、股票列表追加进去方便事后复盘。我在脚本里为了演示简洁没有加logging但实际运行版本是带上的。接口返回数据异常时不要直接退出整个脚本而是捕获异常、打印告警、跳过当前股票继续扫描。全市场5000多只股票里偶尔有几只请求失败完全是正常现象。只要失败率不高就不影响整体结果。最后说点个人的体感。这套脚本真正改变我的是两件事一是把盯盘时间从全天降低到每天只跑几次二是逼我养成了把条件写清楚的习惯——加量能过滤、加尾盘确认、加一字板排除表面上看是在“少选”其实是在“选得更稳”。指标本身不会产生收益纪律才会。如果你打算照着这篇文章搭一套建议先跑几天不要实盘把输出结果和你自己的复盘对照一下感受一下盘中信号和尾盘信号的差异。然后再决定用它来做提醒、还是做正式交易决策的一部分。