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

策略逻辑:

做期货日内,20min为周期,每天14:55的时候把仓位还原,控制每天的交易次数不超过固定数值

如果现在的价格比过去20min的最高价高就买入,比20min最低价低就卖空

利用atr进行移动止损

代码如下,麻烦大神看看哪里错了:

from datetime import datetime
from mimetypes import inited

from numpy import True_
from pymongo import MIN_SUPPORTED_WIRE_VERSION
from vnpy.trader.constant import Interval
from vnpy.trader.object import BarData, OrderData, TickData, TradeData
from vnpy.trader.utility import ArrayManager, BarGenerator
from vnpy_ctastrategy.base import StopOrder
from vnpy_ctastrategy.template import CtaTemplate

class BreakThrough(CtaTemplate):

#参数
signal_window = 20
atr_multiple:int = 2            #atr倍数
atr_window:int = 20                #用多少跟算atr
fixed_pro_tar:int = 10         #止盈,最后除以1000,10%
fixed_stop:int = 15             #止损,每天15%
count_control:int = 6           #每天交易次数
fixed_size = 10

#变量
atr_value:float = 0.0
up_line:float = 0.0
mid_line:float = 0.0
down_line:float = 0.0
pro_tar:float = 0.0             #止盈线
net_stop:float = 0.0            #止损线——平仓线
count_control_num:int = 0
intra_trade_high = 0
intra_trade_low = 0

#参数和变量展示存放区
parameters = [
    "atr_multiple",           
    "atr_window",                
    "fixed_pro_tar",       
    "fixed_stop",    
    "count_control"        
]

variables = [
    "atr_value",
    "up_line",
    "mid_line",
    "down_line",
    "pro_tar",             
    "net_stop",            
    "count_control_num",
    "intra_trade_high",
    "intra_trade_low"
]

def __init__(self, cta_engine, strategy_name, vt_symbol, setting) -> None:
    super().__init__(cta_engine, strategy_name, vt_symbol, setting)
    self.bg = BarGenerator(self.on_bar, 20, self.on_20min_bar, interval=Interval.MINUTE) #合成器
    self.am = ArrayManager()  #K线池
    self.count_control_num = 0
    self.last_bar:BarData = None #刚开始没有K线

def on_init(self) -> None: 
    self.write_log("策略初始化")
    self.load_bar(days=10)  #下载K线

def on_start(self) -> None:
    self.write_log("策略启动")
    self.put_event()

def on_stop(self) -> None:
    self.write_log("策略停止")
    self.put_event()

def on_tick(self, tick: TickData): 
    self.bg.update_tick(tick)

def on_bar(self, bar: BarData): #合成K线
    self.bg.update_bar(bar)

def on_20min_bar(self, bar: BarData) -> None: #合成K线

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

    self.atr_value = self.am.atr(self.atr_window)

    if self.last_bar and str(self.last_bar.datetime)[-14:-6] == "14:55:00":
        self.count_control_num = 0 #交易次数归0

        if self.pos > 0:
            self.sell(bar.close_price, abs(self.pos), stop=True)
        elif self.pos < 0:
            self.cover(bar.close_price, abs(self.pos), stop=True)

    if self.pos == 0 and self.count_control_num < self.count_control:
        self.intra_trade_high = bar.high_price
        self.intra_trade_low = bar.low_price

        self.up_line = max(self.am.high[-20:]) # 20min最高价
        self.down_line = min(self.am.low[-20:]) # 20min最低价
        print('\n')
        print(bar.high_price, bar.low_price, self.up_line, self.down_line)

        if bar.high_price > self.up_line:
            print("多头信号成立")
            # if bar.open_price > self.up_line:
            #     print("买一手")
            #     self.buy(bar.open_price, self.fixed_size)
            # else:
            #     print("买一手")
            #     self.buy(self.up_line, self.fixed_size)
            self.buy(self.up_line, self.fixed_size, stop=True)

        elif bar.low_price < self.down_line:
            print("空头信号成立")
            # if bar.open_price < self.down_line:
            #     print("卖一手")
            #     self.short(bar.open_price, self.fixed_size)
            # else:
            #     print("卖一手")
            #     self.short(self.down_line, self.fixed_size)
            self.short(self.down_line, self.fixed_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.atr_value*self.atr_multiple
        self.sell(self.long_stop, abs(self.pos), stop=True)

    elif self.pos < 0: # 止损单
        self.intra_trade_low = min(self.intra_trade_low, bar.low_price)
        self.short_stop = self.intra_trade_low + self.atr_value*self.atr_multiple
        self.cover(self.short_stop, abs(self.pos), True)

    self.put_event()

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

def on_trade(self, trade: TradeData): 
    """
    Callback of new trade data update.
    """
    if self.pos > 0: #持仓大于0,说明有多单
        self.count_control_num += 1 #当有成交后,成交次数加一
    elif self.pos < 0: #持空单
        self.count_control_num += 1 
    self.put_event()

def on_stop_order(self, stop_order: StopOrder):
    """
    Callback of stop order update.
    """
    pass
Member
avatar
加入于:
帖子: 3357
声望: 226

可以自己打印一下策略trading状态看看

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

沪公网安备 31011502017034号

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