企业官网建设流程全解析

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。各有各的特点。

实时筛选这个需求的核心是“全市场实时行情”，所以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, days=20): end_date = datetime.now().strftime('%Y%m%d') start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d') try: raw = ak.stock_zh_a_hist( symbol=code, period='daily', start_date=start_date, end_date=end_date, adjust='qfq' ) 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(drop=True) # 关键：剔除今日未收盘的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_workers=10) 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'已缓存 {i+1}/{len(stock_codes)} 只股票') return cache

3.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, interval=60): alerted = set() output_file = f"ma5_cross_ma10_{datetime.now().strftime('%Y%m%d')}.csv" if not os.path.exists(output_file): pd.DataFrame(columns=['时间', '代码', '名称', '最新价', '涨跌幅', '换手率']).to_csv( output_file, index=False, encoding='utf-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, mode='a', header=False, index=False, encoding='utf-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, interval=60)

我在实际跑这个脚本时，一般是在开盘前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}}, timeout=5)

我习惯只推“信号出现”这一条关键信息，比如“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) continue

5.3 盘中均线与交易软件显示不一致

很多用户第一次跑出来的结果，和同花顺、通达信里显示的MA5不完全一致，会觉得自己代码错了。这个现象太正常了，原因基本都出在前复权处理上。不同的数据源、不同的复权基准，计算出的均线本身就有微小差异；另外，行情软件在盘中计算MA时用的规则也可能不一样。解决方法是：不要追求和某个软件显示的数字分毫不差，你的目标是判断“是否发生上穿”这个状态命题，几个百分点的差异不影响这个判断。

5.4 常见问题速查

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_dates

6. 从真实使用中得来的几点体会

这个实时筛选工具我用过很长时间，也迭代过好几版。最早一版只是收盘后做一个静态筛选，后来发现信号确认后第二天开盘的表现总是差口气，才下决心做成盘中实时监控。真实的感受是：实时金叉筛选给的最大价值不是“帮你抓到了哪只票”，而是把全市场的状态变化浓缩成了一条流，你盯盘不再是从几千只股票里漫无目的地翻，每一个被提醒的标的都带有明确的逻辑前提，省下的精力是实实在在的。

不过有一个坑我必须直说：别把这个工具的输出当成交信号。MA5上穿MA10在日线级别上只是一次动能切换的提示，它解决的是“注意力分配”问题，也就是今天你应该花时间去看哪些股票，而不是告诉你该买哪只股票。我在实战中至少叠加了两层信心判断——第一层是中期均线方向必须向上，第二层是成交量要有效放大，单纯的金叉信号命中率其实不高。

另外建议你跑一段时间后一定要做复盘记录。把每次信号出现时的代码、时间、价格、后续3日和5日走势都记录下来，用数据验证你设置的过滤条件到底有没有效果。我自己的经验是，任何指标过滤条件的有效性都会随着市场风格变化而变化，震荡市里有效的规则在趋势市里可能反而帮倒忙，反之亦然。定期基于历史信号重新审视你的参数，比盲目优化来得更实际。

最后分享一个操作细节：程序启动后不要马上最小化窗口就不管了，先观察前两轮扫描正确。确认缓存有效、快照能拉到、信号判断正常，再让它连续跑。一次正常的运行应该看到固定的扫描日志，偶尔有命中输出；如果运行了半个多小时一直没动静，大概率是哪一步配置出了问题，不是市场本身没机会。