Quantxで二種類のシグナルを操る

概要

Quantxでは、色々なテクニカル指標を組み合わせて買いシグナルと売りシグナルを出すことができますが、このシグナルは必ずしも一種類である必要はありません。今回は買いシグナルを二種類作って取引をしてみました。

具体的には、ADX（トレンドの強さを示す指標、後述）が上昇していてかつMACDのゴールデンクロスが起こっている状態を「強シグナル」、ADXは上昇していないけれどMACDのゴールデンクロスが起こっている状態を「弱シグナル」としています。強シグナルでたくさん買って弱シグナルでは控えめに買うことでより高い収益性を目指しました。
注意すべき点として、今回は強シグナルが弱シグナルに内包されているような関係（強が出るときは絶対弱も出る）ので、そのまま注文すると同じ銘柄が同じタイミングで二重に買われてしまうことが挙げられます。少しコードを付け足すだけで簡単に回避することができますが、それが本当に動いているか確かめる必要があると思います。

MACDとは

MACDとはトレンド系テクニカル指標のひとつで、Moving Average Coverage Divergenceの略です。短期指数平滑直線と長期指数平滑曲線の差で計られます。今回はMACDとその移動平均であるMACDシグナルを使ってシグナルを出しています。
MACDについて詳細はこちら：https://qiita.com/katakyo/items/63e4c6461a0c8ac861d6

ADXとは

ADXはオシレーター系テクニカル指標の一つで、トレンドの強さを計るために使われます。計算方法の詳細はこちら：https://www.moneypartners.co.jp/support/tech/dmi-adx.html
ADXが上昇しているときは価格上昇・下降いずれかのトレンドが強く出ていることが分かります。反対に、ADXが下降しているときは強いトレンドがあるわけではないことになります。

実際のコード

こちらのリンクから開くことができます。
https://factory.quantx.io/developer/5d00b240f4ff4e90b6fc3dfbd2a5d4f8

import numpy as np
import pandas as pd
import talib as ta


def initialize(ctx):
    # 設定
    ctx.logger.debug("initialize() called")
    ctx.configure(
      channels={          # 利用チャンネル
        "jp.stock": {
          "symbols": [
            "jp.stock.2914", #JT
            "jp.stock.3382", #セブン&アイ
            "jp.stock.4063", #信越化
            "jp.stock.4452", #花王
            "jp.stock.4502", # 武田
            "jp.stock.4503", #アステラス
            "jp.stock.6098", #リクルート
            "jp.stock.6501", #日立
            "jp.stock.6752", #パナソニック
            "jp.stock.6758", #ソニー
            "jp.stock.6861", #キーエンス
            "jp.stock.6954", #ファナック
            "jp.stock.6981", #村田製
            "jp.stock.7203", #トヨタ
            "jp.stock.7267", #ホンダ
            "jp.stock.7751", #キャノン
            "jp.stock.7974", #任天堂
            "jp.stock.8031", #三井物
            "jp.stock.8058", #三菱商
            "jp.stock.8306", #三菱UFJ
            "jp.stock.8316", #三井住友FG
            "jp.stock.8411", #みずほFG
            "jp.stock.8766", #東京海上
            "jp.stock.8802", #三菱地所
            "jp.stock.9020", #JR東日本
            "jp.stock.9022", #JR東海
            "jp.stock.9432", #NTT
            "jp.stock.9433", #KDDI
            "jp.stock.9437", #NTTドコモ
            "jp.stock.9984", #SBG
          ],
          "columns": [
            "high_price_adj",    # 高値(株式分割調整後)
            "low_price_adj",     # 安値(株式分割調整後)
            "close_price_adj",    # 終値(株式分割調整後) 
          ]
        }
      }
    )

    def _my_signal(data):
      hp = data["high_price_adj"].fillna(method="ffill")
      lp = data["low_price_adj"].fillna(method="ffill")
      cp = data["close_price_adj"].fillna(method="ffill")

      #ADXの計算
      adx = pd.DataFrame(data=0, columns=[], index=cp.index)

      for (sym, val) in cp.items():
        adx[sym] = ta.ADX(hp[sym].values.astype(np.double), 
                    lp[sym].values.astype(np.double),
                    cp[sym].values.astype(np.double), timeperiod = 14)

      #macdの計算
      d_macd = dict() 
      d_macdsignal = dict()
      d_macdhist = dict() 

      for symbol in data.minor_axis:
        macd, macdsignal, macdhist = ta.MACD(cp[symbol].values.astype(np.double), fastperiod=12 , slowperiod=26, signalperiod=9)

        d_macd[symbol] = macd
        d_macdsignal[symbol] = macdsignal
        d_macdhist[symbol] = macdhist

      df_macd = pd.DataFrame(d_macd, index=data.major_axis)        
      df_macdsignal = pd.DataFrame(d_macdsignal, index=data.major_axis)        
      df_macdhist = pd.DataFrame(d_macdhist, index=data.major_axis)

      #シグナル１：ゴールデンクロス、デッドクロス
      goldencross = (df_macd > df_macdsignal) &  (df_macd.shift(1) < df_macdsignal.shift(1))
      deadcross = (df_macd < df_macdsignal) &  (df_macd.shift(1) > df_macdsignal.shift(1))

      #シグナル2：ADX
      buy_ADX = (adx.shift(1)<adx)
      sell_ADX = (adx.shift(1)>adx)

      #シグナル設定
      buy_sig1 =  buy_ADX & goldencross
      buy_sig2 = goldencross
      sell_sig = deadcross

      return {
        "buy:sig1":buy_sig1,
        "buy:sig2":buy_sig2,
        "sell:sig":sell_sig,
        "golden_cross":df_goldencross,
        "dead_cross":df_deadcross,
        "ADX:g2": adx,
     }

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

def handle_signals(ctx, date, current):
    '''
    current: pd.DataFrame
    '''

    done_syms = set([])
    for (sym,val) in ctx.portfolio.positions.items():
        returns = val["returns"]
        if returns < -0.03:
          sec = ctx.getSecurity(sym)
          sec.order(-val["amount"], comment="損切り(%f)" % returns)
          done_syms.add(sym)
        elif returns > 0.05:
          sec = ctx.getSecurity(sym)
          sec.order(-val["amount"], comment="利益確定売(%f)" % returns)
          done_syms.add(sym)

    done_buy_syms = set([])
    df_buy1 = current[current["buy:sig1"]]
    if not df_buy1.empty:
      for (sym, val) in df_buy1.iterrows():
        if (sym in done_syms):
          continue
        else:
          sec = ctx.getSecurity(sym)
          msg = "強買いシグナル"
          sec.order_target_percent(0.3, comment=msg)
          done_buy_syms.add(sym)

    df_buy2 = current[current["buy:sig2"]]
    if not df_buy2.empty:
      for (sym, val) in df_buy2.iterrows(): 
        if (sym in done_buy_syms):
          continue
        elif (sym in done_syms):
          continue
        else:
          sec = ctx.getSecurity(sym)
          msg = "弱買いシグナル"
          sec.order_target_percent(0.3, comment = msg)

    df_sell = current[current["sell:sig"]]
    if not df_sell.empty:
      for (sym, val) in df_sell.iterrows(): 
        if(sym in done_syms):
          continue
        else:
          sec = ctx.getSecurity(sym)
          msg = "弱売りシグナル"
          sec.order_target_percent(0, comment = msg)

コードの説明

QuantXが初めての方、initialize(ctx)等のQuantX固有の関数に不慣れな方はインターン生によるQiita記事（ https://qiita.com/katakyo/items/d786599f6638d53a8602 ） や公式ドキュメント（ https://factory.quantx.io/handbook/ja/ ）をご覧ください。

ADX の計算部分

 #ADXの計算
      adx = pd.DataFrame(data=0, columns=[], index=cp.index)

      for (sym, val) in cp.items():
        adx[sym] = ta.ADX(hp[sym].values.astype(np.double), 
                    lp[sym].values.astype(np.double),
                    cp[sym].values.astype(np.double), timeperiod = 14)

まずadxという名前のデータフレームを作り、そこにtalibを使って計算したadxの値を入れています。talibは様々な指標を簡単に計算してくれるとても便利なライブラリです。公式ドキュメントはこちら：https://mrjbq7.github.io/ta-lib/doc_index.html

MACDの計算部分

#macdの計算
      d_macd = dict() 
      d_macdsignal = dict()
      d_macdhist = dict() 

      for symbol in data.minor_axis:
        macd, macdsignal, macdhist = ta.MACD(cp[symbol].values.astype(np.double), fastperiod=12 , slowperiod=26, signalperiod=9)

        d_macd[symbol] = macd
        d_macdsignal[symbol] = macdsignal
        d_macdhist[symbol] = macdhist

      df_macd = pd.DataFrame(d_macd, index=data.major_axis)        
      df_macdsignal = pd.DataFrame(d_macdsignal, index=data.major_axis)        
      df_macdhist = pd.DataFrame(d_macdhist, index=data.major_axis)

同様にMACDを入れる入れ物を作り、そこにtalibで計算した値を入れます。

シグナル生成部分

#シグナル１：ゴールデンクロス、デッドクロス
      goldencross = (df_macd > df_macdsignal) &  (df_macd.shift(1) < df_macdsignal.shift(1))
      deadcross = (df_macd < df_macdsignal) &  (df_macd.shift(1) > df_macdsignal.shift(1))

#シグナル2：ADX
      buy_ADX = (adx.shift(1)<adx)
      sell_ADX = (adx.shift(1)>adx)

MACDがMACDシグナルを下から上に追い抜いたときにgoldencrossシグナルが、上から下に追い抜いた時にはdeadcrossシグナルが出ます。また、ADXが上昇している時にはbuy_ADXシグナルが、下降しているときにはsell_ADXシグナルが出ます。

シグナル設定部分

#シグナル設定
buy_sig1 =  buy_ADX & goldencross
buy_sig2 = goldencross
sell_sig = deadcross

return {
        "buy:sig1":buy_sig1,
        "buy:sig2":buy_sig2,
        "sell:sig":sell_sig,
        "golden_cross":df_goldencross,
        "dead_cross":df_deadcross,
        "ADX:g2": adx,
     }

goldencrossとbuy_ADXが両方出ている状況をbuy_sig1（強シグナル）、goldencrossのみの状況をbuy_sig2（弱シグナル）、deadcrossが出ている状況をsell_sigとしてreturnします。
ADXはサブチャートに表示したいので、名前を"ADX:g2"としています。

売買処理部分

（損切り、利確の処理部分は省略しています。）

def handle_signals(ctx, date, current):
    done_buy_syms = set([])
    df_buy1 = current[current["buy:sig1"]]
    if not df_buy1.empty:
      for (sym, val) in df_buy1.iterrows():
        if (sym in done_syms):
          continue
        else:
          sec = ctx.getSecurity(sym)
          msg = "強買いシグナル"
          sec.order_target_percent(0.3, comment=msg)
          done_buy_syms.add(sym)

    df_buy2 = current[current["buy:sig2"]]
    if not df_buy2.empty:
      for (sym, val) in df_buy2.iterrows(): 
        if (sym in done_buy_syms):
          continue
        elif (sym in done_syms):
          continue
        else:
          sec = ctx.getSecurity(sym)
          msg = "弱買いシグナル"
          sec.order_target_percent(0.3, comment = msg)
    df_sell = current[current["sell:sig"]]
    if not df_sell.empty:
      for (sym, val) in df_sell.iterrows(): 
        if(sym in done_syms):
          continue
        else:
          sec = ctx.getSecurity(sym)
          msg = "売りシグナル"
          sec.order_target_percent(0, comment = msg)

今回のポイントは、冒頭でも述べたように強シグナルと弱シグナルが同時に出た時に買い注文がダブってしまわないようにすることです！
私はdone_buy_symsというセットを用意して強シグナルで買った銘柄はそこに入れておき、弱シグナルでの購入時にまずdone_buy_symsをチェックしてそこに入っていない銘柄のみ買うという手法を取りました。損切りや利確で使われている方法と同じです。
今回は強シグナルで0.3パーセント買い、弱シグナルで0.15パーセント買い、売りシグナルでポジション全清算に設定しています。

シグナルを二個使う必要はあるの？

強で0.3, 弱で0.15買う設定（今回のアルゴ）での結果は以下のようになりました。

弱シグナルのみで0.45買う設定での結果は

強シグナルのみで0.45買う設定だと

（シグナル回数が異なるのでこの比較は適当ではありませんが）損益率としては悪くないと思います。

まとめ

本記事ではシグナルを二種類出す方法を考えてみました。今回使った手法の有効性は微妙なところですが、参考にしていただけると嬉しいです。