ファイナンス機械学習：人工データによるバックテスト オルヌシュタインーウーレンベック過程による最適戦略の決定
OU過程を線形化し、入力パラメータ$${\{\sigma,\phi\}}$$を推定する。
$${P_{i,t}=E_0[P_{i,T_i}]+\phi(P_{i,t-1}-E_0[P_{i,T_i}]) + \xi_t}$$
上記の式の$${\phi}$$の係数を順番に並べたベクトルを$${{\bf X}}$$、各投資機会$${i}$$のTまでの価格を並べたベクトルを$${{\bf Y}}$$、ポジションを取った時点での推定価格を並べたベクトルを$${{\bf Z}}$$とする。
$${X=\begin{bmatrix} P_{0,0} -E_0[P_{0,T_0}] \\P_{0,1} -E_0[P_{0,T_0}] \\ \vdots \\P_{0,T-1} -E_0[P_{0,T_0}] \\ P_{1,0} -E_0[P_{1,T_1}] \\ \vdots \\ P_{1,T-1} -E_0[P_{1,T_1}] \\ \vdots \\ P_{I,T-1} -E_0[P_{I,T_I}] \\ \end{bmatrix}}$$,$${Y=\begin{bmatrix} P_{0,1} \\ P_{0,2} \\ \vdots \\ P_{0,T} \\ P_{1,1} \\ \vdots \\ P_{I,T} \\ \vdots \\ P_{I,T} \\ \end{bmatrix}}$$,$${Z=\begin{bmatrix} E_0[P_{0,T_0}] \\E_0[P_{0,T_0} \\ \vdots \\ E_0[P_{0,T_0}] \\ E_0[P_{1,T_1}] \\ \vdots \\ E_0[P_{I,T_I}] \\ \vdots \\ E_0[P_{I,T_I}] \\ \end{bmatrix}}$$
これらのベクトルを代入し、最小二乗法からOU過程のパラメータを推定する。
$${\hat{\phi}=\displaystyle{\frac{cov[Y,X]}{cov[X,X]}}}$$
$${\hat{\xi}_t=Y-Z-\hat{\phi}X}$$
$${\hat{\sigma}=\sqrt{cov[\hat{\xi}_t,\hat{\xi}_t]}}$$
次に、水平バリアを作る損切り閾値を利益確定閾値のペア$${(\underline{\pi},\overline{\pi})}$$で格子を作る。また、ポジションを無限に取り続けることがないように垂直バリアに当たる最大保有期間を設定する。
推定された$${(\hat{\sigma},\hat{\phi})}$$を適用した$${\pi_{i,t}}$$に対し、十分な数のパスを生成する。パスの初期値は$${\{P_{i,0},E_0[P_{i,T_i}]\}}$$である。このパスを各格子点に適用し、各格子点でトリプルバリアを適用しパスの数だけ得られる$${\pi_{i,T_i}}$$から、シャープレシオを計算する。
ここで得られたシャープレシオから、戦略の偽陽性率を考慮に入れ、不適切な戦略を落とす。
残ったシャープレシオを用い以下の3通りの解析ができる。
入力パラメータ$${\{\hat{\sigma},\hat{\phi}\}}$$と、初期条件$${\{P_{i,0},E_0[P_{i,T_i}]\}}$$に対する最適な$${(\underline{\pi},\overline{\pi})}$$のペアを決定する。
戦略Sが特定の投資機会$${i}$$に対して、利益目標$${\overline{\pi}}$$を固定する場合、最適な損切り$${\underline{\pi}}$$値を決定する。
投資機会投資機会$${i}$$に対して、最大損切り$${\underline{\pi}}$$を決められている時に、その限度内の$${[0,\underline{\pi}]}$$に対する最適な利益確定閾値$${\overline{\pi}}$$を求める。
スニペット13.1,13.2で与えられている実装を、concurrentで並列化し、グラフを出力させるコードは以下の通り。
import numpy as np
from random import gauss
from itertools import product
import concurrent.futures
import pandas as pd
import CoreNum
import matplotlib.pyplot as plt
def batchORG(coeffs, nlter=1e5, maxHP=100,rPT=np.linspace(.5,10,20),
rSLm=np.linspace(.5,10,20),p0=0):
phi,output1=2**(-1./coeffs['hl']),[]
for comb_ in product(rPT,rSLm):
output2=[]
for iter_ in range(int(nlter)):
p,hp,count=p0,0,0
while True:
p=(1-phi)*coeffs['forecast'] + phi*p + coeffs['sigma']*gauss(0,1)
cP=p-p0
hp+=1
if cP > comb_[0] or cP < -comb_[1] or hp>maxHP:
output2.append(cP)
break
mean,std=np.mean(output2),np.std(output2)
#print(comb_[0],comb_[1],mean,std,mean/std)
output1.append((comb_[0],comb_[1],mean,std,mean/std))
return output1
def Fbatch(comb, nlter, p0, coeffs, maxHP):
output2 = []
phi=2**(-1./coeffs['hl'])
for iter_ in range(int(nlter)):
p, hp, count = p0, 0, 0
while True:
p = (1 - phi) * coeffs['forecast'] + phi * p + coeffs['sigma'] * gauss(0, 1)
cP = p - p0
hp += 1
if cP > comb[0] or cP < -comb[1] or hp > maxHP:
output2.append(cP)
break
mean, std = np.mean(output2), np.std(output2)
return (comb[0], comb[1], mean, std, mean/std)
def batchP(coeffs, nlter=1e5, maxHP=100,rPT=np.linspace(.5,10,20),
rSLm=np.linspace(.5,10,20),p0=0,njobs=1):
print(coeffs['forecast'],coeffs['hl'])
output1=[]
with concurrent.futures.ProcessPoolExecutor(max_workers=njobs) as executor:
futures = [executor.submit(Fbatch, comb, nlter, p0, coeffs, maxHP) for comb in product(rPT, rSLm)]
output1 = [future.result() for future in concurrent.futures.as_completed(futures)]
return output1
def main():
nCore=CoreNum.CoreNum()
mT=10
rPT=rSLm=np.linspace(0,mT,21)
E0=[10.,5.,0.,-5.,-10.]
tau=[5.,10.,25.,50.,100.]
columns=['top','bottom','mean','std','sr']
for prod_ in product(E0,tau):
coeffs={'forecast':prod_[0],'hl':prod_[1],'sigma':1}
output=batchP(coeffs, nlter=1e5, maxHP=100,rPT=rPT,rSLm=rSLm,njobs=nCore)
df=pd.DataFrame(output,columns=columns)
#csv_path=f"./OU_SR_E0{prod_[0]}_tau{prod_[1]}.csv"
#df.to_csv(csv_path,index=False)
sr=df.sort_values(by=['top','bottom'])['sr']
sr=sr.to_numpy().reshape(len(rPT),-1)
plt.imshow(sr.T,cmap='viridis', extent=(0, rPT[-1], 0, rSLm[-1]),
aspect='auto',interpolation='nearest',origin='lower')
plt.title(f'Forcast:{coeffs["forecast"] }, tau:{coeffs["hl"]}, sigma:1')
plt.xlabel('Top Barrier')
plt.ylabel('Bottom Barrier')
plt.colorbar()
filename=f"./OU_SR_E0{prod_[0]}_tau{prod_[1]}.pdf"
plt.savefig(filename)
if __name__ == '__main__':
main()