QuantXでコポック買い指標を実装してみた

今回やったこと

投資アルゴリズム開発プラットフォーム「QuantX」で、コポック買い指標を実装してみました。

コポック買い指標とは

コポック買い指標は、主に月足ベースにして計算する指標で、中期的な株価の上下で儲けるのに役立ちます。
まず、それぞれの月で１ヶ月間の平均株価を求め、それが前年の同じ月からどの程度上下したか求めます。この値を対前年騰落率と呼び、これに加重平均をとって、コポック買い指標を求めます。
$$コポック買い指標 = \frac{1}{10}\Bigl( 1×9ヶ月前の対前年騰落率 + 2×8ヶ月前の対前年騰落率 + ・・・ + 9×1ヶ月前の対前年騰落率 + 10×今月の対前年騰落率\Bigr)$$

アルゴリズム

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

import pandas as pd 
import numpy as np

def initialize(ctx):
    # 設定
    ctx.logger.debug("initialize() called")
    ctx.configure(
      channels={          # 利用チャンネル
        "jp.stock": {
          "symbols": [
            "jp.stock.2317", 
            "jp.stock.3150", 
            "jp.stock.3665", 
            "jp.stock.3687",
            "jp.stock.3836", 
            "jp.stock.3923", 
            "jp.stock.3962",
            "jp.stock.6035",
            "jp.stock.6046", 
            "jp.stock.6050",
            "jp.stock.6055", 
            "jp.stock.6067",
            "jp.stock.6200", 
            "jp.stock.6920",
          ],
          "columns": [
            #"open_price_adj",    # 始値(株式分割調整後)
            #"high_price_adj",    # 高値(株式分割調整後)
            #"low_price_adj",     # 安値(株式分割調整後)
            #"volume_adj",         # 出来高
            #"txn_volume",         # 売買代金
            #"close_price",        # 終値
            "close_price_adj",    # 終値(株式分割調整後) 
          ]
        }
      }
    )

    def _my_signal(data):
      cp = data['close_price_adj'].fillna(method='ffill')
      window_length = 10
      cp = cp.set_index([cp.index.year, cp.index.month, cp.index])
      cp.index.names = ['year', 'month', 'date']
      cp_month = cp.mean(level=['year', 'month'])
      cp_month = (cp_month - cp_month.shift(12)) / cp_month * 100
      Cppockindicator = pd.DataFrame(None, index = cp_month.index, columns = cp_month.columns)

      my_weight = [1,2,3,4,5,6,7,8,9,10]
      for (sym,val) in cp_month.items():
        for i in range(len(cp_month)):
          if i < (12+window_length):
            Cppockindicator[sym][i] = None
          else:
            Cppockindicator[sym][i] = np.average(cp_month[sym].iloc[i - window_length:i], weights = my_weight)
      Cppockindicator_diff = Cppockindicator.dropna().diff()

      sig = pd.DataFrame(0.0, index=cp.index, columns=cp.columns)
      for idx1 in Cppockindicator_diff.index.levels[0]:
        for idx2 in Cppockindicator_diff.index.levels[1]:
          if Cppockindicator_diff[(Cppockindicator_diff.index.get_level_values(0) == idx1) & (Cppockindicator_diff.index.get_level_values(1) == idx2)].empty:
            continue
          if (Cppockindicator_diff[(Cppockindicator_diff.index.get_level_values(0) == idx1) & (Cppockindicator_diff.index.get_level_values(1) == idx2)].values != Cppockindicator_diff[(Cppockindicator_diff.index.get_level_values(0) == idx1) & (Cppockindicator_diff.index.get_level_values(1) == idx2)].values).all():
            continue
          insert_value = Cppockindicator_diff[(Cppockindicator_diff.index.get_level_values(0) == idx1) & (Cppockindicator_diff.index.get_level_values(1) == idx2)].values
          sig[(sig.index.get_level_values(0) == idx1) & (sig.index.get_level_values(1) == idx2)] = insert_value[0,:]

      sig = sig.reset_index(level = ('year', 'month'), drop=True)

      buy_sig = sig[(sig > 0) & (sig.shift(30) < 0)]
      sell_sig = sig[(sig < 0) & (sig.shift(30) > 0)]

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

    # シグナル登録
    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"]
        #ctx.logger.debug("%s %f" % (sym, 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_target_percent(0.1, comment="SIGNAL BUY")
        pass

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

        sec = ctx.getSecurity(sym)
        sec.order_target_percent(0, comment="SIGNAL SELL")
        pass

QuantXは日足のデータしか用意されていないので、月足データをMultiIndexを用いて計算しましたが、最後売買シグナルを日足の形に戻さなければならず何気に時間食いました。。
QuantXでは1500日以内のバックテストしか実行できないので、シグナル回数も少なく、あまり有意味な数字にはならなそうです。