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 测试算法“啦
