1. algo应用和CTA应用增加用户策略方法不一样

  • 新建CTA用户策略只要在 [用户命令]\strategies\目录下,创建自己的策略,实现on_start ,on_init, on_tick, on_bar, on_trade, on_order, on_stop_order等推送接口。重新加载cta app,vnpy系统会自动把系统自带策略和用户策略一起编译添加进CTA应用界面,直接选择就OK;
  • algo应用却不是这样的,它的策略是在algo引擎初始化时候,执行load_algo_template()加载的,加载哪些策略是写死的,它没有像CTA app那样考虑用户策略的加载,代码如下:
    def load_algo_template(self):
        """"""
        from .algos.twap_algo import TwapAlgo
        from .algos.iceberg_algo import IcebergAlgo
        from .algos.sniper_algo import SniperAlgo
        from .algos.stop_algo import StopAlgo
        from .algos.best_limit_algo import BestLimitAlgo
        from .algos.grid_algo import GridAlgo
        from .algos.dma_algo import DmaAlgo
        from .algos.arbitrage_algo import ArbitrageAlgo
        from .algos.test_algo import TestAlgo       # hxxjava add

        self.add_algo_template(TwapAlgo)
        self.add_algo_template(IcebergAlgo)
        self.add_algo_template(SniperAlgo)
        self.add_algo_template(StopAlgo)
        self.add_algo_template(BestLimitAlgo)
        self.add_algo_template(GridAlgo)
        self.add_algo_template(DmaAlgo)
        self.add_algo_template(ArbitrageAlgo)
        self.add_algo_template(TestAlgo)    # hxxjava add

        from .genus import (
            GenusVWAP,
            GenusTWAP,
            GenusPercent,
            GenusPxInline,
            GenusSniper,
            GenusDMA
        )

        self.add_algo_template(GenusVWAP)
        self.add_algo_template(GenusTWAP)
        self.add_algo_template(GenusPercent)
        self.add_algo_template(GenusPxInline)
        self.add_algo_template(GenusSniper)
        self.add_algo_template(GenusDMA)

考虑到这一特点,只能像上面带注释的行这么添加algo策略。

2. 复制Iceberg策略代码,修改成TestAlgo策略

from vnpy.trader.constant import Offset, Direction
from vnpy.trader.object import TradeData, OrderData, TickData
from vnpy.trader.engine import BaseEngine

from vnpy.app.algo_trading import AlgoTemplate


class TestAlgo(AlgoTemplate):
    """"""

    display_name = "TestAlgo 测试算法"

    default_setting = {
        "vt_symbol": "",
        "direction": [Direction.LONG.value, Direction.SHORT.value],
        "price": 0.0,
        "volume": 0.0,
        "display_volume": 0.0,
        "interval": 0,
        "offset": [
            Offset.NONE.value,
            Offset.OPEN.value,
            Offset.CLOSE.value,
            Offset.CLOSETODAY.value,
            Offset.CLOSEYESTERDAY.value
        ]
    }

    variables = [
        "traded",
        "timer_count",
        "vt_orderid"
    ]

    def __init__(
        self,
        algo_engine: BaseEngine,
        algo_name: str,
        setting: dict
    ):
        """"""
        super().__init__(algo_engine, algo_name, setting)

        # Parameters
        self.vt_symbol = setting["vt_symbol"]
        self.direction = Direction(setting["direction"])
        self.price = setting["price"]
        self.volume = setting["volume"]
        self.display_volume = setting["display_volume"]
        self.interval = setting["interval"]
        self.offset = Offset(setting["offset"])

        # Variables
        self.timer_count = 0
        self.vt_orderid = ""
        self.traded = 0

        self.last_tick = None

        self.subscribe(self.vt_symbol)
        self.put_parameters_event()
        self.put_variables_event()

    def on_stop(self):
        """"""
        self.write_log("停止算法")

    def on_tick(self, tick: TickData):
        """"""
        self.last_tick = tick

    def on_order(self, order: OrderData):
        """"""
        msg = f"委托号:{order.vt_orderid},委托状态:{order.status.value}"
        self.write_log(msg)

        if not order.is_active():
            self.vt_orderid = ""
            self.put_variables_event()

    def on_trade(self, trade: TradeData):
        """"""
        self.traded += trade.volume

        if self.traded >= self.volume:
            self.write_log(f"已交易数量:{self.traded},总数量:{self.volume}")
            self.stop()
        else:
            self.put_variables_event()

    def on_timer(self):
        """"""
        self.timer_count += 1

        if self.timer_count < self.interval:
            self.put_variables_event()
            return

        self.timer_count = 0

        contract = self.get_contract(self.vt_symbol)
        if not contract:
            return

        print(f"last_tick={self.last_tick}")

        # # If order already finished, just send new order
        # if not self.vt_orderid:
        #     order_volume = self.volume - self.traded
        #     order_volume = min(order_volume, self.display_volume)

        #     if self.direction == Direction.LONG:
        #         self.vt_orderid = self.buy(
        #             self.vt_symbol,
        #             self.price,
        #             order_volume,
        #             offset=self.offset
        #         )
        #     else:
        #         self.vt_orderid = self.sell(
        #             self.vt_symbol,
        #             self.price,
        #             order_volume,
        #             offset=self.offset
        #         )
        # # Otherwise check for cancel
        # else:
        #     if self.direction == Direction.LONG:
        #         if self.last_tick.ask_price_1 <= self.price:
        #             self.cancel_order(self.vt_orderid)
        #             self.vt_orderid = ""
        #             self.write_log(u"最新Tick卖一价,低于买入委托价格,之前委托可能丢失,强制撤单")
        #     else:
        #         if self.last_tick.bid_price_1 >= self.price:
        #             self.cancel_order(self.vt_orderid)
        #             self.vt_orderid = ""
        #             self.write_log(u"最新Tick买一价,高于卖出委托价格,之前委托可能丢失,强制撤单")

        self.put_variables_event()

3. 重新启动algo就可以看见 ”TestAlgo 测试算法“啦

description