first line of backtest system

# 从零搭建量化回测系统：我的第一行代码该写什么？

> 本文是「回测系统开发」系列的第一篇，适合有Python基础、想入门量化系统开发的读者。

## 为什么要自建回测系统？

在使用聚宽、米筐等在线平台时，你可能会有这些困扰：

- **黑盒问题**：不知道回测引擎内部如何运作
- **灵活性受限**：无法自定义数据流和事件处理
- **性能瓶颈**：大规模回测时速度太慢
- **实盘对接**：从回测到实盘需要重新开发

自建回测系统可以解决这些问题，让你完全掌控整个流程。

## 回测系统的核心概念

一个最简单的回测系统需要三个核心组件：

1. **数据管理器（Data Handler）**：提供历史行情数据
2. **策略（Strategy）**：根据数据生成交易信号
3. **执行器（Executor）**：模拟订单执行和记录持仓

## 第一版代码：极简回测框架

让我们从一个最简单的例子开始：

```python
import pandas as pd
import numpy as np

class SimpleBacktest:
    """极简回测框架"""
    
    def __init__(self, data, strategy):
        self.data = data  # 历史数据
        self.strategy = strategy  # 策略函数
        self.positions = []  # 持仓记录
        self.trades = []  # 交易记录
        
    def run(self):
        """运行回测"""
        for i in range(len(self.data)):
            # 获取当前时刻的数据
            current_data = self.data.iloc[:i+1]
            
            # 策略生成信号
            signal = self.strategy(current_data)
            
            # 执行交易
            if signal != 0:
                self.execute_trade(i, signal)
                
        return self.calculate_performance()
    
    def execute_trade(self, idx, signal):
        """执行交易"""
        price = self.data.iloc[idx]['close']
        self.trades.append({
            'date': self.data.index[idx],
            'signal': signal,
            'price': price
        })
    
    def calculate_performance(self):
        """计算绩效"""
        if len(self.trades) < 2:
            return {'total_return': 0}
            
        # 简单计算：买入为+1，卖出为-1
        returns = []
        position = 0
        entry_price = 0
        
        for trade in self.trades:
            if trade['signal'] == 1 and position == 0:  # 买入
                position = 1
                entry_price = trade['price']
            elif trade['signal'] == -1 and position == 1:  # 卖出
                ret = (trade['price'] - entry_price) / entry_price
                returns.append(ret)
                position = 0
                
        total_return = sum(returns) if returns else 0
        
        return {
            'total_return': f"{total_return*100:.2f}%",
            'num_trades': len(self.trades) // 2,
            'avg_return': f"{np.mean(returns)*100:.2f}%" if returns else "0%"
        }


def simple_strategy(data):
    """简单双均线策略"""
    if len(data) < 20:
        return 0
        
    # 计算均线
    ma5 = data['close'].tail(5).mean()
    ma20 = data['close'].tail(20).mean()
    
    # 金叉买入，死叉卖出
    if ma5 > ma20 and data['close'].iloc[-2] <= data['close'].iloc[-7]:
        return 1  # 买入信号
    elif ma5 < ma20 and data['close'].iloc[-2] >= data['close'].iloc[-7]:
        return -1  # 卖出信号
    
    return 0


# 使用示例
if __name__ == "__main__":
    # 模拟数据
    dates = pd.date_range('2024-01-01', periods=100, freq='D')
    prices = 100 + np.cumsum(np.random.randn(100) * 2)
    data = pd.DataFrame({
        'close': prices
    }, index=dates)
    
    # 运行回测
    backtest = SimpleBacktest(data, simple_strategy)
    result = backtest.run()
    
    print("回测结果：")
    for key, value in result.items():
        print(f"  {key}: {value}")
```

## 代码解析

### 1. 数据流设计

```
历史数据 → Data Handler → Strategy → Executor → 绩效计算
```

这个流程是事件驱动回测的基础。每一根K线到来时：
1. 数据管理器推送新数据
2. 策略根据数据生成信号
3. 执行器模拟订单成交
4. 更新持仓和资金

### 2. 策略接口

策略函数接收历史数据，返回交易信号：
- `1`：买入
- `-1`：卖出
- `0`：无操作

### 3. 绩效计算

目前只实现了简单的收益率计算。生产级系统还需要：
- 最大回撤
- 夏普比率
- 胜率
- 盈亏比

## 下一步

这个极简框架帮助我们理解了回测的核心原理。接下来的文章将介绍：

1. **数据管理器优化**：接入真实数据源（Tushare、AKShare）
2. **策略框架设计**：支持多因子、多策略组合
3. **执行器完善**：滑点、手续费、市场冲击
4. **向量化回测**：用Pandas实现高速回测

## 总结

自建回测系统的第一步不是写复杂的代码，而是理解核心概念：

> **数据流 → 策略决策 → 订单执行 → 绩效评估**

掌握这个流程后，你可以逐步添加功能，最终构建出生产级的回测平台。

---

**参考代码**：完整代码已上传至 [GitHub](https://github.com)

**下篇预告**：《数据管理器设计：如何高效处理百万级行情数据》