实盘，版本3.8，用PortfolioBollChannelStrategy测试，一个时间点推送了3次
from typing import List, Dict
from datetime import datetime
from vnpy.trader.utility import ArrayManager, Interval
from vnpy.trader.object import TickData, BarData
from vnpy_portfoliostrategy import StrategyTemplate, StrategyEngine
from vnpy_portfoliostrategy.utility import PortfolioBarGenerator
class PortfolioBollChannel1Strategy(StrategyTemplate):
"""组合布林带通道策略"""
author = "用Python的交易员"
boll_window = 18
boll_dev = 3.4
cci_window = 10
atr_window = 30
sl_multiplier = 5.2
fixed_size = 1
price_add = 5
parameters = [
"boll_window",
"boll_dev",
"cci_window",
"atr_window",
"sl_multiplier",
"fixed_size",
"price_add"
]
variables = []
def __init__(
self,
strategy_engine: StrategyEngine,
strategy_name: str,
vt_symbols: List[str],
setting: dict
) -> None:
"""构造函数"""
super().__init__(strategy_engine, strategy_name, vt_symbols, setting)
self.boll_up: Dict[str, float] = {}
self.boll_down: Dict[str, float] = {}
self.cci_value: Dict[str, float] = {}
self.atr_value: Dict[str, float] = {}
self.intra_trade_high: Dict[str, float] = {}
self.intra_trade_low: Dict[str, float] = {}
self.targets: Dict[str, int] = {}
self.last_tick_time: datetime = None
# 获取合约信息
self.ams: Dict[str, ArrayManager] = {}
for vt_symbol in self.vt_symbols:
self.ams[vt_symbol] = ArrayManager()
self.targets[vt_symbol] = 0
self.pbg = PortfolioBarGenerator(
self.on_bars, 2, self.on_2hour_bars, Interval.HOUR)
def on_init(self) -> None:
"""策略初始化回调"""
self.write_log("策略初始化")
self.load_bars(15)
def on_start(self) -> None:
"""策略启动回调"""
self.write_log("策略启动")
def on_stop(self) -> None:
"""策略停止回调"""
self.write_log("策略停止")
def on_tick(self, tick: TickData) -> None:
"""行情推送回调"""
self.pbg.update_tick(tick)
def on_bars(self, bars: Dict[str, BarData]) -> None:
"""K线切片回调"""
self.pbg.update_bars(bars)
def on_2hour_bars(self, bars: Dict[str, BarData]) -> None:
"""2小时K线回调"""
self.cancel_all()
print("合约个数:", len(bars))
# 更新到缓存序列
for vt_symbol, bar in bars.items():
am: ArrayManager = self.ams[vt_symbol]
am.update_bar(bar)
for vt_symbol, bar in bars.items():
am: ArrayManager = self.ams[vt_symbol]
if not am.inited:
return
print(datetime.now(), vt_symbol, bar.datetime, bar.close_price)
# 推送界面更新
self.put_event()
