cfdaily
128 lines
3.4 KiB
Python
128 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
在Docker容器内直接运行回测 - 简化版
|
|
"""
|
|
|
|
import types
|
|
import sys
|
|
|
|
# vnpy.app兼容性
|
|
vnpy_app = types.ModuleType('vnpy.app')
|
|
sys.modules['vnpy.app'] = vnpy_app
|
|
|
|
for name in ['cta_strategy', 'cta_backtester', 'data_manager']:
|
|
mod = types.ModuleType(f'vnpy.app.{name}')
|
|
sys.modules[f'vnpy.app.{name}'] = mod
|
|
setattr(vnpy_app, name, mod)
|
|
|
|
from vnpy_ctastrategy import CtaTemplate
|
|
sys.modules['vnpy.app.cta_strategy'].CtaTemplate = CtaTemplate
|
|
|
|
from vnpy_ctabacktester import BacktesterEngine
|
|
sys.modules['vnpy.app.cta_backtester'].BacktesterEngine = BacktesterEngine
|
|
|
|
# 导入其他必要模块
|
|
from vnpy.event import EventEngine
|
|
from vnpy.trader.engine import MainEngine
|
|
from vnpy.trader.constant import Exchange, Interval
|
|
from vnpy.trader.database import get_database
|
|
from datetime import datetime
|
|
import traceback
|
|
|
|
# 定义策略
|
|
class TestStrategy(CtaTemplate):
|
|
"""简化测试策略"""
|
|
parameters = []
|
|
variables = []
|
|
|
|
def on_init(self):
|
|
self.write_log("策略初始化")
|
|
self.load_bar(100)
|
|
|
|
def on_bar(self, bar):
|
|
# 简单策略：第一天买入，最后一天卖出
|
|
if self.bar_count == 100:
|
|
self.write_log("首次进入，不操作")
|
|
elif self.bar_count == 101:
|
|
if self.pos == 0:
|
|
self.buy(bar.close_price, 10000)
|
|
self.write_log(f"买入: {bar.close_price}")
|
|
elif self.bar_count == 1044: # 接近最后一天
|
|
if self.pos > 0:
|
|
self.sell(bar.close_price, abs(self.pos))
|
|
self.write_log(f"卖出: {bar.close_price}")
|
|
|
|
print("=" * 80)
|
|
print("🚀 初始化回测引擎...")
|
|
print("=" * 80)
|
|
|
|
event_engine = EventEngine()
|
|
main_engine = MainEngine(event_engine)
|
|
|
|
backtester_engine = BacktesterEngine(main_engine, event_engine)
|
|
backtester_engine.classes["TestStrategy"] = TestStrategy
|
|
|
|
print("✅ BacktesterEngine 初始化完成")
|
|
|
|
# 加载数据
|
|
print("\n加载数据...")
|
|
db = get_database()
|
|
|
|
symbol = "510300"
|
|
exchange = Exchange.SSE
|
|
interval = Interval.DAILY
|
|
start = datetime(2021, 1, 1)
|
|
end = datetime(2026, 3, 1)
|
|
|
|
bars = db.load_bar_data(symbol, exchange, interval, start, end)
|
|
print(f"✅ 加载了 {len(bars)} 条bar数据")
|
|
|
|
if len(bars) == 0:
|
|
print("❌ 数据为空，无法回测")
|
|
sys.exit(1)
|
|
|
|
print(f"时间范围: {bars[0].datetime} ~ {bars[-1].datetime}")
|
|
|
|
# 运行回测
|
|
print("\n运行回测...")
|
|
|
|
try:
|
|
# 直接调用API
|
|
backtester_engine.run_backtesting(
|
|
class_name="TestStrategy",
|
|
vt_symbol="510300.SSE",
|
|
interval="1d",
|
|
start=start,
|
|
end=end,
|
|
rate=0.00003,
|
|
slippage=0.002,
|
|
size=10000,
|
|
pricetick=0.001,
|
|
capital=1000000,
|
|
setting={}
|
|
)
|
|
|
|
print("✅ 回测完成")
|
|
|
|
# 获取结果
|
|
result = backtester_engine.calculate_result()
|
|
|
|
print("\n" + "=" * 80)
|
|
print("回测结果:")
|
|
print("=" * 80)
|
|
|
|
# 打印关键指标
|
|
print(f"总收益率: {result['total_return']:.2%}")
|
|
print(f"年化收益率: {result['annual_return']:.2%}")
|
|
print(f"最大回撤: {result['max_drawdown']:.2%}")
|
|
print(f"夏普比率: {result['sharpe_ratio']:.2f}")
|
|
print(f"总交易次数: {result['total_trades']}")
|
|
print(f"胜率: {result['win_rate']:.2%}")
|
|
|
|
except Exception as e:
|
|
print(f"❌ 回测失败: {e}")
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
print("\n✅ 回测执行完成!")
|