資金管理（取引量の調整）が成績に及ぼす影響のバックテスト｜日経225 FX 4時間足

class Strategy_CycleEMA_Base(StrategyBase) :

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

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

    self.principal = 10000000
    self.unit_pct  = 0.01

    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)

    def calc_stage(s ,m ,l) :
      return 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 ,s ,m ,l)))

    self.stage       = calc_stage(self.ema5 ,self.ema20  ,self.ema40)
    self.upper_stage = calc_stage(self.ema5 ,self.ema120 ,self.ema240)

    self.atrL ,self.atrH = ta.MINMAX(self.atr20 ,1800)
    self.atrM = (self.atrH + self.atrL) / 2

    self.atrUM = 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)))

    self.is_nan = np.isnan(self.ema240)


  def management(self ,i ,Case=None) :

    first    = self.first_time
    not_nan  = not Case == 'is_nan'

    Lot      = Dict_Symbol[self.symbol]['lot']
    Currency = 1 if Dict_Symbol[self.symbol]['currency']=="" else self.np_arr_dic[self.symbol]['Close'][i-1]
    LS       = self.l_s
    b4asset  = np.nan if first else self.asset[i-1]
    b4mgmt   = np.nan if first else self.mgmt[i-1]
    ATR      = np.nan if first else self.atr[i-1]
    ATR_N    = self.atrN
    Entry    = self.entry_price
    Close    = self.c[i-1]

    def mgmt_calc_unit() :
      if first            : self.mgmt[i]['Unit'] = np.nan
      elif Case=='is_nan' : self.mgmt[i]['Unit'] = b4mgmt['Unit']
      else :
        UnitValue      = b4asset['UnitValue'] if Case=='1st' else self.asset[i]['UnitValue']
        Unit           = round( UnitValue / ( ATR * Lot * Currency) )
        self.mgmt[i]['Unit'] = Unit if not_nan else b4mgmt['Unit']

    def mgmt_append() :
      if not_nan and not first :
        Amount          = b4mgmt["Amount"]
        self.count_nums = np.nan if Amount==0 else self.count_nums + 1
        b4_SO_Close     = np.nan if Amount==0 else b4mgmt["SO_Close"]
        b4_SO_Entry     = np.nan if Amount==0 else b4mgmt["SO_Entry"]
        pre_SO_Close    = np.nan if Amount==0 else Close - ATR * ATR_N * LS
        pre_SO_Entry    = np.nan if Amount==0 else Entry - ATR * ATR_N * LS
        SO_Close        = np.nan if Amount==0 else pre_SO_Close if (LS==1 and pre_SO_Close > b4_SO_Close) or (LS==-1 and pre_SO_Close < b4_SO_Close) else b4_SO_Close
        SO_Entry        = np.nan if Amount==0 else pre_SO_Entry if (LS==1 and pre_SO_Entry > b4_SO_Entry) or (LS==-1 and pre_SO_Entry < b4_SO_Entry) else b4_SO_Entry
      self.mgmt[i]['Time']         = self.df.index[i]
      self.mgmt[i]['Symol']        = self.symbol
      self.mgmt[i]['b4Open']       = np.nan if first else self.o[i-1]           if not_nan else b4mgmt['b4Close']
      self.mgmt[i]['b4High']       = np.nan if first else self.h[i-1]           if not_nan else b4mgmt['b4Close']
      self.mgmt[i]['b4Low']        = np.nan if first else self.l[i-1]           if not_nan else b4mgmt['b4Close']
      self.mgmt[i]['b4Close']      = np.nan if first else self.c[i-1]           if not_nan else b4mgmt['b4Close']
      self.mgmt[i]['b4ATR']        = np.nan if first else self.atr[i-1]         if not_nan else b4mgmt['b4ATR']
      self.mgmt[i]['b4EMA5']       = np.nan if first else self.ema5[i-1]        if not_nan else b4mgmt['b4EMA5']
      self.mgmt[i]['b4EMA20']      = np.nan if first else self.ema20[i-1]       if not_nan else b4mgmt['b4EMA20']
      self.mgmt[i]['b4EMA40']      = np.nan if first else self.ema40[i-1]       if not_nan else b4mgmt['b4EMA40']
      self.mgmt[i]['b4EMA120']     = np.nan if first else self.ema120[i-1]      if not_nan else b4mgmt['b4EMA120']
      self.mgmt[i]['b4EMA240']     = np.nan if first else self.ema240[i-1]      if not_nan else b4mgmt['b4EMA240']
      self.mgmt[i]['b4Stage']      = np.nan if first else self.stage[i-1]       if not_nan else b4mgmt['b4Stage']
      self.mgmt[i]['b4UpperStage'] = np.nan if first else self.upper_stage[i-1] if not_nan else b4mgmt['b4UpperStage']
      self.mgmt[i]['Amount']       = 0      if first else Amount                if not_nan else b4mgmt['Amount']
      self.mgmt[i]['CountNums']    = self.count_nums
      self.mgmt[i]['EntrySig']     = 0
      self.mgmt[i]['ExitSig']      = 0
      self.mgmt[i]['L/S']          = LS
      self.mgmt[i]['Lot']          = Lot
      self.mgmt[i]['EntryPrice']   = Entry
      self.mgmt[i]['ExitPrice']    = np.nan
      self.mgmt[i]['b4Currency']   = Currency
      self.mgmt[i]['SO']           = self.sosig
      self.mgmt[i]['SO_Close']     = np.nan if first else SO_Close if not_nan else b4mgmt['SO_Close']
      self.mgmt[i]['SO_Entry']     = np.nan if first else SO_Entry if not_nan else b4mgmt['SO_Entry']
      self.mgmt[i]['MTM_']         = 0
      self.mgmt[i]['MTM']          = 0
      self.mgmt[i]['PL_']          = 0
      self.mgmt[i]['PL']           = 0


    def asset_count_n() :
      pass

    def asset_append() :
      if not_nan and not first :
        Total_PL    = self.mgmt[i]['PL']
        Total_MTM   = self.mgmt[i]['MTM']
        Cash        = b4asset['Cash'] + Total_PL
        Net         = Cash + Total_MTM
        MAX_Cash    = Cash if Cash > b4asset['MAX_Cash'] else b4asset['MAX_Cash']
        MAX_Net     = Net  if Net  > b4asset['MAX_Net']  else b4asset['MAX_Net']
        DD_Cash     = Cash - MAX_Cash
        DD_Net      = Net - MAX_Net
        DD_pct_Cash = DD_Cash / MAX_Cash
        DD_pct_Net  = DD_Net / MAX_Net
        UnitValue   = self.principal * self.unit_pct
      self.asset[i]['Time']        = self.df.index[i]
      self.asset[i]['Total_PL']    = 0              if first else Total_PL    if not_nan else b4asset['Total_PL']
      self.asset[i]['Total_MTM']   = 0              if first else Total_MTM   if not_nan else b4asset['Total_MTM']
      self.asset[i]['Cash']        = self.principal if first else Cash        if not_nan else b4asset['Cash']
      self.asset[i]['Net']         = self.principal if first else Net         if not_nan else b4asset['Net']
      self.asset[i]['MAX_Cash']    = self.principal if first else MAX_Cash    if not_nan else b4asset['MAX_Cash']
      self.asset[i]['MAX_Net']     = self.principal if first else MAX_Net     if not_nan else b4asset['MAX_Net']
      self.asset[i]['DD_Cash']     = 0              if first else DD_Cash     if not_nan else b4asset['DD_Cash']
      self.asset[i]['DD_Net']      = 0              if first else DD_Net      if not_nan else b4asset['DD_Net']
      self.asset[i]['DD_pct_Cash'] = 0              if first else DD_pct_Cash if not_nan else b4asset['DD_pct_Cash']
      self.asset[i]['DD_pct_Net']  = 0              if first else DD_pct_Net  if not_nan else b4asset['DD_pct_Net']
      self.asset[i]['UnitValue']   = self.principal * self.unit_pct if first else UnitValue if not_nan else b4asset['UnitValue']

    if Case=='is_nan' or first:
      mgmt_append()
      mgmt_calc_unit()
      asset_append()
      asset_count_n()

      if first and not self.is_nan[i+1] : self.first_time = False

    elif Case=='1st' :
      mgmt_append()
      asset_count_n()

    elif Case=='2nd' :
      self.mark_to_market(i)
      asset_append()
      mgmt_calc_unit()
      asset_count_n()
      self.count_correl_risk(i)


  def entry_sig(self ,i) :
    if self.FILTER=='none' :
      buy_filter = sell_filter = True
    elif self.FILTER=='UpperCycle63' :
      buy_filter  = self.mgmt[i]['b4UpperStage'] == 6 or self.mgmt[i]['b4UpperStage'] == 1
      sell_filter = self.mgmt[i]['b4UpperStage'] == 3 or self.mgmt[i]['b4UpperStage'] == 4

    buy_sig  = self.mgmt[i]['b4Stage'] == 6 or self.mgmt[i]['b4Stage'] == 1
    sell_sig = self.mgmt[i]['b4Stage'] == 3 or self.mgmt[i]['b4Stage'] == 4

    if   all([buy_filter  ,buy_sig  ,self.count_buy()==0])  : self.opsig = self.mgmt[i]['EntrySig'] = 1
    elif all([sell_filter ,sell_sig ,self.count_sell()==0]) : self.opsig = self.mgmt[i]['EntrySig'] = -1


  def exit_sig(self ,i) :
    self.sosig = np.nan

    if self.count_buy() > 0 :
      clbuy = self.mgmt[i-1]['b4Stage'] < 3 and ( self.mgmt[i]['b4Stage'] == 3 or self.mgmt[i]['b4Stage'] == 4 )
      if not self.EXIT2=='none' and self.mgmt[i]['b4Low'] < self.mgmt[i-1][self.EXIT2] : self.sosig = 1
      if any([clbuy ,self.sosig==1 ,self.opsig==-1]) : self.clsig = self.mgmt[i]['ExitSig'] = 1

    elif self.count_sell() > 0 :
      clsell = ( self.mgmt[i-1]['b4Stage'] < 6 and self.mgmt[i-1]['b4Stage'] >= 4 ) and ( self.mgmt[i]['b4Stage'] == 6 or self.mgmt[i]['b4Stage'] == 1 )
      if not self.EXIT2=='none' and self.mgmt[i]['b4High'] > self.mgmt[i-1][self.EXIT2] : self.sosig = 1
      if any([clsell ,self.sosig==1 ,self.opsig==1]) : self.clsig = self.mgmt[i]['ExitSig'] = 1


  def onTick_Base(self ,i) :
    self.append_dict(i)

    if self.is_nan[i] or self.first_time :
      self.management(i ,Case='is_nan')
      return

    self.management(i ,Case='1st')
    self.entry_sig(i)
    self.exit_sig(i)
    self.exit(i)


    self.management(i ,Case='2nd')
    self.entry(i)

class ENTRY_Cycle63__FILTER_none__EXIT1_Cycle36__EXIT2_none(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atrUM
    self.atrN   = 3
    self.EXIT2  = 'none'
    self.FILTER = 'none'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_none__EXIT1_Cycle36__EXIT2_SO_Entry_3_UM(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atrUM
    self.atrN   = 3
    self.EXIT2  = 'SO_Entry'
    self.FILTER = 'none'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_none__EXIT1_Cycle36__EXIT2_SO_Entry_3_20(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atr20
    self.atrN   = 3
    self.EXIT2  = 'SO_Entry'
    self.FILTER = 'none'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_none__EXIT1_Cycle36__EXIT2_SO_Entry_2_UM(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atrUM
    self.atrN   = 2
    self.EXIT2  = 'SO_Entry'
    self.FILTER = 'none'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_none__EXIT1_Cycle36__EXIT2_SO_Entry_2_20(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atr20
    self.atrN   = 2
    self.EXIT2  = 'SO_Entry'
    self.FILTER = 'none'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_none__EXIT1_Cycle36__EXIT2_SO_Close_3_UM(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atrUM
    self.atrN   = 3
    self.EXIT2  = 'SO_Close'
    self.FILTER = 'none'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_none__EXIT1_Cycle36__EXIT2_SO_Close_3_20(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atr20
    self.atrN   = 3
    self.EXIT2  = 'SO_Close'
    self.FILTER = 'none'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_none__EXIT1_Cycle36__EXIT2_SO_Close_2_UM(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atrUM
    self.atrN   = 2
    self.EXIT2  = 'SO_Close'
    self.FILTER = 'none'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_none__EXIT1_Cycle36__EXIT2_SO_Close_2_20(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atr20
    self.atrN   = 2
    self.EXIT2  = 'SO_Close'
    self.FILTER = 'none'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_UpperCycle63__EXIT1_Cycle36__EXIT2_none(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atrUM
    self.atrN   = 3
    self.EXIT2  = 'none'
    self.FILTER = 'UpperCycle63'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_UpperCycle63__EXIT1_Cycle36__EXIT2_SO_Entry_3_UM(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atrUM
    self.atrN   = 3
    self.EXIT2  = 'SO_Entry'
    self.FILTER = 'UpperCycle63'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_UpperCycle63__EXIT1_Cycle36__EXIT2_SO_Entry_3_20(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atr20
    self.atrN   = 3
    self.EXIT2  = 'SO_Entry'
    self.FILTER = 'UpperCycle63'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_UpperCycle63__EXIT1_Cycle36__EXIT2_SO_Entry_2_UM(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atrUM
    self.atrN   = 2
    self.EXIT2  = 'SO_Entry'
    self.FILTER = 'UpperCycle63'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_UpperCycle63__EXIT1_Cycle36__EXIT2_SO_Entry_2_20(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atr20
    self.atrN   = 2
    self.EXIT2  = 'SO_Entry'
    self.FILTER = 'UpperCycle63'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_UpperCycle63__EXIT1_Cycle36__EXIT2_SO_Close_3_UM(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atrUM
    self.atrN   = 3
    self.EXIT2  = 'SO_Close'
    self.FILTER = 'UpperCycle63'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_UpperCycle63__EXIT1_Cycle36__EXIT2_SO_Close_3_20(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atr20
    self.atrN   = 3
    self.EXIT2  = 'SO_Close'
    self.FILTER = 'UpperCycle63'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_UpperCycle63__EXIT1_Cycle36__EXIT2_SO_Close_2_UM(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atrUM
    self.atrN   = 2
    self.EXIT2  = 'SO_Close'
    self.FILTER = 'UpperCycle63'
  def onTick(self ,i) : self.onTick_Base(i)


class ENTRY_Cycle63__FILTER_UpperCycle63__EXIT1_Cycle36__EXIT2_SO_Close_2_20(Strategy_CycleEMA_Base) :
  def __init__(self ,symbol ,np_arr_dic ,df) :
    Strategy_CycleEMA_Base.__init__(self ,symbol ,np_arr_dic ,df)
    self.first_time = True

    self.opcl   = 'close'     
    self.atr    = self.atr20
    self.atrN   = 2
    self.EXIT2  = 'SO_Close'
    self.FILTER = 'UpperCycle63'
  def onTick(self ,i) : self.onTick_Base(i)