运行 run.py 进行无界面测试。
显示如下:
但CTA策略中就是收不到数据,因为是周日,选的第三套simnow账号,
run.py完整代码如下:
import multiprocessing
from time import sleep
from datetime import datetime, time
from logging import INFO
from vnpy.event import EventEngine
from vnpy.trader.setting import SETTINGS
from vnpy.trader.engine import MainEngine
from vnpy.gateway.ctp import CtpGateway
from vnpy.gateway.xtp import XtpGateway
from vnpy.app.cta_strategy import CtaEngine
from vnpy.app.cta_strategy import CtaStrategyApp
from vnpy.app.cta_strategy.base import EVENT_CTA_LOG
SETTINGS["log.active"] = True
SETTINGS["log.level"] = INFO
SETTINGS["log.console"] = True
CTP接口连接设置
ctp_setting = {
"用户名": "XXXXX",
"密码": "XXXXX",
"经纪商代码": "9999",
"交易服务器": "180.168.146.187:10130",
"行情服务器": "180.168.146.187:10131",
"产品名称": "simnow_client_test",
"授权编码": "0000000000000000",
"产品信息": ""
}
CTA策略信息
class_name = "DualThrustStrategy"
strategy_name = "DT_p2009"
vt_symbol = "p2009.DCE"
strategy_setting = {
"risk_percent": 0.02,
"long_window": 21,
"mobile_stop": 0
}
def run_child():
"""
Running in the child process.
"""
SETTINGS["log.file"] = True
event_engine = EventEngine()
main_engine = MainEngine(event_engine)
main_engine.add_gateway(CtpGateway)
main_engine.add_app(CtaStrategyApp)
main_engine.write_log("主引擎创建成功")
log_engine = main_engine.get_engine("log")
event_engine.register(EVENT_CTA_LOG, log_engine.process_log_event)
main_engine.write_log("注册日志事件监听")
main_engine.connect(ctp_setting, "CTP")
main_engine.write_log("连接CTP接口")
sleep(10)
# 创建CTA策略引擎
cta_engine = CtaEngine(main_engine, event_engine)
# 初始化CTA策略引擎, 会依次调用init_rqdata(), load_strategy_class()等函数
cta_engine.init_engine()
# 创建属于我们自己的策略,首次创建成功后会将参数写入到C:\Users\Administrator\.vntrader文件夹下的cta_strategy_setting.json文件内
if strategy_name not in cta_engine.strategies:
main_engine.write_log(f"创建{strategy_name}策略")
cta_engine.add_strategy(class_name, strategy_name, vt_symbol, strategy_setting)
else:
cta_engine.update_strategy_setting(strategy_name, strategy_setting)
main_engine.write_log(f"更新{strategy_name}策略")
# 初始化刚创建的策略
cta_engine.init_strategy(strategy_name)
# 留有足够的时间来进行策略初始化
sleep(10)
# 启动刚创建的策略
cta_engine.start_strategy(strategy_name)
main_engine.write_log(f"启动{strategy_name}策略")
sleep(20) # Leave enough time to complete strategy initialization
main_engine.write_log(f"启动{strategy_name}策略成功")
sleep(2) # Leave enough time
print("正在交易中...")
while True:
sleep(1)
def run_parent():
"""
Running in the parent process.
"""
print("启动CTA策略守护父进程")
# Chinese futures market trading period (day/night)
DAY_START = time(8, 45)
DAY_END = time(15, 10)
NIGHT_START = time(20, 45)
NIGHT_END = time(2, 45)
child_process = None
while True:
current_time = datetime.now().time()
trading = False
# Check whether in trading period
if (
(current_time >= DAY_START and current_time <= DAY_END)
or (current_time >= NIGHT_START)
or (current_time <= NIGHT_END)
):
trading = True
# Start child process in trading period
if trading and child_process is None:
print("启动子进程")
child_process = multiprocessing.Process(target=run_child)
child_process.start()
print("子进程启动成功")
# 非记录时间则退出子进程
if not trading and child_process is not None:
print("关闭子进程")
child_process.terminate()
child_process.join()
child_process = None
print("子进程关闭成功")
sleep(5)
if name == "main":
run_parent()
CTA策略在 jupyter notebook 回测能收到数据。
重启机器,修改 DAY_START时间都不管用。
请问各位大神帮我看看,哪儿出问题了?