1 假如你需要扩展CtaTemplate,扩展后的模板叫MyCtaTemplate
下面的代码位于vnpy\usertools\my_cta_template.py中
from typing import Any,List,Dict,Tuple
import copy
from vnpy.app.cta_strategy import (
CtaTemplate,
BarGenerator,
ArrayManager,
StopOrder,
Direction
)
from vnpy.trader.engine import MainEngine,EventEngine
from vnpy.app.cta_strategy.engine import CtaEngine
from vnpy.event.engine import Event
from vnpy.trader.object import (
LogData,
TickData,
BarData,
TradeData,
OrderData,
)
from vnpy.app.cta_strategy import StopOrder
from vnpy.app.cta_strategy.base import EngineType
from vnpy.trader.constant import Interval
from vnpy.app.cta_strategy.base import (
APP_NAME,
EVENT_CTA_LOG,
EVENT_CTA_TICK,
EVENT_CTA_HISTORY_BAR,
EVENT_CTA_BAR,
EVENT_CTA_ORDER,
EVENT_CTA_TRADE,
EVENT_CTA_STOPORDER,
EVENT_CTA_STRATEGY,
)
from vnpy.usertools.kx_chart import (
NewChartWidget,
CandleItem,
VolumeItem,
LineItem,
SmaItem,
RsiItem,
MacdItem,
)
from vnpy.usertools.my_strategy_tool import FixedBarGenerator
from vnpy.trader.engine import SamEngine
class MyCtaTemplate(CtaTemplate):
"""
一个包含可视化K线图表和策略账户的策略模板
"""
init_money:float = 100000.0 # 初始资金
kx_interval:int = 5
show_chart = False # 显示K线图表
kx_count:int = 0
def __init__(
self,
cta_engine: Any,
strategy_name: str,
vt_symbol: str,
setting: dict,
):
super().__init__(cta_engine,strategy_name,vt_symbol,setting)
self.bg = FixedBarGenerator(self.on_bar,self.kx_interval,self.on_Nmin_bar,vt_symbol=self.vt_symbol)
self.am = ArrayManager()
cta_engine:CtaEngine = self.cta_engine
self.engine_type = cta_engine.engine_type
self.even_engine = cta_engine.main_engine.event_engine
# 必须在这里声明,因为它们是实例变量
self.all_bars:List[BarData] = []
self.cur_window_bar:[BarData] = None
self.bar_updated = False
# 策略账户引擎
self.sam_engine:SamEngine = cta_engine.main_engine.get_engine('sam')
def on_init(self):
"""
Callback when strategy is inited.
"""
# 创建或者获得策略账户
if self.sam_engine.create_strategy_account(strategy_name=self.strategy_name,vt_symbols=[self.vt_symbol],
init_money=self.init_money,pickup_position=True):
self.sam_engine.notify_strategy_ui(self.strategy_name)
self.write_log(f"策略账户{self.strategy_name}创建成功!")
account_info = self.sam_engine.get_strategy_total_money(self.strategy_name)
self.write_log(f"策略账户{account_info}")
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
self.bar_updated = False
self.current_tick = tick # 记录最新tick
# 再更新tick,产生1分钟K线乃至N 分钟线
self.bg.update_tick(tick)
if self.inited:
# 先产生当前临时K线
self.cur_window_bar = self.get_cur_window_bar()
if self.cur_window_bar:
# 发送当前临时K线更新消息
self.send_event(EVENT_CTA_BAR,self.cur_window_bar)
self.send_event(EVENT_CTA_TICK,tick)
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
if self.inited:
self.write_log(f"I got a 1min BarData at {bar.datetime}")
self.bg.update_bar(bar)
self.bar_updated = True
def on_Nmin_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.all_bars.append(bar)
self.kx_count = len(self.all_bars)
if self.inited:
self.write_log(f"I got a {self.kx_interval}min BarData at {bar.datetime}")
self.send_event(EVENT_CTA_BAR,bar)
self.put_event()
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
self.send_event(EVENT_CTA_TRADE,trade)
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
self.send_event(EVENT_CTA_ORDER,order)
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
self.send_event(EVENT_CTA_STOPORDER,stop_order)
def get_cur_window_bar(self):
window_bar = copy.deepcopy(self.bg.window_bar)
bar = self.bg.bar
if not(window_bar): # 刚产生过window_bar
return None
if self.bar_updated: # 刚更新过window_bar
return window_bar
# 上一分钟window_bar和当前bar合成出临时window bar
window_bar.high_price = max(window_bar.high_price, bar.high_price)
window_bar.low_price = min(window_bar.low_price, bar.low_price)
# Update close price/volume into window bar
window_bar.close_price = bar.close_price
window_bar.volume += int(bar.volume)
window_bar.open_interest = bar.open_interest
return window_bar
def send_event(self,event_type:str,data:Any):
"""
只在实盘引擎并且配置为显示K线图表的情况下发送小线
"""
if self.engine_type==EngineType.LIVE and self.show_chart: # "如果显示K线图表"
self.even_engine.put(Event(event_type,(self.strategy_name,data)))
def init_kx_chart(self,kx_chart:NewChartWidget=None): # 提供给外部调用
if kx_chart:
kx_chart.add_plot("candle", hide_x_axis=True)
kx_chart.add_plot("volume", maximum_height=150)
kx_chart.add_item(CandleItem, "candle", "candle")
kx_chart.add_item(VolumeItem, "volume", "volume")
2 假如MyCtaTemplate需要参数化和变量输出
MyCtaTemplate的这3个成员需要参数化:
init_money:float = 100000.0 # 初始资金
kx_interval:int = 5
show_chart = False # 显示K线图表
MyCtaTemplate的这个变量需要输出:
kx_count:int = 0
3 创建一个策略ChartStrategy,这样做就可以把MyCtaTemplate参数化了
保存下面的代码到 [用户目录]\strategies\chart_strategy.py
from typing import Any,List,Dict,Tuple
from vnpy.usertools.my_cta_template import MyCtaTemplate
from vnpy.app.cta_strategy.base import (
APP_NAME,
EVENT_CTA_LOG,
EVENT_CTA_TICK,
EVENT_CTA_HISTORY_BAR,
EVENT_CTA_BAR,
EVENT_CTA_ORDER,
EVENT_CTA_TRADE,
EVENT_CTA_STOPORDER,
EVENT_CTA_STRATEGY,
)
from vnpy.usertools.kx_chart import (
NewChartWidget,
ChartItem,
CandleItem,
VolumeItem,
LineItem,
SmaItem,
RsiItem,
MacdItem,
)
import pyqtgraph as pg
from PyQt5 import QtGui
class ChartStrategy(MyCtaTemplate):
author = "hxxjava"
atr_window = 20
atr_value = 0.0
parameters = [
"init_money",
"kx_interval",
"show_chart",
"atr_window",
]
variables = [
"kx_count",
"atr_value",
]
def __init__(
self,
cta_engine: Any,
strategy_name: str,
vt_symbol: str,
setting: dict,
):
super().__init__(cta_engine,strategy_name,vt_symbol,setting)
def on_init(self):
"""
Callback when strategy is inited.
"""
super().on_init()
self.load_bar(20)
if len(self.all_bars)>0 and self.show_chart:
self.send_event(EVENT_CTA_HISTORY_BAR,self.all_bars)
def on_start(self):
""" """
self.write_log("已开始")
def on_stop(self):
""""""
self.write_log("_kx_strategy 已停止")
def init_kx_chart(self,kx_chart:NewChartWidget=None): # 提供给外部调用
# self.write_log("init_kx_chart executed !!!")
super().init_kx_chart(kx_chart)
if kx_chart:
kx_chart.add_plot("rsi", maximum_height=150)
kx_chart.add_plot("macd", maximum_height=150)
kx_chart.add_item(LineItem, "line", "candle")
kx_chart.add_item(SmaItem, "sma1", "candle")
kx_chart.add_item(SmaItem, "sma2", "candle")
sma1:SmaItem = kx_chart.get_item("sma1")
sma2:SmaItem = kx_chart.get_item("sma2")
blue_pen:QtGui.QPen = pg.mkPen(color=(100, 100, 255), width=2)
red_pen:QtGui.QPen = pg.mkPen(color=(255, 0, 0), width=2)
sma1.set_pen(blue_pen)
sma1.set_sma_window(5)
sma2.set_pen(red_pen)
sma2.set_sma_window(20)
kx_chart.add_item(RsiItem, "rsi", "rsi")
kx_chart.add_item(MacdItem, "macd", "macd")
kx_chart.add_last_price_line()
kx_chart.add_cursor()
注意上面的策略类的成员parameters和variables的赋值内容,这是本贴的重点!!!
4 创建ChartStrategy的实例,如图所示:
5 总结:
对CtaTemplate的扩展策略模板MyCtaTemplate,必须在继承MyCtaTemplate的最终策略中完成参数化和变量输出。