sanguo_vnpy/research/docker/automated-backtest-service-design.md

自动化回测服务设计方案

需求背景

三国量化团队需要支持自动化批量回测场景：

• 将军们提交回测脚本和参数
• 系统自动运行回测
• 完成后返回结果给将军
• 不需要手动干预

现有环境基础

当前 Docker 容器环境已经提供：

• ✅ Python 3.10
• ✅ 完整 vnpy 框架
• ✅ vnpy.trader.backtesting.BacktestingEngine 原生回测引擎
• ✅ FastAPI（已被 vnpy_webtrader 依赖）
• ✅ SSH 远程访问
• ✅ Jupyter Lab 结果分析

设计原则

1. 尽量基于vnpy原生：核心回测使用原生 BacktestingEngine，不重复造轮子
2. 轻量级实现：不引入过重的第三方任务队列（如 Redis/RabbitMQ）
3. 保持简单：够用就好，便于维护
4. 不强制自动启动：作为可选服务，需要时手动启动

架构设计

┌─────────────┐
│   将军      │
└──────┬──────┘
       │ POST /api/backtest/submit
       ↓
┌──────────────────────────┐
│  FastAPI 回测服务        │ ← 监听 8088 端口
│  - 接收任务              │
│  - 任务排队              │
│  - 进程池执行            │
│  - 保存结果              │
└──────┬───────────────────┘
       │ 调用
       ↓
┌──────────────────────────┐
│  vnpy 原生 BacktestingEngine ← 核心回测计算
│  - 加载历史数据          │
│  - 运行策略回测          │
│  - 计算统计结果          │
│  - 输出图表              │
└──────┬───────────────────┘
       │ 保存结果
       ↓
┌──────────────────────────┐
│  文件系统存储            │
│  - 任务元数据 JSON       │
│  - 回测结果 CSV/JSON    │
│  - 收益曲线图片          │
└──────┬───────────────────┘
       │ GET /api/backtest/result/{task_id}
       ↓
┌─────────────┐
│   将军      │ ← 获取结果分析
└─────────────┘

端口分配

服务	端口	用途	状态
SSH	22 → 2222	远程登录	✅ 已映射
Jupyter Lab	8888	Notebook开发	✅ 已映射
code-server	8080	Web VS Code	✅ 已映射
VNPY Web Trader	8000	交易Web API	✅ 已映射
VNPY RPC 交易核心	2018	RPC 请求	✅ 已映射
VNPY RPC 交易核心	4102	RPC 订阅	✅ 已映射
自动化回测服务	8088	回测API	✅ 已映射

API 接口设计

1. 提交回测任务

接口： POST /api/backtest/submit

请求体：

{
  "strategy_name": "double_ma",
  "strategy_code": "from vnpy.trader.strategy import ...",
  "parameters": {
    "fast_window": 5,
    "slow_window": 20
  },
  "start_date": "2020-01-01",
  "end_date": "2024-01-01",
  "symbol": "IF888",
  "interval": "1h"
}

响应：

{
  "code": 0,
  "msg": "success",
  "data": {
    "task_id": "a1b2c3d4-1234-5678-90ef-ghijklmnopqr",
    "status": "pending"
  }
}

2. 查询任务状态

接口： GET /api/backtest/status/{task_id}

响应：

{
  "code": 0,
  "msg": "success",
  "data": {
    "task_id": "a1b2c3d4-1234-5678-90ef-ghijklmnopqr",
    "status": "running|completed|failed",
    "progress": 75,
    "message": "正在计算统计结果"
  }
}

3. 获取回测结果

接口： GET /api/backtest/result/{task_id}

响应：

{
  "code": 0,
  "msg": "success",
  "data": {
    "task_id": "...",
    "status": "completed",
    "statistics": {
      "total_trades": 100,
      "win_rate": 0.6,
      "sharpe_ratio": 1.8,
      "max_drawdown": 0.15,
      "total_return": 0.35,
      "annual_return": 0.08
    },
    "result_file": "/app/backtest_results/.../result.csv",
    "chart_file": "/app/backtest_results/.../equity_curve.png",
    "completed_at": "2026-04-11T22:00:00Z"
  }
}

4. 列出所有任务

接口： GET /api/backtest/list

响应：

{
  "code": 0,
  "msg": "success",
  "data": {
    "tasks": [
      {
        "task_id": "...",
        "strategy_name": "...",
        "status": "...",
        "created_at": "..."
      }
    ]
  }
}

目录结构

/app/
├── scripts/
│   └── backtest-service/
│       ├── main.py           # FastAPI 入口
│       ├── task_queue.py     # 任务队列实现
│       ├── executor.py       # 任务执行器
│       ├── models.py         # 数据模型
│       └── README.md         # 使用说明
└── backtest_jobs/
    ├── pending/              # 等待执行
    ├── running/              # 执行中
    ├── completed/            # 已完成
    └── failed/               # 执行失败

任务队列实现

使用 Python 内置 multiprocessing.Pool 做进程池，限制并发回测数量，避免资源耗尽：

# 配置示例
MAX_WORKERS = 2  # 最大并发回测数，根据CPU核数调整

每个回测任务在独立进程中运行，互不干扰，一个失败不影响其他任务。

启动方式

手动启动（开发环境）

ssh -p 2222 vnpy@192.168.2.153
cd /app/scripts/backtest-service
python main.py

后台运行

nohup python main.py > backtest-service.log 2>&1 &

后续实现步骤

1. 创建目录结构
2. 编写 Pydantic 数据模型
3. 编写任务队列
4. 编写任务执行器
5. 编写 FastAPI 路由
6. 编写 README 使用说明
7. 测试验证

参考文档

• vnpy-webtrader-research-report.md
• docker-dependency-cache-scheme.md

---

文档记录时间： 2026-04-11 22:21 GMT+8 记录人： 姜维 伯约 状态： 设计完成，待实现