cfdaily
103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
在Docker容器内直接执行关羽的策略回测
|
||
不经过HTTP API，直接在容器内运行
|
||
"""
|
||
|
||
# ============================================
|
||
# 导入完整RPC服务代码（复用run_strategy_backtest函数）
|
||
# ============================================
|
||
import types
|
||
import sys
|
||
|
||
# 复制完整的RPC服务代码
|
||
with open("/Users/chufeng/.openclaw/workspace-jiangwei/final_rpc_correct.py", 'r', encoding='utf-8') as f:
|
||
rpc_code = f.read()
|
||
|
||
# 执行RPC服务代码，但不启动服务
|
||
exec(rpc_code, globals())
|
||
|
||
# 不会自动启动RPC服务，因为我们已经定义了run_strategy_backtest
|
||
print("=" * 80)
|
||
print("🚀 RPC服务代码加载完成，准备执行回测")
|
||
print("=" * 80)
|
||
|
||
# ============================================
|
||
# 读取关羽的策略代码
|
||
# ============================================
|
||
risk_control_file = "/Users/chufeng/.openclaw/sanguo_projects/sanguo_quant_live/strategies/factors-dynamic-weight-timing-20260327/risk_control.py"
|
||
strategy_file = "/Users/chufeng/.openclaw/sanguo_projects/sanguo_quant_live/strategies/factors-dynamic-weight-timing-20260327/main_strategy_single_file.py"
|
||
|
||
with open(risk_control_file, 'r', encoding='utf-8') as f:
|
||
risk_control_code = f.read()
|
||
|
||
with open(strategy_file, 'r', encoding='utf-8') as f:
|
||
strategy_code = f.read()
|
||
|
||
# 合并代码
|
||
full_strategy_code = f"{risk_control_code}\n\n# ========== 主策略 ==========\n\n{strategy_code}"
|
||
|
||
print(f"✅ 风控代码: {len(risk_control_code)} 字符")
|
||
print(f"✅ 主策略代码: {len(strategy_code)} 字符")
|
||
|
||
# ============================================
|
||
# 执行回测
|
||
# ============================================
|
||
print("\n" + "=" * 80)
|
||
print("开始执行回测...")
|
||
print("=" * 80)
|
||
|
||
result = run_strategy_backtest(
|
||
strategy_code=full_strategy_code,
|
||
symbol="510300.SSE",
|
||
interval="1d",
|
||
start=1609459200, # 2021-01-01
|
||
end=1772515200, # 2026-03-01
|
||
capital=1000000,
|
||
rate=3e-5,
|
||
slippage=0.002,
|
||
size=10000,
|
||
pricetick=0.001,
|
||
data_source="sqlite"
|
||
)
|
||
|
||
# ============================================
|
||
# 打印结果
|
||
# ============================================
|
||
print("\n" + "=" * 80)
|
||
print("回测结果:")
|
||
print("=" * 80)
|
||
|
||
if "error" in result:
|
||
print(f"❌ 回测失败: {result.get('error')}")
|
||
if "traceback" in result:
|
||
print("\n错误堆栈:")
|
||
print(result["traceback"])
|
||
sys.exit(1)
|
||
|
||
if "statistics" in result:
|
||
stats = result["statistics"]
|
||
|
||
print(f"\n📊 绩效指标:")
|
||
print(f" 总收益率: {stats.get('total_return', 0):.2%}")
|
||
print(f" 年化收益率: {stats.get('annual_return', 0):.2%}")
|
||
print(f" 最大回撤: {stats.get('max_drawdown', 0):.2%}")
|
||
print(f" 夏普比率: {stats.get('sharpe_ratio', 0):.2f}")
|
||
print(f" 总交易次数: {stats.get('total_trades', 0)}")
|
||
print(f" 胜率: {stats.get('win_rate', 0):.2%}")
|
||
print(f" 盈亏比: {stats.get('profit_loss_ratio', 0):.2f}")
|
||
|
||
if "trades" in result:
|
||
trades = result["trades"]
|
||
print(f"\n📝 交易记录: 共 {len(trades)} 笔")
|
||
|
||
for idx, trade in enumerate(trades[:10], 1):
|
||
print(f" {idx}. {trade.get('datetime')} {trade.get('direction')} {trade.get('symbol')} @ {trade.get('price')} × {trade.get('volume')}")
|
||
|
||
if len(trades) > 10:
|
||
print(f" ... 还有 {len(trades) - 10} 笔")
|
||
|
||
print("\n" + "=" * 80)
|
||
print("✅ 回测执行完成！")
|
||
print("=" * 80)
|