VeighNa量化社区
你的开源社区量化交易平台 | vn.py | vnpy
Member
avatar
加入于:
帖子: 35
声望: 0

用的还是2.4.0版本。实盘几年了,没有问题(由衷感谢陈老师和他的团队,让我看到了量化的方向)。
现在按要求重新做穿透式测试。用的飞泰柜台。只有交易服务器,没有行情服务器。
因为交易服务器有自己的测试逻辑,我在手动下单时,任何价格(如豆粕,从1到10000)都能成交。
但是用程序测试时,因为没有行情,无法触发on_tick(self, tick: TickData)等函数,所以在on_start(self)函数里面写了 a=self.buy(self.Price,self.lots,stop=False,lock=self.is_locked)这个句子。但下单的句子在各项条件都满足时完全没有动作,a也等于【】。
请问该怎么办才能在这种测试环境下触发交易呢?
感谢感谢!

Super Moderator
avatar
加入于:
帖子: 111
声望: 12

我理解您遇到的问题。在穿透式测试环境下,只有交易服务器而没有行情服务器,导致无法通过正常的tick数据触发交易逻辑。让我查询相关的文档和解决方案。

让我进一步查询CTA策略模块的相关文档,了解在没有行情数据的情况下如何处理交易。

让我查询一下CTA策略模板的源代码,了解on_start函数的执行时机和交易状态的控制逻辑。

现在让我查看CTA引擎的源代码,了解trading状态的控制逻辑:

根据查询到的源代码和文档,我现在可以为您提供解决方案。

问题分析

您的问题核心在于:在穿透式测试环境下,没有行情服务器,导致无法触发 on_tickon_bar 回调函数。虽然您在 on_start 函数中尝试下单,但返回空列表 [],这是因为:

  1. trading状态控制:从 template.pysend_order 方法可以看到,只有当 self.trading == True 时才会真正发送订单,否则返回空列表。

  2. on_start执行时机:从 engine.pystart_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

重要注意事项

  1. 合约信息:确保您的 vt_symbol 格式正确,如 "m2401.DCE"(豆粕2401合约,大商所)

  2. 交易时间:穿透式测试通常在交易时间内进行,确保您的测试时间在交易所的交易时段内

  3. 价格范围:虽然测试环境接受任意价格,但建议使用合理的市场价格范围

  4. 日志监控:密切关注策略日志,查看是否有错误信息

推荐使用方案一,因为它最符合VeighNa的事件驱动架构,且能确保在正确的 trading 状态下执行交易。

Member
avatar
加入于:
帖子: 35
声望: 0

太荣幸了!感谢大佬!

根据您提供的思路,我尝试在on_start函数中强制增加了一句self.strategy.trading = True,然后再下单,也成功了。
供其他小伙伴参考。

© 2015-2022 上海韦纳软件科技有限公司
备案服务号:沪ICP备18006526号

沪公网安备 31011502017034号

【用户协议】
【隐私政策】
【免责条款】