简介这是一份面向金融数据分析初学者与Python进阶学习者的股市分析实战项目基于Jupyter Notebook构建聚焦标普500指数探索性分析、超级投资者持仓追踪及模拟持有策略回测等典型量化分析场景。资源采用规范的机器学习项目结构组织含dataexternal/interim/processed三级数据目录、notebooks3个核心.ipynb分析脚本、src模块化代码、models与docs等完整模块体现工程化数据科学实践范式。压缩包共34个文件以14个Python脚本支撑底层逻辑、3个Jupyter Notebook承载交互式分析流程、6个.gitkeep保障目录完整性并辅以Makefile自动化命令、配置模板与文档体系整体仅1.75MB轻量易部署。已有470人学习下载读者可直接复用其标准化数据处理流水线、EDA分析框架与模拟交易逻辑快速掌握从数据获取、清洗、特征构建到策略验证的全流程实现方法。1. 股市分析不是画几条K线就完事它本质是把行情数据、财务逻辑和交易行为拧成一股可验证的决策流很多人第一次跑stock_analysis这个词以为只是调个yfinance下点股价、画个matplotlib折线图——结果发现模型在回测里年化收益32%实盘三个月亏穿底裤。这不是玄学是漏掉了三个刚性环节数据清洗的时序对齐陷阱、因子计算的滚动窗口偏移、信号生成的滞后性校准。真正的股市分析项目核心不是“看涨跌”而是构建一个能经受住多周期验证日线/周线切换、多标的比对沪深300 vs 中证500成分股、多场景压力测试分红除权、ST摘帽、熔断恢复的分析流水线。它适合两类人一是想把课堂上学的CAPM、Fama-French三因子落地到A股的某高校金融工程方向学生二是手头有券商Level-2逐笔委托数据、但卡在特征工程无法复用的某量化团队初级工程师。本文不讲宏观择时不碰高频做市只聚焦「如何用Python本地跑通一个可复现、可调试、可上线的轻量级股市分析框架」——从原始数据接入开始到生成带仓位建议的日报PDF结束每一步都踩过坑、调过参、压过测。2. 用 yfinance akshare 搭建双源行情管道为什么单靠一个库永远缺关键字段2.1 为什么必须双源并行——yfinance 的免费红利与 akshare 的本土补丁yfinance是目前最稳定的美股港股部分A股免费数据源但它对A股存在三个硬伤无前复权价格yfinance.Ticker(600519.SS).history(periodmax)返回的是后复权价直接用于技术指标计算会导致均线系统整体下移缺失分红信息无法获取现金分红金额、股权登记日、除权日导致股息率回测失真停牌期间数据断层连续涨停/跌停时Close字段常为 NaN但实际应沿用前一日收盘价交易所规则要求。而akshare2024年v1.12.82起补上了这些缺口它通过爬取上交所/深交所公告页解析分红送转表用akshare.stock_zh_a_daily(symbolsh600519, start_date20200101, end_date20241231)可直接拿到前复权价且停牌日自动填充为前一日收盘价。但它的缺点是无分钟级数据、无港股通持股变动、无融资融券余额。提示不要试图用akshare替代yfinance获取美股数据——其美股接口已停更两年返回数据为静态缓存最新日期停留在2022年12月。2.2 构建双源同步校验管道用 checksum 对齐关键字段我们不信任任何单一数据源。做法是用 yfinance 获取原始OHLCV开盘/最高/最低/收盘/成交量用 akshare 获取前复权收盘价与分红事件再用 checksum 校验两者在非分红日的收盘价一致性。# data_pipeline.py import yfinance as yf import akshare as ak import pandas as pd import numpy as np def fetch_dual_source(symbol: str, start_date: str, end_date: str) - pd.DataFrame: # 步骤1yfinance 原始数据后复权 ticker yf.Ticker(f{symbol}.SS) if symbol.isdigit() else yf.Ticker(symbol) yf_df ticker.history(startstart_date, endend_date) yf_df yf_df[[Open, High, Low, Close, Volume]].copy() # 步骤2akshare 前复权数据含停牌填充 ak_df ak.stock_zh_a_daily( symbolfsh{symbol} if symbol.startswith(6) else fsz{symbol}, start_datestart_date.replace(-, ), end_dateend_date.replace(-, ) ) ak_df[date] pd.to_datetime(ak_df[date]) ak_df ak_df.set_index(date)[[close, open, high, low, volume]].rename( columns{close: close_adj, open: open_adj, high: high_adj, low: low_adj, volume: volume_adj} ) # 步骤3合并并校验仅校验非分红日 merged yf_df.join(ak_df, howinner, rsuffix_adj) # 计算 checksum取非NaN的Close与close_adj差值绝对值中位数 valid_mask ~merged[Close].isna() ~merged[close_adj].isna() diff_median np.median(np.abs(merged.loc[valid_mask, Close] - merged.loc[valid_mask, close_adj])) if diff_median 0.05: # 允许0.05元误差A股最小变动单位 raise ValueError(f双源价格校验失败中位数偏差 {diff_median:.3f}元请检查symbol格式或网络状态) return merged # 示例调用 df fetch_dual_source(600519, 2023-01-01, 2024-12-31) print(df[[Close, close_adj, Volume, volume_adj]].tail())参数说明symbolA股必须传纯数字代码如600519函数内部自动补sh/sz前缀港股传00700.HK格式美股传AAPLstart_date/end_date必须为YYYY-MM-DD格式akshare接口要求YYYYMMDD函数内自动转换diff_median 0.05是硬阈值A股价格精度为0.01元0.05元误差意味着至少5个价位偏差大概率是代码映射错误如把000001当成深市代码却用了sh前缀。2.3 分红事件注入把公告文本变成可计算的现金流时间点akshare的stock_zh_a_dividend接口返回的是结构化分红表但字段名是中文且存在空值。我们需要将其转为标准时间序列事件标记# dividend_enricher.py def inject_dividends(df: pd.DataFrame, symbol: str) - pd.DataFrame: # 获取分红数据 div_df ak.stock_zh_a_dividend(symbolsymbol) # 清洗只保留有效分红排除送转股、配股等非现金事件 div_df div_df[div_df[分红类型].str.contains(现金分红, naFalse)] div_df[除权除息日] pd.to_datetime(div_df[除权除息日]) div_df div_df.set_index(除权除息日)[[每股分红(税后)]] # 注入到主DataFrame用reindex实现左连接缺失值填0 df df.join(div_df, howleft).fillna({每股分红(税后): 0.0}) return df # 使用示例 df_with_div inject_dividends(df, 600519) print(df_with_div[df_with_div[每股分红(税后)] 0][[Close, 每股分红(税后)]].head())关键逻辑reindex比merge更安全避免因日期精度差异如2023-06-15vs2023-06-15 00:00:00导致匹配失败fillna统一设为0.0后续计算股息率时dividend_yield df[每股分红(税后)] / df[Close]自动得0无需额外判断不过滤“税前”字段A股分红税后金额已扣20%红利税直接使用即可。3. 构建可复用的因子计算引擎滚动窗口、时序对齐与内存优化三原则3.1 为什么不能直接用 talib——A股特有的停牌与涨跌停处理TA-Lib是技术指标计算的事实标准但它默认按自然日滚动如talib.SMA(close, timeperiod20)而A股存在大量停牌日。若某股票连续停牌10天talib.SMA会把这10个NaN当作有效窗口导致20日均线在复牌后突然跳变。正确做法是用交易日历对齐滚动窗口且窗口内只计有效交易日。我们弃用talib改用pandas原生rolling 自定义apply并绑定上交所/深交所交易日历# factor_engine.py import pandas as pd import numpy as np from datetime import datetime, timedelta # 预加载2020-2030年A股交易日历真实项目中从akshare.stock_zh_a_hist获取 TRADING_DAYS pd.date_range(2020-01-01, 2030-12-31, freqD) # 过滤出工作日再剔除已知节假日此处简化实际需调用akshare.stock_zh_a_sina_lhb TRADING_DAYS TRADING_DAYS[TRADING_DAYS.weekday 5] # 剔除周末 TRADING_DAYS TRADING_DAYS[~TRADING_DAYS.isin(pd.to_datetime([2023-01-21, 2023-01-27]))] # 示例剔除春节 def rolling_sma_adjusted(series: pd.Series, window: int) - pd.Series: 适配A股停牌的SMA窗口内只计交易日自动跳过NaN def _sma_window(x): valid_vals x.dropna() if len(valid_vals) window // 2: # 至少一半有效数据才计算 return np.nan return valid_vals.mean() # 用交易日历重采样确保索引为连续交易日 series_aligned series.reindex(TRADING_DAYS, methodffill) # 停牌日用前一日收盘价填充 return series_aligned.rolling(windowwindow, min_periods1).apply(_sma_window) # 应用示例 df[ma20] rolling_sma_adjusted(df[Close], window20)参数说明window20指20个交易日非自然日min_periods1允许首20日部分填充避免开头全NaNmethodffill停牌日用前一日收盘价填充——这是交易所规则如上交所《交易规则》第3.1.7条不是插值玄学_sma_window内len(valid_vals) window // 2是防呆若20日窗口内有效数据不足10个说明该股长期停牌SMA失去意义返回NaN。3.2 多因子合成用 rank-based normalization 解决量纲冲突PE、PB、ROE、换手率等因子量纲差异极大PE可能100倍ROE仅15%直接加权会淹没低量纲因子。常见误用是MinMaxScaler但它对异常值敏感如某ST股PE10000会压缩其他股票全部范围。我们采用分位数秩标准化Rank-based Normalizationdef rank_normalize(series: pd.Series, method: str average) - pd.Series: 用分位数秩替代原始值消除量纲与异常值影响 # 处理NaN赋予中位秩不影响排序 series_clean series.fillna(series.median()) # 计算秩ascendingTrue 表示越小越好如PE需根据因子经济意义反转 ranks series_clean.rank(methodmethod, ascendingTrue) # 转为0-1区间(rank - 1) / (n - 1) n len(ranks) normalized (ranks - 1) / (n - 1) if n 1 else 0.5 return normalized # 示例PE越小越好ROE越大越好 → ROE需反转 df[pe_rank] rank_normalize(df[pe_ttm]) # 小→好直接用 df[roe_rank] 1 - rank_normalize(df[roe_ttm]) # 大→好取反 df[factor_score] 0.4 * df[pe_rank] 0.6 * df[roe_rank]为什么选rank而非zscorezscore假设分布正态但A股PE分布严重右偏大量小市值股PE20少数妖股PE1000rank是非参数方法对分布形状无假设且天然鲁棒——即使加入一个PE10000的离群值其余股票秩仅微调1位methodaverage处理并列值如多只股票PE同为15.2避免人为制造差异。3.3 内存优化用 dtype downcast 避免 pandas 默认 float64 浪费一个含10年日线2500行×50因子的DataFrame若全用float64内存占用约1MB但多数因子如涨跌幅%、换手率%精度到小数点后4位足矣。我们用pd.to_numeric(..., downcastfloat)自动降级def optimize_dtypes(df: pd.DataFrame) - pd.DataFrame: 将float64降为float32/float16int64降为int32/int16 for col in df.columns: if df[col].dtype float64: df[col] pd.to_numeric(df[col], downcastfloat) elif df[col].dtype int64: df[col] pd.to_numeric(df[col], downcastinteger) return df # 应用后内存对比 df_orig df.copy() df_opt optimize_dtypes(df) print(f原始内存: {df_orig.memory_usage(deepTrue).sum() / 1024**2:.2f} MB) print(f优化后内存: {df_opt.memory_usage(deepTrue).sum() / 1024**2:.2f} MB) # 典型结果1.23 MB → 0.65 MB节省47%注意downcastfloat会优先尝试float32若精度损失超容忍阈值如四舍五入误差1e-6则回落到float64。对A股日线数据float32完全够用价格精度0.01元float32可精确表示±16777216范围内的整数。4. 回测引擎的三大避坑指南信号滞后、滑点模拟与幸存者偏差4.1 现象回测收益高得离谱实盘一买就跌原因信号生成未考虑执行滞后。例如用df[Close].shift(-1)生成次日买入信号但实际交易需在当日收盘后、次日开盘前决策而次日开盘价常跳空。解决所有信号必须基于T-1日收盘数据生成T日开盘价执行。修改信号列命名规范signal_t1表示T-1日生成的信号对应T日操作。# 错误示范导致未来函数 df[signal] np.where(df[ma5] df[ma20], 1, 0) # 用当日ma5/ma20生成当日信号 # 正确示范严格滞后 df[ma5_t1] df[Close].rolling(5).mean().shift(1) # T-1日的5日均值 df[ma20_t1] df[Close].rolling(20).mean().shift(1) # T-1日的20日均值 df[signal_t1] np.where(df[ma5_t1] df[ma20_t1], 1, 0) # T-1日生成T日执行4.2 现象回测中频繁满仓实盘总因流动性不足无法成交原因未模拟滑点slippage与冲击成本。小市值股日均成交额5000万你下单1000万必然推高价格。解决按成交额占比动态计算滑点。公式slippage 0.001 * (order_amount / daily_volume)0.1%基础滑点 × 成交额占比def calculate_slippage(df: pd.DataFrame, order_amount: float 1e6) - pd.Series: 计算每笔交易的预估滑点小数形式 # daily_volume 单位元order_amount 单位元 volume_ratio order_amount / df[Volume] / df[Close] # 换算成股数占比 # 设定上限单笔成交不超过日成交额5% volume_ratio np.clip(volume_ratio, 0, 0.05) return 0.001 * volume_ratio # 基础滑点0.1% df[slippage] calculate_slippage(df, order_amount5e5) df[exec_price] df[Open] * (1 df[slippage]) # 开盘价基础上浮滑点4.3 现象用中证全指回测十年年化25%但2023年新上市的股票从未被选中原因幸存者偏差Survivorship Bias。回测时用了当前成分股列表但十年前这些股票可能尚未上市或已被ST。解决必须使用指数历史成分股快照。akshare的stock_zh_a_hist不提供此功能需手动维护或调用聚宽JoinQuantAPI。简易方案用akshare.index_stock_cons获取各季度末成分股再按时间切片# 获取2023Q4中证500成分股示例 cons_df ak.index_stock_cons(indexCSI500) # cons_df 包含 symbol, code, date成分股纳入日期 # 回测时对每个交易日T只从 cons_df[cons_df[date] T] 中筛选当前成分股注意akshare.index_stock_cons返回的是当前最新成分股历史快照需自行存档。生产环境必须建立成分股变更日志表否则回测毫无意义。4.4 现象回测净值曲线平滑如镜但实盘单月回撤30%原因未考虑手续费与印花税。A股卖出收0.1%印花税双边交易佣金约0.025%。解决在每次买卖时扣除费用def apply_transaction_cost(position_change: float, price: float, fee_rate: float 0.00025, tax_rate: float 0.001) - float: 计算交易成本买入佣金卖出佣金印花税 cost abs(position_change) * price * fee_rate if position_change 0: # 卖出 cost abs(position_change) * price * tax_rate return cost # 在回测循环中调用 cost apply_transaction_cost(delta_shares, exec_price) cash - cost5. 生成可交付的分析报告从 DataFrame 到 PDF 的自动化流水线5.1 用 jinja2 模板引擎解耦数据与排版硬编码matplotlib画图再pdfkit转PDF会导致每次改图表就要重写逻辑。我们用jinja2将数据渲染为HTML再用weasyprint转PDF——好处是前端同事可直接改CSS样式无需碰Python代码。!-- report_template.html -- !DOCTYPE html html head meta charsetutf-8 title{{ stock_name }} {{ report_date }} 分析报告/title style body { font-family: Microsoft YaHei, sans-serif; margin: 2cm; } .chart { width: 100%; height: 400px; border: 1px solid #eee; } .factor-table { width: 100%; border-collapse: collapse; } .factor-table th, .factor-table td { border: 1px solid #ccc; padding: 4px; } /style /head body h1{{ stock_name }} ({{ symbol }}) {{ report_date }} 分析报告/h1 h2核心指标/h2 table classfactor-table trth指标/thth值/th/tr trtdPE-TTM/tdtd{{ pe_ttm|round(2) }}/td/tr trtd股息率/tdtd{{ dividend_yield|round(4)*100 }}%/td/tr trtd20日均线/tdtd{{ ma20|round(2) }}/td/tr /table h2价格走势图/h2 img src{{ chart_path }} classchart / /body /html5.2 自动生成带水印的PDF用 weasyprint 插入公司标识与防伪码weasyprint支持CSSpage规则可添加页眉页脚。我们插入动态水印含生成时间哈希# report_generator.py from weasyprint import HTML, CSS import hashlib from datetime import datetime def generate_pdf_report(df: pd.DataFrame, symbol: str, output_path: str): # 准备数据 context { stock_name: 贵州茅台, symbol: symbol, report_date: datetime.now().strftime(%Y年%m月%d日), pe_ttm: df[pe_ttm].iloc[-1], dividend_yield: df[dividend_yield].iloc[-1], ma20: df[ma20].iloc[-1], chart_path: charts/price_chart.png # 先保存图表 } # 渲染HTML template open(report_template.html).read() html_content template.render(context) # 添加水印CSS watermark_css CSS(stringf page {{ bottom-center {{ content: Report ID: {hashlib.md5((symbol str(datetime.now())).encode()).hexdigest()[:8]} | Generated {datetime.now().strftime(%Y-%m-%d %H:%M)}; font-size: 10px; color: #aaa; }} }} ) # 生成PDF HTML(stringhtml_content).write_pdf( output_path, stylesheets[watermark_css] ) print(fPDF报告已生成{output_path}) # 调用 generate_pdf_report(df, 600519, reports/600519_20241231.pdf)关键设计Report ID由symbol datetime的MD5前8位生成确保每份报告唯一可追溯bottom-center位置固定在页脚中央避免与正文重叠font-size: 10px和color: #aaa保证水印不喧宾夺主但复印时仍可见。5.3 批量生成多股票报告用 concurrent.futures 加速IO密集型任务生成100只股票报告若串行执行每份耗时3秒图表渲染PDF生成总耗时5分钟。用进程池并发from concurrent.futures import ProcessPoolExecutor, as_completed def process_single_stock(symbol: str): try: df fetch_dual_source(symbol, 2023-01-01, 2024-12-31) df inject_dividends(df, symbol) df[dividend_yield] df[每股分红(税后)] / df[Close] df[ma20] rolling_sma_adjusted(df[Close], 20) generate_pdf_report(df, symbol, freports/{symbol}_20241231.pdf) return f{symbol}: success except Exception as e: return f{symbol}: failed - {str(e)} # 并发执行 symbols [600519, 000858, 300750, 601318] # 实际读取自CSV with ProcessPoolExecutor(max_workers4) as executor: futures [executor.submit(process_single_stock, s) for s in symbols] for future in as_completed(futures): print(future.result()) # 输出示例 # 600519: success # 000858: success # 300750: failed - ValueError: 双源价格校验失败...为什么用ProcessPoolExecutor而非ThreadPoolExecutorweasyprint和matplotlib是CPU密集型PDF渲染、图像绘制GIL会阻塞线程max_workers4是经验阈值超过4个进程会导致内存竞争反而降低吞吐as_completed保证失败任务即时反馈不因某只股票报错而中断全部流程。6. 我的血泪经验三个让分析真正落地的细节技巧6.1 把「无效日期」显式标为 NaN而不是用 fillna 填充——这是调试的后悔药新手常犯的错看到df[Close]有NaN立刻df[Close].fillna(methodffill)。这会让调试变得地狱级困难——你永远不知道这个值是真实停牌填充的还是原始数据缺失。我的做法是所有数据源接入后先用pd.isna()标记原始缺失再用df[Close].mask(invalid_mask, np.nan)显式置空# 正确姿势保留原始缺失痕迹 invalid_mask (df[Volume] 0) | (df[Close].isna()) # 量为0且价为空判定为无效 df[Close_raw] df[Close].copy() # 保留原始列 df[Close] df[Close].mask(invalid_mask, np.nan) # 显式置空 # 后续fill时加注释说明原因 df[Close] df[Close].ffill() # 停牌日填充符合交易所规则这样当你发现某日Close异常时可以快速定位若Close_raw也是NaN → 数据源本身缺失若Close_raw有值但Close被填充 → 是你主动填充的检查填充逻辑若Close_raw有值且Close未变 → 是有效交易日。没有这一步你在回测里看到一个诡异的收益峰值可能要花两小时排查是数据问题还是策略bug。6.2 用 pytest 写数据校验测试而不是靠肉眼盯Excel我见过太多团队把数据校验写成“打开Excel看一眼”。正确的做法是为每个数据源写独立的pytest测试失败时直接打印差异行# test_data_integrity.py import pytest import pandas as pd def test_yfinance_akshare_alignment(): df fetch_dual_source(600519, 2023-01-01, 2023-01-31) # 校验非分红日yfinance Close 与 akshare close_adj 偏差 0.05 non_div_mask df[每股分红(税后)] 0 diffs abs(df.loc[non_div_mask, Close] - df.loc[non_div_mask, close_adj]) assert diffs.max() 0.05, f最大偏差 {diffs.max():.3f} 0.05异常行{diffs.idxmax()} # 运行pytest test_data_integrity.py -v # 输出FAILED test_data_integrity.py::test_yfinance_akshare_alignment - AssertionError: 最大偏差 0.123 0.05异常行2023-01-15这个测试每天凌晨自动运行一旦失败企业微信机器人立刻推送告警并附上diffs.nlargest(3)的具体日期和偏差值。比人工抽查可靠100倍。6.3 在因子计算中强制加入「业务含义注释」让三个月后的自己能看懂写因子公式时别只写df[factor] (df[roa] - df[roa].shift(1)) / df[roa].shift(1)。要像这样# ✅ 正确每行都有业务注释 df[roa_qoq_change] ( df[roa] - df[roa].shift(1) # 当前季度ROA - 上季度ROA ) / df[roa].shift(1) # 除以上季度ROA得环比增长率 # 注ROA净利润/总资产此处用TTM数据故上季度实为上期TTM # 来源akshare.stock_fundamental_em(600519) 的 roa 字段为什么重要因为三个月后你翻代码看到roa_qoq_change第一反应是“这是季报数据吗”而不是“这玩意儿到底怎么算的”。我在某跨平台系统里见过一个叫score_v3的因子没人知道v1和v2是什么最后花了两天时间反向工程出它其实是PE倒数乘以ROE——这种黑匣子早该在诞生时就被注释封印。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/16 18:19:00 · 阅读 1.2k Python构建可复现股市分析框架：双源数据校验与因子引擎
简介这是一份面向金融数据分析初学者与Python进阶学习者的股市分析实战项目基于Jupyter Notebook构建聚焦标普500指数探索性分析、超级投资者持仓追踪及模拟持有策略回测等典型量化分析场景。资源采用规范的机器学习项目结构组织含dataexternal/interim/processed三级数据目录、notebooks3个核心.ipynb分析脚本、src模块化代码、models与docs等完整模块体现工程化数据科学实践范式。压缩包共34个文件以14个Python脚本支撑底层逻辑、3个Jupyter Notebook承载交互式分析流程、6个.gitkeep保障目录完整性并辅以Makefile自动化命令、配置模板与文档体系整体仅1.75MB轻量易部署。已有470人学习下载读者可直接复用其标准化数据处理流水线、EDA分析框架与模拟交易逻辑快速掌握从数据获取、清洗、特征构建到策略验证的全流程实现方法。1. 股市分析不是画几条K线就完事它本质是把行情数据、财务逻辑和交易行为拧成一股可验证的决策流很多人第一次跑stock_analysis这个词以为只是调个yfinance下点股价、画个matplotlib折线图——结果发现模型在回测里年化收益32%实盘三个月亏穿底裤。这不是玄学是漏掉了三个刚性环节数据清洗的时序对齐陷阱、因子计算的滚动窗口偏移、信号生成的滞后性校准。真正的股市分析项目核心不是“看涨跌”而是构建一个能经受住多周期验证日线/周线切换、多标的比对沪深300 vs 中证500成分股、多场景压力测试分红除权、ST摘帽、熔断恢复的分析流水线。它适合两类人一是想把课堂上学的CAPM、Fama-French三因子落地到A股的某高校金融工程方向学生二是手头有券商Level-2逐笔委托数据、但卡在特征工程无法复用的某量化团队初级工程师。本文不讲宏观择时不碰高频做市只聚焦「如何用Python本地跑通一个可复现、可调试、可上线的轻量级股市分析框架」——从原始数据接入开始到生成带仓位建议的日报PDF结束每一步都踩过坑、调过参、压过测。2. 用 yfinance akshare 搭建双源行情管道为什么单靠一个库永远缺关键字段2.1 为什么必须双源并行——yfinance 的免费红利与 akshare 的本土补丁yfinance是目前最稳定的美股港股部分A股免费数据源但它对A股存在三个硬伤无前复权价格yfinance.Ticker(600519.SS).history(periodmax)返回的是后复权价直接用于技术指标计算会导致均线系统整体下移缺失分红信息无法获取现金分红金额、股权登记日、除权日导致股息率回测失真停牌期间数据断层连续涨停/跌停时Close字段常为 NaN但实际应沿用前一日收盘价交易所规则要求。而akshare2024年v1.12.82起补上了这些缺口它通过爬取上交所/深交所公告页解析分红送转表用akshare.stock_zh_a_daily(symbolsh600519, start_date20200101, end_date20241231)可直接拿到前复权价且停牌日自动填充为前一日收盘价。但它的缺点是无分钟级数据、无港股通持股变动、无融资融券余额。提示不要试图用akshare替代yfinance获取美股数据——其美股接口已停更两年返回数据为静态缓存最新日期停留在2022年12月。2.2 构建双源同步校验管道用 checksum 对齐关键字段我们不信任任何单一数据源。做法是用 yfinance 获取原始OHLCV开盘/最高/最低/收盘/成交量用 akshare 获取前复权收盘价与分红事件再用 checksum 校验两者在非分红日的收盘价一致性。# data_pipeline.py import yfinance as yf import akshare as ak import pandas as pd import numpy as np def fetch_dual_source(symbol: str, start_date: str, end_date: str) - pd.DataFrame: # 步骤1yfinance 原始数据后复权 ticker yf.Ticker(f{symbol}.SS) if symbol.isdigit() else yf.Ticker(symbol) yf_df ticker.history(startstart_date, endend_date) yf_df yf_df[[Open, High, Low, Close, Volume]].copy() # 步骤2akshare 前复权数据含停牌填充 ak_df ak.stock_zh_a_daily( symbolfsh{symbol} if symbol.startswith(6) else fsz{symbol}, start_datestart_date.replace(-, ), end_dateend_date.replace(-, ) ) ak_df[date] pd.to_datetime(ak_df[date]) ak_df ak_df.set_index(date)[[close, open, high, low, volume]].rename( columns{close: close_adj, open: open_adj, high: high_adj, low: low_adj, volume: volume_adj} ) # 步骤3合并并校验仅校验非分红日 merged yf_df.join(ak_df, howinner, rsuffix_adj) # 计算 checksum取非NaN的Close与close_adj差值绝对值中位数 valid_mask ~merged[Close].isna() ~merged[close_adj].isna() diff_median np.median(np.abs(merged.loc[valid_mask, Close] - merged.loc[valid_mask, close_adj])) if diff_median 0.05: # 允许0.05元误差A股最小变动单位 raise ValueError(f双源价格校验失败中位数偏差 {diff_median:.3f}元请检查symbol格式或网络状态) return merged # 示例调用 df fetch_dual_source(600519, 2023-01-01, 2024-12-31) print(df[[Close, close_adj, Volume, volume_adj]].tail())参数说明symbolA股必须传纯数字代码如600519函数内部自动补sh/sz前缀港股传00700.HK格式美股传AAPLstart_date/end_date必须为YYYY-MM-DD格式akshare接口要求YYYYMMDD函数内自动转换diff_median 0.05是硬阈值A股价格精度为0.01元0.05元误差意味着至少5个价位偏差大概率是代码映射错误如把000001当成深市代码却用了sh前缀。2.3 分红事件注入把公告文本变成可计算的现金流时间点akshare的stock_zh_a_dividend接口返回的是结构化分红表但字段名是中文且存在空值。我们需要将其转为标准时间序列事件标记# dividend_enricher.py def inject_dividends(df: pd.DataFrame, symbol: str) - pd.DataFrame: # 获取分红数据 div_df ak.stock_zh_a_dividend(symbolsymbol) # 清洗只保留有效分红排除送转股、配股等非现金事件 div_df div_df[div_df[分红类型].str.contains(现金分红, naFalse)] div_df[除权除息日] pd.to_datetime(div_df[除权除息日]) div_df div_df.set_index(除权除息日)[[每股分红(税后)]] # 注入到主DataFrame用reindex实现左连接缺失值填0 df df.join(div_df, howleft).fillna({每股分红(税后): 0.0}) return df # 使用示例 df_with_div inject_dividends(df, 600519) print(df_with_div[df_with_div[每股分红(税后)] 0][[Close, 每股分红(税后)]].head())关键逻辑reindex比merge更安全避免因日期精度差异如2023-06-15vs2023-06-15 00:00:00导致匹配失败fillna统一设为0.0后续计算股息率时dividend_yield df[每股分红(税后)] / df[Close]自动得0无需额外判断不过滤“税前”字段A股分红税后金额已扣20%红利税直接使用即可。3. 构建可复用的因子计算引擎滚动窗口、时序对齐与内存优化三原则3.1 为什么不能直接用 talib——A股特有的停牌与涨跌停处理TA-Lib是技术指标计算的事实标准但它默认按自然日滚动如talib.SMA(close, timeperiod20)而A股存在大量停牌日。若某股票连续停牌10天talib.SMA会把这10个NaN当作有效窗口导致20日均线在复牌后突然跳变。正确做法是用交易日历对齐滚动窗口且窗口内只计有效交易日。我们弃用talib改用pandas原生rolling 自定义apply并绑定上交所/深交所交易日历# factor_engine.py import pandas as pd import numpy as np from datetime import datetime, timedelta # 预加载2020-2030年A股交易日历真实项目中从akshare.stock_zh_a_hist获取 TRADING_DAYS pd.date_range(2020-01-01, 2030-12-31, freqD) # 过滤出工作日再剔除已知节假日此处简化实际需调用akshare.stock_zh_a_sina_lhb TRADING_DAYS TRADING_DAYS[TRADING_DAYS.weekday 5] # 剔除周末 TRADING_DAYS TRADING_DAYS[~TRADING_DAYS.isin(pd.to_datetime([2023-01-21, 2023-01-27]))] # 示例剔除春节 def rolling_sma_adjusted(series: pd.Series, window: int) - pd.Series: 适配A股停牌的SMA窗口内只计交易日自动跳过NaN def _sma_window(x): valid_vals x.dropna() if len(valid_vals) window // 2: # 至少一半有效数据才计算 return np.nan return valid_vals.mean() # 用交易日历重采样确保索引为连续交易日 series_aligned series.reindex(TRADING_DAYS, methodffill) # 停牌日用前一日收盘价填充 return series_aligned.rolling(windowwindow, min_periods1).apply(_sma_window) # 应用示例 df[ma20] rolling_sma_adjusted(df[Close], window20)参数说明window20指20个交易日非自然日min_periods1允许首20日部分填充避免开头全NaNmethodffill停牌日用前一日收盘价填充——这是交易所规则如上交所《交易规则》第3.1.7条不是插值玄学_sma_window内len(valid_vals) window // 2是防呆若20日窗口内有效数据不足10个说明该股长期停牌SMA失去意义返回NaN。3.2 多因子合成用 rank-based normalization 解决量纲冲突PE、PB、ROE、换手率等因子量纲差异极大PE可能100倍ROE仅15%直接加权会淹没低量纲因子。常见误用是MinMaxScaler但它对异常值敏感如某ST股PE10000会压缩其他股票全部范围。我们采用分位数秩标准化Rank-based Normalizationdef rank_normalize(series: pd.Series, method: str average) - pd.Series: 用分位数秩替代原始值消除量纲与异常值影响 # 处理NaN赋予中位秩不影响排序 series_clean series.fillna(series.median()) # 计算秩ascendingTrue 表示越小越好如PE需根据因子经济意义反转 ranks series_clean.rank(methodmethod, ascendingTrue) # 转为0-1区间(rank - 1) / (n - 1) n len(ranks) normalized (ranks - 1) / (n - 1) if n 1 else 0.5 return normalized # 示例PE越小越好ROE越大越好 → ROE需反转 df[pe_rank] rank_normalize(df[pe_ttm]) # 小→好直接用 df[roe_rank] 1 - rank_normalize(df[roe_ttm]) # 大→好取反 df[factor_score] 0.4 * df[pe_rank] 0.6 * df[roe_rank]为什么选rank而非zscorezscore假设分布正态但A股PE分布严重右偏大量小市值股PE20少数妖股PE1000rank是非参数方法对分布形状无假设且天然鲁棒——即使加入一个PE10000的离群值其余股票秩仅微调1位methodaverage处理并列值如多只股票PE同为15.2避免人为制造差异。3.3 内存优化用 dtype downcast 避免 pandas 默认 float64 浪费一个含10年日线2500行×50因子的DataFrame若全用float64内存占用约1MB但多数因子如涨跌幅%、换手率%精度到小数点后4位足矣。我们用pd.to_numeric(..., downcastfloat)自动降级def optimize_dtypes(df: pd.DataFrame) - pd.DataFrame: 将float64降为float32/float16int64降为int32/int16 for col in df.columns: if df[col].dtype float64: df[col] pd.to_numeric(df[col], downcastfloat) elif df[col].dtype int64: df[col] pd.to_numeric(df[col], downcastinteger) return df # 应用后内存对比 df_orig df.copy() df_opt optimize_dtypes(df) print(f原始内存: {df_orig.memory_usage(deepTrue).sum() / 1024**2:.2f} MB) print(f优化后内存: {df_opt.memory_usage(deepTrue).sum() / 1024**2:.2f} MB) # 典型结果1.23 MB → 0.65 MB节省47%注意downcastfloat会优先尝试float32若精度损失超容忍阈值如四舍五入误差1e-6则回落到float64。对A股日线数据float32完全够用价格精度0.01元float32可精确表示±16777216范围内的整数。4. 回测引擎的三大避坑指南信号滞后、滑点模拟与幸存者偏差4.1 现象回测收益高得离谱实盘一买就跌原因信号生成未考虑执行滞后。例如用df[Close].shift(-1)生成次日买入信号但实际交易需在当日收盘后、次日开盘前决策而次日开盘价常跳空。解决所有信号必须基于T-1日收盘数据生成T日开盘价执行。修改信号列命名规范signal_t1表示T-1日生成的信号对应T日操作。# 错误示范导致未来函数 df[signal] np.where(df[ma5] df[ma20], 1, 0) # 用当日ma5/ma20生成当日信号 # 正确示范严格滞后 df[ma5_t1] df[Close].rolling(5).mean().shift(1) # T-1日的5日均值 df[ma20_t1] df[Close].rolling(20).mean().shift(1) # T-1日的20日均值 df[signal_t1] np.where(df[ma5_t1] df[ma20_t1], 1, 0) # T-1日生成T日执行4.2 现象回测中频繁满仓实盘总因流动性不足无法成交原因未模拟滑点slippage与冲击成本。小市值股日均成交额5000万你下单1000万必然推高价格。解决按成交额占比动态计算滑点。公式slippage 0.001 * (order_amount / daily_volume)0.1%基础滑点 × 成交额占比def calculate_slippage(df: pd.DataFrame, order_amount: float 1e6) - pd.Series: 计算每笔交易的预估滑点小数形式 # daily_volume 单位元order_amount 单位元 volume_ratio order_amount / df[Volume] / df[Close] # 换算成股数占比 # 设定上限单笔成交不超过日成交额5% volume_ratio np.clip(volume_ratio, 0, 0.05) return 0.001 * volume_ratio # 基础滑点0.1% df[slippage] calculate_slippage(df, order_amount5e5) df[exec_price] df[Open] * (1 df[slippage]) # 开盘价基础上浮滑点4.3 现象用中证全指回测十年年化25%但2023年新上市的股票从未被选中原因幸存者偏差Survivorship Bias。回测时用了当前成分股列表但十年前这些股票可能尚未上市或已被ST。解决必须使用指数历史成分股快照。akshare的stock_zh_a_hist不提供此功能需手动维护或调用聚宽JoinQuantAPI。简易方案用akshare.index_stock_cons获取各季度末成分股再按时间切片# 获取2023Q4中证500成分股示例 cons_df ak.index_stock_cons(indexCSI500) # cons_df 包含 symbol, code, date成分股纳入日期 # 回测时对每个交易日T只从 cons_df[cons_df[date] T] 中筛选当前成分股注意akshare.index_stock_cons返回的是当前最新成分股历史快照需自行存档。生产环境必须建立成分股变更日志表否则回测毫无意义。4.4 现象回测净值曲线平滑如镜但实盘单月回撤30%原因未考虑手续费与印花税。A股卖出收0.1%印花税双边交易佣金约0.025%。解决在每次买卖时扣除费用def apply_transaction_cost(position_change: float, price: float, fee_rate: float 0.00025, tax_rate: float 0.001) - float: 计算交易成本买入佣金卖出佣金印花税 cost abs(position_change) * price * fee_rate if position_change 0: # 卖出 cost abs(position_change) * price * tax_rate return cost # 在回测循环中调用 cost apply_transaction_cost(delta_shares, exec_price) cash - cost5. 生成可交付的分析报告从 DataFrame 到 PDF 的自动化流水线5.1 用 jinja2 模板引擎解耦数据与排版硬编码matplotlib画图再pdfkit转PDF会导致每次改图表就要重写逻辑。我们用jinja2将数据渲染为HTML再用weasyprint转PDF——好处是前端同事可直接改CSS样式无需碰Python代码。!-- report_template.html -- !DOCTYPE html html head meta charsetutf-8 title{{ stock_name }} {{ report_date }} 分析报告/title style body { font-family: Microsoft YaHei, sans-serif; margin: 2cm; } .chart { width: 100%; height: 400px; border: 1px solid #eee; } .factor-table { width: 100%; border-collapse: collapse; } .factor-table th, .factor-table td { border: 1px solid #ccc; padding: 4px; } /style /head body h1{{ stock_name }} ({{ symbol }}) {{ report_date }} 分析报告/h1 h2核心指标/h2 table classfactor-table trth指标/thth值/th/tr trtdPE-TTM/tdtd{{ pe_ttm|round(2) }}/td/tr trtd股息率/tdtd{{ dividend_yield|round(4)*100 }}%/td/tr trtd20日均线/tdtd{{ ma20|round(2) }}/td/tr /table h2价格走势图/h2 img src{{ chart_path }} classchart / /body /html5.2 自动生成带水印的PDF用 weasyprint 插入公司标识与防伪码weasyprint支持CSSpage规则可添加页眉页脚。我们插入动态水印含生成时间哈希# report_generator.py from weasyprint import HTML, CSS import hashlib from datetime import datetime def generate_pdf_report(df: pd.DataFrame, symbol: str, output_path: str): # 准备数据 context { stock_name: 贵州茅台, symbol: symbol, report_date: datetime.now().strftime(%Y年%m月%d日), pe_ttm: df[pe_ttm].iloc[-1], dividend_yield: df[dividend_yield].iloc[-1], ma20: df[ma20].iloc[-1], chart_path: charts/price_chart.png # 先保存图表 } # 渲染HTML template open(report_template.html).read() html_content template.render(context) # 添加水印CSS watermark_css CSS(stringf page {{ bottom-center {{ content: Report ID: {hashlib.md5((symbol str(datetime.now())).encode()).hexdigest()[:8]} | Generated {datetime.now().strftime(%Y-%m-%d %H:%M)}; font-size: 10px; color: #aaa; }} }} ) # 生成PDF HTML(stringhtml_content).write_pdf( output_path, stylesheets[watermark_css] ) print(fPDF报告已生成{output_path}) # 调用 generate_pdf_report(df, 600519, reports/600519_20241231.pdf)关键设计Report ID由symbol datetime的MD5前8位生成确保每份报告唯一可追溯bottom-center位置固定在页脚中央避免与正文重叠font-size: 10px和color: #aaa保证水印不喧宾夺主但复印时仍可见。5.3 批量生成多股票报告用 concurrent.futures 加速IO密集型任务生成100只股票报告若串行执行每份耗时3秒图表渲染PDF生成总耗时5分钟。用进程池并发from concurrent.futures import ProcessPoolExecutor, as_completed def process_single_stock(symbol: str): try: df fetch_dual_source(symbol, 2023-01-01, 2024-12-31) df inject_dividends(df, symbol) df[dividend_yield] df[每股分红(税后)] / df[Close] df[ma20] rolling_sma_adjusted(df[Close], 20) generate_pdf_report(df, symbol, freports/{symbol}_20241231.pdf) return f{symbol}: success except Exception as e: return f{symbol}: failed - {str(e)} # 并发执行 symbols [600519, 000858, 300750, 601318] # 实际读取自CSV with ProcessPoolExecutor(max_workers4) as executor: futures [executor.submit(process_single_stock, s) for s in symbols] for future in as_completed(futures): print(future.result()) # 输出示例 # 600519: success # 000858: success # 300750: failed - ValueError: 双源价格校验失败...为什么用ProcessPoolExecutor而非ThreadPoolExecutorweasyprint和matplotlib是CPU密集型PDF渲染、图像绘制GIL会阻塞线程max_workers4是经验阈值超过4个进程会导致内存竞争反而降低吞吐as_completed保证失败任务即时反馈不因某只股票报错而中断全部流程。6. 我的血泪经验三个让分析真正落地的细节技巧6.1 把「无效日期」显式标为 NaN而不是用 fillna 填充——这是调试的后悔药新手常犯的错看到df[Close]有NaN立刻df[Close].fillna(methodffill)。这会让调试变得地狱级困难——你永远不知道这个值是真实停牌填充的还是原始数据缺失。我的做法是所有数据源接入后先用pd.isna()标记原始缺失再用df[Close].mask(invalid_mask, np.nan)显式置空# 正确姿势保留原始缺失痕迹 invalid_mask (df[Volume] 0) | (df[Close].isna()) # 量为0且价为空判定为无效 df[Close_raw] df[Close].copy() # 保留原始列 df[Close] df[Close].mask(invalid_mask, np.nan) # 显式置空 # 后续fill时加注释说明原因 df[Close] df[Close].ffill() # 停牌日填充符合交易所规则这样当你发现某日Close异常时可以快速定位若Close_raw也是NaN → 数据源本身缺失若Close_raw有值但Close被填充 → 是你主动填充的检查填充逻辑若Close_raw有值且Close未变 → 是有效交易日。没有这一步你在回测里看到一个诡异的收益峰值可能要花两小时排查是数据问题还是策略bug。6.2 用 pytest 写数据校验测试而不是靠肉眼盯Excel我见过太多团队把数据校验写成“打开Excel看一眼”。正确的做法是为每个数据源写独立的pytest测试失败时直接打印差异行# test_data_integrity.py import pytest import pandas as pd def test_yfinance_akshare_alignment(): df fetch_dual_source(600519, 2023-01-01, 2023-01-31) # 校验非分红日yfinance Close 与 akshare close_adj 偏差 0.05 non_div_mask df[每股分红(税后)] 0 diffs abs(df.loc[non_div_mask, Close] - df.loc[non_div_mask, close_adj]) assert diffs.max() 0.05, f最大偏差 {diffs.max():.3f} 0.05异常行{diffs.idxmax()} # 运行pytest test_data_integrity.py -v # 输出FAILED test_data_integrity.py::test_yfinance_akshare_alignment - AssertionError: 最大偏差 0.123 0.05异常行2023-01-15这个测试每天凌晨自动运行一旦失败企业微信机器人立刻推送告警并附上diffs.nlargest(3)的具体日期和偏差值。比人工抽查可靠100倍。6.3 在因子计算中强制加入「业务含义注释」让三个月后的自己能看懂写因子公式时别只写df[factor] (df[roa] - df[roa].shift(1)) / df[roa].shift(1)。要像这样# ✅ 正确每行都有业务注释 df[roa_qoq_change] ( df[roa] - df[roa].shift(1) # 当前季度ROA - 上季度ROA ) / df[roa].shift(1) # 除以上季度ROA得环比增长率 # 注ROA净利润/总资产此处用TTM数据故上季度实为上期TTM # 来源akshare.stock_fundamental_em(600519) 的 roa 字段为什么重要因为三个月后你翻代码看到roa_qoq_change第一反应是“这是季报数据吗”而不是“这玩意儿到底怎么算的”。我在某跨平台系统里见过一个叫score_v3的因子没人知道v1和v2是什么最后花了两天时间反向工程出它其实是PE倒数乘以ROE——这种黑匣子早该在诞生时就被注释封印。希望帮到你。本文还有配套的精品资源点击获取