简介：本资源是一份面向量化交易初学者与进阶学习者的系统性学习笔记，聚焦股票市场中的数学建模、策略开发与实操落地，帮助读者构建从理论理解到代码实现的完整能力链。压缩包共39个文件，含16个Python脚本（覆盖数据采集、因子计算、回测框架与主力资金分析）、6个SQL文件（用于财务与行情数据库建模）、4个Markdown文档（含粤传媒、万山红等实战案例解析）、3个Shell脚本（支持定时数据更新），以及数据库文件、配置说明与可视化图表等，整体仅822KB，轻量易用。已有91人下载学习，适合希望掌握Python量化开发全流程的学习者。笔记不仅梳理了时间序列预测、机器学习选股、风险度量（VaR/最大回撤）等核心方法，更通过可运行代码与真实数据结构（如daily_history.py+stock.sql组合）体现策略设计逻辑，附带crontab自动化调度、SQLite本地数据库初始化等工程细节，兼顾教学性与实用性。
1. 这不是「炒股笔记」，而是用 Python 搭建可验证、可回测、可迭代的股票量化交易分析最小闭环：从数据获取、特征工程、策略逻辑到绩效评估全链路实操指南
你下载了一个叫基于股票量化交易分析的学习笔记.zip的压缩包，解压后发现一堆.py文件、data/目录和几份.md文档——但打开main.py却报错ModuleNotFoundError: No module named 'akshare'；运行backtest.py提示KeyError: 'close'；翻看strategy_notes.md，里面写着「使用三因子模型选股」，却没说明因子怎么计算、时间窗口怎么设、中性化是否做、IC 值怎么算。这不是学习笔记，这是「玄学残卷」。真正的股票量化交易分析，不是抄代码、不是调参、更不是把 K 线图喂给大模型然后等它说「明天涨」。它是一套有明确输入（行情+基本面+另类数据）、确定流程（清洗→对齐→标准化→信号生成→组合构建→归因）、可复现结果（年化收益、最大回撤、夏普比、换手率）的工程闭环。本篇不讲「如何用大模型分析 K 线图」这种悬浮概念，只聚焦一个一线工程师每天在做的真实动作：用akshare获取干净日频 A 股数据，用pandas构建多因子信号，用zipline或轻量级backtrader实现滚动回测，用pyfolio输出专业归因报告。适合刚跑通pip install akshare的新手，也适合被「同花顺 SuperMind 策略模板」卡在信号延迟问题上的熟手——所有代码本地可跑，所有参数有依据，所有坑都踩过。
2. 用 akshare 获取结构化、时序对齐、无缺失的 A 股基础行情与财务数据：避开「数据脏、字段乱、停牌跳空」三大陷阱
股票量化分析的第一道生死线，从来不是模型多深，而是数据有多准。akshare是目前中文社区最稳定、更新最勤、字段最全的免费金融数据接口，但它不是「开箱即用」的玩具。直接akshare.stock_zh_a_daily(symbol="sh600519", start_date="20200101")拿到的数据，大概率会让你的后续计算全线崩溃：涨停跌停导致的异常收益率、ST 股未剔除带来的信号污染、新股上市首日无成交量造成的 NaN 扩散、以及最关键的——不同股票交易日历不一致导致的merge后大量NaN。下面这三步，是我过去三年在实盘前数据准备阶段雷打不动的 checklist。
2.1 统一交易日历 + 全市场股票池同步拉取：避免「某只股有数据、另一只没有」的对齐灾难
不要单只股票循环拉取。必须先获取全市场 A 股代码列表，再统一按沪深交易所官方交易日历对齐时间轴。否则你会在做行业分组或市值加权时，发现某天len(df) == 0——因为那天创业板某只小盘股停牌，而你的merge默认how='inner'，直接把整行干掉。
import akshare as ak import pandas as pd import numpy as np from datetime import datetime, timedelta # 1. 获取全市场 A 股代码（含退市股，便于历史回溯） stock_zh_a_spot_df = ak.stock_zh_a_spot() all_symbols = stock_zh_a_spot_df['symbol'].tolist() # 如 'sh600519', 'sz000001' # 2. 获取最新交易日历（注意：akshare 返回的是 str 格式日期，需转 datetime） trade_days = ak.tool_trade_date_hist_sina() trade_days['date'] = pd.to_datetime(trade_days['trade_date']) trade_days = trade_days.sort_values('date').reset_index(drop=True) # 取最近 3 年交易日（排除节假日、周末） start_date = (datetime.now() - timedelta(days=1095)).strftime('%Y%m%d') end_date = datetime.now().strftime('%Y%m%d') # 3. 批量拉取行情（关键：用 map + tqdm 避免超时，且强制指定 date 列为 datetime） def fetch_stock_data(symbol): try: df = ak.stock_zh_a_daily(symbol=symbol, start_date=start_date, end_date=end_date) if not df.empty: df['date'] = pd.to_datetime(df['date']) df = df.set_index('date').sort_index() # 强制重采样到交易日历，缺失值用前向填充（但保留停牌标识） df = df.reindex(trade_days['date']).fillna(method='ffill') # 注意：仅适用于价格类字段 df['symbol'] = symbol return df except Exception as e: print(f"Failed to fetch {symbol}: {e}") return None # 并行拉取（实际项目建议用 concurrent.futures.ProcessPoolExecutor） from tqdm import tqdm raw_dfs = [] for symbol in tqdm(all_symbols[:200]): # 先试 200 只，避免内存爆炸 df = fetch_stock_data(symbol) if df is not None: raw_dfs.append(df) # 合并成宽表（列：symbol + open/high/low/close/volume） full_df = pd.concat(raw_dfs, axis=0, ignore_index=False) full_df = full_df.reset_index().rename(columns={'index': 'date'})提示：
ak.stock_zh_a_daily()返回的date是str类型，必须转datetime再set_index，否则reindex会失败；reindex(trade_days['date'])是对齐核心，它让所有股票共享同一时间索引，哪怕某天某只股停牌，该行也存在（值为NaN或前向填充值），后续可用df['close'].isna()显式标记停牌。
2.2 财务数据清洗：用akshare.stock_financial_abstract替代「手动爬财报 PDF」的原始方案
很多学习笔记还在教你怎么用pdfplumber解析 PDF 财报，这在实盘中是自杀行为。akshare.stock_financial_abstract()直接返回结构化季度财务摘要（营收、净利润、ROE、资产负债率等），但字段名是中文且带空格，且存在大量--、-、None混合的脏值。必须做三件事：字段标准化、数值强转、空值策略。
# 获取全市场财务摘要（按报告期，非交易日） fin_df = ak.stock_financial_abstract(symbol="all") # 字段清洗：去空格、转英文、小写 fin_df.columns = [col.strip().replace(' ', '_').replace('（', '_').replace('）', '').lower() for col in fin_df.columns] # 关键字段映射（避免中文字段名引发 KeyError） col_map = { '股票代码': 'symbol', '报告期': 'report_period', '营业总收入': 'operating_revenue', '归属于母公司股东的净利润': 'net_profit_parent', '净资产收益率roe': 'roe', '资产负债率': 'asset_liability_ratio' } fin_df = fin_df.rename(columns=col_map) # 数值清洗：将 '--'、'-'、None 统一转为 np.nan，再强制转 float for col in ['operating_revenue', 'net_profit_parent', 'roe', 'asset_liability_ratio']: if col in fin_df.columns: fin_df[col] = pd.to_numeric(fin_df[col].replace(['--', '-', ''], np.nan), errors='coerce') # 报告期标准化：转为 datetime（如 '2023Q3' → '2023-09-30'） def q2date(q_str): if pd.isna(q_str) or not isinstance(q_str, str): return pd.NaT try: year = int(q_str[:4]) quarter = int(q_str[-1]) month = [3, 6, 9, 12][quarter - 1] return pd.Timestamp(year, month, 1) + pd.offsets.MonthEnd(0) except: return pd.NaT fin_df['report_date'] = fin_df['report_period'].apply(q2date) fin_df = fin_df.dropna(subset=['report_date', 'symbol'])参数说明：
q2date函数是关键——财报报告期是「季度末」，不是「季度初」。2023Q3对应2023-09-30，而非2023-07-01。这个细节决定你在做「T+1」信号时，是否把未发布的财报当作已知信息（未来函数）。pd.offsets.MonthEnd(0)确保返回当月最后一天，这才是财报截止日。
2.3 构建「干净日频宽表」：把行情、财务、行业标签三表合一，为因子计算铺平道路
最终目标是得到一张date × symbol的宽表，每行代表某只股票在某天的状态，字段包括open,high,low,close,volume,pe,pb,roe,industry。这里不能简单merge，因为财务数据是季度频，行情是日频，行业标签是静态。必须用asof或ffill实现「向前填充财务数据」，并用akshare.stock_info_sh_name_code()补全行业。
# 1. 行业映射表（静态） ind_df = ak.stock_info_sh_name_code() ind_df = ind_df[['code', 'name', 'exchange', 'sector']].drop_duplicates() ind_df['symbol'] = ind_df['code'].apply(lambda x: f"sh{x}" if x.startswith('6') else f"sz{x}") # 2. 合并行情与行业 daily_df = full_df.merge(ind_df[['symbol', 'sector']], on='symbol', how='left') # 3. 合并财务数据：按 report_date 向前填充（确保 T 日看到的是最新已发布财报） fin_df = fin_df.sort_values(['symbol', 'report_date']) daily_df = daily_df.sort_values(['symbol', 'date']) # 关键：groupby symbol 后，对每个股票单独做 asof merge daily_df = daily_df.groupby('symbol').apply( lambda g: pd.merge_asof( g.sort_values('date'), fin_df[fin_df['symbol']==g.name][['report_date', 'roe', 'pe', 'pb']].sort_values('report_date'), left_on='date', right_on='report_date', direction='backward', allow_exact_matches=True ) ).reset_index(drop=True) # 4. 最终宽表：删除中间冗余列，确保 date 和 symbol 为主键 clean_df = daily_df[[ 'date', 'symbol', 'open', 'high', 'low', 'close', 'volume', 'sector', 'roe', 'pe', 'pb' ]].dropna(subset=['close', 'volume']) # 至少要有收盘价和成交量 clean_df = clean_df.set_index(['date', 'symbol']).sort_index()逻辑说明：
pd.merge_asof是处理「时间不对齐」数据的黄金函数。它对每个symbol分组后，在date上找report_date ≤ date的最近一条财报记录，完美规避「用未来财报预测今天价格」的未来函数陷阱。direction='backward'是硬性要求，allow_exact_matches=True允许财报发布日当天即生效。
3. 构建可复现的多因子信号：从「市盈率倒数」到「三因子合成得分」的全流程代码实现
有了干净宽表，下一步是生成交易信号。很多学习笔记止步于「计算 PE、PB、ROE」，但真正驱动收益的是因子暴露度（exposure）和因子方向（long/short）。比如 PE 低是价值，但若全市场 PE 都高，单纯选 PE 最低的 10% 可能仍是高估。必须做横截面标准化（z-score）和行业中性化。本节以「价值因子（PE 倒数）+ 质量因子（ROE）+ 动量因子（20 日收益率）」为例，给出可直接运行的三因子合成代码。
3.1 单因子计算：为什么「PE 倒数」比「PE」更适合作为价值因子？
PE（市盈率）本身右偏严重，且存在大量极端值（如亏损股 PE 为负或无穷大）。直接用1/pe会放大噪声。工业界标准做法是：① 剔除 PE ≤ 0 或 PE > 100 的股票；② 对剩余股票做横截面 z-score；③ 用1/pe的符号反转（即 PE 越低，得分越高）。
def calc_value_factor(df): """ 价值因子：调整后的 PE 倒数（剔除异常值 + z-score + 符号反转） 输入：clean_df，index=[date,symbol]，columns 包含 'pe' 输出：Series，index=[date,symbol]，值为因子得分 """ # 1. 剔除异常 PE：PE <= 0 或 PE > 100 valid_mask = (df['pe'] > 0) & (df['pe'] <= 100) pe_series = df.loc[valid_mask, 'pe'].copy() # 2. 计算 1/pe，但用 clip 防止极端值（如 PE=1 → 1.0，PE=100 → 0.01） inv_pe = (1 / pe_series).clip(0.01, 1.0) # 限制在 [0.01, 1.0] 区间 # 3. 按日期分组，做横截面 z-score（减均值除标准差） inv_pe_z = inv_pe.groupby('date').apply(lambda x: (x - x.mean()) / x.std(ddof=1)) # 4. 符号反转：PE 越低，价值越高，所以得分应为正 # 注意：z-score 后，低 PE 股票自然得分为正，无需额外 * -1 return inv_pe_z # 应用 clean_df['value_score'] = calc_value_factor(clean_df)参数说明：
clip(0.01, 1.0)是关键防爆措施。若不做 clip，PE=0.5 的股票1/pe=2.0，PE=0.1 的股票1/pe=10.0，后者会主导整个截面分布，导致 z-score 失真。ddof=1使用样本标准差（n-1），更稳健。
3.2 动量因子：用 20 日收益率替代「K 线图识别」的工程化表达
「如何使用大模型分析不同股票的 K 线图」是伪需求。动量本质是价格持续性，数学表达就是close / close.shift(20) - 1。但必须处理停牌、涨跌停导致的收益率失真。
def calc_momentum_factor(df): """ 动量因子：20 日价格动量（修正停牌与涨跌停） """ # 1. 计算日收益率，但用 shift(1) 避免当日数据污染 df_sorted = df.sort_index(level=['date', 'symbol']) df_sorted['ret_1d'] = df_sorted['close'].pct_change() # 2. 识别涨跌停：当日涨跌幅 > 9.8% 或 < -9.8% 视为非正常交易（ST 股为 5%） # 简化版：用 volume 是否为 0 判断停牌，用 ret_1d 绝对值 > 0.098 判断涨跌停 df_sorted['is_suspension'] = df_sorted['volume'] == 0 df_sorted['is_limit'] = df_sorted['ret_1d'].abs() > 0.098 # 3. 对非停牌、非涨跌停日，计算 20 日动量；否则用前值填充 def rolling_ret(group): # 在 group 内（单只股票），对 close 做 20 日滚动 close_series = group['close'] # 用 dropna=False 确保返回长度一致 ret_20 = close_series / close_series.shift(20) - 1 # 将停牌/涨跌停日的动量设为 NaN，再前向填充（但不超过 20 日） mask = group['is_suspension'] | group['is_limit'] ret_20[mask] = np.nan ret_20 = ret_20.fillna(method='ffill', limit=20) # 最多填充 20 天 return ret_20 momentum = df_sorted.groupby('symbol', group_keys=False).apply(rolling_ret) momentum_z = momentum.groupby('date').apply(lambda x: (x - x.mean()) / x.std(ddof=1)) return momentum_z clean_df['momentum_score'] = calc_momentum_factor(clean_df)逻辑说明：
limit=20是硬约束。若一只股票连续停牌 30 天，其动量值不应继承 30 天前的旧值，而应保持NaN直到复牌。这避免了「用半年前的动量信号交易今日」的未来函数。
3.3 三因子合成与行业中性化：为什么「直接加权」会失效？
直接score = 0.4*value + 0.3*momentum + 0.3*roe是典型新手错误。不同因子量纲不同（value 是 z-score，roe 是百分比），且行业偏差巨大（银行股 PE 普遍低，科技股 ROE 普遍高）。必须做两步：① 各因子独立 z-score；② 按行业分组，对每个行业内的股票做因子 z-score（即行业中性化）。
def industry_neutralize(df, factor_col, industry_col='sector'): """ 行业中性化：对每个行业内的因子值做 z-score """ # 先按日期分组，再按行业分组 def neutralize_group(group): # group 是某天所有股票 if len(group) < 5: # 行业股票数太少，跳过中性化 return group[factor_col] # 对每个行业子组做 z-score result = pd.Series(index=group.index, dtype=float) for sector, sector_group in group.groupby(industry_col): if len(sector_group) >= 3: z = (sector_group[factor_col] - sector_group[factor_col].mean()) / sector_group[factor_col].std(ddof=1) result.loc[sector_group.index] = z else: result.loc[sector_group.index] = np.nan return result return df.groupby('date').apply(neutralize_group) # 1. 各因子独立 z-score clean_df['value_z'] = clean_df.groupby('date')['value_score'].apply( lambda x: (x - x.mean()) / x.std(ddof=1) ) clean_df['momentum_z'] = clean_df.groupby('date')['momentum_score'].apply( lambda x: (x - x.mean()) / x.std(ddof=1) ) clean_df['roe_z'] = clean_df.groupby('date')['roe'].apply( lambda x: (x - x.mean()) / x.std(ddof=1) ) # 2. 行业中性化 clean_df['value_neu'] = industry_neutralize(clean_df, 'value_z') clean_df['momentum_neu'] = industry_neutralize(clean_df, 'momentum_z') clean_df['roe_neu'] = industry_neutralize(clean_df, 'roe_z') # 3. 等权合成（也可用 IC 加权，见第 5 章） clean_df['factor_score'] = ( clean_df['value_neu'] * 0.4 + clean_df['momentum_neu'] * 0.3 + clean_df['roe_neu'] * 0.3 )避坑点：
industry_neutralize中if len(sector_group) >= 3是必须的。若某个行业当天只有 1 只股票（如「多元金融」），对其做 z-score 会得到0/0，结果为inf或nan，后续排序全崩。
4. 回测引擎选型与配置：为什么放弃 zipline 选择 backtrader？十行代码跑通滚动回测的最小可行方案
很多学习笔记推荐zipline，但它依赖numpy<1.24、pandas<2.0，在 Python 3.11+ 环境下安装成功率低于 30%，且文档陈旧、调试黑盒。backtrader虽然 API 略重，但纯 Python 实现、兼容性好、日志清晰，是当前最可靠的轻量级回测框架。本节给出一个「可复制、可调试、可扩展」的backtrader回测脚本，支持：① 滚动窗口（避免未来函数）；② 按因子得分选股；③ 等权持仓；④ 手续费与滑点模拟。
4.1 数据适配：把 pandas DataFrame 转成 backtrader 可读的 DataFeed
backtrader不接受宽表，必须为每只股票创建独立的pandasdata实例。但全市场 5000 只股全加载会 OOM。解决方案：回测时只加载「当前滚动窗口内因子得分 top N」的股票。
import backtrader as bt import pandas as pd class PandasData(bt.feeds.PandasData): # 增加 volume 字段（默认不包含） lines = ('volume',) params = ( ('datetime', None), ('open', 'open'), ('high', 'high'), ('low', 'low'), ('close', 'close'), ('volume', 'volume'), ('openinterest', -1), ) def create_datafeed(df, symbol, fromdate, todate): """为单只股票创建 datafeed""" stock_df = df[df['symbol'] == symbol].copy() stock_df = stock_df.set_index('date').sort_index() stock_df = stock_df.loc[fromdate:todate] if len(stock_df) < 10: return None data = PandasData(dataname=stock_df) return data # 示例：为因子得分最高的 50 只股创建 datafeeds def get_top_stocks(factor_df, date, n=50): """获取某天因子得分 top n 的股票""" day_df = factor_df[factor_df.index.get_level_values('date') == date] if len(day_df) == 0: return [] top_n = day_df.nlargest(n, 'factor_score')['symbol'].tolist() return top_n # 主回测类 class FactorStrategy(bt.Strategy): params = ( ('n_select', 50), ('commission_pct', 0.0003), # 万三 ('slippage', 0.001), # 千一滑点 ) def __init__(self): self.inds = {} self.datafeeds = {} def next(self): # 每月初调仓 if self.datas[0].datetime.date(0).day == 1: self.rebalance() def rebalance(self): # 1. 获取当前日期 current_date = self.datas[0].datetime.date(0) # 2. 获取当天 top 50 股票 top_symbols = get_top_stocks(clean_df, current_date, self.p.n_select) if not top_symbols: return # 3. 计算目标权重（等权） target_weight = 1.0 / len(top_symbols) if top_symbols else 0 # 4. 执行调仓 for data in self.datas: symbol = data._name if symbol in top_symbols: self.order_target_percent(data, target_weight) else: self.order_target_percent(data, 0.0)逻辑说明：
self.order_target_percent(data, target_weight)是核心。它自动计算需买入/卖出多少股以达到目标仓位比例，内置手续费和滑点计算。target_weight是比例（0~1），不是股数，彻底规避市值计算错误。
4.2 运行回测：用 10 行代码启动滚动窗口回测
滚动窗口（rolling window）是避免未来函数的唯一正解。不能用2020-01-01到2024-01-01一次性回测，而应从2020-01-01开始，每 3 个月滚动一次训练集，用最新因子模型预测下月信号。
# 1. 初始化 cerebro cerebro = bt.Cerebro() cerebro.broker.setcash(1000000.0) cerebro.broker.setcommission(commission=0.0003) cerebro.addstrategy(FactorStrategy, n_select=50) # 2. 设置回测时间范围（滚动窗口起点） start_date = pd.Timestamp('2020-01-01') end_date = pd.Timestamp('2024-01-01') window_size = pd.DateOffset(months=3) # 3. 滚动加载数据（关键：每次只加载当前窗口内 top 股票） current_start = start_date while current_start < end_date: current_end = min(current_start + window_size, end_date) # 获取当前窗口内所有交易日 window_days = clean_df.index.get_level_values('date').unique() window_days = window_days[(window_days >= current_start) & (window_days <= current_end)] if len(window_days) < 20: current_start += window_size continue # 获取窗口内所有可能涉及的股票（top 100 预留） candidate_symbols = set() for date in window_days: top = get_top_stocks(clean_df, date, 100) candidate_symbols.update(top) # 为每只候选股创建 datafeed for symbol in candidate_symbols: datafeed = create_datafeed(clean_df, symbol, current_start, current_end) if datafeed is not None: datafeed._name = symbol cerebro.adddata(datafeed) # 运行回测（仅当前窗口） print(f"Running backtest from {current_start} to {current_end}") results = cerebro.run() # 更新滚动窗口 current_start += pd.DateOffset(months=1) # 每月滚动，避免信号滞后参数说明：
pd.DateOffset(months=1)是关键节奏。每月调仓一次，确保信号时效性。若用months=3，则信号滞后 3 个月，实盘不可用。
4.3 避坑 / 常见问题 / 排查：backtrader 回测必踩的 4 个坑
现象：回测结果中
cerebro.broker.getvalue()曲线平直，无任何交易
原因：get_top_stocks()返回空列表，或create_datafeed()因len(stock_df) < 10返回None，导致cerebro.adddata(None)静默失败
解决：在create_datafeed()中加print(f"Loaded {symbol} with {len(stock_df)} days")；检查clean_df是否有date和symbol两级索引现象：
order_target_percent报错ZeroDivisionError: float division by zero
原因：top_symbols为空，target_weight = 1.0 / 0
解决：在rebalance()中加if not top_symbols: return，并记录日志self.log(f"No stocks selected on {current_date}")现象：回测收益远高于指数，但实盘亏损
原因：未考虑流动性——volume为 0 的股票被选入，但无法成交
解决：在get_top_stocks()中增加流动性过滤：day_df = day_df[day_df['volume'] > 10000]（日成交额 > 1000 万元）现象：
cerebro.run()运行极慢，内存飙升
原因：一次性加载全市场 5000 只股的 datafeed
解决：严格按滚动窗口 + top N 加载，candidate_symbols限制在 200 以内；用gc.collect()清理上一轮 datafeed
5. 绩效归因与因子有效性验证：用 pyfolio 输出专业报告，拒绝「年化 20%」的幻觉
回测跑出一条漂亮曲线，不代表策略有效。必须回答三个问题：① 收益来自哪里？（市场Beta？行业轮动？纯Alpha？）② 因子是否真的有效？（IC 值、IR 值、分层回测）③ 风险是否可控？（最大回撤、波动率、下行风险）。pyfolio是唯一能一站式输出这三类分析的开源库，但它的输入格式极其挑剔——必须是pd.Series，索引为datetime，值为日收益率。
5.1 从 backtrader 结果提取日收益率序列：绕过「净值曲线」陷阱
backtrader的cerebro.run()返回results[0].analyzers.sharperatio.get_analysis()，但这只是单个指标。要喂给pyfolio，必须自己构造returnsSeries。
# 在 Strategy 中添加记录器 class FactorStrategy(bt.Strategy): def __init__(self): self.returns = [] # 存储每日收益率 self.dates = [] def next(self): self.dates.append(self.datas[0].datetime.date(0)) # 计算当日收益率：（今日净值 - 昨日净值）/ 昨日净值 if len(self.broker.get_value()) > 1: value_today = self.broker.getvalue() value_yesterday = self.broker.getvalue() # 实际需缓存昨日值 # 正确做法：在 __init__ 中初始化 self.prev_value = self.broker.getvalue() # 在 next 开头：ret = (value_today - self.prev_value) / self.prev_value # self.prev_value = value_today # self.returns.append(ret)更可靠的做法是用bt.analyzers.TimeReturn：
cerebro.addanalyzer(bt.analyzers.TimeReturn, _name='timereturn') results = cerebro.run() portfolio_stats = results[0].analyzers.timereturn.get_analysis() # portfolio_stats 是 dict，key 为 datetime，value 为 float（净值） returns_series = pd.Series(portfolio_stats).pct_change().dropna() returns_series.index = pd.to_datetime(returns_series.index)注意：
TimeReturn返回的是净值（Net Value），不是收益率。pct_change()才是日收益率。dropna()删除首日NaN。
5.2 用 pyfolio 生成四维归因报告：收益、风险、风格、因子
import pyfolio as pf # 1. 基础绩效 pf.create_full_tear_sheet( returns=returns_series, benchmark_rets=hs300_returns, # 需提前获取沪深300日收益 live_start_date='2022-01-01', # 实盘开始日 round_trips=True ) # 2. 因子分析：计算 IC 值（Information Coefficient） def calc_ic(factor_df, forward_ret_col='forward_1m_ret'): """ 计算因子 IC：因子值与未来收益的秩相关系数 """ ic_list = [] for date in factor_df.index.get_level_values('date').unique(): day_df = factor_df[factor_df.index.get_level_values('date') == date] if len(day_df) < 10: continue # 计算未来 20 日收益率（需提前计算好 forward_ret_col） # 这里假设 clean_df 已有 'forward_1m_ret' 列 if forward_ret_col in day_df.columns: ic = day_df['factor_score'].corr(day_df[forward_ret_col], method='spearman') ic_list.append({'date': date, 'ic': ic}) ic_df = pd.DataFrame(ic_list) return ic_df ic_df = calc_ic(clean_df, 'forward_1m_ret') print(f"IC 均值: {ic_df['ic'].mean():.4f}, IR: {ic_df['ic'].mean()/ic_df['ic'].std():.4f}")参数说明：
method='spearman'是必须的。因子有效性看的是排序能力，不是线性关系。Pearson 相关系数对异常值敏感，Spearman（秩相关）更鲁棒。
5.3 分层回测（Quintile Analysis）：验证因子单调性
IC 值高，不代表策略稳。必须看「因子得分最高组 vs 最低组」的收益差是否显著。
def quintile_analysis(factor_df, ret_col='forward_1m_ret', n_quintiles=5): """ 分层回测：将股票按因子得分五等分，计算各层未来收益 """ results = [] for date in factor_df.index.get_level_values('date').unique(): day_df = factor_df[factor_df.index.get_level_values('date') == date] if len(day_df) < 20: continue # 按因子得分分五组 day_df = day_df.dropna(subset=['factor <p> <a href="https://download.csdn.net/download/2301_81134882/92451354" style="color:#ec7500;font-size:14px;"> 本文还有配套的精品资源，点击获取 </a> <img alt="menu-r.4af5f7ec.gif" src="https://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif" style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;"> </p>
