量化回测框架使用文档

一、项目概述

本框架支持日级别、分钟级别、Tick级别三个粒度的回测，采用向量化计算和高效数据读取技术，提供完整的策略开发、因子分析和绩效评估功能。

二、目录结构

回测框架/
├── 日级别/
│   ├── Data/                    # 日频数据目录
│   │   └── 2026/               # 按年份组织
│   ├── data_loader.py          # 数据加载模块
│   ├── backtest_engine.py      # 回测引擎
│   ├── strategy_template.py    # 策略模板
│   ├── factor_base.py          # 因子框架
│   ├── performance_analyzer.py # 绩效分析
│   └── backtest_framework.py   # 主入口
├── 分钟级别/
│   ├── Data/                    # 分钟数据目录
│   │   └── 2025_1min/          # 按年份组织
│   ├── data_loader.py
│   ├── backtest_engine.py
│   ├── strategy_template.py
│   └── backtest_framework.py
├── tick级别/
│   ├── Data/                    # Tick数据目录
│   │   └── 2026/01/            # 按年月组织
│   ├── data_loader.py
│   ├── backtest_engine.py
│   ├── strategy_template.py
│   └── backtest_framework.py
├── have_done.txt               # 已完成工作
├── to_do.txt                   # 待办事项
└── 回测框架技术栈细节.txt       # 技术规范

三、快速开始

3.1 环境依赖

pip install numpy polars pyarrow

3.2 日级别回测

import sys
sys.path.insert(0, '日级别路径')

from backtest_framework import create_framework

# 创建框架实例
framework = create_framework('日级别目录路径')

# 查看可用策略
print(framework.list_strategies())
# 输出: ['MA', 'MACD', 'RSI', 'Boll']

# 快速回测
result = framework.quick_backtest(
    strategy_name='MA',
    year=2026,
    stock_count=20,
    fast_period=5,
    slow_period=20
)

# 查看结果
print(f"总收益率: {result['metrics']['total_return']:.2%}")
print(f"夏普比率: {result['metrics']['sharpe_ratio']:.2f}")
print(f"最大回撤: {result['metrics']['max_drawdown']:.2%}")

3.3 分钟级别回测

import sys
sys.path.insert(0, '分钟级别路径')

from backtest_framework import create_framework
from strategy_template import MinuteMAStrategy

framework = create_framework('分钟级别目录路径')

# 获取股票列表
stocks = framework.get_stock_list(2025)[:5]

# 创建策略
strategy = MinuteMAStrategy(fast_period=5, slow_period=20)

# 运行回测
result = framework.run_backtest(strategy, stocks, year=2025)

# 分析绩效
metrics = framework.analyze_performance(result)
print(f"总收益率: {metrics['total_return']:.2%}")

3.4 Tick级别回测

import sys
sys.path.insert(0, 'tick级别路径')

from backtest_framework import create_framework
from strategy_template import OrderImbalanceStrategy

framework = create_framework('tick级别目录路径')

# 获取可用日期
dates = framework.get_available_dates(2026, 1)
print(f"可用日期: {dates}")

# 获取股票列表
stocks = framework.get_stock_list(dates[0])[:5]

# 创建策略
strategy = OrderImbalanceStrategy(threshold=0.3)

# 运行回测
result = framework.run_backtest(strategy, stocks, date=dates[0])

# 分析绩效
metrics = framework.analyze_performance(result)
print(f"总收益率: {metrics['total_return']:.2%}")

四、自定义策略开发

4.1 策略基类

from abc import ABC, abstractmethod
import numpy as np

class StrategyBase(ABC):
    def __init__(self, name: str):
        self.name = name
        self.params = {}
    
    def set_params(self, **kwargs):
        self.params.update(kwargs)
    
    @abstractmethod
    def generate_signals(self, kline) -> np.ndarray:
        """
        生成交易信号
        
        返回值:
        - 1: 买入信号
        - -1: 卖出信号
        - 0: 无信号
        """
        pass
    
    def get_param(self, key: str, default=None):
        return self.params.get(key, default)

4.2 自定义策略示例

import numpy as np
from strategy_template import StrategyBase
from data_loader import FactorCalculator

class MyMAStrategy(StrategyBase):
    def __init__(self, fast_period=5, slow_period=20):
        super().__init__("My_MA_Strategy")
        self.set_params(fast_period=fast_period, slow_period=slow_period)
    
    def generate_signals(self, kline) -> np.ndarray:
        close = kline.close
        fast = self.get_param('fast_period', 5)
        slow = self.get_param('slow_period', 20)
        
        # 计算均线
        ma_fast = FactorCalculator.calc_ma(close, fast)
        ma_slow = FactorCalculator.calc_ma(close, slow)
        
        # 生成信号
        signals = np.zeros(len(close))
        for i in range(1, len(close)):
            if not np.isnan(ma_fast[i]) and not np.isnan(ma_slow[i]):
                if not np.isnan(ma_fast[i-1]) and not np.isnan(ma_slow[i-1]):
                    # 金叉买入
                    if ma_fast[i] > ma_slow[i] and ma_fast[i-1] <= ma_slow[i-1]:
                        signals[i] = 1
                    # 死叉卖出
                    elif ma_fast[i] < ma_slow[i] and ma_fast[i-1] >= ma_slow[i-1]:
                        signals[i] = -1
        
        return signals

4.3 注册自定义策略

framework.register_strategy('MyMA', MyMAStrategy)

# 使用自定义策略
result = framework.quick_backtest('MyMA', 2026, fast_period=10, slow_period=30)

五、因子分析

5.1 内置因子

因子名称	说明	参数
MA	移动平均	window
EMA	指数移动平均	window
RSI	相对强弱指标	window
VolRatio	成交量比率	window
Momentum	动量因子	window
Volatility	波动率因子	window

5.2 因子测试

from factor_base import MAFactor, FactorTester

# 创建因子测试器
tester = FactorTester(framework.data_loader)

# 创建因子
ma_factor = MAFactor(window=5)

# 测试因子
result = tester.test_factor_batch(ma_factor, stocks[:20], year=2026)

print(f"平均IC: {result['avg_ic']:.4f}")
print(f"平均Rank IC: {result['avg_rank_ic']:.4f}")

5.3 因子比较

comparison = tester.compare_factors(
    ['MA', 'RSI', 'Momentum'],
    stocks[:50],
    year=2026,
    params_list=[
        {'window': 5},
        {'window': 14},
        {'window': 20}
    ]
)

for name, metrics in comparison.items():
    print(f"{name}: IC={metrics['avg_ic']:.4f}")

六、绩效指标

6.1 收益指标

指标	说明
total_return	总收益率
annual_return	年化收益率

6.2 风险指标

指标	说明
max_drawdown	最大回撤
sharpe_ratio	夏普比率
sortino_ratio	索提诺比率
calmar_ratio	卡玛比率

6.3 交易统计

指标	说明
total_trades	总交易次数
win_rate	胜率
profit_loss_ratio	盈亏比
avg_holding_days	平均持仓天数

七、数据格式

7.1 日级别数据格式 (Parquet)

字段	类型	说明
ts_code	str	股票代码
trade_date	str	交易日期 (YYYYMMDD)
open	float	开盘价
high	float	最高价
low	float	最低价
close	float	收盘价
vol	float	成交量
amount	float	成交额

7.2 分钟级别数据格式 (CSV)

字段	类型	说明
时间	str	时间戳
代码	str	股票代码
开盘价	float	开盘价
收盘价	float	收盘价
最高价	float	最高价
最低价	float	最低价
成交量	int	成交量
成交额	float	成交额

7.3 Tick级别数据格式 (CSV)

字段	类型	说明
time	int	时间戳
current	float	当前价
a1_p	float	卖一价
a1_v	float	卖一量
b1_p	float	买一价
b1_v	float	买一量
volume	float	成交量
b/s	str	买卖方向

八、API参考

8.1 BacktestFramework

class BacktestFramework:
    def __init__(self, data_dir, initial_cash=1000000, 
                 commission_rate=0.0003, stamp_duty=0.001):
        """
        初始化回测框架
        
        参数:
            data_dir: 数据目录路径
            initial_cash: 初始资金
            commission_rate: 佣金率
            stamp_duty: 印花税率
        """
    
    def get_stock_list(self, year) -> List[str]:
        """获取股票列表"""
    
    def run_backtest(self, strategy, ts_codes, year, position_size=0.1) -> Dict:
        """运行回测"""
    
    def quick_backtest(self, strategy_name, year, stock_count=50, **params) -> Dict:
        """快速回测"""
    
    def analyze_performance(self, result) -> PerformanceMetrics:
        """分析绩效"""
    
    def register_strategy(self, name, strategy_class):
        """注册策略"""
    
    def list_strategies(self) -> List[str]:
        """列出可用策略"""

8.2 DataLoader

class DataLoader:
    def load_single_stock(self, ts_code, year) -> KlineData:
        """加载单只股票数据"""
    
    def load_batch_stocks(self, ts_codes, year) -> Dict[str, KlineData]:
        """批量加载股票数据"""
    
    def load_by_date_range(self, ts_code, start_date, end_date, year) -> KlineData:
        """按日期范围加载数据"""
    
    def clear_cache(self):
        """清除缓存"""

九、注意事项

1. 数据格式: 确保数据文件格式与框架要求一致
2. 内存管理: 处理大量数据时注意调用 clear_cache() 释放内存
3. 交易时间: 分钟/Tick级别回测会自动过滤非交易时间
4. 手续费: 默认佣金率0.03%，印花税0.1%（仅卖出）
5. 最小交易单位: A股最小交易单位为100股

十、性能优化建议

1. 减少数据加载: 使用 columns 参数只加载需要的列
2. 批量处理: 使用 load_batch_stocks 批量加载数据
3. 缓存利用: 重复使用的数据会被自动缓存
4. 向量化计算: 自定义策略尽量使用NumPy向量化操作

十一、常见问题

Q: 如何处理停牌股票？ A: 框架会自动跳过数据缺失的股票，不会影响回测进程。

Q: 如何设置不同的仓位大小？ A: 在 run_backtest 中设置 position_size 参数，如 0.1 表示每只股票最大占用10%资金。

Q: 如何获取详细的交易记录？ A: 回测结果中的 trades 字段包含所有交易记录。

Q: Tick级别回测速度慢怎么办？ A: 建议减少股票数量或使用更短的时间范围进行测试。

---

版本: v1.0.0
更新日期: 2026-02-25