cfdaily
110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试 510300.SSE 回测
|
|
验证数据是否正确加载
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import time
|
|
|
|
def main():
|
|
print("🚀 测试 510300.SSE 回测")
|
|
print("="*60)
|
|
|
|
url = "http://192.168.2.154:8088/api/backtest/run"
|
|
|
|
# 简单均线策略
|
|
strategy_code = '''
|
|
from vnpy.app.cta_strategy import CtaTemplate
|
|
|
|
class SimpleMAStrategy(CtaTemplate):
|
|
"""简单均线策略"""
|
|
author = "关羽将军"
|
|
|
|
parameters = ["fast_window", "slow_window"]
|
|
variables = ["fast_ma", "slow_ma"]
|
|
|
|
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
|
|
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
|
|
|
|
self.fast_window = 5
|
|
self.slow_window = 20
|
|
|
|
self.fast_ma = 0
|
|
self.slow_ma = 0
|
|
|
|
def on_init(self):
|
|
self.write_log("✅ 策略初始化完成")
|
|
self.put_event()
|
|
|
|
def on_bar(self, bar):
|
|
# 这里简单回测逻辑
|
|
if self.fast_ma > self.slow_ma and not self.pos:
|
|
self.buy(bar.close, 1)
|
|
self.write_log(f"买入: {bar.close}")
|
|
elif self.fast_ma < self.slow_ma and self.pos > 0:
|
|
self.sell(bar.close, self.pos)
|
|
self.write_log(f"卖出: {bar.close}")
|
|
|
|
self.put_event()
|
|
'''
|
|
|
|
payload = {
|
|
"strategy_code": strategy_code,
|
|
"symbol": "510300.SSE",
|
|
"interval": "1d",
|
|
"start": 20200101,
|
|
"end": 20220926,
|
|
"capital": 1000000,
|
|
}
|
|
|
|
print(f"请求参数:")
|
|
print(f" 标的: {payload['symbol']}")
|
|
print(f" 时间: {payload['start']} - {payload['end']}")
|
|
print(f" 初始资金: {payload['capital']}")
|
|
print()
|
|
|
|
try:
|
|
start_time = time.time()
|
|
response = requests.post(url, json=payload, timeout=30)
|
|
elapsed = time.time() - start_time
|
|
|
|
print(f"请求完成，耗时: {elapsed:.2f} 秒")
|
|
print(f"状态码: {response.status_code}")
|
|
print()
|
|
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
print(f"响应内容:")
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
code = result.get('code')
|
|
if code == 200:
|
|
print("\n✅ 回测成功完成！")
|
|
data = result.get('data', {})
|
|
if data:
|
|
if 'statistics' in data:
|
|
stats = data['statistics']
|
|
print(f"统计数据: {list(stats.keys())}")
|
|
else:
|
|
print(f"数据包含: {list(data.keys())}")
|
|
else:
|
|
print("回测完成，获得绩效数据")
|
|
else:
|
|
error = result.get('error', '未知错误')
|
|
print(f"\n❌ 回测失败: {error}")
|
|
else:
|
|
print(f"❌ HTTP错误: {response.status_code}")
|
|
print(f"响应: {response.text[:200]}")
|
|
|
|
except requests.exceptions.Timeout:
|
|
print("❌ 请求超时")
|
|
except Exception as e:
|
|
print(f"❌ 异常: {e}")
|
|
|
|
print("\n" + "="*60)
|
|
print("测试完成")
|
|
|
|
if __name__ == "__main__":
|
|
main() |