VeighNa量化社区
你的开源社区量化交易平台
Member
avatar
加入于:
帖子: 419
声望: 170

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的实例,如图所示:

description

5 总结:

对CtaTemplate的扩展策略模板MyCtaTemplate,必须在继承MyCtaTemplate的最终策略中完成参数化和变量输出。

Administrator
avatar
加入于:
帖子: 4500
声望: 320

给你加个精华

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

大神,# 策略账户引擎 self.sam_engine:SamEngine = cta_engine.main_engine.get_engine('sam') 是自定义的?

Member
avatar
加入于:
帖子: 419
声望: 170

vnpytrader wrote:

大神,# 策略账户引擎 self.sam_engine:SamEngine = cta_engine.main_engine.get_engine('sam') 是自定义的?

是的,策略账户引擎现在测试中。

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

期待大神的分享!

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

沪公网安备 31011502017034号

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