用Python构建A股板块打分模型：从行情数据到回测实战-平芜编程栈

8月中旬做 A 股短线的人，自选列表里通常都离不开四个方向：白酒、煤炭、电力、消费。它们并不是同一个风格，却经常被放在一起观察。原因在于这几条线的资金属性完全不同——白酒和消费代表内需基本盘，煤炭代表周期与红利，电力代表防御与盈利稳定性。把它们放在同一张表格里看，本质上是在观察市场的风险偏好处于什么位置：资金是愿意进攻，还是只想防守。

但如果你每天只是看涨跌幅、刷消息面解读，第二天很容易被打脸。因为“涨了才知道为什么涨”是信息落后者的常态。要解决这个问题，不能靠更努力地盯盘，而要建立一套自己的板块分析方法。具体来说，就是拿到公开行情数据，对每个板块做趋势、动量、量能、技术形态四个维度的打分，再用回测验证这套逻辑是否有效。这件事可以用 Python 完整实现。

本文就以 8 月 13 日收盘后的数据为基准，把酒、煤炭、电力、消费四个板块的处理流程完整拆开。内容包括行情数据的获取与清洗、技术指标的计算、板块打分模型的构建、基于打分结果的“明天策略”应该怎么定，以及回测验证和风险控制。整篇文章的重点不是告诉你明天一定买什么，而是给你一套可以长期复用、并且能根据自己偏好调整的量化分析流程。对于已经熟悉 Pandas 的开发者，可以直接跳到第 4 节看打分模型；对于刚开始接触量化分析的朋友，建议按顺序读完，每一步都能跟上。

1. 为什么把“酒、煤炭、电力、消费”放在一起看

这四类资产放在一起，不是因为它们会同步涨跌，而是因为它们分别代表了 A 股市场中几类典型的风格。白酒和消费属于大消费风格，与居民收入预期、消费数据和行业库存周期关系密切；煤炭属于周期资源风格，定价受到煤价、库存和长协价等产业因素影响；电力属于公用事业风格，资金更看重盈利稳定性、现金流和高分红属性。

在 8 月中旬这个时间节点，关注这几个板块还有另一层原因：中报是资金重新定价的窗口。业绩超预期的板块容易被持续买入，业绩不达预期的题材则可能被快速抛弃。四类板块的风险收益特征差异明显，正好可以用来观察资金在“求稳”和“求进攻”之间的切换方向。如果市场整体风险偏好下降，资金倾向于流向电力这类稳定型资产；如果风险偏好回升，消费和白酒这类超跌方向往往会出现修复机会。

从技术分析角度看，板块指数或者代表性 ETF 的日线走势，已经包含了对以上产业逻辑的大部分预期。我们不需要去预测“明天的新闻”，只需要把走势背后的信息量化出来。常用的工具就是均线、MACD、RSI、成交量这些技术指标，它们共同回答一个问题：当前这个板块处于多头状态还是空头状态，强势特征是否得到量能确认。

这里先给出四类板块的定位观察表，后文的数据和策略都以这张表为基础。

板块	代表ETF	定价核心变量	技术面观察点
白酒	512690	动销、批价、库存周期	是否重新站回 20 日均线
煤炭	515220	煤价、库存、分红预期	高位是否出现放量滞涨
电力	159611	来水、供需、盈利稳定性	是否形成持续上升趋势
消费	159928	社零数据、CPI、估值修复	底部是否放量企稳

选择 ETF 而不是个股作为观察对象，是因为个股容易受到单一消息冲击，出现非基本面驱动的异常波动；而 ETF 代表了一篮子股票的总体走势，数据连续性更好，也更适合做板块层面的横向比较。后面所有代码都围绕这些 ETF 展开。

2. 数据准备：从零开始获取四类板块行情数据

做量化分析的第一步，是拿到干净、完整、时间对齐的行情数据。A 股行情数据获取，社区里最常用的工具是 AkShare，它封装了东方财富、新浪财经等多个公开数据源，接口比较统一，对个人研究和学习很友好。本文使用 AkShare 获取场内 ETF 的日线数据，重点演示完整流程。

2.1 安装依赖库

建议使用独立环境安装，避免和现有项目依赖冲突。这里需要安装的库包括 AkShare、Pandas、NumPy，以及用于可视化的 Matplotlib。如果你只想跑通本文的分析流程，前三个就够用了。

pip install akshare pandas numpy matplotlib

AkShare 的接口更新速度比较快，不同版本的参数可能略有差异。安装完成后，建议先打印版本号，确认环境是否正常。

import akshare as ak print("akshare version:", ak.__version__)

2.2 拉取 ETF 日线行情

AkShare 中获取场内 ETF 历史行情的接口是fund_etf_hist_em。需要传入 ETF 代码、周期、起止日期和复权方式。这里选择前复权，目的是消除分红除权对价格连续性的影响，让技术指标的计算结果更准确。

# data_loader.py import os import akshare as ak import pandas as pd # 与四大板块高度相关的场内 ETF targets = { "白酒": "512690", "煤炭": "515220", "电力": "159611", "消费": "159928", } def load_etf_history(symbol: str, start_date: str, end_date: str) -> pd.DataFrame: """ 通过 AkShare 获取场内 ETF 日线行情 """ df = ak.fund_etf_hist_em( symbol=symbol, period="daily", start_date=start_date, end_date=end_date, adjust="qfq", # 前复权 ) # 统一列名为英文，方便后续计算 col_mapping = { "日期": "date", "开盘": "open", "收盘": "close", "最高": "high", "最低": "low", "成交量": "volume", "成交额": "amount", "振幅": "amplitude", "涨跌幅": "pct_chg", "涨跌额": "change", "换手率": "turnover", } df.rename(columns=col_mapping, inplace=True) df["date"] = pd.to_datetime(df["date"]) df = df[["date", "open", "close", "high", "low", "volume", "amount", "pct_chg", "turnover"]].copy() df = df.sort_values("date").reset_index(drop=True) return df def load_all(data_dir: str = "data") -> dict: """批量拉取四个板块数据，并保存为 CSV 备查""" os.makedirs(data_dir, exist_ok=True) result = {} for name, symbol in targets.items(): df = load_etf_history(symbol, "20240101", "20240813") df.to_csv(f"{data_dir}/{name}.csv", index=False, encoding="utf-8-sig") result[name] = df print(f"{name}: {len(df)} 个交易日, 最新收盘 {df['close'].iloc[-1]:.3f}") return result

这段代码的核心逻辑并不复杂：先定义要观察的板块和对应 ETF，然后封装一个加载函数，把 AkShare 返回的中文列名改成英文列名，再按日期排序。之所以要保存 CSV，是为了避免反复请求网络接口，也方便后续离线调试。

从实践角度看，start_date建议至少提前半年以上，因为计算 60 日均线需要前 60 个交易日的数据，如果起始时间太晚，会有一大段指标是空值。

如果当前 AkShare 版本不支持fund_etf_hist_em，可以退而求其次，使用个股行情接口stock_zh_a_hist，传入同样的 ETF 代码。ETF 在交易所的代码与股票代码规则一致，这个方法也能获取到日线数据。

# 备选方案：部分 AkShare 版本可用该接口 df = ak.stock_zh_a_hist( symbol="512690", period="daily", start_date="20240101", end_date="20240813", adjust="qfq", )

需要注意的是，AkShare 的数据源来自公开市场，接口偶尔会出现限流或字段调整。遇到问题优先检查版本，其次查看返回数据的列名结构，不要盲目修改代码逻辑。

3. 计算核心技术指标：趋势、动量与量能

拿到日线数据之后，下一步是计算技术指标。这里的核心思路是：把“看着强不强”变成“由几个可量化的指标共同描述”。我们不依赖单一指标，而是让趋势、动量、量能、情绪四个维度相互验证。

3.1 常用指标的含义与计算

本文用到的指标如下：

MA20：20 日均线，代表中期趋势方向。收盘价在 MA20 上方，通常认为中期偏强。
MACD：由 DIF、DEA、柱状值组成。DIF 上穿 DEA 是金叉，下穿是死叉，用于衡量动量的变化。
RSI14：14 日相对强弱指数，衡量买卖力量的强弱。数值在 70 以上代表超买，30 以下代表超卖。
成交量均线：用 5 日均量与 20 日均量的关系判断资金活跃度，放量代表新增资金参与。

这些指标本身都不复杂，但组合在一起，就能从不同角度刻画板块状态。下面是完整的指标计算函数。

# indicators.py import numpy as np import pandas as pd def add_indicators(df: pd.DataFrame) -> pd.DataFrame: """ 在原始行情数据基础上增加技术指标列 """ df = df.sort_values("date").reset_index(drop=True) # 均线 df["ma5"] = df["close"].rolling(5).mean() df["ma10"] = df["close"].rolling(10).mean() df["ma20"] = df["close"].rolling(20).mean() df["ma60"] = df["close"].rolling(60).mean() # RSI(14) delta = df["close"].diff() gain = delta.clip(lower=0) loss = -delta.clip(upper=0) avg_gain = gain.rolling(14).mean() avg_loss = loss.rolling(14).mean() # 处理 avg_loss 为 0 的情况，避免除零错误 rs = avg_gain / avg_loss.replace(0, np.nan) df["rsi14"] = 100 - (100 / (1 + rs)) df["rsi14"] = df["rsi14"].fillna(100).clip(0, 100) # MACD ema12 = df["close"].ewm(span=12, adjust=False).mean() ema26 = df["close"].ewm(span=26, adjust=False).mean() df["dif"] = ema12 - ema26 df["dea"] = df["dif"].ewm(span=9, adjust=False).mean() df["macd"] = (df["dif"] - df["dea"]) * 2 # 布林带 df["boll_mid"] = df["close"].rolling(20).mean() df["boll_std"] = df["close"].rolling(20).std() df["boll_upper"] = df["boll_mid"] + 2 * df["boll_std"] df["boll_lower"] = df["boll_mid"] - 2 * df["boll_std"] return df

这里比较容易被忽略的是 RSI 的计算。在持续上涨行情中，avg_loss会等于 0，直接用除法会得到无穷大。代码里的处理方式是把 0 替换为NaN，再在最后用 100 填充，这表示“持续强势，没有任何回调”，对趋势策略来说是有效的状态描述。

四个板块都套用这个函数后，我们可以快速查看某一天的数据长什么样。

data = load_all() for name, df in data.items(): data[name] = add_indicators(df) # 查看白酒板块最新 5 行 print(data["白酒"].tail(5).to_string())

输出会包含日期、收盘价、均线、RSI、MACD、布林带等多个字段。到这一步，数据层面的准备工作已经完成，接下来就可以构建打分模型了。

4. 构建板块打分模型：让“强弱”可以被比较

很多人在复盘时会遇到一个问题：看着四张 K 线图，觉得这个也强、那个也强，却说不清到底哪个更强。涨幅是最直观的指标，但它只反映过去，且容易被一天的大阳线扭曲。更好的做法是做一个多维度打分，让每个板块的强弱状态变成一个可比较的分数。

4.1 四个维度的打分规则

本文采用一个简单但逻辑完整的打分模型，每个维度 1 分，总分 0 到 4 分。

趋势维度：收盘价位于 MA20 上方，记 1 分。这是最基础的趋势过滤条件，避免在下跌趋势中盲目接飞刀。
动量维度：MACD 的 DIF 位于 DEA 上方，记 1 分。表示短期动能强于长期动能，处于多头排列状态。
量能维度：近 5 日均量大于近 20 日均量，记 1 分。表示近期资金活跃度高于前期，放量才有持续性。
情绪维度：RSI14 处于 50 到 70 之间，记 1 分。50 到 70 代表强势但未进入超买区，短线仍有空间，同时避免了追在 80 以上高位的风险。

为什么这样设计？因为每个维度都对应一个具体的市场状态。趋势维度过滤下跌趋势，动量维度确认动能方向，量能维度验证资金参与度，情绪维度防止追高。四个条件同时满足，说明板块处于相对健康的多头共振状态；只有 1 分甚至 0 分，则说明当前没有形成合力。

4.2 打分函数实现

下面是完整的打分代码。打分基于最后一根 K 线，也就是 8 月 13 日收盘后的状态，因此所有数据都应该包含到该日期。

# score_model.py def calculate_score(df: pd.DataFrame) -> dict: """根据最后一天的指标状态计算板块得分""" last = df.iloc[-1] # 1. 趋势：价格在 MA20 上方 trend = 1 if last["close"] > last["ma20"] else 0 # 2. 动量：DIF 在 DEA 上方 momentum = 1 if last["dif"] > last["dea"] else 0 # 3. 量能：5日均量 > 20日均量 vol_ma5 = df["volume"].tail(5).mean() vol_ma20 = df["volume"].tail(20).mean() volume = 1 if vol_ma5 > vol_ma20 else 0 # 4. 情绪：RSI 在 50~70 之间 emotion = 1 if 50 <= last["rsi14"] <= 70 else 0 total = trend + momentum + volume + emotion return { "trend": trend, "momentum": momentum, "volume": volume, "emotion": emotion, "total": total, } scores = {} for name, df in data.items(): scores[name] = calculate_score(df) score_df = pd.DataFrame(scores).T print(score_df.sort_values("total", ascending=False))

运行这段代码后，会输出每个板块在四个维度上的得分明细。为了方便理解，这里给出一个示例输出，并非 8 月 13 日的真实数据，但可以说明表格应该怎么读：

trend momentum volume emotion total 电力 1 1 1 1 4 白酒 1 1 0 1 3 煤炭 0 1 1 0 2 消费 0 0 1 0 1

总分越高，说明板块当前的技术状态越强。4 分代表趋势、动量、量能、情绪四个维度都偏多，是强势共振状态；2 分到 3 分说明部分维度走强，需要结合其他因素判断；0 分到 1 分说明整体处于弱势，反弹缺乏持续性。

这里要特别强调：打分模型给出的是“状态”，不是“涨跌预测”。它告诉你当前市场在哪个板块上形成了合力，但不会告诉你明天一定继续涨。这正是技术分析的边界——我们只能做概率判断，不能做确定性预测。

4.3 为什么不用机器学习打分

很多读者可能会问：既然已经有 Python 了，为什么不直接上个机器学习模型，让模型自动选板块？

这个问题值得展开。机器学习模型确实可以处理高维特征，但它需要大量样本。A 股一年只有约 240 个交易日，四个板块也不过 960 条日线数据，样本量不足以支撑一个稳定的选板块模型。更重要的是，机器学习模型的可解释性差，在策略出现亏损时，很难定位到底是特征问题、样本问题还是过拟合问题。

因此，对于个人研究和中小团队，先用规则型打分模型把流程跑通，比直接上复杂模型更实际。规则模型逻辑透明，方便调整权重，也能在遇到问题时快速排查。等积累足够多的数据和策略经验后，再考虑用机器学习做增强，才是更稳妥的路径。

5. 8月13日盘后：把“明天策略”变成一套可执行流程

有了打分结果，接下来的问题就是：明天怎么办？这里需要建立一套流程，而不是简单地说“买得分最高的”。

5.1 先确定观察优先级

打分的第一作用是排序。总分最高的板块，应该被放在优先级最高的位置。比如，如果电力是 4 分，其他板块低于 3 分，那说明市场当前在电力方向上形成了技术面的多头共振，复盘时就应该重点看电力板块的资金流向、成交量变化和龙头股表现。

反过来，如果消费只有 1 分，那说明它虽然处于低位，但还没有出现企稳信号。此时不应该因为“跌了这么多”就去抄底，而应等待它站上 20 日均线、MACD 金叉等信号出现之后再考虑。这就是“不预测、只应对”的交易思路。

5.2 制定“如果……那么……”的规则

“明天策略”不应该是一句“明天看电力”，而应该是一组提前写好的条件规则。比如：

如果高分板块次日继续放量上行，且 RSI 没有超过 75，说明趋势延续，可以继续跟踪。
如果高分板块次日出现高开低走或放量滞涨，说明资金分歧加大，应该降低预期。
如果低分板块突然放量长阳，站上 MA10，则需要重新评估，因为它可能正在从弱势转向强势。

这种写法的好处是，你不用在盘中临时决策。所有的可能性都在盘前列好了，当天只需要对照执行。

5.3 形成自己的复盘记录

更进一步，可以把每天的打分结果、次日走势、策略执行情况记录下来。这样持续一段时间后，你就能统计出“4 分板块次日上涨的概率”到底是多少，从而验证这套模型是否适合当前市场。记录可以很简单，用 CSV 就行。

# record_daily.py import csv from datetime import datetime def save_daily_record(date: str, scores: dict): """把每日打分结果追加到记录文件""" row = {"date": date} for name, sc in scores.items(): row[f"{name}_total"] = sc["total"] row[f"{name}_trend"] = sc["trend"] row[f"{name}_momentum"] = sc["momentum"] row[f"{name}_volume"] = sc["volume"] row[f"{name}_emotion"] = sc["emotion"] file_name = "daily_scores.csv" with open(file_name, "a", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=list(row.keys())) if f.tell() == 0: writer.writeheader() writer.writerow(row)

这个记录文件的积累价值很大。一个月后，你手里的就不再是模糊的盘感，而是一张可以统计、可以分析、可以优化的表格。

6. 回测验证：给策略装一个“后悔药”

趋势型策略最怕震荡行情。某些策略在 2024 年上半年表现很好，但切换到震荡市后可能频繁亏损。因此，任何策略在实盘之前，都必须先做回测。回测的核心目的不是得到“这个策略能赚钱”的结论，而是回答三个问题：胜率大概多少？平均收益如何？信号会不会太频繁？

6.1 一个最小可用的持有期回测

下面给出一个简单的回测函数：当收盘价在 MA20 上方且 MACD 金叉时，记录为一个买入信号，持有 5 个交易日后卖出，统计每次交易的收益率。

# backtest.py import numpy as np import pandas as pd def add_indicators(df): # 为了保持示例独立，这里简写，实际可直接复用第 3 节函数 ... def hold_period_backtest(df: pd.DataFrame, hold_days: int = 5) -> dict: """基于 MA20 + MACD 金叉信号的持有期回测""" df = df.sort_values("date").reset_index(drop=True) df["signal"] = (df["close"] > df["ma20"]) & (df["dif"] > df["dea"]) rets = [] i = 0 while i < len(df) - hold_days: if df.loc[i, "signal"]: ret = df.loc[i + hold_days, "close"] / df.loc[i, "close"] - 1 rets.append(ret) i += hold_days else: i += 1 if len(rets) == 0: return {"count": 0, "avg_ret": 0.0, "win_rate": 0.0, "total_ret": 0.0} avg_ret = float(np.mean(rets)) win_rate = float(np.mean([r > 0 for r in rets])) total_ret = float(np.prod([1 + r for r in rets]) - 1) return { "count": len(rets), "avg_ret": avg_ret, "win_rate": win_rate, "total_ret": total_ret, } # 运行回测 for name, df in data.items(): result = hold_period_backtest(df, hold_days=5) print(f"{name}: {result}")

6.2 回测结果怎么看

输出中的count代表信号次数。如果信号次数太少，说明策略参与机会不多；如果太多，说明信号在震荡市中反复触发，容易磨损收益。win_rate代表胜率，avg_ret代表平均单次收益率，total_ret代表累计收益率。

需要特别提醒的是，这个回测函数存在明显局限。首先，它没有考虑手续费和滑点，实际交易中成本会显著拉低收益。其次，固定持有 5 天，没有动态止盈止损，也没有处理信号在持有期内消失的情况。最后，样本区间只有半年多，行情特征相对单一，回测结果不能直接外推到未来。

因此，回测的价值在于“排除明显无效的策略”，而不是“证明策略一定有效”。如果某个策略在回测中都无法盈利，实盘大概率更差；但如果回测表现不错，还需要经过样本外验证和小仓位试运行才能逐步投入使用。

对于更完整的回测，可以考虑引入 backtrader、vectorbt 这类专业框架。它们支持手续费、滑点、多标的组合，但学习成本也会更高。建议先把本文的最小流程跑通，理解信号与收益之间的关系，再逐步增加复杂度。

7. 常见问题与排查方法

在实际运行代码的过程中，读者可能会遇到各种问题。这里把最常见的问题和排查思路整理成表格。

问题现象	可能原因	排查方式	解决方案
调用`fund_etf_hist_em`报错	AkShare 版本过旧或接口更名	打印`ak.__version__`，查看官方接口文档	`pip install akshare --upgrade`升级到最新版
返回 DataFrame 的列是中文，代码报错	接口返回字段与示例不一致	打印`df.columns`查看列名	使用`rename`做列名映射，不要直接用位置索引
指标出现大量 NaN	数据起始时间太短	查看数据行数，确认是否少于 60 行	延长回看区间，至少从半年前开始
RSI 恒为 100	上涨阶段没有亏损，`avg_loss`为 0	检查`loss`列是否全为 0	用`replace(0, np.nan)`处理除零，再填充
回测信号过多，频繁交易	震荡市信号反复金叉死叉	统计`signal`次数	增加过滤条件，例如要求 MA20 斜率向上
回测结果非常好	可能存在未来函数或过拟合	检查信号是否用到未来数据	用样本外数据验证，同时加入手续费和滑点
四个板块数据日期不一致	ETF 上市时间不同，停牌或数据缺失	检查各 DataFrame 的日期范围	用`reindex`对齐日期，缺失值做前向填充