cfdaily
75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
简化测试脚本
|
|
"""
|
|
|
|
import requests
|
|
import time
|
|
|
|
def test_minimal_backtest():
|
|
"""最小化回测测试"""
|
|
|
|
url = "http://192.168.2.154:8088/api/backtest/run"
|
|
|
|
# 空策略
|
|
empty_strategy = '''
|
|
from vnpy_ctastrategy import CtaTemplate
|
|
|
|
class EmptyStrategy(CtaTemplate):
|
|
author = "Test"
|
|
|
|
def on_init(self):
|
|
self.write_log("✅ 策略初始化完成")
|
|
self.inited = True
|
|
'''
|
|
|
|
payload = {
|
|
"strategy_code": empty_strategy,
|
|
"symbol": "rb8888.SHFE",
|
|
"interval": "1m",
|
|
"start": 20240101,
|
|
"end": 20240101, # 只测试1天
|
|
"capital": 100000,
|
|
}
|
|
|
|
print("测试最小化回测请求...")
|
|
print(f"API地址: {url}")
|
|
|
|
try:
|
|
start_time = time.time()
|
|
response = requests.post(url, json=payload, timeout=10) # 只等10秒
|
|
|
|
elapsed = time.time() - start_time
|
|
print(f"响应时间: {elapsed:.2f}秒")
|
|
print(f"状态码: {response.status_code}")
|
|
|
|
if response.status_code == 200:
|
|
print("✅ 回测API正常工作！")
|
|
result = response.json()
|
|
print(f"返回码: {result.get('code')}")
|
|
print(f"消息: {result.get('msg')}")
|
|
return True
|
|
else:
|
|
print(f"❌ HTTP错误: {response.status_code}")
|
|
print(response.text)
|
|
return False
|
|
|
|
except requests.exceptions.Timeout:
|
|
print("❌ 请求超时 (10秒)")
|
|
return False
|
|
except requests.exceptions.ConnectionError:
|
|
print("❌ 连接失败")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ 其他错误: {e}")
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = test_minimal_backtest()
|
|
if success:
|
|
print("\n🎉 回测API已修复，可以开始测试！")
|
|
exit(0)
|
|
else:
|
|
print("\n❌ 回测API仍有问题，需要进一步排查")
|
|
exit(1) |