上位足を考慮した移動平均線のバックテスト｜日経225 FX 4時間足

class Strategy_SycleEMA_Base(StrategyBase) :

  def __init__(self ,symbol ,np_arr_dic ,df) :

    StrategyBase.__init__(self ,symbol ,np_arr_dic ,df)

    self.ATR_N = 3

    self.ema5   = ta.EMA(self.c ,5)
    self.ema20  = ta.EMA(self.c ,20)
    self.ema40  = ta.EMA(self.c ,40)
    self.ema120 = ta.EMA(self.c ,120)
    self.ema240 = ta.EMA(self.c ,240)

    self.stage = np.array(list(map(lambda s ,m ,l :
                                  1 if s > m and m > l else \
                                  2 if m > s and s > l else \
                                  3 if m > l and l > s else \
                                  4 if l > m and m > s else \
                                  5 if l > s and s > m else \
                                  6 if s > l and l > m else \
                                  np.nan
                                  ,self.ema5 ,self.ema20 ,self.ema40)))

    self.upper_stage = np.array(list(map(lambda s ,m ,l :
                                    1 if s > m and m > l else \
                                    2 if m > s and s > l else \
                                    3 if m > l and l > s else \
                                    4 if l > m and m > s else \
                                    5 if l > s and s > m else \
                                    6 if s > l and l > m else \
                                    np.nan
                                    ,self.ema5 ,self.ema120 ,self.ema240)))

    self.atrL ,self.atrH = ta.MINMAX(self.atr20 ,1800)
    self.atrM = (self.atrH + self.atrL) / 2
    self.atr  = np.array(list(map(lambda a ,b :
                                 a if a > b else \
                                 b if a < b else \
                                 a if np.isnan(b)==True and np.isnan(a)==False else \
                                 b if np.isnan(a)==True and np.isnan(b)==False else \
                                 np.nan
                                 ,self.atr20 ,self.atrM)))


  def Base_record_mgmt(self ,i) :

    Pos             = self.count_pos()
    self.count_days = np.nan if Pos==0 else self.count_days + 1
    b4_EntrySig     = False  if Pos==0 else abs(self.mgmt[-1]["EntrySig"])==1
    MTM             = 0      if Pos==0 else ((self.o[i] - self.entry_price) * self.l_s)
    b4_SO_Close     = np.nan if Pos==0 else self.mgmt[-1]["SO_Close"]
    b4_SO_Entry     = np.nan if Pos==0 else self.mgmt[-1]["SO_Entry"]
    pre_SO_Close    = np.nan if Pos==0 and not b4_EntrySig else self.c[i-1] - self.atr[i-1] * self.ATR_N * self.l_s
    pre_SO_Entry    = np.nan if Pos==0 and not b4_EntrySig else self.entry_price - self.atr[i-1] * self.ATR_N * self.l_s
    SO_Close        = np.nan if Pos==0 and not b4_EntrySig else pre_SO_Close if b4_EntrySig or (self.l_s==1 and pre_SO_Close > b4_SO_Close) or (self.l_s==-1 and pre_SO_Close < b4_SO_Close) else b4_SO_Close
    SO_Entry        = np.nan if Pos==0 and not b4_EntrySig else pre_SO_Entry if b4_EntrySig or (self.l_s==1 and pre_SO_Entry > b4_SO_Entry) or (self.l_s==-1 and pre_SO_Entry < b4_SO_Entry) else b4_SO_Entry
    PL              = 0      if len(self.hst)==0 else self.hst[-1]['profit'] if self.mgmt[-1]['ExitSig']==1 else 0
    self.mgmt.append({
        'Time'        : self.df.index[i]
        ,'Open'       : self.o[i]
        ,'High'       : self.h[i]
        ,'Low'        : self.l[i]
        ,'Close'      : self.c[i]
        ,'ATR'        : self.atr[i]
        ,'EMA5'       : self.ema5[i]
        ,'EMA20'      : self.ema20[i]
        ,'EMA40'      : self.ema40[i]
        ,'EMA120'     : self.ema120[i]
        ,'EMA240'     : self.ema240[i]
        ,'Pos'        : Pos
        ,'CountDays'  : self.count_days
        ,'EntrySig'   : 1 if self.opbuy else -1 if self.opsell else 0
        ,'ExitSig'    : 1 if self.clbuy or self.clsell else 0
        ,'L/S'        : self.l_s
        ,'EntryPrice' : self.entry_price
        ,'MTM'        : MTM
        ,'PL'         : PL
        ,'Stage'      : self.stage[i]
        ,'UpperStage' : self.upper_stage[i]
        ,'SO'         : self.SO
        ,'SO_Close'   : SO_Close
        ,'SO_Entry'   : SO_Entry
    })



class ENTRY_CycleStage14_FILTER_UpperCycleStage14__EXIT1_CycleStage36__EXIT2_SO_Entry(Strategy_SycleEMA_Base) :

  def __init__(self ,symbol ,np_arr_dic ,df) :

    Strategy_SycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)


  def onTick(self ,i) :

    if self.is_nan[i] : return

    if any([self.opbuy ,self.opsell ,self.clbuy ,self.clsell]) :
      if self.SO :
        self.order_proccesing(i ,Entry_Price=self.c[i-1] ,Exit_Price=self.mgmt[-2]['SO_Entry'] ,ATR=self.atr[i-1])
        self.SO = False
      else :
        self.order_proccesing(i ,Entry_Price=self.c[i-1] ,Exit_Price=self.c[i-1] ,ATR=self.atr[i-1])

    opbuy = opsell = clbuy_1 = clbuy_2 = clsell_1 = clsell_2 = False

    buy_filter  = self.upper_stage[i] == 1
    buy_sig     = self.stage[i]       == 1

    sell_filter = self.upper_stage[i] == 4
    sell_sig    = self.stage[i]       == 4

    self.opbuy  = buy_filter  and buy_sig  and self.count_buy()==0
    self.opsell = sell_filter and sell_sig and self.count_sell()==0

    if self.count_buy() > 0 :
      SO_Entry   = self.entry_price - self.atr[i-1] * self.ATR_N * self.l_s if self.count_days==0 else self.mgmt[-1]['SO_Entry']
      clbuy_1    = self.stage[i-1] < 3 and (self.stage[i] == 3 or self.stage[i] == 4)
      clbuy_2    = self.l[i] < SO_Entry
      self.clbuy = any([clbuy_1 ,clbuy_2 ,self.opsell])
      self.SO    = clbuy_2

    if self.count_sell() > 0 :
      SO_Entry    = self.entry_price - self.atr[i-1] * self.ATR_N * self.l_s if self.count_days==0 else self.mgmt[-1]['SO_Entry']
      clsell_1    = (self.stage[i-1] < 6 and self.stage[i-1] >= 4) and (self.stage[i] == 6 or self.stage[i] == 1)
      clsell_2    = self.h[i] > SO_Entry
      self.clsell = any([clsell_1 ,clsell_2 ,self.opbuy])
      self.SO     = clsell_2

    self.Base_record_mgmt(i)