我理解您遇到的问题。在穿透式测试环境下,只有交易服务器而没有行情服务器,导致无法通过正常的tick数据触发交易逻辑。让我查询相关的文档和解决方案。
让我进一步查询CTA策略模块的相关文档,了解在没有行情数据的情况下如何处理交易。
让我查询一下CTA策略模板的源代码,了解on_start函数的执行时机和交易状态的控制逻辑。
现在让我查看CTA引擎的源代码,了解trading状态的控制逻辑:
根据查询到的源代码和文档,我现在可以为您提供解决方案。
问题分析
您的问题核心在于:在穿透式测试环境下,没有行情服务器,导致无法触发 on_tick 或 on_bar 回调函数。虽然您在 on_start 函数中尝试下单,但返回空列表 [],这是因为:
trading状态控制:从 template.py 的 send_order 方法可以看到,只有当 self.trading == True 时才会真正发送订单,否则返回空列表。
on_start执行时机:从 engine.py 的 start_strategy 方法可以看到,on_start() 是在设置 strategy.trading = True 之前调用的,所以此时 trading 状态还是 False。
解决方案
方案一:使用定时器触发交易(推荐)
在 on_init 中设置一个定时器,在策略启动后延迟执行下单:
from vnpy_ctastrategy import CtaTemplate
from vnpy.trader.object import TickData, BarData
import time
import threading
class TestStrategy(CtaTemplate):
author = "穿透式测试"
# 策略参数
Price = 3000.0 # 您的测试价格
lots = 1 # 手数
is_locked = False
parameters = ["Price", "lots", "is_locked"]
variables = []
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.timer_thread = None
def on_init(self):
"""策略初始化"""
self.write_log("策略初始化")
self.put_event()
def on_start(self):
"""策略启动"""
self.write_log("策略启动")
# 启动一个定时器线程,在1秒后执行下单
self.timer_thread = threading.Timer(1.0, self.execute_test_order)
self.timer_thread.start()
def on_stop(self):
"""策略停止"""
self.write_log("策略停止")
if self.timer_thread and self.timer_thread.is_alive():
self.timer_thread.cancel()
def execute_test_order(self):
"""执行测试订单"""
self.write_log(f"执行测试订单: 价格={self.Price}, 手数={self.lots}")
try:
# 此时 trading 状态应该已经是 True
a = self.buy(self.Price, self.lots, stop=False, lock=self.is_locked)
self.write_log(f"下单结果: {a}")
except Exception as e:
self.write_log(f"下单异常: {str(e)}")
def on_tick(self, tick: TickData):
"""收到行情tick"""
pass
def on_bar(self, bar: BarData):
"""收到K线"""
pass
方案二:手动触发on_tick模拟
如果您知道要交易的合约信息,可以在 on_start 中手动创建一个 TickData 对象并调用 on_tick:
from vnpy_ctastrategy import CtaTemplate
from vnpy.trader.object import TickData
from vnpy.trader.constant import Exchange
from datetime import datetime
import pytz
class TestStrategy(CtaTemplate):
author = "穿透式测试"
Price = 3000.0
lots = 1
is_locked = False
parameters = ["Price", "lots", "is_locked"]
variables = []
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
def on_init(self):
self.write_log("策略初始化")
self.put_event()
def on_start(self):
self.write_log("策略启动")
# 手动创建tick数据并触发on_tick
symbol, exchange_str = self.vt_symbol.split(".")
exchange = Exchange(exchange_str)
tick = TickData(
symbol=symbol,
exchange=exchange,
datetime=datetime.now(pytz.timezone("Asia/Shanghai")),
name="测试合约",
volume=1000,
last_price=self.Price, # 使用您的测试价格
limit_up=10000,
limit_down=1,
ask_price_1=self.Price + 10,
ask_volume_1=10,
bid_price_1=self.Price - 10,
bid_volume_1=10,
gateway_name="TEST"
)
# 直接调用on_tick处理
self.on_tick(tick)
def on_tick(self, tick: TickData):
"""收到行情tick - 现在可以正常下单了"""
self.write_log(f"收到tick数据,价格: {tick.last_price}")
# 在这里执行您的下单逻辑
a = self.buy(tick.last_price, self.lots, stop=False, lock=self.is_locked)
self.write_log(f"下单结果: {a}")
def on_bar(self, bar: BarData):
pass
方案三:在on_init中延迟初始化(简单但有效)
import time
import threading
class TestStrategy(CtaTemplate):
author = "穿透式测试"
Price = 3000.0
lots = 1
is_locked = False
parameters = ["Price", "lots", "is_locked"]
variables = []
def on_init(self):
"""策略初始化"""
self.write_log("策略初始化")
# 延迟2秒后执行下单(确保策略已经完全启动)
def delayed_order():
time.sleep(2)
self.write_log("延迟下单开始")
a = self.buy(self.Price, self.lots, stop=False, lock=self.is_locked)
self.write_log(f"下单结果: {a}")
threading.Thread(target=delayed_order).start()
self.put_event()
def on_start(self):
self.write_log("策略启动")
def on_tick(self, tick: TickData):
pass
def on_bar(self, BarData):
pass
重要注意事项
合约信息:确保您的 vt_symbol 格式正确,如 "m2401.DCE"(豆粕2401合约,大商所)
交易时间:穿透式测试通常在交易时间内进行,确保您的测试时间在交易所的交易时段内
价格范围:虽然测试环境接受任意价格,但建议使用合理的市场价格范围
日志监控:密切关注策略日志,查看是否有错误信息
推荐使用方案一,因为它最符合VeighNa的事件驱动架构,且能确保在正确的 trading 状态下执行交易。
基于RAG技术开发的VeighNa AI助手,你的个人专属 Desk Quant