基于 Python 和 Qlib 的 TopkDropout 股票回测系统实现

1. 导入库和模块

from pprint import pprint
import qlib
import pandas as pd
from qlib.utils.time import Freq
from qlib.utils import flatten_dict
from qlib.contrib.evaluate import risk_analysis
from qlib.contrib.strategy import TopkDropoutStrategy
from qlib.backtest import backtest, executor

• pprint：美化输出字典或复杂数据结构
• qlib：量化金融库，提供数据管理和回测功能
• pandas：用于数据处理和分析
• Freq：时间频率解析工具
• flatten_dict：展平嵌套字典
• risk_analysis：计算风险指标（年化收益、波动率等）
• TopkDropoutStrategy：基于预测得分选择并定期调整股票组合的策略
• backtest, executor：回测执行模块

2. 初始化 Qlib

qlib.init(provider_uri=<qlib data dir>)

• 加载股票数据（价格、交易量等）
• <qlib data dir>需替换为实际数据路径

3. 基准指数设置

CSI300_BENCH = "SH000300"

• 使用沪深300指数（SH000300）作为基准
• 用于计算超额回报(excess return)

4. 策略配置

FREQ = "day"
STRATEGY_CONFIG = {
    "topk": 50,
    "n_drop": 5,
    "signal": pred_score,
}

• 交易频率：每日(day)
• 策略参数：
  • topk：选择预测得分前50的股票
  • n_drop：定期剔除表现最差的5只股票
  • signal：需用户提供预测得分数据(pred_score)

5. 执行器配置

EXECUTOR_CONFIG = {
    "time_per_step": "day",
    "generate_portfolio_metrics": True,
}

• 交易步长：每日
• 生成投资组合指标

6. 回测参数

backtest_config = {
    "start_time": "2017-01-01",
    "end_time": "2020-08-01",
    "account": 100000000,
    "benchmark": CSI300_BENCH,
    "exchange_kwargs": {
        "freq": FREQ,
        "limit_threshold": 0.095,
        "deal_price": "close",
        "open_cost": 0.0005,
        "close_cost": 0.0015,
        "min_cost": 5,
    },
}

• 回测时间：2017-01-01至2020-08-01
• 初始资金：1亿元
• 交易设置：
  • 涨跌停限制：9.5%
  • 交易价格：收盘价
  • 交易成本：买入0.05%，卖出0.15%
  • 最低交易成本：5元

7. 创建策略和执行器

strategy_obj = TopkDropoutStrategy(**STRATEGY_CONFIG)
executor_obj = executor.SimulatorExecutor(**EXECUTOR_CONFIG)

8. 执行回测

portfolio_metric_dict, indicator_dict = backtest(
    executor=executor_obj, 
    strategy=strategy_obj, 
    **backtest_config
)

• 输出：
  • portfolio_metric_dict：投资组合指标
  • indicator_dict：交易记录

9. 获取回测结果

analysis_freq = "{0}{1}".format(*Freq.parse(FREQ))
report_normal, positions_normal = portfolio_metric_dict.get(analysis_freq)

• 解析交易频率（如"1day"）
• 获取收益报告和持仓信息

10. 风险分析

analysis = dict()
analysis["excess_return_without_cost"] = risk_analysis(
    report_normal["return"] - report_normal["bench"], 
    freq=analysis_freq
)
analysis["excess_return_with_cost"] = risk_analysis(
    report_normal["return"] - report_normal["bench"] - report_normal["cost"], 
    freq=analysis_freq
)

• 计算两种超额回报：
  • 不含交易成本
  • 含交易成本
• 分析风险指标（年化收益、夏普比率等）

11. 结果整理

analysis_df = pd.concat(analysis)  # type: pd.DataFrame
analysis_dict = flatten_dict(analysis_df["risk"].unstack().T.to_dict())

12. 输出结果

pprint(f"基准收益分析结果({analysis_freq})：")
pprint(risk_analysis(report_normal["bench"], freq=analysis_freq))
pprint(f"不含交易成本的超额回报分析({analysis_freq})：")
pprint(analysis["excess_return_without_cost"])
pprint(f"含交易成本的超额回报分析({analysis_freq})：")
pprint(analysis["excess_return_with_cost"])

系统总结

实现了一个基于Qlib的股票回测系统，核心功能：

1. 使用TopkDropout策略，每日调整股票组合
2. 完整回测流程，包括风险分析和绩效评估