1. 先搞清楚需求本质MA5上穿MA10究竟在找什么1.1 均线金叉的技术含义把标题拆开看其实就两个核心要素MA5、MA10以及它们之间的“上穿”关系。MA5是指最近5个交易日收盘价的平均值它对短线价格波动更敏感反应快MA10是最近10个交易日收盘价的平均值代表的是稍长一点的持仓成本。当MA5从下方往上穿过MA10在技术分析里叫“金叉”通常被看作短期趋势由弱转强的信号。这个信号的核心逻辑是短周期价格的平均水平开始跑赢中周期平均水平说明最近几个交易日的买入力量在增强。很多刚入门的股友容易把“金叉”当成无脑买入信号这点我必须提醒一下。均线交叉本质上是对历史价格的一种平滑处理它描述的是“已经发生的事实”而不是预测未来的水晶球。单独靠一根金叉去交易胜率并不会很理想它在震荡市里尤其容易反复打脸——价格横盘时MA5和MA10会在粘合状态下频繁上下穿插制造大量假信号。所以真正的用法是把它当做一个“初筛条件”用来把全市场几千只股票快速压缩到一个很小的候选池再结合成交量、行情背景、个股基本面去做二次过滤。1.2 为什么交易时间内的实时筛选有价值收盘后做MA5上穿MA10筛选很简单拉历史数据算一遍就行。但这里存在一个很现实的问题金叉信号一旦在收盘后确认第二天开盘往往已经高开你看到的成本就不是信号出现那一刻的成本了。盘中价格在不断波动MA5和MA10的值也在跟着最新价实时变动同一个信号可能在上午10点出现、下午又消失或者下午再次出现。如果只做收盘后的静态筛选等于放弃了盘中信号出现时最有利的观察窗口。“实时筛选”这四个字背后还有一个容易被忽略的需求把筛选做成一个持续运行的监控程序而不是手动刷一下就跑。这样一旦某只股票的MA5在盘中上穿MA10程序能在第一时间把代码、名称、最新价、涨幅这些信息输出到你面前你再去结合自己的判断决策。这才是完整的“实时”意义。2. 方案设计用什么技术栈、选什么数据源2.1 语言与第三方库选型这个场景用Python是最合适的没有太多争议。原因很直接Python在量化数据处理上有大量现成库pandas处理时间序列极其顺手而A股实时行情、历史K线都有免费接口可以直接拉。整个工具核心就两步——拿数据、算均线用Python写起来代码量可以压得很小。数据处理上主要用pandas和numpy。pandas的DataFrame天然支持滚动计算rolling(5).mean()一行就能算出MA5不需要自己写循环取平均值效率高也不容易出错。实时监控部分用标准库的time做循环调度就够了没必要上celery这类重型任务系统——这个工具的运行频率本来就是分钟级不是秒级。2.2 数据接口的三个选择A股行情数据接口有不少我实际折腾过的有三个akshare、tushare、baostock。各有各的特点。接口是否免费实时全市场快照历史日K稳定性适合场景akshare免费支持一次拉全市场支持需逐只请求中等数据源偶尔波动个人工具、学习研究tushare部分免费需要高权限积分支持且批量效率高较好有积分积累的量化玩家baostock免费不支持实时快照支持且批量效率高较好只做历史回测实时筛选这个需求的核心是“全市场实时行情”所以akshare在这一点上最有优势。ak.stock_zh_a_spot_em()可以直接返回全市场A股的最新价、涨跌幅、换手率、量比等字段一次请求全出来省得自己拼接。唯一的问题是历史K线需要逐只拉取全市场5000多只股票跑一遍比较耗时但这只是启动阶段的一次性成本完全可以接受。2.3 整体处理流程这个工具我设计成两个阶段。第一阶段是盘前或程序启动时拉取所有A股代码然后逐只获取最近15个交易日的日K线数据在本地构建一个缓存第二阶段进入盘中循环每隔一分钟左右拉一次全市场实时快照把每只股票的实时价格代入缓存的历史数据里算出当前的MA5和MA10再和昨天的情况做对比判断是否发生上穿。这个设计的好处是把耗时操作逐只拉历史K线隔离到了启动阶段盘中循环只做轻量计算不会出现“一边盯盘一边等数据”的卡顿。3. 核心代码实现从历史缓存到实时监控3.1 环境准备与依赖安装先把依赖装好pandas和akshare是必须的如果你想把缓存保存到本地再加载顺便装个pyarrow。pip install pandas akshare如果你不需要保存缓存pyarrow可以不装。我这里的实现会把历史K线缓存放在内存里程序启动时现拉简单直接。3.2 获取全市场股票代码首先要拿到所有A股的代码列表。akshare有一个接口专门干这个事import akshare as ak import pandas as pd import time from datetime import datetime, timedelta from concurrent.futures import ThreadPoolExecutor, as_completed def get_stock_list(): df ak.stock_info_a_code_name() return df[code].tolist()stock_info_a_code_name()返回的是沪深京三市全部A股代码和名称包含科创板、创业板、北交所。如果你是做实际交易建议后续把ST、上市不足半年的次新股过滤掉这部分我会在参数优化章节展开。3.3 构建历史K线缓存拿到代码列表后就是对每只股票拉最近15个交易日的日K线。注意这里要取“截至昨日”的数据因为盘中拉到的当日K线是不完整的我需要在实时计算时用最新价去替代当日的收盘价这样算出来的MA5和MA10才是“实时状态下的均线值”。def fetch_hist(code, days20): end_date datetime.now().strftime(%Y%m%d) start_date (datetime.now() - timedelta(daysdays)).strftime(%Y%m%d) try: raw ak.stock_zh_a_hist( symbolcode, perioddaily, start_datestart_date, end_dateend_date, adjustqfq ) if raw is None or raw.empty: return code, None raw raw.rename(columns{日期: date, 收盘: close}) raw[date] pd.to_datetime(raw[date]) raw raw[[date, close]].sort_values(date).reset_index(dropTrue) # 关键剔除今日未收盘的K线只用已收盘历史数据 today_str pd.Timestamp(datetime.now().date()) raw raw[raw[date] today_str] return code, raw except Exception as e: return code, None这里用到了前复权qfq原因是A股存在除权除息不复权的价格会出现人为跳空影响均线的连续性。对于均线计算来说前复权是更合理的选择。构建缓存时我建议用线程池并发拉取全市场5000多只股票串行要跑很久并发可以把时间压缩到几分钟内。但注意并发数不要开太大我在实际操作中并发数调到20以上就容易触发对方数据源的请求限制导致大批请求返回异常或空数据片。稳妥起见线程数控制在8到12之间。def build_cache(stock_codes): cache {} with ThreadPoolExecutor(max_workers10) as executor: futures {executor.submit(fetch_hist, code): code for code in stock_codes} for i, future in enumerate(as_completed(futures)): code, df future.result() cache[code] df if (i 1) % 500 0: print(f已缓存 {i1}/{len(stock_codes)} 只股票) return cache3.4 实时计算MA5与MA10并判断上穿这是整个工具的核心算法。先说思路昨天的MA5和MA10用昨天之前的历史收盘价来计算今天的MA5和MA10把最新价当作“今日收盘价”代入计算。判断上穿的条件是昨天的MA5小于等于昨天的MA10而今天的MA5已经大于今天的MA10。def calc_ma_cross(hist, latest_price): if hist is None or len(hist) 11: return False closes hist[close].values # 昨日状态 ma5_prev closes[-5:].mean() ma10_prev closes[-10:].mean() # 今日实时状态用最新价替代今日收盘 ma5_now (latest_price closes[-4:].sum()) / 5 ma10_now (latest_price closes[-9:].sum()) / 10 # 上穿昨日MA5在上方或持平今日MA5已经上穿 if ma5_prev ma10_prev and ma5_now ma10_now: return True return False这里有一个细节很多人容易写错计算今天的MA5时closes[-4:]取的是最近4个历史收盘再加上最新价总共5个数值。同理MA10是最近9个历史收盘加最新价。如果直接用历史数据里包含当天的不完整K线去算不仅多算一天还会把实时价格的变化弄混算出来的信号会偏移。3.5 主循环与结果输出实时监控主循环的作用是定时拉全市场快照对每只股票执行上穿判断把命中的股票输出到控制台同时追加保存到CSV文件里方便收盘后复盘。def is_trading_time(now): if now.weekday() 5: return False am (now.hour 9 and now.minute 30) or (now.hour 10) or (now.hour 11 and now.minute 30) pm (now.hour 13) or (now.hour 14) or (now.hour 15 and now.minute 0) return am or pm def realtime_scan(cache, interval60): alerted set() output_file fma5_cross_ma10_{datetime.now().strftime(%Y%m%d)}.csv if not os.path.exists(output_file): pd.DataFrame(columns[时间, 代码, 名称, 最新价, 涨跌幅, 换手率]).to_csv( output_file, indexFalse, encodingutf-8-sig ) while True: now datetime.now() if not is_trading_time(now): print(f[{now.strftime(%H:%M:%S)}] 非交易时段等待下一分钟) time.sleep(60) continue try: snapshot ak.stock_zh_a_spot_em() hit_list [] for _, row in snapshot.iterrows(): code str(row[代码]) if code not in cache: continue latest row[最新价] if latest is None or pd.isna(latest) or latest 0: continue if calc_ma_cross(cache[code], latest): hit_list.append(( now.strftime(%H:%M:%S), code, row[名称], latest, row[涨跌幅], row[换手率] )) if hit_list: df_hit pd.DataFrame(hit_list, columns[时间, 代码, 名称, 最新价, 涨跌幅, 换手率]) df_hit.to_csv(output_file, modea, headerFalse, indexFalse, encodingutf-8-sig) for item in hit_list: if item[1] not in alerted: alerted.add(item[1]) print(f 实时金叉: {item[0]} {item[1]} {item[2]} 最新价{item[3]} 涨幅{item[4]}% 换手{item[5]}%) else: print(f[{now.strftime(%H:%M:%S)}] 本轮筛选完成未发现上穿个股) except Exception as e: print(f出现异常: {e}) time.sleep(interval)主流程串起来很简单if __name__ __main__: codes get_stock_list() print(f共获取 {len(codes)} 只A股开始构建历史缓存...) cache build_cache(codes) realtime_scan(cache, interval60)我在实际跑这个脚本时一般是在开盘前9点启动历史缓存构建大概需要3到5分钟刚好赶上9点30分集合竞价结束后的第一轮扫描。设置成60秒扫描一次即可——均线是日线级别指标5分钟内的变化幅度有限60秒已经足够及时同时又不会给数据源造成过大压力。4. 参数优化与实用扩展4.1 用成交量过滤假信号MA5上穿MA10这个信号单独用最让人头疼的就是震荡行情里的假突破。价格在均线附近反复摩擦时金叉死叉会连环出现跟着信号进去基本就是来回打脸。一个非常有效且简单的过滤方法是叠加量能条件。具体到代码里可以在realtime_scan的判断中增加一个条件要求当日量比大于1或者换手率在某个区间内。量比代表当日成交量与过去5日平均水平的比值小于1说明今天交投清谈即使均线金叉了也可能是无量空涨可信度不高。if row[量比] and row[量比] 1.0: # 再判断是否上穿不过要注意盘中早盘阶段量比天然偏高9点35分的量比和14点30分的量比不具备可比性。这是实时监控工具特有的问题静态回测不会碰到。我的处理方式是只在盘中判断量比与时段匹配的锚定值比如上午用大于1.2下午用大于0.8更精细的做法是把量比和自身历史均值做偏离度判断。4.2 叠加均线排列条件均线交叉本身只能说明短期动能盖过了中期并不能确认趋势方向。一个经典的做法是叠加中长期均线过滤要求MA20整体走平或向上甚至要求MA5、MA10、MA20呈多头排列再考虑金叉信号。这样筛出来的股票首先处于中期趋势向上的环境里金叉的边际意义更大。在代码里给calc_ma_cross增加一个trend_filter参数需要历史缓存里包含至少20个交易日的收盘价。计算方式也很简单用closes[-20:].mean()得到MA20对比MA10的大小关系def calc_ma_cross_with_filter(hist, latest_price): if hist is None or len(hist) 21: return False closes hist[close].values ma5_prev closes[-5:].mean() ma10_prev closes[-10:].mean() ma5_now (latest_price closes[-4:].sum()) / 5 ma10_now (latest_price closes[-9:].sum()) / 10 ma20_prev closes[-20:].mean() return ma5_prev ma10_prev and ma5_now ma10_now and ma10_now ma20_prev这个过滤条件加上之后候选股票数量会明显减少但信号的质量会好很多。它本质上是在说股票处于中期上涨趋势中短期均线回踩后重新抬头而不是在下跌趋势里抢反弹。4.3 引入信号状态管理同一个交易日里一只股票可能多次上穿下穿。如果每次都弹提醒不仅吵还容易让人在信号反复时心态失衡。我的做法是用一个alerted集合记录当天已经发送过提醒的股票代码同一天内只提醒一次。如果你希望重置提醒状态可以记录每条信号的发布时间超过一定时间窗口后允许再次提醒。class SignalState: def __init__(self): self.alerted {} def should_alert(self, code): today datetime.now().date() if self.alerted.get(code) today: return False self.alerted[code] today return True这个状态管理还有另一个好处它可以帮你收盘后复盘。把每次触发信号时对应的价格、时间保存下来过几天回头看这些信号的后续表现就知道这个工具在当前行情下的准确率到底如何。我在实测中会把交易结果导到Excel里做统计一个月下来就能总结出哪些附加条件真正有用。4.4 消息推送控制台输出适合自己盯盘时用但如果你开着代码又不盯着终端信号出来了也不知道。更实际的做法是把命中结果推到手机。实现方式很多最省事的是用Server酱或者企业微信机器人webhook代码就是一次HTTP POST。import requests def push_wechat(content): webhook_url 你的企业微信机器人地址 requests.post(webhook_url, json{msgtype: text, text: {content: content}}, timeout5)我习惯只推“信号出现”这一条关键信息比如“10:32 000001平安银行 MA5上穿MA10 价11.23 涨2.8%”。不要每次都推全量扫描结果信息噪音多了人反而会麻木真正的信号也就失去了冲击力。5. 常见问题与排查心得5.1 缓存里大量股票为None我自己第一次跑全市场的时候碰上过几百只股票的历史数据拉取失败缓存里全是None。原因大概有三种一是上市时间太短历史K线不足15个交易日二是网络请求被数据源临时拒绝三是某些代码在接口里返回了空DataFrame。处理方式比较务实读完缓存后统计一下有效股票数量低于一定比例就重新构建对于个股拉取失败的情况直接跳过不影响整体运行。5.2 盘中快照接口偶发超时或返回空stock_zh_a_spot_em()是全市场快照接口数据量大在早盘开盘那一分钟会非常拥挤偶发超时是常态。我的应对是两个措施并用一是把请求包在try/except里失败了就休眠重试二是每次循环固定间隔不要因为某次异常就立刻连发请求反而更容易触发限流。except Exception as e: print(f快照拉取失败: {e}) time.sleep(5) continue5.3 盘中均线与交易软件显示不一致很多用户第一次跑出来的结果和同花顺、通达信里显示的MA5不完全一致会觉得自己代码错了。这个现象太正常了原因基本都出在前复权处理上。不同的数据源、不同的复权基准计算出的均线本身就有微小差异另外行情软件在盘中计算MA时用的规则也可能不一样。解决方法是不要追求和某个软件显示的数字分毫不差你的目标是判断“是否发生上穿”这个状态命题几个百分点的差异不影响这个判断。5.4 常见问题速查现象可能原因处理方法缓存构建极慢串行拉取、网络波动用线程池并发线程数控制在8到12盘中一直提示非交易时段节假日判断缺失在is_trading_time里增加交易日历判断信号一直没有输出过滤条件过严、缓存无足够历史先去掉附加过滤条件直接跑逐步加条件验证重复弹同一个信号未维护提醒状态增加alerted集合同一天只提醒一次CSV乱码中文编码问题用encodingutf-8-sig写入5.5 非交易日与长假的处理如果程序是从某个周一跑到周六的而你没有做节假日判断那is_trading_time会把工作日外的所有时间都排除这个没问题。但遇到法定节假日落在工作日比如国庆前的调休、清明、端午我的简单写法就会把非交易日当成交易日去拉快照虽然不会出错但会白跑一天。严谨的做法是引入一个交易日历akshare里有ak.tool_trade_date_hist_sina()可以直接拿到当年所有交易日把这个列表缓存下来再用它作为交易时段判断的基础。trade_dates set(ak.tool_trade_date_hist_sina()[trade_date].astype(str)) def is_trade_date(dt): return dt.strftime(%Y-%m-%d) in trade_dates6. 从真实使用中得来的几点体会这个实时筛选工具我用过很长时间也迭代过好几版。最早一版只是收盘后做一个静态筛选后来发现信号确认后第二天开盘的表现总是差口气才下决心做成盘中实时监控。真实的感受是实时金叉筛选给的最大价值不是“帮你抓到了哪只票”而是把全市场的状态变化浓缩成了一条流你盯盘不再是从几千只股票里漫无目的地翻每一个被提醒的标的都带有明确的逻辑前提省下的精力是实实在在的。不过有一个坑我必须直说别把这个工具的输出当成交信号。MA5上穿MA10在日线级别上只是一次动能切换的提示它解决的是“注意力分配”问题也就是今天你应该花时间去看哪些股票而不是告诉你该买哪只股票。我在实战中至少叠加了两层信心判断——第一层是中期均线方向必须向上第二层是成交量要有效放大单纯的金叉信号命中率其实不高。另外建议你跑一段时间后一定要做复盘记录。把每次信号出现时的代码、时间、价格、后续3日和5日走势都记录下来用数据验证你设置的过滤条件到底有没有效果。我自己的经验是任何指标过滤条件的有效性都会随着市场风格变化而变化震荡市里有效的规则在趋势市里可能反而帮倒忙反之亦然。定期基于历史信号重新审视你的参数比盲目优化来得更实际。最后分享一个操作细节程序启动后不要马上最小化窗口就不管了先观察前两轮扫描正确。确认缓存有效、快照能拉到、信号判断正常再让它连续跑。一次正常的运行应该看到固定的扫描日志偶尔有命中输出如果运行了半个多小时一直没动静大概率是哪一步配置出了问题不是市场本身没机会。