ブル・ベアファンド自動裁定 〜レンジ相場を乗り切る〜

導入

人気のあるETFブルベア型の取引の自動裁定システムをQuantX Factory上で開発します。
シリーズ化していますので

「QuantXって何？」
「ブルベア取引って？」
「まず名を名乗れよ。」

って方は以下のURLの1を参照下さい。

1. ブル・ベアファンド自動裁定 〜はじめの一歩〜
2. ブル・ベアファンド自動裁定 〜利確損切り機能実装〜
3. ブル・ベアファンド自動裁定 〜QuantX APIで買い増し機能実装〜
4. ブル・ベアファンド自動裁定 〜QuantX APIでリスク回避・保有日数編〜
5. ブル・ベアファンド自動裁定 〜買い増し機能を柔軟に〜
6. ブル・ベアファンド自動裁定 〜QuantX APIでリスク回避・保有率編〜（前回）

今回の目標

前回ではそろそろhandle_signalをいじることでの改善はそろそろ限界な気がしてきたので(実は飽きた)
今回はシグナルの出し方を工夫してみようかと思います。
そこで今回は

レンジ相場を認識して回避して取れるときに利益をだす
前回のコードはこちら

という機能の実装を目指します。

具体的には以下のようなアルゴリズムです

• ある指標から計算された数値からレンジ相場かどうかを判断する
• レンジ相場と考えられる時は取引を行わない

レンジ相場について

早速やっていく前にレンジ相場について説明します。

レンジ相場

上下のトレンドのどちらかであり、ある程度の判断が効くトレンド相場と比べて
レンジ相場とはもみ合い相場やボックス相場ともいいチャートが一定の価格帯(レンジ)
を行き来しているどちらのトレンドだか読みきれないよく分からない相場のことです。イメージ図で言うなれば以下の四角い（ボックス）範囲の中の相場のことでしょうか。

この相場は上昇下落どちらのトレンドもなくつかみどころのないもみ合いの様な状態となりこの
相場にエントリーするのは買いエントリーだろうが売りエントリーだろうがリスキーです。

逆にこのレンジ相場（ボックス）を抜け出す時はチャンス（ブレイク）でどちらかの上昇下落どちらかの方向へ動きます。（上図参照）

レンジ相場の認識のしかた

移動平均線のもつれ具合

複数の期間の移動平均線がもつれている時をレンジ相場とします。
イメージとしては以下の画像の赤い範囲でしょうか。

これを実装するにはどういう状況を「もつれている」とプログラムに判断させるかが課題でしょう。
例えば、一定期間内に何回交差したかとか、それぞれの線の距離とか？線の傾きによる平行さ加減とか？

これは一旦置いといて今後考える事にします。

RSIによる判断

RSIの値が４０％〜６０％で推移しているとレンジ相場であると考えられる。
RSIに関してはこちら
イメージとしては以下の様

これは何日連続で４０％〜６０％を推移するかで実装できそうですね。

高値低値によるチャートパターン認識

過去の高値低値からチャートパターンを判断します。

ここでは割愛しますがこのパターン認識をさせる事で細かな値動きを判断することが出来そうです。
詳しくはこちら

BBANDsのバンド幅

ボリンジャーバンドの線の幅が狭まり、横ばいになっている時レンジ相場だと考えられる。
イメージとしては画像の様でしょうか。
バンドが狭ばれば狭ばる程市場のエネルギーが蓄積しておりブレイクしトレンド相場移行した際に大きな動きを示すことが多いです。
ちなみにボリンジャーバンドについては終値が上のバンドに近づいたら売り、下のバンドに近づいたら買いです。

今回はこれの幅に関して実装、判断をします。

じゃあやっていく

一応コードの全体像を眺めておきましょう、今回はこのコードの

my_signals

の部分をいじって行く形になります。

コードの全体像

#必要なライブラリの読み込み
import numpy as np
import pandas as pd
import talib as ta

#RCIを計算する関数
def get_rci(close, period):
      rank_period = np.arange(period, 0, -1)
      length = len(close)
      rci = np.zeros(length)
      for i in range(length):
          if i < period - 1:
              rci[i] = 0
          else :
              rank_price = close[i - period + 1: i + 1].rank(method='min', ascending = False).values
              rci[i] = (1 - (6 * sum((rank_period - rank_price)**2)) / (period**3 - period)) * 100
      return rci

def initialize(ctx):
    # 設定
    ctx.logger.debug("initialize() called")
    ctx.configure(
      channels={          # 利用チャンネル
            "jp.stock": {
                "symbols": [
                    "jp.stock.1321", #日経225連動型上場投資信託
                    "jp.stock.1357", #日経ダブルインバース上場投信 bear
                    "jp.stock.1570", #日経平均レバレッジ上場投信 bull
                ],
                "columns": [
                  "close_price_adj"
                ]
            }

      }
    )
    ctx.localStorage = {
      "jp.stock.1570":0,
      "jp.stock.1357":0
    }



    def _my_signal(data):
      cp = data["close_price_adj"].fillna(method='ffill')
      # ctx.logger.debug(cp)
      rci9 = pd.DataFrame(0, index=cp.index, columns=cp.columns)
      rsi7 = pd.DataFrame(0,index=cp.index, columns=cp.columns)

      for (sym,val) in cp.items():
        rci9[sym]=get_rci(cp[sym], 9)
        rsi7[sym] = ta.RSI(cp[sym].values.astype(np.double), timeperiod=7)
      # ctx.logger.debug(rci9)
      # ctx.logger.debug(rsi7) 
      buy_sig= (rci9 < -80) | (rsi7 < 30)
      sell_sig= (rci9 > 80) | (rsi7 > 70)

      # ctx.logger.debug(buy_sig)
      # ctx.logger.debug(buy_sig.shift(1))
      return {
        "incbull:sig":buy_sig.shift(1),
        "incbear:sig":sell_sig.shift(1),
        "buy:sig":buy_sig,
        "sell:sig":sell_sig,
        "rci9:g2":rci9,
        "rsi7:g2":rsi7,
      }

    # シグナル登録
    ctx.regist_signal("my_signal", _my_signal)

def handle_signals(ctx, date, current):
    '''
    current: pd.DataFrame
    '''
    done_syms = set([])

    # ctx.logger.debug(current)
    # ctx.logger.debug(ctx.localStorage["bulldone"])
    bear = ctx.getSecurity("jp.stock.1357")
    bull = ctx.getSecurity("jp.stock.1570")
    df_buy = current["buy:sig"][0]
    df_sell = current["sell:sig"][0]
    df_inc_bull = current["incbull:sig"][0]
    df_inc_bear = current["incbear:sig"][0]

    # ctx.logger.debug(ctx.portfolio.positions.keys())
    # 取引されないと辞書ctx.portfolio.positionsの中に銘柄と各種データは入らず参照しようとするとKeyErrorになる
    # (sym,val) = ctx.portfolio.positions
    # if ctx.portfolio.positions != {}:
    #   ctx.logger.debug(ctx.portfolio.positions["jp.stock.1357"])

    #利確損切り
    for (sym,val) in ctx.portfolio.positions.items():
        if (sym == 'jp.stock.1321') | ('jp.stock.1357' in done_syms) | ('jp.stock.1570' in done_syms):
          continue
        if val["portfolio_ratio"] > 0.65:
          ctx.profit_taking = 0.02
          ctx.losscut = -0.01
        else:
          ctx.profit_taking = 0.048
          ctx.losscut = -0.02
        returns = val["returns"]
        if returns > ctx.profit_taking:
          sec = ctx.getSecurity(sym)
          sec.order(-val["amount"], comment="利益確定売(%f)" % returns)
          ctx.localStorage[sym] = 0
          done_syms.add(sym)
          # ctx.logger.debug(ctx.localStorage[sym])
        elif returns < ctx.losscut:
          sec = ctx.getSecurity(sym)
          sec.order(-val["amount"], comment="損切り(%f)" % returns)
          ctx.localStorage[sym] = 0
          done_syms.add(sym)
    # 買いシグナル
    # ctx.logger.debug(df_buy)

    # ctx.logger.debug(ctx.localStorage["jp.stock.1570"])

    if df_buy & df_inc_bull & (ctx.localStorage["jp.stock.1570"] == 1):
      if not ('jp.stock.1357' in done_syms) | ('jp.stock.1570' in done_syms):
        # ctx.logger.debug(ctx.portfolio.positions.keys())
        if 'jp.stock.1570' in ctx.portfolio.positions:
          bull.order(ctx.portfolio.positions['jp.stock.1570']['amount']*0.08, comment="BULL買い増し")
          # bull.order(bull.unit() * 10, comment="BULL買い増し")
          done_syms.add("jp.stock.1570")
    if df_buy & (ctx.localStorage["jp.stock.1570"] == 0):
      if not ('jp.stock.1357' in done_syms) | ('jp.stock.1570' in done_syms):
        bull.order_target_percent(0.50, comment="BULL BUY")
        bear.order_target_percent(0, comment="BEAR SELL")
        done_syms.add("jp.stock.1570")
        ctx.localStorage["jp.stock.1570"] = 1
        ctx.localStorage["jp.stock.1357"] = 0

    # 売りシグナル
    if df_sell & df_inc_bear & (ctx.localStorage["jp.stock.1357"] == 1):
      if not ('jp.stock.1357' in done_syms) | ('jp.stock.1570' in done_syms):
        if 'jp.stock.1357' in ctx.portfolio.positions:
          # ctx.logger.debug(ctx.portfolio.positions.keys())
          bear.order(ctx.portfolio.positions['jp.stock.1357']['amount']*0.08, comment="BEAR買い増し")
          # bear.order(bear.unit() * 10, comment="BEAR買い増し")
          done_syms.add("jp.stock.1357")
    if df_sell & (ctx.localStorage["jp.stock.1357"] == 0):
      if not ('jp.stock.1357' in done_syms) | ('jp.stock.1570' in done_syms):
        bear.order_target_percent(0.50, comment="BEAR BUY")
        bull.order_target_percent(0, comment="BULL SELL") 
        done_syms.add("jp.stock.1357")
        ctx.localStorage["jp.stock.1357"] = 1
        ctx.localStorage["jp.stock.1570"] = 0

ボリンジャーバンドの実装

とりあえずアッパーバンドとミドルバンド、ロワーバンドとかの容れ物を作ります

upperband={}
lowerband={}
middleband={}

      bb_ratio_buy = pd.DataFrame(data=0, columns=[], index=cp.index) 
      bb_ratio_sell = pd.DataFrame(data=0, columns=[], index=cp.index) 
      df_bb_buy = pd.DataFrame(data=0, columns=[], index=cp.index)
      df_bb_sell = pd.DataFrame(data=0, columns=[], index=cp.index)
      uband = pd.DataFrame(data=0, columns=[], index=cp.index) 
      lband = pd.DataFrame(data=0, columns=[], index=cp.index)
      range_bands = pd.DataFrame(data=0, columns=[], index=cp.index)
      df_range_bands = pd.DataFrame(data=0, columns=[], index=cp.index)

Talibのライブラリによって
アッパーバンドとミドルバンド、ロワーバンドの計算をします、for文の中で以下を書き加えます。

upperband[sym], middleband[sym], lowerband[sym] = ta.BBANDS(cp[sym].values.astype(np.double),
             timeperiod=13, nbdevup=2.2, nbdevdn=2.2)

ここで計算しているのは13日の「平均値±標準偏差×2.2」である±2.2σの上下バンドですこれを

        bb_ratio_buy[sym] = (lowerband[sym]/ cp[sym])
        bb_ratio_sell[sym] = (cp[sym]/ upperband[sym])
        df_bb_buy[sym] = bb_ratio_buy[sym] >= 0.99
        df_bb_sell[sym] = bb_ratio_sell[sym] >= 0.9998
        lband[sym] = lowerband[sym]
        uband[sym] = upperband[sym]
        range_bands[sym] = upperband[sym] - lowerband[sym]

として
bb_ratio_buy[sym] = (lowerband[sym]/ cp[sym])が0.99以上つまり上のバンドに終値がかなり近づいている時に買い
bb_ratio_sell[sym] = (cp[sym]/ upperband[sym])が0.9998以上つまり下のバンドに終値がかなり近づいている時に売り
とします。

また、ここでTalibの上下バンドの計算結果をのちにデバッグ用のグラフに描画するために
lband,ubandとして格納する。
また後に使う上下バンドの差をrange_bandsとして格納します。

ボリンジャーバンドの幅の収縮の判断

まずボリンジャーバンドの幅の移動平均を出します。

range_average = range_bands.apply(lambda s: ta.SMA(s.values.astype(np.double), timeperiod=30), axis=0)

ここでは30日間の単純移動平均線を計算しています。
次は幅によるレンジ相場の判断を行います。

df_range_bands_sell = range_bands / range_average > 0.97
df_range_bands_buy = range_bands / range_average > 0.97

ここではrange_bands / range_averageが0.97より大きい時(ほぼ平均のバンド幅かそれより大きい時)のみシグナルを出しそれ以下の場合はバンドが収縮している(レンジ相場)として取引は行わないということです。

buy_sig= df_bb_buy & df_range_bands_buy 
sell_sig= df_bb_sell & df_range_bands_sell 

以上の様にボリンジャーバンドの売買シグナルが出ている時かつバンドが収縮していない(レンジ相場でない)時に取引を行うというシグナルの出し方になります。

最後にreturnでそれぞれ使うものを返しておきます。

return {
        "incbull:sig":buy_sig.shift(1),
        "incbear:sig":sell_sig.shift(1),
        "buy:sig":buy_sig,
        "sell:sig":sell_sig,
        # "ma20":ma20_cp,
        # "ma50":ma50_cp,
        # "ma100":ma100_cp,
        "rci9:g2":rci9,
        "rsi14:g2":rsi14,
        "upperband:g2":uband, #この形で渡さないとエラる
        "lowerband:g2":lband, #この形で渡さないとエラる
      }

(ma20とかは気にしないでください)

これで完成です。
完成のコードはこちら

結果

レンジ相場判断なし

レンジ相場判断あり

考察

レンジ相場も回避出来ているがそれより大きく損益に貢献しているのはブレイクする時を上手く捉えられているところだと考えられる。
pandasのpct_change関数でバンド幅の変化具合を調べてより正確にブレイクを捉えられるか？
当たり前だがある程度のバンドの幅以下(レンジ相場?)は取引しないので取引回数は少なくなっている。

今後の展望

ある程度のそこまでバンド幅が縮まっていない時(ある程度のレンジ相場)でも取引を行う様なものにして利益をさらに上げることを目指す。

宣伝

もくもく会もやってるよ

日時：毎週水曜日18時〜
場所：神田　千代田共同ビル４階　SmartTrade社オフィス
内容：基本黙々と自習しながら猛者の方に質問して強くなっていく会
備考：お菓子と終わりにお酒が出るよ

詳細はこちらだよ

Pythonアルゴリズム勉強会HP：https://python-algo.connpass.com/
(connpassって言うイベントサイトに飛びます)

免責注意事項

このコード・知識を使った実際の取引で生じた損益に関しては一切の責任を負いかねますので御了承下さい。

materials

bull-bear:Designed by Rawpixel.com

参考文献

単純明快！レンジ相場を簡単に判断するインジゲーターとは？
9つの代表的なパターン