## 概述 本框架是一个基于聚宽数据的量化策略回测模板，提供了完整的策略开发、回测和分析流程。该框架采用模块化设计，只需替换核心策略逻辑即可快速进行策略回测。为解决研究环境内的回测，写了这个模板供大家参考。 ## 框架特性 - **完整的回测流程**：包含数据获取、策略执行、收益计算等完整环节 - **灵活的调仓频率**：支持日度、周度、月度等不同调仓频率 - **风险控制**：内置ST股、停牌股过滤机制 - **交易成本**：考虑换仓成本的收益计算 - **标准化输出**：统一的回测结果格式 ## 核心组件说明 ### 1. 配置参数 ```python start_date = datetime.date(2020, 1, 1) # 回测开始日期 end_date = datetime.date(2024, 11, 29) # 回测结束日期 investment_horizon = 'W' # 调仓频率：'D'(日)、'W'(周)、'M'(月) number_of_periods_per_year = 52 # 年化频率：252(日)、52(周)、12(月) ``` ### 2. 股票池过滤函数 ```python def get_st_or_paused_stock_set(decision_date): ``` 自动过滤ST股票和停牌股票，确保交易可行性。 ### 3. **核心策略函数** ```python def cal_portfolio_weight_series(decision_date, old_portfolio_weight_series): ``` 这是策略的核心逻辑函数，可以根据自己的策略逻辑进行替换。 **函数参数说明：** - `decision_date`：决策日期（调仓日前一个交易日） - `old_portfolio_weight_series`：当前持仓权重序列 **函数返回值：** - `new_portfolio_weight_series`：新的持仓权重序列 ## 如何替换策略？ ### 步骤1：理解策略框架 原模板采用多因子选股+技术面择时的策略： - **股票池**：中证1000成分股 - **初筛**：基于基本面因子（市值、市盈率、流动性等） - **择时**：基于布林带突破信号 ### 步骤2：替换策略逻辑 在 `cal_portfolio_weight_series` 函数中替换以下关键部分： #### 2.1 定义股票池 ```python # 原代码：中证1000成分股 all_stocks = get_index_stocks('000852.XSHG', date=decision_date) # 替换为你的股票池 all_stocks = get_index_stocks('000300.XSHG', date=decision_date) # 沪深300 # 或自定义股票列表 all_stocks = ['000001.XSHE', '000002.XSHE', ...] ``` #### 2.2 替换选股逻辑 ```python # 原代码：多因子筛选 factor_df_filtered = factor_df.loc[ (factor_df.loc[:, 'book_to_price_ratio']>factor_df.loc[:, 'book_to_price_ratio'].quantile(0.3)) &\ # ... 其他条件 ] # 替换为你的选股逻辑 if your_strategy_condition: selected_stocks = apply_your_selection_method(all_stocks) else: selected_stocks = all_stocks ``` #### 2.3 替换持仓构建逻辑 ```python # 原代码：基于技术指标择时 for next_stock in all_stocks: if technical_condition: stocks_to_buy.append(next_stock) # 替换为你的持仓构建方法 new_portfolio_weight_series = your_position_sizing_method(selected_stocks) ``` ### 步骤3：策略示例 #### 示例1：价值投资策略 ```python def cal_portfolio_weight_series(decision_date, old_portfolio_weight_series): # 选择中证800成分股 all_stocks = get_index_stocks('000906.XSHG', date=decision_date) all_stocks = list(set(all_stocks) - get_st_or_paused_stock_set(decision_date)) # 获取价值因子 pe_ratio = get_factor_values(securities=all_stocks, factors='pe_ratio', start_date=decision_date, end_date=decision_date)['pe_ratio'].T pb_ratio = get_factor_values(securities=all_stocks, factors='pb_ratio', start_date=decision_date, end_date=decision_date)['pb_ratio'].T # 选择低估值股票 selected_stocks = pe_ratio[pe_ratio < pe_ratio.quantile(0.2)].index.intersection( pb_ratio[pb_ratio < pb_ratio.quantile(0.2)].index).tolist() # 等权重分配 new_portfolio_weight_series = pd.Series(1.0/len(selected_stocks) if selected_stocks else 0, index=selected_stocks) return new_portfolio_weight_series ``` #### 示例2：动量策略 ```python def cal_portfolio_weight_series(decision_date, old_portfolio_weight_series): # 获取股票池 all_stocks = get_index_stocks('000300.XSHG', date=decision_date) all_stocks = list(set(all_stocks) - get_st_or_paused_stock_set(decision_date)) # 计算20日收益率 prices = get_price(all_stocks, count=21, end_date=decision_date, fields=['close'], fq='post', panel=False) returns_20d = prices.groupby('code')['close'].apply(lambda x: x.iloc[-1]/x.iloc[0] - 1) # 选择高动量股票 top_momentum = returns_20d.nlargest(20).index.tolist() # 等权重分配 new_portfolio_weight_series = pd.Series(1.0/20, index=top_momentum) return new_portfolio_weight_series ``` ## 运行回测 ```python # 执行回测 wealth_process, allocation_dict = simulate_wealth_process(start_date, end_date) # 查看结果 print("累计收益:", wealth_process.iloc[-1] - 1) print("年化收益率:", (wealth_process.iloc[-1]/wealth_process.iloc[0])**(252/len(wealth_process)) - 1) ``` ## 扩展功能 可以根据需要扩展以下功能： - **风险指标计算**：夏普比率、最大回撤、波动率等 - **策略分析**：换手率、行业分布、因子暴露等 - **参数优化**：网格搜索、遗传算法等参数调优方法