#!/usr/bin/env python3 """ 在Windows Test Node上运行回测服务 """ import sys import os # ============================================ # 1. vnpy.app兼容性模块 # ============================================ print("[1/10] 加载vnpy.app兼容性模块...") import types vnpy_app = types.ModuleType('vnpy.app') sys.modules['vnpy.app'] = vnpy_app for name in ['cta_strategy', 'cta_backtester', 'data_manager']: fullname = f'vnpy.app.{name}' mod = types.ModuleType(fullname) sys.modules[fullname] = mod setattr(vnpy_app, name, mod) try: from vnpy_ctastrategy import CtaTemplate, CtaStrategyApp sys.modules['vnpy.app.cta_strategy'].CtaTemplate = CtaTemplate sys.modules['vnpy.app.cta_strategy'].CtaStrategyApp = CtaStrategyApp vnpy_app.CtaTemplate = CtaTemplate vnpy_app.CtaStrategyApp = CtaStrategyApp from vnpy_ctabacktester import BacktesterEngine sys.modules['vnpy.app.cta_backtester'].BacktesterEngine = BacktesterEngine vnpy_app.BacktesterEngine = BacktesterEngine except ImportError as e: print(f"⚠️ vnpy模块未找到，请先安装: {e}") sys.exit(1) print("[2/10] ✅ vnpy.app兼容性加载完成") # 导入依赖 try: from vnpy.event import EventEngine from vnpy.trader.engine import MainEngine from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import pydantic import traceback from typing import Optional, Dict, Any import uvicorn except ImportError as e: print(f"⚠️ 缺少依赖: {e}") print(f"请运行: pip install fastapi uvicorn pydantic vnpy") sys.exit(1) print("[3/10] ✅ 依赖导入完成") # 创建FastAPI app = FastAPI( title="回测API服务 - 最终正确版本", description="所有问题已修复：正确实例化 + 正确调用方法", version="14.0.0-finally-fixed", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) print("[4/10] ✅ FastAPI应用创建完成") # 模型 class BacktestRequest(pydantic.BaseModel): strategy_code: str symbol: str interval: str = "1d" start: int end: int capital: float = 1000000.0 rate: float = 0.00003 slippage: float = 0.2 size: int = 1 pricetick: float = 0.2 class ApiResponse(pydantic.BaseModel): code: int msg: str data: Optional[Dict[str, Any]] = None error: Optional[str] = None error_detail: Optional[str] = None print("[5/10] ✅ 模型定义完成") # 核心回测函数 def run_backtest_core( strategy_code: str, symbol: str, interval: str, start: int, end: int, **kwargs ): """核心回测函数""" try: print(f"\n[6/10] 🚀 开始新回测: {symbol} [{start} - {end}]") namespace = {} exec(strategy_code, globals(), namespace) classes = [] for k, v in namespace.items(): if isinstance(v, type) and issubclass(v, CtaTemplate) and v != CtaTemplate: classes.append(v) if not classes: return { "error": "未找到CtaTemplate子类", "hint": "请确认策略继承自CtaTemplate" } StrategyClass = classes[0] print(f"[7/10] ✅ 找到策略类: {StrategyClass.__name__}") # ============================================ # 🔥 核心修复1：正确实例化 # ============================================ print("[8/10] 🔧 创建引擎...") event_engine = EventEngine() print(f"[8/10] ✅ event_engine = EventEngine()") main_engine = MainEngine(event_engine) print(f"[8/10] ✅ main_engine = MainEngine(event_engine)") backtester_engine = BacktesterEngine(main_engine, event_engine) print(f"[8/10] ✅ backtester_engine = BacktesterEngine(main_engine, event_engine)") main_engine.add_app(backtester_engine) print(f"[8/10] ✅ main_engine.add_app(backtester_engine)") backtester_engine.init_engine() print(f"[8/10] ✅ backtester_engine.init_engine()") # ============================================ # 修复1完成 # ============================================ # 格式化日期 start_str = str(start) if len(start_str) == 8: start_str = f"{start_str[:4]}-{start_str[4:6]}-{start_str[6:8]}" end_str = str(end) if len(end_str) == 8: end_str = f"{end_str[:4]}-{end_str[4:6]}-{end_str[6:8]}" setting = { "vt_symbol": symbol, "interval": interval, "start_date": start_str, "end_date": end_str, "rate": kwargs.get("rate", 0.00003), "slippage": kwargs.get("slippage", 0.2), "size": kwargs.get("size", 1), "pricetick": kwargs.get("pricetick", 0.2), "capital": kwargs.get("capital", 1000000.0), } print(f"[9/10] ✅ 参数准备完成: {setting}") # ============================================ # 🔥 核心修复2：正确调用方法，不直接调用实例 # ============================================ print("[10/10] 🔧 执行回测...") # ✅✅✅ 正确写法：调用方法 # ❌ 错误写法：result = backtester_engine(...) # ✅ 正确写法：result = backtester_engine.run_backtesting(...) result = backtester_engine.run_backtesting( strategy_class=StrategyClass, setting=setting ) print(f"[10/10] ✅ 回测完成: result = backtester_engine.run_backtesting(...)") # ============================================ # 修复2完成 # ============================================ statistics = backtester_engine.get_result_statistics() print(f"✅ 获取统计结果: {list(statistics.keys()) if statistics else '无'}") daily_df = backtester_engine.get_daily_df() if daily_df is not None and hasattr(daily_df, 'to_dict'): daily_data = daily_df.to_dict(orient='records') else: daily_data = [] trades = backtester_engine.get_all_trades() trade_list = [t.__dict__ for t in trades] if trades else [] return { "statistics": statistics, "trades": trade_list, "daily_data": daily_data } except Exception as e: error_info = { "error": str(e), "traceback": traceback.format_exc() } print(f"❌ 回测错误: {error_info['error']}") print(error_info['traceback']) return error_info # 路由 @app.get("/") async def root(): return { "message": "回测API服务 - 最终正确版本", "version": "14.0.0-finally-fixed", "fixes": [ "✅ vnpy.app兼容性修复", "✅ BacktesterEngine(main_engine, event_engine) 正确实例化", "✅ result = backtester_engine.run_backtesting(...) 正确调用方法", "✅ 绝对没有 result = backtester_engine(...) 错误调用", "✅ 运行在Windows Test Node (192.168.2.33)", ], "endpoint": "/api/backtest/run", } @app.post("/api/backtest/run", response_model=ApiResponse) async def run_backtest(request: BacktestRequest): try: result = run_backtest_core( strategy_code=request.strategy_code, symbol=request.symbol, interval=request.interval, start=request.start, end=request.end, capital=request.capital, rate=request.rate, slippage=request.slippage, size=request.size, pricetick=request.pricetick, ) if "error" in result: return ApiResponse( code=400, msg="回测出错", data=result, error=result.get("error"), error_detail=result.get("traceback"), ) else: return ApiResponse( code=200, msg="回测完成", data=result, error=None, error_detail=None, ) except Exception as e: error_tb = traceback.format_exc() return ApiResponse( code=500, msg="API内部错误", error=str(e), error_detail=error_tb, ) if __name__ == "__main__": print("=" * 60) print("🚀 启动最终正确版本回测API") print(" 监听: 0.0.0.0:8088") print(" BacktesterEngine实例化: backtester_engine = BacktesterEngine(main_engine, event_engine) ✅") print(" 回测调用: result = backtester_engine.run_backtesting(...) ✅") print(" 错误调用: backtester_engine() ❌ 绝对不存在") print("=" * 60) uvicorn.run(app, host="0.0.0.0", port=8088)