在开源社区里daily_stock_analysis 是一个出现频率很高的项目命名方式。它直译过来就是“每日股票数据分析”通常意味着项目会按时拉取行情数据计算常用技术指标再根据预先写好的规则筛出一批值得关注的标的。这类项目非常适合作为 Python 数据处理、工程化编码和量化研究入门的练习。实际工作中难点往往不在 pandas 语法而在于如何保证数据干净、指标没有未来函数、任务每天都能稳定运行。这篇文章围绕 daily_stock_analysis 这个主题搭一个最小可运行的工程。读者可以把它当模板先跑通数据获取、指标计算、规则筛选、结果输出四个环节再逐步扩展数据源、回测和可视化。文章会解释每一步为什么这样做并给出常见问题排查和工程化建议。所有代码只用于技术演示不构成投资建议。1. 先理解 daily_stock_analysis 到底要解决什么问题1.1 从命名拆需求daily / stock / analysis项目名daily_stock_analysis可以拆成三个关键词daily按日频运行。一次任务处理一天或一段历史交易日的数据。stock数据对象是股票行情通常包含开盘价、收盘价、最高价、最低价、成交量等字段。analysis不只是展示行情而是要做分析例如计算技术指标、生成候选池、输出报告。拆开看之后需求边界就清晰了。这个项目不是实时行情系统不需要对接 tick 级数据也不是完整回测框架不需要处理复杂的撮合逻辑。它最核心的任务是在每天收盘后把一批股票的日线数据拉下来计算指标再按照固定规则输出分析结果。这个定位很重要。很多初学者看到“股票分析”就想着要做预测、做自动交易最终被复杂需求拖垮。先从每日数据整理开始反而能快速形成一个可以迭代的闭环。1.2 一个最小闭环应该包含哪四层一个可运行的 daily_stock_analysis 工程至少包含四层层次职责输入输出数据获取层拉取股票日线数据股票代码、起止日期原始 DataFrame数据清洗层处理列名、日期、缺失值、复权原始 DataFrame标准格式 DataFrame指标计算层计算 MA、RSI、MACD 等标准行情数据带指标的 DataFrame结果输出层按规则筛选并生成文件带指标的 DataFrameCSV、Excel、Markdown 报告这四个层次可以分别用独立模块实现。这样即使数据源接口变化或者筛选规则调整影响范围也能被限制住不用把整个脚本推倒重写。1.3 技术方案选型为什么用 Python pandas akshare对于这种日频数据分析项目Python 生态是最稳妥的选择。pandas 提供 DataFrame 和滚动窗口计算天然适合处理日期序列akshare 提供国内股票日线数据的免费接口适合学习阶段快速拿数据。选择依赖时不追求大而全。最小工程只需要Python 3.9 或更高版本。pandas数据处理和指标计算。numpy部分数学计算。akshare行情数据获取。openpyxl输出 Excel 文件。PyYAML读取配置文件。直接使用 akshare 有一个前提接口会随数据源调整而变化。所以在代码里要把数据获取隔离到独立模块并做好异常捕获这样接口变动时只需要改一个文件。2. 环境准备先把项目目录、依赖和配置对齐2.1 本地环境要求学习环境建议使用 Python 3.9 以上版本。命令如下python --version pip --version如果系统里同时存在 python3 和 python可以先确认默认解释器。推荐在项目目录下创建虚拟环境避免污染全局环境python -m venv .venvWindows 激活方式.venv\Scripts\activatemacOS / Linux 激活方式source .venv/bin/activate激活后执行安装命令依赖才能安装到当前虚拟环境中。2.2 项目目录结构一个清晰的项目结构能减少后续维护成本。推荐使用下面的布局daily_stock_analysis/ ├── config.yaml ├── requirements.txt ├── run.py ├── output/ │ └── .gitkeep └── src/ ├── __init__.py ├── data_fetcher.py ├── indicators.py ├── screener.py └── report.py目录说明config.yaml股票代码、日期范围、筛选参数等配置。run.py主入口串联数据获取、指标计算、筛选、输出。src/data_fetcher.py数据获取层。src/indicators.py指标计算层。src/screener.py筛选规则。src/report.py结果输出。output/存放生成的报告文件。这种拆分方式不是过度设计。数据源、指标、筛选、输出是四个不同维度的变化点提前隔离开后面改动会更轻松。2.3 依赖安装和配置外置requirements.txt内容如下。这里给出的版本号是示例实际安装时以本机可用版本为准akshare1.12.0 pandas1.5.0 numpy1.23.0 openpyxl3.0.10 PyYAML6.0安装命令pip install -r requirements.txt安装完成后把股票代码和规则参数放到config.yaml中而不是直接写在代码里。示例配置symbols: - 000001 - 600036 - 000858 start_date: 20240101 end_date: 20241231 adjust: qfq output_dir: output csv_filename: daily_stock_analysis.csv excel_filename: daily_stock_analysis.xlsx markdown_filename: daily_stock_analysis.md screen: volume_ratio: 1.5 ma_window: 20这样做的原因是股票池和筛选参数会经常调整如果写死在代码里每次调整都要改源码并重新测试。配置外置后只需要修改 YAML 文件。2.4 环境检查清单开始写代码前可以快速做一轮检查Python 版本是否满足 3.9。虚拟环境是否激活。pip list中是否能看到 pandas、akshare 等依赖。网络是否能正常访问行情数据源。output/目录是否存在。检查命令python -c import pandas, akshare; print(ok)如果这条命令没有报错说明核心依赖已经可用。3. 代码实现搭出可运行的日频分析流水线3.1 数据获取层先把数据源接口隔离出来数据获取层负责拉取单只股票的日线数据。以 akshare 的stock_zh_a_hist接口为例返回的是一个包含中文列名的 DataFrame。在src/data_fetcher.py中实现一个类import pandas as pd import akshare as ak class DataFetcher: def __init__(self, symbol: str, start_date: str, end_date: str, adjust: str qfq): self.symbol symbol self.start_date start_date self.end_date end_date self.adjust adjust def fetch(self) - pd.DataFrame: raw ak.stock_zh_a_hist( symbolself.symbol, perioddaily, start_dateself.start_date, end_dateself.end_date, adjustself.adjust, ) if raw is None or raw.empty: raise RuntimeError(f{self.symbol} 返回空数据) return raw为什么要单独写类而不是在入口脚本里直接调用接口因为后续可能切换数据源或者增加本地区缓存。如果入口脚本到处调用 akshare替换时会非常痛苦。隔离之后外部只依赖DataFetcher.fetch()返回一个 DataFrame。3.2 数据清洗日期、列名、排序和缺失值akshare 返回的字段通常包含“日期、开盘、收盘、最高、最低、成交量、成交额、振幅、涨跌幅、涨跌额、换手率”。为了让后续代码稳定需要把字段名改成英文小写并只保留核心列。在fetch()方法中增加清洗逻辑def fetch(self) - pd.DataFrame: raw ak.stock_zh_a_hist( symbolself.symbol, perioddaily, start_dateself.start_date, end_dateself.end_date, adjustself.adjust, ) if raw is None or raw.empty: raise RuntimeError(f{self.symbol} 返回空数据) df raw.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, }) df df[[date, open, high, low, close, volume]].copy() df[date] pd.to_datetime(df[date]) df df.sort_values(date).reset_index(dropTrue) numeric_cols [open, high, low, close, volume] for col in numeric_cols: df[col] pd.to_numeric(df[col], errorscoerce) return df这里有几个容易忽略的点sort_values(date)必须执行因为后续滚动指标依赖时间顺序。数据乱序会让 MA、RSI 全部算错。reset_index(dropTrue)让索引从 0 开始后续按位置取最后一行时更直观。pd.to_numeric(..., errorscoerce)可以把--或空字符串转成NaN避免后续计算报错。列明使用英文小写可以避免中文字段在不同版本的数据源中列名变化。3.3 技术指标计算MA、RSI、MACD 与成交量均线指标计算是分析的核心。在src/indicators.py中实现一个函数接收清洗后的 DataFrame返回新增指标列后的 DataFrame。import pandas as pd import numpy as np def add_indicators(df: pd.DataFrame) - pd.DataFrame: result df.copy() result result.sort_values(date).reset_index(dropTrue) # 均线 result[ma5] result[close].rolling(window5).mean() result[ma10] result[close].rolling(window10).mean() result[ma20] result[close].rolling(window20).mean() # 成交量均线 result[volume_ma5] result[volume].rolling(window5).mean() result[volume_ma20] result[volume].rolling(window20).mean() # MACD ema12 result[close].ewm(span12, adjustFalse).mean() ema26 result[close].ewm(span26, adjustFalse).mean() result[dif] ema12 - ema26 result[dea] result[dif].ewm(span9, adjustFalse).mean() result[macd] (result[dif] - result[dea]) * 2 # RSI 14使用 Wilder 平滑 delta result[close].diff() gain delta.clip(lower0) loss -delta.clip(upper0) avg_gain gain.ewm(alpha1 / 14, min_periods14).mean() avg_loss loss.ewm(alpha1 / 14, min_periods14).mean() rs avg_gain / avg_loss.replace(0, np.nan) result[rsi14] 100 - (100 / (1 rs)) return result每个指标的计算都依赖历史数据因此要和当前行的北京时间对应清楚。rolling默认窗口包含当前行意味着 T 日的 MA20 使用的是 T 日及之前 19 个交易日的收盘价没有使用未来数据这样才符合日频分析的“收盘后计算”场景。MACD 中adjustFalse表示使用 EMA 常用的递归计算方式与国内行情软件的一致性更高。RSI 计算中把avg_loss的 0 替换成NaN是为了避免除零。3.4 规则筛选把“分析”转化为可执行条件技术指标计算出来后需要写筛选规则。在src/screener.py中定义两个函数def check_candidate(df: pd.DataFrame, volume_ratio: float 1.5, ma_window: int 20) - bool: if df is None or df.empty: return False latest df.iloc[-1] ma_col fma{ma_window} if pd.isna(latest.get(ma_col)) or pd.isna(latest.get(volume_ma5)): return False # 示例规则收盘价站上均线同时成交量明显放大 price_ok latest[close] latest[ma_col] volume_ok latest[volume_ma5] latest[volume_ma20] * volume_ratio return price_ok and volume_ok这段代码里的规则只是示例实际项目应当根据自己研究的目标调整。关键点在于判断NaN必须放在最前面否则缺失值会直接导致比较结果恒为 False。使用iloc[-1]取最新交易日保证这个筛选是在当天收盘后看到完整日线才做出的。规则参数通过函数参数传入方便从配置文件中读取。3.5 结果输出CSV、Excel、Markdown 三种方式筛选结果需要落盘。在src/report.py中实现输出函数import pandas as pd def save_results(df: pd.DataFrame, output_dir: str, csv_filename: str, excel_filename: str, markdown_filename: str) - None: import os os.makedirs(output_dir, exist_okTrue) csv_path os.path.join(output_dir, csv_filename) excel_path os.path.join(output_dir, excel_filename) markdown_path os.path.join(output_dir, markdown_filename) df.to_csv(csv_path, indexFalse, encodingutf-8-sig) df.to_excel(excel_path, indexFalse) with open(markdown_path, w, encodingutf-8) as f: f.write(df.to_markdown(indexFalse))几个实现细节os.makedirs(output_dir, exist_okTrue)保证输出目录存在。CSV 使用utf-8-sig而不是普通utf-8这样用 Excel 打开 CSV 时中文不会乱码。Excel 输出依赖openpyxl如果没有安装会在这里报错。to_markdown依赖 pandas 和 tabulate 库。如果不想额外依赖也可以手动拼接表格。如果不想引入 Markdown 输出依赖可以把to_markdown换成简单的手动拼接。这里为了演示保留实际项目中按需求取舍。3.6 主入口 run.py将流程串起来run.py负责读取配置遍历股票池调用各模块最后生成汇总表。import argparse import logging from datetime import datetime import pandas as pd import yaml from src.data_fetcher import DataFetcher from src.indicators import add_indicators from src.screener import check_candidate from src.report import save_results logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s, ) logger logging.getLogger(__name__) def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def main(config_path: str) - None: config load_config(config_path) rows [] for symbol in config[symbols]: try: logger.info(开始处理 %s, symbol) fetcher DataFetcher( symbolsymbol, start_dateconfig[start_date], end_dateconfig[end_date], adjustconfig.get(adjust, qfq), ) df fetcher.fetch() df add_indicators(df) screen_config config.get(screen, {}) is_candidate check_candidate( df, volume_ratioscreen_config.get(volume_ratio, 1.5), ma_windowscreen_config.get(ma_window, 20), ) latest df.iloc[-1] rows.append({ symbol: symbol, trade_date: latest[date].strftime(%Y-%m-%d), close: latest[close], ma20: latest.get(ma20), rsi14: latest.get(rsi14), volume_ma5: latest.get(volume_ma5), is_candidate: is_candidate, }) except Exception as e: logger.exception(处理 %s 失败: %s, symbol, e) result_df pd.DataFrame(rows) if result_df.empty: logger.warning(没有生成任何结果请检查股票代码或数据源) return save_results( result_df, output_dirconfig[output_dir], csv_filenameconfig[csv_filename], excel_filenameconfig[excel_filename], markdown_filenameconfig[markdown_filename], ) logger.info(分析完成生成文件%s, config[output_dir]) if __name__ __main__: parser argparse.ArgumentParser(descriptiondaily stock analysis) parser.add_argument(--config, defaultconfig.yaml, help配置文件路径) args parser.parse_args() main(args.config)这个入口脚本把流程串成了明确的三段遍历股票、计算指标、汇总输出。单只股票失败不会中断整个任务而是记录异常后继续处理下一只。这对跑批任务非常重要。4. 运行验证不仅要能跑还要确认结果可信4.1 验证流程和预期输出运行命令python run.py --config config.yaml正常情况下的预期输出2025-01-02 10:00:00 [INFO] 开始处理 000001 2025-01-02 10:00:01 [INFO] 分析完成生成文件output在output/目录下会生成三个文件daily_stock_analysis.csvdaily_stock_analysis.xlsxdaily_stock_analysis.mdCSV 内容大致如下symbol,trade_date,close,ma20,rsi14,volume_ma5,is_candidate 000001,2024-12-31,10.5,10.2,55.3,1200000,True验证时不能只看文件是否生成还要检查日期是否为最近一个交易日。收盘价是否和行情软件一致。is_candidate列是否存在明确 True / False。单只股票失败时结果表里是否仍然包含其他股票。4.2 指标自检方式最直接的验证方式是对比行情软件。打开同花顺、东方财富等软件的日线界面选择同一只股票、同一个复权方式查看对应日期的 MA20、MACD 和 RSI。由于不同软件的计算细节不同数值可能出现小幅偏差尤其是 MACD 的adjust参数和 RSI 的平滑方式。如果偏差在 1% 以内通常可以接受。也可以写一段自检代码输出最后 3 行结果import pandas as pd from src.data_fetcher import DataFetcher from src.indicators import add_indicators fetcher DataFetcher(000001, 20240101, 20241231, adjustqfq) df fetcher.fetch() df add_indicators(df) print(df.tail(3)[[date, close, ma20, macd, rsi14]])检查点ma20在NaN后面开始出现正常数值。rsi14范围在 0 到 100 之间。df的行数大于指标窗口大小。如果数据量太少例如只有 10 条MA20 计算结果全是NaN这是正常的但筛选结果就不完整。这也是为什么配置日期范围时要保证历史数据数量足够。4.3 警惕未来函数和复权导致的偏差“未来函数”是股票数据分析里最常见也最隐蔽的错误。简单说T 日的分析只能用 T 日及以前的数据不能用到 T1 日或更晚的数据。在滚动指标计算中rolling(20).mean()默认包含当前行和之前 19 行不包含未来数据。但如果写成close.shift(-1).rolling(20).mean()就会引入下一天数据属于未来函数。复权问题同样要警惕。akshare 下载数据时可以选择qfq前复权、hfq后复权、空字符串表示不复权。不同的选择历史价格不同指标也会不同。复权方式特点适用场景不复权保留原始价格除权日有跳空查看真实成交价、填权缺口前复权 qfq最新价格不变历史价格调整技术指标、选股筛选后复权 hfq历史价格不变最新价格调整长期收益计算、回测日频选股分析通常使用前复权因为均线和 MACD 需要连续价格序列。但要注意前复权价格会随着后续除权而不断修正导致同一段历史数据“每除一次权就变化一次”。因此如果以后做回测还需要考虑复权基准点的处理。5. 常见问题排查从现象出发定位问题5.1 数据源返回空数据或请求失败现象ak.stock_zh_a_hist返回空 DataFrame或者抛出网络异常。可能原因股票代码错误例如在 A 股接口传了 6 位不含市场前缀的数字但代码本身不在交易所。日期范围不合法例如start_date大于end_date。非交易时段拉取日线数据数据源可能还没有生成当天记录。数据源接口升级或服务器限频。检查方式import akshare as ak raw ak.stock_zh_a_hist(symbol000001, perioddaily, start_date20240101, end_date20241231, adjustqfq) print(raw.shape) print(raw.head())如果单独执行仍然为空优先确认股票代码和日期范围。如果单次执行正常但批量执行失败多半是请求频率过高可以在循环中加入time.sleep(0.5)。5.2 中文列名、日期格式和索引错位现象df[date]报 KeyError或者按iloc[-1]取到的是最早一行。可能原因没有执行 rename列名仍是中文。数据源返回的日期是object类型没有转成 datetime。清洗后没有排序索引顺序和日期顺序不一致。检查方式print(df.columns) print(df.dtypes) print(df.head())解决方案统一在清洗阶段完成列名转换、日期解析、排序和重置索引。只要清洗层规范后续层都不会遇到这个问题。5.3 指标出现全 NaN 或长期是 None现象MA20、MACD 等在最终结果里全部是NaN或者is_candidate全是 False。可能原因历史数据量小于指标窗口。清洗时列名不对导致close列实际是字符串类型。数据源复权参数错误返回了空集合。check_candidate中pd.isna判断缺失后直接返回 False。检查方式print(df[close].dtype) print(df[close].notna().sum())如果close是 object 类型先执行pd.to_numeric。如果数据量不足需要调大start_date往前推的时间范围。5.4 输出文件乱码或 Excel 打不开现象用 Excel 打开 CSV 乱码或者生成的 xlsx 文件报错。可能原因CSV 使用utf-8而不是utf-8-sig。没有安装openpyxl。输出目录没有写权限。检查方式打开文件时确认编码命令行查看目录权限。解决方案写入 CSV 时固定使用encodingutf-8-sig写入 Excel 前确认import openpyxl不报错尽量把输出目录放在项目内部并用os.makedirs(..., exist_okTrue)创建。问题现象常见原因检查方式处理建议数据源返回空表代码错误、日期错误、限频单独执行接口语句核对代码和日期加入重试和延时中文列名报 KeyError未重命名列print(df.columns)在清洗层统一 rename指标全 NaN数据量不足或类型错误查看 dtype 和行数增加历史日期范围强转数值类型CSV 中文乱码编码写错用文本编辑器查看文件头使用utf-8-sigExcel 打不开缺少 openpyxlpip show openpyxl安装 openpyxl6. 工程化最佳实践从每日脚本升级成稳定任务6.1 拆分模块与保持幂等很多人的日频分析项目一开始只有一个stock.py后面不断往里面加参数、加规则最终变成一个大脚本。维护成本高的主要原因不是代码量多而是没有拆模块。推荐保持四层拆分data_fetcher.py只管数据获取。indicators.py只管指标计算。screener.py只管筛选规则。report.py只管输出。模块之间的通信对象统一为 DataFrame。每个模块都可以独立测试出现问题也能直接定位到对应文件。幂等性是指“同一份输入无论跑多少次输出结果都一样”。在日频任务中这个特性很重要。每天定时任务可能在凌晨跑两遍或者手动补跑。如果每次运行都对同一个 CSV 追加写入最终文件会出现重复行。推荐的做法是每次运行覆盖输出文件并在文件名中加入日期例如daily_stock_analysis_20241231.csv。6.2 日志、重试和异常处理批处理任务里一只股票失败不应该导致整个任务退出。在run.py中已经用try except包裹单只股票的处理逻辑这是最低要求。更好的做法是使用logging记录每只股票的开始、成功、失败。在logger.exception中输出完整堆栈方便排查。对数据源请求增加重试机制例如失败后等待 1 秒再试一次最多重试 3 次。最终生成一个failed_symbols列表输出到单独文件便于补跑。failed [] for symbol in config[symbols]: try: ... except Exception as e: failed.append(symbol) logger.exception(处理 %s 失败, symbol)6.3 定时调度与生产环境要求学习阶段可以手动执行python run.py --config config.yaml。进入生产环境后需要定时任务。常见方案有cronLinux 系统自带适合固定时间执行。Windows 任务计划程序适合 Windows 服务器。Airflow / DolphinScheduler适合复杂依赖和多人协作。云函数 / 容器定时任务适合云环境。cron 示例每天早上 18:00 执行0 18 * * 1-5 cd /path/to/daily_stock_analysis .venv/bin/python run.py logs/run_$(date \%Y\%m\%d).log 21生产环境建议额外考虑配置外置化不要把数据库密码或数据源 token 写进代码。数据源不可用时是否有备用数据源。历史数据是否需要落地到数据库而不是每次重新拉取。运行日志是否采集到统一平台例如 ELK。任务失败时是否触发告警。6.4 可复用清单发布每日分析任务前可以按这份清单逐项检查股票代码列表是否符合当前需求。日期范围是否覆盖足够的指标预热期。复权方式是否统一。指标计算是否使用了未来数据。输出文件是否为幂等覆盖。单只股票失败时是否只记录异常并继续处理。日志是否记录关键步骤和异常堆栈。定时任务时间是否在收盘后且数据源已完成当日更新。是否保留了最近几天的报告以便回溯。是否明确该输出只是技术参考不构成投资建议。7. 扩展方向从每日分析走向策略研究与可视化7.1 回测阶段要修正哪些偏差从 daily_stock_analysis 继续往前走最常见的方向是策略回测。回测比“每日筛选”复杂得多因为它需要模拟交易规则、手续费、滑点和持仓周期。回测阶段需要特别修正几个偏差信号使用偏差回测中必须确定信号是在收盘后产生下单和成交也只能在下一周期不能假设信号产生当天就能精确成交。复权基准偏差前复权价格会随最新除权变化导致历史收益计算失真。更严谨的方式是使用后复权或自行处理复权因子。幸存者偏差股票池如果只保留当前还在上市的股票回测结果会偏乐观。历史分析要考虑当时存在的股票。交易成本偏差如果忽略手续费、印花税和滑点回测收益会被明显高估。这些偏差不是一次能全部解决的但需要在一开始就有意识。7.2 可视化与数据存储当前工程输出的是表格文件后续可以增加图表。例如使用 matplotlib 绘制 K 线图和均线。使用 plotly 生成交互式走势图。在图上标注 MA20 突破、RSI 超买超卖区域。数据存储方面如果数据量增大可以引入 SQLite 或 PostgreSQL。存储结构至少包含CREATE TABLE daily_price ( symbol TEXT NOT NULL, trade_date DATE NOT NULL, open REAL, high REAL, low REAL, close REAL, volume INTEGER, PRIMARY KEY (symbol, trade_date) );主键使用(symbol, trade_date)可以保证同一只股票同一天不会重复存储。每日任务可以先删除当天数据再写入或者使用INSERT OR REPLACE实现增量更新。7.3 学习路径和练习建议如果读者刚接触这个主题建议按下面的路径练习先用单只股票跑通完整流程。再扩展到 10 只股票验证批处理稳定性。修改筛选规则观察is_candidate的变化。手动计算某一天的 MA20 和 RSI和代码结果对比。加入失败重试和日志模块。把结果写入 SQLite接入定时任务。最后再考虑回测和可视化。每完成一步都做一次小范围验证。不要一次性把所有功能都堆进去否则出现问题很难定位。daily_stock_analysis 表面上是做一个每日分析脚本实际操作中却涉及网络请求、数据处理、指标计算、配置管理、任务调度和数据可靠性。把这些基础能力练扎实再往量化方向发展就会顺畅很多。最关键的不是追求复杂模型而是保证每天输出的结果干净、可解释、可验证。
