Home >  > VNPY学习记录11(微信curtro策略)

VNPY学习记录11(微信curtro策略)

0

一、如何将成交信息发到微信
原理:利用vnpy的内置函数发送邮件到qq邮箱,微信绑定QQ邮箱,在微信收到信息。

1、设定邮箱,注意password是邮箱授权码。

2、发送邮件的代码,写在on_trade

def on_trade(self.trade:TradeData):
    msg = f" 新的成交,策略{self.strategy_name},方向{trade.direction},开平{trade.offset},当前仓位{self.pos}"

    self.send_email(msg)

二、curtro策略
复习的知识要点:
1.布林的代码(包括参数、使用)
2.RSI的代码
3.双均线的代码
4.移动上操的代码

from vnpy.app.cta_strategy import (
    CtaTemplate,
    BarGenerator,
    ArrayManager,
    TickData,
    BarData,
    OrderData,
    TradeData,
    StopOrder
)


class CuatroStrategy(CtaTemplate):
    """"""
    author = "vn_trader"

    boll_window = 20
    boll_dev = 1.8

    rsi_window = 14
    rsi_signal = 19

    fast_window = 4
    slow_window = 26

    trailing_long = 0.5
    trailing_short = 0.3

    fixed_size = 1

    boll_up = 0
    boll_down = 0

    rsi_value = 0
    rsi_long = 0
    rsi_short = 0

    fast_ma = 0
    slow_ma = 0
    ma_trend = 0

    intra_trade_high = 0
    intra_trade_low = 0
    long_stop = 0
    short_stop = 0

    parameters = [
        "boll_window",
        "boll_dev",
        "rsi_window",
        "rsi_signal",
        "fast_window",
        "slow_window",
        "trailing_long",
        "trailing_short",
        "fixed_size"
    ]
    
    variables = [
        "boll_up",
        "boll_down",
        "rsi_value",
        "rsi_long",
        "rsi_short",
        "fast_ma",
        "slow_ma",
        "ma_trend",
        "intra_trade_high",
        "intra_trade_low",
        "long_stop",
        "short_stop"
    ]

    def __init__(
        self,
        cta_engine,
        strategy_name: str,
        vt_symbol: str,
        setting: dict,
    ):
        """"""
        super().__init__(cta_engine, strategy_name, vt_symbol, setting)

        self.rsi_long = 50 + self.rsi_signal
        self.rsi_short = 50 - self.rsi_signal

        self.bg5 = BarGenerator(self.on_bar, 5, self.on_5min_bar)   #BarGenerator,ArrayManager几乎是标配。
        self.bg15 = BarGenerator(self.on_bar, 15, self.on_15min_bar)

        self.am5 = ArrayManager()
        self.am15 = ArrayManager()        

    def on_init(self):
        """
        Callback when strategy is inited.
        """
        self.write_log("策略初始化")

        self.load_bar(10)  #这个不要漏掉,不然有时会报错。

    def on_start(self):
        """
        Callback when strategy is started.
        """
        self.write_log("策略启动")

    def on_stop(self):
        """
        Callback when strategy is stopped.
        """
        self.write_log("策略停止")

    def on_tick(self, tick: TickData):
        """
        Callback of new tick data update.
        """
        self.bg5.update_tick(tick)   #将tick推送给bargenerator,合成1分钟bar

    def on_bar(self, bar: BarData):
        """
        Callback of new bar data update.
        """
        self.bg5.update_bar(bar)
        self.bg15.update_bar(bar)

    def on_5min_bar(self, bar: BarData):
        """"""
        self.cancel_all()

        self.am5.update_bar(bar)
        if not self.am5.inited or not self.am15.inited:
            return

        self.boll_up, self.boll_down = self.am5.boll(self.boll_window, self.boll_dev)
        self.rsi_value = self.am5.rsi(self.rsi_window)
        boll_width = self.boll_up - self.boll_down

        # No position
        if self.pos == 0:
            self.intra_trade_high = bar.high_price
            self.intra_trade_low = bar.low_price
            self.long_stop = 0
            self.short_stop = 0

            if self.ma_trend > 0 and self.rsi_value >= self.rsi_long:
                self.buy(self.boll_up, self.fixed_size, stop=True)

            if self.ma_trend < 0 and self.rsi_value <= self.rsi_short:
                self.short(self.boll_down, self.fixed_size, stop=True)
        
        # Long position
        elif self.pos > 0:
            self.intra_trade_high = max(self.intra_trade_high, bar.high_price)
            self.long_stop = (self.intra_trade_high - self.trailing_long * boll_width)
            self.sell(self.long_stop, abs(self.pos), stop=True)
        
        # Short position
        else:
            self.intra_trade_low = min(self.intra_trade_low, bar.low_price)
            self.short_stop = (self.intra_trade_low + self.trailing_short * boll_width)
            self.cover(self.short_stop, abs(self.pos), stop=True)
        
        self.put_event()

    def on_15min_bar(self, bar: BarData):
        """"""
        self.am15.update_bar(bar)
        if not self.am15.inited:
            return
        
        self.fast_ma = self.am15.sma(self.fast_window)
        self.slow_ma = self.am15.sma(self.slow_window)

        if self.fast_ma > self.slow_ma:
            self.ma_trend = 1
        elif self.fast_ma < self.slow_ma:
            self.ma_trend = -1
        else:
            self.ma_trend = 0
        
        self.put_event()

    def on_trade(self, trade: TradeData):
        """
        Callback of new trade data update.
        """
        self.put_event()

    def on_order(self, order: OrderData):
        """
        Callback of new order data update.
        """
        pass
    
    def on_stop_order(self, stop_order: StopOrder):
        """
        Callback of stop order update.
        """
        pass

三、cinco策略(动态仓位管理)

代码

from vnpy.app.cta_strategy import (
    CtaTemplate,
    BarGenerator,
    ArrayManager,
    TickData,
    BarData,
    OrderData,
    TradeData,
    StopOrder
)


class CincoStrategy(CtaTemplate):
    """"""

    author = "vnpy"

    boll_window = 42
    boll_dev = 2.2
    trailing_long = 0.65
    trailing_short = 0.65
    atr_window = 4
    risk_level = 300

    boll_up = 0
    boll_down = 0
    trading_size = 0
    intra_trade_high = 0
    intra_trade_low = 0
    long_stop = 0
    short_stop = 0
    atr_value = 0

    parameters = [
        "boll_window",
        "boll_dev",
        "trailing_long",
        "trailing_short",
        "atr_window",
        "risk_level"
    ]

    variables = [
        "boll_up",
        "boll_down",
        "trading_size",
        "intra_trade_high",
        "intra_trade_low",
        "long_stop",
        "short_stop",
        "atr_value"
    ]

    def __init__(
        self,
        cta_engine,
        strategy_name: str,
        vt_symbol: str,
        setting: dict,
    ):
        """"""
        super().__init__(cta_engine, strategy_name, vt_symbol, setting)

        self.bg = BarGenerator(self.on_bar, 15, self.on_15min_bar)
        self.am = ArrayManager()

    def on_init(self):
        """
        Callback when strategy is inited.
        """
        self.write_log("策略初始化")

        self.load_bar(10)
    
    def on_start(self):
        """
        Callback when strategy is started.
        """
        self.write_log("策略启动")

    def on_stop(self):
        """
        Callback when strategy is stopped.
        """
        self.write_log("策略停止")
    
    def on_tick(self, tick: TickData):
        """
        Callback of new tick data update.
        """
        self.bg.update_tick(tick)

    def on_bar(self, bar: BarData):
        """
        Callback of new bar data update.
        """
        self.bg.update_bar(bar)
    
    def on_15min_bar(self, bar: BarData):
        """"""
        self.cancel_all()

        self.am.update_bar(bar)
        if not self.inited:
            return

        self.boll_up, self.boll_down = self.am.boll(self.boll_window, self.boll_dev)
        boll_width = self.boll_up - self.boll_down

        if not self.pos:
            self.atr_value = self.am.atr(self.atr_window)
            self.trading_size = int(self.risk_level / self.atr_value)

            self.intra_trade_high = bar.high_price
            self.intra_trade_low = bar.low_price
            self.long_stop = 0
            self.short_stop = 0

            self.buy(self.boll_up, self.trading_size, stop=True)
            self.short(self.boll_down, self.trading_size, stop=True)
        
        elif self.pos > 0:
            self.intra_trade_high = max(self.intra_trade_high, bar.high_price)
            self.long_stop = self.intra_trade_high - self.trailing_long * boll_width
            self.sell(self.long_stop, abs(self.pos), stop=True)
        
        else:
            self.intra_trade_low = min(self.intra_trade_low, bar.low_price)
            self.short_stop = self.intra_trade_low + self.trailing_short * boll_width
            self.cover(self.short_stop, abs(self.pos), stop=True)
        
        self.put_event()

    def on_trade(self, trade: TradeData):
        """
        Callback of new trade data update.
        """
        self.put_event()
    
    def on_order(self, order: OrderData):
        """
        Callback of new order data update.
        """
        pass
    
    def on_stop_order(self, stop_order: StopOrder):
        """
        Callback of stop order update.
        """
        pass

本文暂无标签

发表评论

*

*