ring520/strategy_backtest

Fork 0

Files

lintaogood fe50ea935a 新建回测系统，并提交

2026-01-17 21:37:42 +08:00

10 KiB

Raw Permalink Blame History

策略回测平台使用说明

1. 项目结构

当前目录 strategy_backtest/ 结构如下：

main.py：回测入口脚本。
requirements.txt：Python 依赖列表。
config/settings.py：全局配置：路径、日期、初始资金、策略参数等。
utils/：工具模块。
- logger.py：统一日志模块，所有文件使用 setup_logger 获取 logger。
- data_loader.py：行情数据加载模块，从 data/day/*.txt 读取日线数据。
- performance.py：根据资金曲线计算收益、夏普、最大回撤等指标。
- plotter.py：根据资金曲线 CSV 绘制资金曲线图。
strategies/：策略模块。
- base_strategy.py：抽象基类，封装账户与持仓、主回测循环等。
- ma_cross.py：示例策略：均线交叉（持有 5 天，最多 2 只）。
results/：回测输出。
- logs/app.log：统一运行日志。
- equity/{strategy_name}_equity.csv：资金曲线 CSV。
- plots/{strategy_name}_curve.png：资金曲线图。

2. 环境准备

安装 Python 3.10+（推荐）。
在 strategy_backtest/ 目录执行：
```
pip install -r requirements.txt
```

3. 行情数据准备

将日线 TXT 行情文件放在项目根目录上一级的 data/day/ 下（即 d:/data/jq_hc/data/day/）：
- 文件名格式：{ts_code}_daily_data.txt，例如：000001.SZ_daily_data.txt；
- 编码：UTF-8 无 BOM；
- 分隔符：制表符 \t；
- 首行必须包含至少这些列：ts_code, trade_date, open, high, low, close, vol；
- 日期列 trade_date 格式为 YYYYMMDD。
可选：在 data/code/all_stock_codes.txt 中按行列出股票代码（含后缀），例如：
```
000001.SZ
600000.SH
```
若该文件存在，则回测股票池从该文件读取；否则会扫描 data/day/ 目录下的文件名自动推断。

4. 配置说明（config/settings.py）

主要配置项说明如下：

BASE_DIR：项目根目录（d:/data/jq_hc/）。
DATA_DAY_DIR：日线数据目录（BASE_DIR / "data" / "day"）。
RESULTS_DIR：回测结果目录（BASE_DIR / "strategy_backtest" / "results"）。
START_DATE / END_DATE：回测起止日期（YYYYMMDD）。
INITIAL_CASH：初始资金金额。
MAX_POSITIONS：最多持仓股票数。
HOLD_DAYS：单只股票持有天数。
MA_SHORT / MA_LONG：均线参数。
STRATEGY：策略配置，包含模块名、类名与参数。

如需调整策略参数或回测时间窗口，直接修改 settings.py 中对应变量即可。

5. 运行回测

5.1 单策略回测模式

在 strategy_backtest/ 目录执行：

python main.py

程序将自动执行以下步骤：

加载股票代码列表（优先使用 data/code/all_stock_codes.txt，否则扫描 data/day/）。
对股票池中的每只股票，从 DATA_DAY_DIR 加载日线数据，并按日期升序整理。
性能优化：预计算日期索引（O(1) 查询）、均线指标、交易信号（金叉+放量布尔掩码）。
动态加载 STRATEGY 中配置的策略类（默认 MaCrossStrategy），并运行回测。
将资金曲线保存到 results/equity/{strategy_name}_equity.csv。
计算绩效指标（累计收益、年化收益、夏普比率、最大回撤、胜率、盈亏比等）并记录到日志中。
生成资金曲线图 results/plots/{strategy_name}_curve.png。

5.2 参数优化模式

使用多进程网格搜索找到最优参数组合：

# 使用默认配置（4核，保存前20组，按夏普比率排序）
python main.py --optimize

# 自定义配置
python main.py --optimize --jobs 8 --top 50 --metric sharpe

# 可选排序指标：sharpe / total_return / max_drawdown / annual_return
python main.py --optimize --metric total_return --top 10

参数说明：

--optimize：启用参数优化模式
--strategy：策略名称（默认：ma_cross）
--jobs：并行进程数（默认：4）
--top：保存前N个最优结果（默认：20）
--metric：排序指标（默认：sharpe）

优化结果：

优化过程会显示进度条和实时完成数
结果保存在 results/optimization/grid_search_{timestamp}.csv
日志中打印 Top 5 最优参数组合及其绩效

6. 日志说明

日志格式：YYYY-MM-DD HH:MM:SS [LEVEL] 文件名.py:行号 - 消息内容；
输出位置：
- 终端标准输出；
- 文件：results/logs/app.log；
关键日志：
- utils/data_loader.py：加载成功/失败/缺失文件；
- strategies/base_strategy.py：回测起止；
- strategies/ma_cross.py：买卖信号；
- main.py：主流程异常等。

7. 参数优化配置（config/settings.py）

7.1 参数空间定义

在 config/settings.py 中配置参数空间：

PARAM_SPACES: dict = {
    "ma_cross": {
        "ma_short": [3, 20, 2],      # 短期均线：3-20，步长2 -> [3,5,7,9,...,19]
        "ma_long": [20, 60, 5],       # 长期均线：20-60，步长5 -> [20,25,30,...,55]
        "hold_days": [3, 10, 1],      # 持有天数：3-10，步长1 -> [3,4,5,...,9]
        "position_pct_per_stock": [0.4, 0.5, 0.6],  # 单股仓位：离散值
    },
}

格式说明：

范围格式：[min, max, step] - 生成等差数列
离散格式：[value1, value2, ...] - 直接指定取值
支持整数和浮点数

7.2 参数约束条件

使用 lambda 函数过滤无意义的参数组合：

PARAM_CONSTRAINTS: dict = {
    "ma_cross": lambda params: (
        # 约束1：短期均线必须小于长期均线
        params.get("ma_short", 0) < params.get("ma_long", 999)
        # 可以添加更多约束，例如：
        # and params.get("hold_days", 0) >= 3
        # and params.get("position_pct_per_stock", 0) <= 0.6
    ),
}

约束示例：

避免 ma_short >= ma_long（短期≥长期无意义）
限制参数范围（如仓位不超过60%）
参数之间的逻辑关系

优化效果：

自动过滤无效组合，减少计算量
例：648 组 -> 过滤后 400 组（节省 38% 时间）

7.3 优化方法

当前实现的是 网格搜索（Grid Search） 方法：

特性	说明
方法	全遍历（笛卡尔积）
优点	简单直观，保证找到全局最优
缺点	参数多时计算量大（指数增长）
适用	参数维度低（2-4个）、参数范围小
加速	多进程并行（默认4核，可配置）

参数组合数计算：

总数 = ma_short的取值数 × ma_long的取值数 × hold_days的取值数 × ...
例：9 × 9 × 7 × 3 = 1701 组

其他可选方法（未实现）：

贝叶斯优化：高维空间高效，智能搜索
遗传算法：适合复杂约束条件
随机搜索：快速探索可行域

8. 性能优化详解

8.1 优化策略

回测系统实现了多层次性能优化：

① 日期索引预计算（Date Index Dict）

问题：原始方法使用 df[df['trade_date'] == date] 过滤，每次 O(n) 复杂度

解决：预先构建 {ts_code: {date: row_index}} 字典，查询时间降为 O(1)

date_index_dict = {
    "000001.SZ": {"20230101": 0, "20230102": 1, ...},
    "600000.SH": {"20230101": 0, "20230102": 1, ...},
}

# 使用：
idx = date_index_dict[ts_code][current_date]
row = df.iloc[idx]  # 直接定位，无需过滤

提升：727天 × 3776股 = 274万次查询，每次从 O(n) 降为 O(1)

② 信号预计算（Signal Precomputation）

问题：回测时每天重复计算 rolling_mean()、pct_change() 等

解决：使用 pandas 向量化操作一次性计算所有信号，存储为布尔列

# 预计算金叉信号
df["golden_cross"] = (
    (df["ma_short"] > df["ma_long"]) & 
    (df["ma_short"].shift(1) <= df["ma_long"].shift(1))
)

# 预计算放量信号
df["volume_surge"] = df["vol"] > df["vol"].rolling(5).mean() * 1.5

# 组合买入信号
df["buy_signal"] = df["golden_cross"] & df["volume_surge"]

提升：避免 274万次重复计算，约节省 80% CPU 时间

③ 买入信号索引（Buy Signal Index）

问题：每天遍历所有 3776 只股票检查是否有信号

解决：构建反向索引 {date: [stock_list]}，只遍历有信号的股票

buy_signal_index = {
    "20230103": ["000001.SZ", "600000.SH"],  # 只有20只股票有信号
    "20230104": ["000002.SZ", "600519.SH"],
}

# 使用：
signal_stocks = buy_signal_index.get(current_date, [])
for ts_code in signal_stocks:  # 只遍历 20-50 只，而非 3776 只
    # 处理买入逻辑

提升：每天从遍历 3776 只减少到 20-50 只，减少 99% 循环

④ 避免 DataFrame.attrs deepcopy

问题：将索引存储在 df.attrs 中，iloc 切片时触发 deepcopy

解决：将索引字典作为独立参数传递，不使用 df.attrs

# 错误做法：
df.attrs["date_index"] = date_index  # 切片时 deepcopy 整个字典

# 正确做法：
strategy = Strategy(date_index_dict=date_index_dict)  # 作为参数传递

提升：消除 deepcopy 开销，避免程序卡顿

8.2 性能对比

优化阶段	回测耗时	提升倍数
原始版本	约 40 分钟	-
+日期索引	约 5 分钟	8x
+信号预计算	约 1 分钟	40x
+信号索引	约 40 秒	60x

注：其中数据加载约 28 秒，实际回测执行 < 1 秒

8.3 使用建议

大规模回测（1000+ 股票）：必须启用所有优化
参数优化：利用多核并行，8核可将 1701 组从 40 分钟降至 5 分钟
内存优化：优化后自动 gc.collect()，但大数据集建议 16GB+ 内存

9. 扩展新策略

在 strategies/ 新建文件，例如 my_strategy.py，继承 BaseStrategy 并实现 on_bar 方法；

在 config/settings.py 中修改 STRATEGY：

STRATEGY = {
    "name": "MyStrategy",
    "module": "strategies.my_strategy",
    "params": {
        # 自定义参数
    },
}

重新运行 python main.py 即可用新策略回测。

10 KiB Raw Permalink Blame History Unescape Escape