QuantXでBlack-Litterman ModelとBollinger Bandsを組み合わせてみた

はじめに

株式会社Smart Tradeが提供している投資アルゴリズム開発プラットフォームQuantX Factoryで，ポートフォリオ選択についての数理モデルであるBlack-Litterman Modelとテクニカル指標であるBollinger Bandsを用いた取引アルゴリズムを実装しました．

Black-Litterman Modelとは

前回の記事で紹介しているので，以下を参照してください．
[QuantXでBlack-Litterman Modelを実装してみた]
(https://qiita.com/ropomopo/items/0e250697d91b3b74cbf9)

Bollinger Bandsとは

Bollinger Bandsとは，Bollinger Capital Management創立者のJohn Bollingerが1980年代前半に考案した，移動平均線と標準偏差であるσバンドから価格の幅を見ることのできるテクニカル指標で，価格の変動範囲やトレンドにおける過熱のサイン，トレンドの反転を判断する目安になります．¹

以下が参考になります．
[ボリンジャーバンドの使い方・見方　順張りと逆張り両方で使えるテクニカル分析]
(https://ottopilotmedia.com/bollinger/bollinger-use)

実装

コードの全体像

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

def judge_expand(upperband, lowerband, n, m):
  volatility = upperband - lowerband
  vol = volatility.tolist()
  status = [0] * len(vol)
  for i in range(m, len(vol)):
    if m < n:
      if vol[i] > 3 * min(vol[i-n:i-1]):
        status[i] = 1
    else:
      if vol[i] > 3 * min(vol[i-m:i-1]):
        status[i] = 1
  return np.array(status)
	
def judge_plus_two_sigma(upperband, price):
  pts = np.greater_equal(price, upperband)
  return pts.astype(int)

def judge_minus_two_sigma(lowerband, price):
  mts = np.less_equal(price, lowerband)
  return mts.astype(int)

def initialize(ctx):
  ctx.configure(
    channels={
      "jp.stock": {
        "symbols": [
          "jp.stock.2914", #日本たばこ産業
          "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", #三井住友フィナンシャルグループ
          "jp.stock.8411", #みずほフィナンシャルグループ
          "jp.stock.8766", #東京海上ホールディングス
          "jp.stock.8802", #三菱地所
          "jp.stock.9020", #東日本旅客鉄道
          "jp.stock.9022", #東海旅客鉄道
          "jp.stock.9432", #日本電信電話
          "jp.stock.9433", #KDDI
          "jp.stock.9437", #NTTドコモ
          "jp.stock.9984", #ソフトバンクグループ
        ],
        "columns": [
          #"open_price_adj", # 始値(株式分割調整後)
          #"high_price_adj", # 高値(株式分割調整後)
          #"low_price_adj",  # 安値(株式分割調整後)
          #"volume_adj",     # 出来高
          #"txn_volume",     # 売買代金
          "close_price_adj", # 終値(株式分割調整後)
          "market_capitalization", # 時価総額(概算)
        ]
      }
    }
  )
  
  def _my_signal(data):
    cp = data["close_price_adj"].fillna(method="ffill")
    mkt_cap = data["market_capitalization"].fillna(method="ffill")
    
    upperband = {}
    middleband = {}
    lowerband = {}
    
    buy_sig = pd.DataFrame(data=0,columns=[], index=cp.index)
    sell_sig = pd.DataFrame(data=0,columns=[], index=cp.index)
    df_uband = pd.DataFrame(data=0,columns=[], index=cp.index)
    df_lband = pd.DataFrame(data=0,columns=[], index=cp.index)
    expand = pd.DataFrame(data=0,columns=[], index=cp.index)
    minus_two_sigma = pd.DataFrame(data=0,columns=[], index=cp.index)
    plus_two_sigma = pd.DataFrame(data=0,columns=[], index=cp.index)
    
    for (sym, val) in cp.items():
      upperband[sym], middleband[sym], lowerband[sym] = ta.BBANDS(
        cp[sym].values.astype(np.double),
        timeperiod=24,
        nbdevup=2.3,
        nbdevdn=2.3,
        matype=0
      )
      
      expand[sym] = judge_expand(upperband[sym],lowerband[sym],24,48)
      df_lband[sym] = lowerband[sym]
      df_uband[sym] = upperband[sym]
      minus_two_sigma[sym] = judge_minus_two_sigma(df_lband[sym],cp[sym])
      plus_two_sigma[sym] = judge_plus_two_sigma(df_uband[sym],cp[sym])
      
      buy_sig[sym] = minus_two_sigma[sym] * (1 - expand[sym]) + plus_two_sigma[sym]*expand[sym]
      sell_sig[sym] = -1 * (plus_two_sigma[sym] * (1 - expand[sym]) + minus_two_sigma[sym]*expand[sym])
      
    sum_mkt_cap = pd.Series(data=0, index=cp.index)
    weight = pd.DataFrame(data=0, index=cp.index, columns=[])
    daily = pd.DataFrame(data=0, index=cp.index, columns=cp.columns)
    df_w_posterior = pd.DataFrame(data=0, index=cp.index[245:], columns=cp.columns)
    
    for sym, val in cp.items():
      sum_mkt_cap += mkt_cap[sym]
    
    for sym, val in cp.items():
      weight[sym] = mkt_cap[sym] / sum_mkt_cap
      
    # 2x30 (投資家のビュー)
    P = np.array(
          [
            [-0.5,-0.25,0,0,0,0,0,-0.25,-0.75,0,0,0,0,0,-0.75,0,0,0,0,0,-0.25,0,0.25,0.25,0.5,0.75,0,0,0,1],
            [-0.25,-0.5,0,0,0,0,0,0,0,0,1,0,-0.75,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,-0.75,1]
          ]
        )
        
    # 2×1（投資家のビュー）
    Q = np.array([[0.05], [0.07]])
    
    # 2×1（投資家のビューへの確信度）
    omega = np.array(
              [
                [0.001065383, 0],
                [0, 0.001851738]
              ]
            )
            
    for i in daily.columns:
      daily[i] = cp[i].pct_change()
    
    sigma = daily[:245].cov().as_matrix()
    
    delta = 25
    tau = 0.05
    ganma = 0.3
    
    count = 0
    w_matrix =[]
    
    for index, row in weight[245:].iterrows():
      row_list = list(row)
      count += 1
      w_out = []
      for num, i in enumerate(row_list):
        w_in = []
        w_in.append(row_list[num])
        w_out.append(w_in)
      w = np.array(w_out)
      
      r_eq = delta * np.dot(sigma, w)
      r_posterior = r_eq + np.dot(np.dot(tau*np.dot(sigma, P.T), np.linalg.inv(tau*np.dot(np.dot(P, sigma), P.T) + omega)), (Q - np.dot(P, r_eq)))
      sigma_posterior = sigma + tau * sigma - tau * np.dot(np.dot(np.dot(sigma, P.T), np.linalg.inv(tau*np.dot(np.dot(P, sigma), P.T)+omega)), tau*np.dot(P, sigma))
      w_posterior = np.dot(np.linalg.inv(delta * sigma_posterior), r_posterior).T
      w_matrix = np.append(w_matrix, w_posterior)
      
    w_matrix_rt = w_matrix.reshape(count, 30).T
    
    for sym, val in zip(weight[245:].keys(), w_matrix_rt):
      post = pd.Series(val, index=cp[245:], name = sym)
      df_w_posterior[sym] = post.values
      
    return {
      "upperband:price": df_uband,
      "lowerband:price": df_lband,
      "buy:sig": buy_sig,
      "sell:sig": sell_sig,
      "weight:g2": weight,
      "weight_post:g2": df_w_posterior,
      "weight_comp:g2": (1 - ganma) * weight + ganma * df_w_posterior,
      "weight_diff:g2": weight - df_w_posterior,
    }
    
  ctx.regist_signal("my_signal", _my_signal)
  
def handle_signals(ctx, date, current):
  df = current.copy()
  buy = df["buy:sig"].dropna()
  sell = df["sell:sig"].dropna()
  buy = buy[buy != 0]
  sell = sell[sell != 0]
  
  if not buy.empty:
    for (sym, val) in buy.items():
      sec = ctx.getSecurity(sym)
      sec.order_target_percent(df["weight_comp:g2"][sym])
      
  if not sell.empty:
    for (sym, val) in sell.items():
      sec = ctx.getSecurity(sym)
      sec.order_target_percent(0)

def judge_expand(upperband, lowerband, n, m):
  volatility = upperband - lowerband
  vol = volatility.tolist()
  status = [0] * len(vol)
  for i in range(m, len(vol)):
    if m < n:
      if vol[i] > 3 * min(vol[i-n:i-1]):
        status[i] = 1
    else:
      if vol[i] > 3 * min(vol[i-m:i-1]):
        status[i] = 1
  return np.array(status)
	
def judge_plus_two_sigma(upperband, price):
  pts = np.greater_equal(price, upperband)
  return pts.astype(int)

def judge_minus_two_sigma(lowerband, price):
  mts = np.less_equal(price, lowerband)
  return mts.astype(int)

以下の記事にアルゴリズムの説明があります．
[QuantXでボリンジャーバンドのアルゴリズムを改良してみる]
(https://qiita.com/geoha6/items/94bc8b134c06f8b51911)

def handle_signals(ctx, date, current):
  df = current.copy()
  buy = df["buy:sig"].dropna()
  sell = df["sell:sig"].dropna()
  buy = buy[buy != 0]
  sell = sell[sell != 0]
  
  if not buy.empty:
    for (sym, val) in buy.items():
      sec = ctx.getSecurity(sym)
      sec.order_target_percent(df["weight_comp:g2"][sym])
      
  if not sell.empty:
    for (sym, val) in sell.items():
      sec = ctx.getSecurity(sym)
      sec.order_target_percent(0)

買いシグナルが出ているときは，その銘柄にBlack-Litterman Modelによる割合で注文を行います．
また，売りシグナルが出ているときは，その銘柄の保有する割合を0%にします．

実験

期間 2018-05-21 ~ 2019-05-21
銘柄 TOPIX Core30

結果

時価総額による単純な重みづけのみ
https://factory.quantx.io/developer/3c90e539586d4ec3a0903877b8f3a9cc

Black-Litterman Modelのみ
https://factory.quantx.io/developer/9deac3f08ff44b4ca49be39f1888578f

Bollinger Bandsのみ
https://factory.quantx.io/developer/54e674cce92f4e3bb94c8d7fcf273a53

時価総額による単純な重みづけとBollinger Bandsの組み合わせ
https://factory.quantx.io/developer/c6a1803d3af843638e4462c665b220f6

Black-Litterman ModelとBollinger Bandsの組み合わせ
https://factory.quantx.io/developer/ed5868112cdd48fe9fa81998f9ee93c1

おわりに

Black-Litterman Modelとテクニカル分析の組み合わせの可能性を感じました．また，投資家のビューをそれなりに適当なものにできると，効果を発揮することが実感できました．

下記のURLからQuantX Factory上で今回紹介したコードのバックテストを行うことが出来ます．
https://factory.quantx.io/developer/ed5868112cdd48fe9fa81998f9ee93c1

参考文献

以下の記事を参考にしています．
ブラック・リッターマンモデルによる資産配分を解説してみる（Pythonによる実行例つき）

#免責注意事項
この記事のコードや知識を使った実際の取引で生じた損益に関しては一切の責任を負いかねますので御了承下さい．

1. https://www.bollingerbands.com/ ↩