在实际的行业板块研究中经常需要同时盯住酒、煤炭、电力、消费这几类代表性行业分析它们的行情走势、成交活跃度和趋势状态并在收盘后生成一份“次日策略清单”。这里的次日策略不是预测涨跌也不构成投资建议而是把人工复盘过程工程化自动拉取行情、计算指标、按规则筛选出值得重点跟踪的标的池为后续研究提供数据依据。这篇文章会从零搭建一个可运行的 Python 小工具用于获取行业板块成分股、计算常用技术指标、输出板块观察清单。你可以直接复用到自己的行业研究流程中也可以在此基础上加入更多因子和回测逻辑。文章会包含依赖安装、代码实现、运行验证、常见问题排查和工程化建议适合有一定 Python 基础、想了解量化研究数据流程的开发者阅读。1. 先把“板块分析次日策略”拆成可执行的工程技术任务很多人一看到“次日策略”就想到复杂的量化模型或人工智能预测实际上可以先从工程角度拆解问题。所谓次日策略本质上是一套流程在某个交易日的收盘数据出来后对选定行业内的股票重新计算状态筛选出符合条件的样本再按板块汇总输出名单。这个流程不是预测而是“基于历史数据的状态扫描”。1.1 酒、煤炭、电力、消费在数据分析上的共性这四个板块在业务上差异很大但在数据处理上高度相似都是由若干只股票组成的样本集合每只股票都有每日的开盘价、最高价、最低价、收盘价、成交量、成交额、换手率等基础数据。分析这些板块时通常会做以下几件事获取板块成分股列表确定样本范围。拉取成分股的日线行情保证数据时间范围一致。计算趋势类指标比如均线、动量。计算量能类指标比如量比、换手率变化。根据一组规则或打分模型输出关注名单。只要把这四步抽象成代码就可以复用同一套逻辑处理任何行业板块。本文示例中酒使用“酿酒行业”板块煤炭使用“煤炭行业”板块电力使用“电力行业”板块消费则可以使用“食品饮料”或“商业百货”等自定义股票池。配置文件可以随时调整。1.2 策略生成的整体流程整个工具的工作流程可以分成三层第一层是数据层负责获取板块成分股和日线行情最终得到一张干净的宽表。每一行是“某只股票在某个交易日的交易数据”包含代码、名称、日期、收盘价、成交量、成交额、换手率等字段。第二层是因子层在宽表基础上按股票和时间窗口计算指标。常见指标包括 5 日均线、10 日均线、20 日均线、20 日涨跌幅、量比等。这些指标是后续筛选规则的基础。第三层是信号层把因子值套入一组规则判断每只股票是否满足条件。满足条件的股票进入“次日观察池”最后按板块输出统计结果。这里的重点是数据对齐。不同股票可能有停牌、新上市、退市等情况计算指标时不能直接用原始数据必须按时间排序并用滚动窗口处理缺失值。1.3 技术选型为什么用 Python akshare pandas这个项目不需要重型的量化平台用 Python 配合开源库就够。核心原因有三个Python 的 pandas 非常适合做表格数据处理、时间序列分组和滚动计算。akshare 提供了免费的 A 股数据接口可以获取行业板块成分股和个股历史行情适合个人研究和学习。整个逻辑可以写成纯本地脚本容易调试也容易集成到定时任务中。需要说明的是akshare 的接口名称和返回字段可能随版本变化你自己落地时要以安装版本的文档为准。下面的代码用于展示完整思路不是对某个特定版本的绑定。2. 环境准备依赖安装和数据源确认在写代码之前先把运行环境准备好。本节会给出建议的 Python 版本、依赖清单以及一个最小接口验证方法。环境准备不充分后面所有结果都可能失真。2.1 Python 环境与依赖版本本项目使用 Python 3.9 或更高版本即可。主要依赖如下依赖库用途建议版本akshare获取板块成分股和股票历史行情1.12.0 及以上pandas数据清洗、分组计算和指标计算1.5.0 及以上numpy数值计算处理空值和向量化运算1.23.0 及以上PyYAML读取配置文件6.0 及以上schedule可选用于定时执行策略生成1.2.0 及以上建议使用虚拟环境安装避免污染系统环境python -m venv .venv source .venv/bin/activate pip install akshare pandas numpy pyyaml schedule如果你的网络环境安装较慢可以指定国内镜像源pip install akshare pandas numpy pyyaml schedule -i https://pypi.tuna.tsinghua.edu.cn/simple安装完成后确认版本python -c import akshare, pandas; print(akshare.__version__, pandas.__version__)2.2 akshare 数据接口的稳定使用方式akshare 是一个爬虫类数据接口库它从东方财富、新浪财经等公开网站抓取数据。不同版本接口可能变化因此代码中不要写死字段名而是先打印列名确认。常用的两个接口获取行业板块成分股ak.stock_board_industry_cons_em(symbol酿酒行业)获取个股历史行情ak.stock_zh_a_hist(symbol600519, perioddaily, start_date20240101, end_date20250201, adjustqfq)注意板块名称要和东方财富行业板块的名称一致比如“酿酒行业”“煤炭行业”“电力行业”。消费不是一个标准板块可以在配置文件中定义一个自定义股票池例如食品饮料、零售、家电等行业中的若干股票。2.3 数据落地本地缓存还是每次拉取策略扫描通常要求所有股票使用相同的截止日期。直接每次运行时全量拉取数据量大且容易触发接口限流。建议在项目目录下建立data/文件夹把每次拉到的行情按{board}_{symbol}.csv保存。当天运行之前先判断本地缓存日期如果缓存文件最后更新日期是当天交易日直接读取。如果不是则重新拉取并覆盖文件。这是一个简单有效的增量缓存策略。生产环境可以换成 SQLite 或 MySQL但小项目中 CSV 已经足够。2.4 环境验证先跑通一个最小接口调用不要直接写完整项目先验证数据源是否可用。创建一个临时脚本test_data.pyimport akshare as ak df ak.stock_board_industry_cons_em(symbol酿酒行业) print(df.head()) print(df.columns)如果运行后能看到成分股代码和名称说明接口可用。如果报错优先检查网络和 akshare 版本pip install --upgrade akshare这一步很重要。先验证最小链路再进入完整实现可以省去大量后期排错时间。注意接口返回的列名可能包含中文也可能变化建议在代码里用columns动态获取不要写死索引。3. 核心实现行业板块行情获取与指标计算这一节进入主体代码。我们会实现一个最小可运行的工具分为数据获取、指标计算、策略输出三个模块。为了保持示例简单每个板块最多取前 10 只流动性较好的股票分析最近 120 个交易日的数据。3.1 构建板块与股票池映射新建config.yaml文件用于管理板块名称和参数boards: 酿酒行业: max_stocks: 10 煤炭行业: max_stocks: 10 电力行业: max_stocks: 10 custom_消费: stocks: - 600887 - 603288 - 000333 - 601888 analysis: ma_windows: [5, 10, 20] momentum_days: 20 volume_ratio_days: 5 limit: 10 data: start_date: 20240101 end_date: 20250201 cache_dir: data这里把“消费”定义为自定义股票池。实际使用中你可以把custom_消费换成任意名称并在stocks列表中填入你关注的股票代码。3.2 数据获取模块新建data_fetcher.py负责获取板块成分股和个股日线行情并写入本地缓存。import os import akshare as ak import pandas as pd def get_board_stocks(board_name, max_stocks10): 获取行业板块成分股最多保留 max_stocks 只。 df ak.stock_board_industry_cons_em(symbolboard_name) df df[[代码, 名称]].copy() df.columns [symbol, name] # 尽量按代码排序这里不引入市值过滤仅作示例 df df.head(max_stocks) return df def get_custom_stocks(stocks): 把自定义股票池转成统一结构。 return pd.DataFrame({symbol: stocks, name: stocks}) def fetch_stock_daily(symbol, start_date, end_date, cache_dirdata): 获取个股日线数据优先读取本地缓存。 os.makedirs(cache_dir, exist_okTrue) cache_file os.path.join(cache_dir, f{symbol}.csv) try: if os.path.exists(cache_file): df_cache pd.read_csv(cache_file) last_date str(df_cache[日期].max()) # 简单判断如果缓存包含 end_date 当天的数据则直接使用 if last_date end_date: return df_cache except Exception: pass df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq, ) df.to_csv(cache_file, indexFalse) return df这段代码的关键点有三个使用前复权qfq避免分红送股导致价格不连续。先读缓存当缓存最新日期覆盖到结束日期时不再请求网络。统一返回包含中文列名的 DataFrame后续处理时再重命名。3.3 行情数据标准化与指标计算新建analyzer.py实现数据处理。首先把中文列名转成英文再按股票分组计算指标。import pandas as pd import numpy as np def normalize_daily(df): 将 akshare 返回的日线数据重命名并标准化。 df df.rename( columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 换手率: turnover, } ) df[date] pd.to_datetime(df[date]) df df.sort_values(date).reset_index(dropTrue) return df def add_indicators(df, ma_windows(5, 10, 20), momentum_days20, volume_ratio_days5): 为单只股票计算均线、动量和量比。 for window in ma_windows: df[fma{window}] df[close].rolling(window).mean() # 动量过去 momentum_days 个交易日的累计涨跌幅 df[momentum] df[close].pct_change(periodsmomentum_days) * 100 # 量比最近5日平均成交量 / 前5日平均成交量 df[vol_ma] df[volume].rolling(volume_ratio_days).mean() df[vol_prev] df[volume].shift(volume_ratio_days).rolling(volume_ratio_days).mean() df[volume_ratio] df[vol_ma] / df[vol_prev] return df关于量比的计算这里要说明通常行情软件中的量比是当日每分钟平均成交量与过去 5 日每分钟平均成交量的比值这里为了简化用“最近 5 日平均成交量 / 前 5 日平均成交量”代替。这是一种研究样例不是行情软件标准定义。如果你需要完全复现软件量比需要引入盘中分钟数据。指标计算完成后需要剔除不完整的数据。通常最后一行是当前最新数据但由于均线和动量需要历史长度前 20 行左右会是 NaN在筛选时直接丢弃。3.4 用趋势和量能规则生成筛选信号策略规则采用最经典的趋势跟踪思路不追求复杂。设计以下三个条件收盘价大于 5 日均线和 20 日均线代表短期和中期趋势向上。20 日动量大于 0 且小于 30代表上涨趋势存在但不过热。量比大于 1.2代表最近成交量相对于前一段有所放大。在analyzer.py中加入筛选函数def generate_signals(df, limit10): 对已经添加指标的完整宽表执行规则筛选。 latest ( df.dropna() .sort_values(date) .groupby(symbol) .tail(1) .reset_index(dropTrue) ) conditions ( (latest[close] latest[ma5]) (latest[close] latest[ma20]) (latest[momentum] 0) (latest[momentum] 30) (latest[volume_ratio] 1.2) ) result latest[conditions].copy() return result.head(limit)注意dropna()会把最早期的数据删掉最终每只股票只保留最新一行。这样得到的表就是“截至最新交易日”的信号结果。3.5 输出“明日策略”的结构与字段策略输出不需要太长但字段必须清晰。建议包含以下字段字段含义board所属行业板块symbol股票代码name股票名称date数据最新交易日close最新收盘价ma55日均线ma2020日均线momentum20日涨跌幅%volume_ratio量比reason满足条件的说明reason可以用一个统一字符串例如“站上5日和20日均线量能放大”也可以写成可读的规则摘要。4. 让策略工具可复用配置文件与命令行入口把参数写死在代码里不适合长期使用。本节会把板块配置、参数配置和主流程组合起来提供命令行入口让同一个工具可以处理不同板块、不同日期和不同规则阈值。4.1 项目目录结构建议项目目录如下sector_daily/ ├── config.yaml ├── requirements.txt ├── data_fetcher.py ├── analyzer.py ├── main.py └── data/ └── 缓存CSV文件data/目录存放个股日线缓存main.py是程序入口。4.2 主流程代码main.py负责读取配置、遍历板块、获取成分股、计算指标、输出结果。import argparse import yaml import pandas as pd from data_fetcher import get_board_stocks, get_custom_stocks, fetch_stock_daily from analyzer import normalize_daily, add_indicators, generate_signals def load_config(pathconfig.yaml): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def process_board(board_name, config): analysis_cfg config[analysis] data_cfg config[data] max_stocks config[boards].get(board_name, {}).get(max_stocks, 10) if board_name.startswith(custom_): stock_list config[boards][board_name][stocks] stock_df get_custom_stocks(stock_list) else: stock_df get_board_stocks(board_name, max_stocksmax_stocks) all_data [] for _, row in stock_df.iterrows(): symbol row[symbol] raw fetch_stock_daily( symbol, start_datedata_cfg[start_date], end_datedata_cfg[end_date], cache_dirdata_cfg[cache_dir], ) if raw is None or raw.empty: continue df normalize_daily(raw) df add_indicators( df, ma_windowstuple(analysis_cfg[ma_windows]), momentum_daysanalysis_cfg[momentum_days], volume_ratio_daysanalysis_cfg[volume_ratio_days], ) df[symbol] symbol df[name] row[name] df[board] board_name all_data.append(df) if not all_data: return pd.DataFrame() full_df pd.concat(all_data, ignore_indexTrue) signals generate_signals(full_df, limitanalysis_cfg[limit]) return signals def main(): parser argparse.ArgumentParser(description行业板块次日观察清单生成器) parser.add_argument(--config, defaultconfig.yaml) parser.add_argument(--board, defaultNone, help只处理指定板块默认处理全部) args parser.parse_args() config load_config(args.config) boards list(config[boards].keys()) if args.board: boards [args.board] all_signals [] for board_name in boards: signals process_board(board_name, config) if not signals.empty: all_signals.append(signals) print(f{board_name}: 生成 {len(signals)} 条信号) if all_signals: result pd.concat(all_signals, ignore_indexTrue) result.to_csv(daily_strategy.csv, indexFalse, encodingutf-8-sig) print(结果已写入 daily_strategy.csv) else: print(没有生成任何信号) if __name__ __main__: main()运行方式python main.py执行后会在项目根目录生成daily_strategy.csv。如果只关心某个板块可以指定--board 酿酒行业。4.3 参数调整和利弊配置文件中可以调整的常用参数包括参数默认值调小的影响调大的影响ma_windows[5,10,20]均线更贴近短期价格易频繁变化均线更平滑反应更慢momentum_days20动量周期短噪声较多动量周期长信号滞后volume_ratio1.2更容易触发信号更严格信号更少limit10输出更少输出更多max_stocks10分析范围小速度快分析范围大速度慢但更全面这些参数没有绝对好坏关键是要和你的研究周期一致。短线研究可缩小均线窗口和动量周期中长线研究则适合增大周期。5. 运行验证与结果解读代码写完后需要验证每个环节是否正常。不要只关注程序是否运行成功还要检查输出数据是否符合规则。5.1 执行一次分析的结果示例假设执行python main.py控制台输出可能如下酿酒行业: 生成 3 条信号 煤炭行业: 生成 2 条信号 电力行业: 生成 4 条信号 custom_消费: 生成 1 条信号 结果已写入 daily_strategy.csv打开daily_strategy.csv可以看到类似这样的内容boardsymbolnamedateclosema5ma20momentumvolume_ratio酿酒行业000858五粮液2025-01-24130.55129.10125.304.121.35煤炭行业600123兰花科创2025-01-2412.3012.0511.806.201.42这里日期和价格只是示例实际结果由你拉取的数据决定。需要特别注意date列应该是所有股票一致的交易日如果某只股票停牌导致日期不一致在最终输出中会混入不同日期需要在处理时过滤。5.2 如何判断信号是否有效信号生成后不要直接用它做任何交易决策。对于技术研究可以从以下角度评估检查信号数量是否稳定。如果某天信号大量出现可能是市场整体上涨或参数过宽。检查次日表现。可以手动记录次日收盘价观察信号池平均涨幅与全市场基准的差异。检查信号重合度。多个板块同时出现同一类信号往往说明市场风格变化而不是单只股票的逻辑。更严谨的方式是引入回测模块把历史数据切成训练集和测试集计算信号的胜率、平均收益、最大回撤。这部分是扩展方向本文不展开。5.3 学习环境与生产环境的差异当前代码适合学习环境也就是在自己电脑上手动运行脚本。如果要在测试或生产环境定时执行需要额外处理关注点学习环境生产环境数据源手动运行失败可重试需要重试和异常告警数据存储CSV 缓存数据库或对象存储策略参数手动改 YAML配置中心或环境变量日志print 输出结构化日志、监控权限本机本地文件服务账号、最小权限回滚重新运行脚本版本化模型与参数如果只是个人研究print 和 CSV 已经够用如果部署到服务器每日自动运行至少要加上日志、异常捕获和运行状态监控。注意不要在一个生产环境脚本里省略异常处理。网络波动、接口限流、股票停牌、新股上市等都是常见异常。6. 常见问题和排查路径实际运行中最容易遇到的问题集中在数据源、日期、字段和计算边界上。下面按排查顺序列出。6.1 akshare 接口失败或返回空数据现象调用stock_board_industry_cons_em或stock_zh_a_hist时抛异常或者返回空的 DataFrame。可能原因akshare 版本过旧接口路径或参数已变化。网络无法访问目标数据源。板块名称不存在或已改名。检查方式pip show akshare python -c import akshare; print(akshare.__version__)再单独测试接口。如果接口名称变了可以在 akshare 官方文档或源码中搜索新的调用方式。解决方案升级 akshare确认板块名称或者把接口调用包在 try except 中并记录错误。6.2 日期对齐和未来数据问题现象输出策略表中不同股票的date不一致或者信号里出现了还没有发生的“未来数据”。常见原因部分股票停牌最新交易日与其它股票不同end_date设置为含当天但当天盘中数据不完整使用了向前填充错误字段。检查方式输出df.groupby(symbol)[date].max()查看每只股票的最新日期。解决方案在generate_signals前统一过滤出全市场最大交易日latest_date full_df[date].max() full_df full_df[full_df[date] latest_date].copy()这样能避免停牌股票把旧日期带入结果。需要注意盘中运行脚本时当天数据可能还在变化建议只在收盘后运行。6.3 指标计算中的 NaN 和除零问题现象ma5、momentum、volume_ratio大量为 NaN或者筛选结果为空。可能原因行情数据不足 20 个交易日无法计算动量。前 5 日成交量为 0导致量比为无穷大或 NaN。停牌股票成交量为 0量比失真。检查方式打印计算后的数据描述print(df[[volume_ratio, momentum]].describe())解决方案在add_indicators中把量比无穷大替换为 NaN或者只保留有限值df[volume_ratio] df[volume_ratio].replace([np.inf, -np.inf], np.nan)筛选前再dropna()。6.4 策略结果不稳定怎么办现象每天输出的信号差异很大或者参数稍微改动结果就完全不同。可能原因规则阈值太窄样本数量少板块只取了前 10 只股票随机性大没有过滤上市时间过短的新股没有控制市值或流动性。解决方案增加max_stocks让样本更有代表性。增加流动性过滤例如剔除成交额过小的股票。使用多因子打分替代硬性阈值减少某一指标微调带来的影响。增加时间平滑比如要求连续 N 天满足条件才进入信号池。6.5 一套完整的排查清单运行脚本后如果结果异常按下面的顺序排查检查config.yaml中板块名称是否存在。检查网络是否正常akshare 是否安装最新版。单独拉取一只股票观察返回字段和日期范围。检查缓存数据是否更新删除data/目录后重新运行。检查add_indicators后 DataFrame 的列名和 NaN 数量。检查最终筛选条件中是否有除零或类型错误。对照daily_strategy.csv的日期确认不是盘中数据。7. 最佳实践和扩展方向这一节给出可以长期落地的建议。核心思想是让代码更健壮、数据更可追溯、策略更容易迭代。7.1 数据增量更新和缓存策略当前示例用 CSV 缓存简单但不够严谨。推荐改进为按“交易日”维度管理每个交易日拉取一次增量数据追加到历史表中。保存时带上数据源、拉取时间、版本号。定期清理异常数据防止污染后续指标计算。如果使用数据库可以设计一张行情表CREATE TABLE stock_daily ( symbol VARCHAR(10), date DATE, open DECIMAL(10, 2), high DECIMAL(10, 2), low DECIMAL(10, 2), close DECIMAL(10, 2), volume BIGINT, amount BIGINT, turnover DECIMAL(10, 4), PRIMARY KEY (symbol, date) );主键使用symbol date可以保证同一只股票同一交易日只有一条记录。7.2 用多因子打分替代单指标筛选硬性条件容易出现边界问题。比如某只股票动量是 30.5%只比阈值高 0.5 就被排除这并不合理。更好的办法是把每个因子映射成分数再计算综合得分。示例打分逻辑收盘价在 5 日和 20 日均线之上计 1 分每满足一个条件加 1 分。动量按 0 到 30% 区间线性映射到 0 到 2 分。量比大于 1 的部分最大值截止到 3 分。最后按综合得分排序取前 N 只。这样参数敏感度会降低不少。7.3 引入回测框架验证策略生成信号只是第一步。要评价“策略”是否值得继续研究必须回测。回测时要注意统一使用前复权价格避免分红影响收益计算。信号在第 T 日收盘后生成第 T1 日开盘或收盘买入不能在 T 日当天买入。考虑手续费、滑点和停牌无法交易等现实因素。使用滚动窗口避免使用未来数据。回测框架可以使用backtrader或vectorbt也可以自己用 pandas 实现简单事件回测。自己实现的优点是容易理解缺点是容易引入未来函数需要格外小心。7.4 合规与数据边界个人研究者在使用公开行情数据时需要注意数据来源的授权和使用限制。akshare 抓取的是公开网页数据适合学习和研究如果用于商业产品必须获得数据方的正式授权或使用商业数据源。最后重申一次本文生成的“次日策略”只是用代码自动完成板块和个股状态扫描输出的是“次日观察候选清单”不是买卖建议。任何基于该清单的交易决策都需要结合基本面、政策面、市场情绪和风险控制并遵守当地法律法规。如果你刚接触这个方向可以从修改config.yaml开始把板块换成你熟悉的行业把max_stocks调小逐步理解每一条数据是如何流动的。熟悉之后再加入数据库、日志、回测和自动任务最终形成一个属于自己的行业研究数据流水线。