QuantXでサイコロジカルラインを実装してみた

今回やったこと

投資アルゴリズム開発プラットフォーム「QuantX」で、サイコロジカルラインを実装してみました。

サイコロジカルラインとは

サイコロジカルラインとは、人間の心理に着目したテクニカル指標です。
株価が何日も連続して上昇すると、「そろそろ上昇が止まるのではないか」と考える人が増え、実際に上昇が止まることが多くなります。逆に、株価が何日も連続して下落すると、「そろそろ上がるのでは」と考える人が増え、実際に上昇に転じることが多くなります。

そこで、サイコロジカルラインでは、ある期間のうち、株価が上昇した日数の割合を求めて、売買を判断します。

$$サイコロジカルライン = \frac{計算対象期間の株価上昇日の日数}{計算対象期間の日数} × 100$$
ここでの計算期間は、12日にするのが一般的です。
例えば、12日中9日株価が上昇した場合には、サイコロジカルラインは75%となります。

アルゴリズム

今回書いたアルゴリズムはこちら

import pandas as pd 
import numpy as np 

def initialize(ctx):
    # 設定
    ctx.logger.debug("initialize() called")
    ctx.configure(
      channels={          # 利用チャンネル
        "jp.stock": {
          "symbols": [
            "jp.stock.1305",
            "jp.stock.9984",
            "jp.stock.9983",
            "jp.stock.7201",
            "jp.stock.9201",
            "jp.stock.9202",
            "jp.stock.7203"
          ],
          "columns": [
            #"open_price_adj",    # 始値(株式分割調整後)
            #"high_price_adj",    # 高値(株式分割調整後)
            #"low_price_adj",     # 安値(株式分割調整後)
            #"volume_adj",         # 出来高
            #"txn_volume",         # 売買代金
            "close_price",        # 終値
            "close_price_adj",    # 終値(株式分割調整後) 
          ]
        }
      }
    )

    def count_plus(x):
      return np.sum(x>0)

    def _my_signal(data):
      cp = data['close_price_adj'].fillna(method = 'ffill')
      window_length = 12
      phycological_line = pd.DataFrame(0, index=cp.index, columns=cp.columns)
      phycological_line = cp.diff()
      ctx.logger.debug(phycological_line)
      for (sym,val) in cp.items():
        phycological_line[sym] = pd.rolling_apply(phycological_line[sym], window_length, count_plus)/window_length*100

      ctx.logger.debug(phycological_line)
      buy_sig = phycological_line[phycological_line < 30]
      sell_sig = phycological_line[phycological_line > 70]

      return {
        'buy:sig':buy_sig,
        'sell:sig':sell_sig,
        'phycological_line:g2':phycological_line
      }

    # シグナル登録
    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)


    buy = current["buy:sig"].dropna()
    for (sym,val) in buy.items():
        if sym in done_syms:
          continue

        sec = ctx.getSecurity(sym)
        sec.order(sec.unit() * 1, comment="SIGNAL BUY")
        ctx.logger.debug("BUY: %s,  %f" % (sec.code(), val))
        pass

    sell = current["sell:sig"].dropna()
    for (sym,val) in sell.items():
        if sym in done_syms:
          continue

        sec = ctx.getSecurity(sym)
        sec.order(sec.unit() * -1, comment="SIGNAL SELL")
        ctx.logger.debug("SELL: %s,  %f" % (sec.code(), val))
        pass

今回は勝率70%以上のとき、すなわち9勝3敗のときに売り、3勝9敗のときに買いとして、シグナルを設定しています。
このあたりの数字は銘柄によって範囲を調整する必要があるようです。