「2位じゃダメなんでしょうか」戦略

nelo

「2位じゃダメなんでしょうか」の答えを株式市場に求めてみる


・2位じゃダメかどうかの評価方法

・業種ごとに浮動株ベースの時価総額が１位の銘柄群、２位の銘柄群をそれぞれの投資ユニバースとする
・３種類の投資戦略でバックテストを行いパフォーマンスを比較。

・データ

浮動株ベースの時価総額はTOPIXウェイトを参考にする。
https://www.jpx.co.jp/markets/indices/topix/index.html

import matplotlib.pyplot as  plt
import pandas as pd
import numpy as np
import requests 
import io
import yfinance as yf
from scipy.optimize import minimize
#ウェイト取得
url = 'https://www.jpx.co.jp/markets/indices/topix/tvdivq00000030ne-att/topixweight_j.csv'
res = requests.get(url)
df = pd.read_csv(io.StringIO(res.text),sep=',').dropna()
df["TOPIXに占める個別銘柄のウエイト"] = df["TOPIXに占める個別銘柄のウエイト"].str.strip("%").astype(float)
df = df.drop(["日付","ニューインデックス区分"],axis=1)

# ユニバース
max_d = df.groupby('業種')['TOPIXに占める個別銘柄のウエイト'].max().reset_index()
first = pd.merge(max_d, df, on=['業種', 'TOPIXに占める個別銘柄のウエイト'], how='left')
max_d = df[['TOPIXに占める個別銘柄のウエイト',"業種"]].groupby("業種").apply(lambda x: x.nlargest(2,"TOPIXに占める個別銘柄のウエイト").iloc[-1]).drop("業種",axis=1).reset_index()
second = pd.merge(max_d, df, on=['業種', 'TOPIXに占める個別銘柄のウエイト'], how='left')
ticker = pd.concat([first["コード"],second["コード"]])

左：時価総額業種１位の銘柄、　右：2位の銘柄


株価データはyahoo finance apiを利用。

#価格取得
st_dt = '2000-01-01'
ed_dt = '2024-12-31'
data = pd.DataFrame()
for i in ticker.values:
    tmp = yf.download(i+'.T', start=st_dt, end=ed_dt)["Adj Close"]
    try:
        data = pd.concat([data,tmp],axis=1)
    except:
        data = tmp
data.columns = ticker.values
data =data.applymap(lambda x:float(x))
data = data.sort_index()
data.index = pd.to_datetime(data.index)
data= data.dropna()
###インデックス
idx = pd.date_range(data.index[0],data.index[-1],freq="BM")

### リターン
ret_d =  data.pct_change().dropna()
ret_m = data.resample("BM").last().pct_change().dropna()

first_ret_d=  ret_d[first["コード"]]
second_ret_d =  ret_d[second["コード"]]
first_ret_m=  ret_m[first["コード"]]
second_ret_m =  ret_m[second["コード"]]

def stats(ret):
    def rtn(ret):
        out_ret = (1+ret).cumprod().iloc[-1] ** (12/len(ret))-1
        return out_ret
    def rsk(ret):
        out_risk = ret.std() *np.sqrt(12)
        return out_risk
    return pd.DataFrame([rtn(ret),rsk(ret),rtn(ret)/rsk(ret)],index=["Return","Risk","Ratio"])


・バックテスト

TOPIXの業種ウェイト、最小分散、リスクパリティの３種類の投資戦略でバックテストを実施してパフォーマンスを比較。

・TOPIX業種ウェイト

直近のTOPIXウェイトで固定してリターン計測。

↓業種別ウェイト

#業種ウェイト
indus = df.drop(["コード","銘柄名"],axis=1).groupby("業種").sum()
w_indus = indus.copy().T/100

ret_bm_first = (first_ret_m.loc[idx[12*3:]] *w_indus.values).sum(axis=1)
ret_bm_second = (second_ret_m.loc[idx[12*3:]] *w_indus.values).sum(axis=1)
cumret_bm_first = (1+ret_bm_first).cumprod()
cumret_bm_second = (1+ret_bm_second).cumprod()

plt.plot(cumret_bm_first,label="first")
plt.plot(cumret_bm_second,label="second")
plt.legend()
plt.show()
stat_bm = pd.concat([stats(ret_bm_first),stats(ret_bm_second)],axis=1)
stat_bm.columns = ["1stBM","2ndBM"]
stat_bm

統計量。リターンは2位が勝利。レシオは１位が勝利


・最小分散

def minvar(ret):
    cov = ret.cov()
    w =  np.array([1/len(cov)] * len(cov))

    def Var(w,cov):
        var = np.dot(np.dot(w.T, cov), w)
        return var 

    cons = ({'type': 'ineq', 'fun': lambda w: w},{'type': 'eq', 'fun': lambda w: w.sum()-1})

    result =  minimize(Var, w, args=(cov,),constraints=cons, tol=1e-10)
    out  = pd.DataFrame(result.x)
    out.index = ret.columns
    out.columns = ["最小分散ウェイト"]
    return out
w_minvar_first = pd.DataFrame(data=None, index=idx[12*3:], columns=first_ret_d.columns)
w_minvar_second = pd.DataFrame(data=None, index=idx[12*3:], columns=second_ret_d.columns)
for i in range(12*3,len(idx)):
   w_minvar_first.loc[idx[i]] = minvar(first_ret_d.loc[idx[i-12*3]:idx[i]]).T.values
   w_minvar_second.loc[idx[i]] = minvar(second_ret_d.loc[idx[i-12*3]:idx[i]]).T.values

ret_minvar_first = (first_ret_m.shift(1).dropna() *w_minvar_first).sum(axis=1)
ret_minvar_second = (second_ret_m.shift(1).dropna() *w_minvar_second).sum(axis=1)
cumret_minvar_first = (1+ret_minvar_first).cumprod()
cumret_minvar_second = (1+ret_minvar_second).cumprod()

plt.plot(cumret_minvar_first.loc[idx[12*3:]],label="first")
plt.plot(cumret_minvar_second.loc[idx[12*3:]],label="second")
plt.legend()
plt.show()
stat_minvar = pd.concat([stats(ret_minvar_first),stats(ret_minvar_second)],axis=1)
stat_minvar.columns = ["1stMinVar","2ndMinVar"]
stat_minvar

リターン、レシオともに２位が勝利


・リスクパリティ

def riskparity(ret):
    
    def calc_weight(w):
        std = np.sqrt(np.dot(np.dot(w.T,sigma),w))
        rc_t = std / len(w)
        rc_i = w * np.dot(sigma,w)/std 
        obj = sum((rc_i-rc_t)**2)
        return obj
    
    sigma = np.cov(ret.T)
    w = np.array([1e-15]*len(sigma))
    
    cons = [{'type': 'eq', 'fun': lambda w: np.sum(w) - 1}]
    bnds = [(0, None)] * len(w) 

    opts = minimize(calc_weight, x0=w, method='SLSQP', bounds=bnds, constraints=cons,tol=1e-15)
    w_opt = opts["x"]
    w_opt = pd.DataFrame(np.round(w_opt,3)).T
    w_opt.columns = ret.columns
    return w_opt
w_rp_first = pd.DataFrame(data=None, index=idx[12*3:], columns=first_ret_d.columns)
w_rp_second = pd.DataFrame(data=None, index=idx[12*3:], columns=second_ret_d.columns)
for i in range(12*3,len(idx)):
   w_rp_first.loc[idx[i]] = riskparity(first_ret_d.loc[idx[i-12*3]:idx[i]]).values
   w_rp_second.loc[idx[i]] = riskparity(second_ret_d.loc[idx[i-12*3]:idx[i]]).values

ret_rp_first = (first_ret_m.shift(1).dropna() *w_rp_first).sum(axis=1)
ret_rp_second = (second_ret_m.shift(1).dropna() *w_rp_second).sum(axis=1)
cumret_rp_first = (1+ret_rp_first).cumprod()
cumret_rp_second = (1+ret_rp_second).cumprod()

plt.plot(cumret_rp_first.loc[idx[12*3:]],label="first")
plt.plot(cumret_rp_second.loc[idx[12*3:]],label="second")
plt.legend()
plt.show()


stat_rp = pd.concat([stats(ret_rp_first),stats(ret_rp_second)],axis=1)
stat_rp.columns = ["1stRP","2ndRP"]
stat_rp

リターン、レシオともに若干２位が勝利

・結論

2位でもいい。

いいなと思ったら応援しよう！