AIシステムトレード バックテスト編

ひとまず完成した、ＡＩ学習モデルを使ってバックテストを試みることにした。

参考資料　→　https://qiita.com/fjunya/items/f2562ad237d35c2b5fc2

リポジトリ　→　https://github.com/chanmoto/eurusd_forex_vgg16

#STEP1　データ取得

バックテストにあたり、データは評価期間れαになるように、期間を直近の期間に限定する。
前処理は全く同じ内容

backtest.py

get_ipython().system('pip install pandas_datareader')
get_ipython().system('pip install --upgrade pandas')

import pandas as pd
import datetime as dt
from pandas_datareader import data as datard
import mplfinance as mpf

import os
os.environ['KMP_DUPLICATE_LIB_OK']='True'

apikey = '取得したAPIキー'
symbol = 'EURUSD'
start = dt.date(2021,1,1)
end = dt.date(2022,2,3)
df = datard.DataReader(symbol,'av-daily',start,end, api_key=apikey)
df['volume']=0.1 

#STEP2　イメージ変換

backtest.py

import numpy
from scipy.stats import boxcox

df2,_ = boxcox(df['close'])
df=df.assign(open=df2,high=df2,low=df2,close=df2)

wsize=60 #サンプル期間
after=15 #最終レコード＋α
change_large = 5 #変化大　4%

df.index = pd.DatetimeIndex(df.index)

alldata =[]

for time in range(len(df)-wsize-1):
    try:
        dfspan = df[time:time+wsize]
        alldata.append({"df":dfspan,"date":dfspan.tail(1)})
    except:
        pass

class ImageTransform():
  def __init__(self, mean, std):
    self.data_transform = transforms.Compose([
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(mean, std)
    ])

  def __call__(self, img):
    return self.data_transform(img)

#STEP3　モデルのロード
学習プログラムで完成したパラメータをロードする

backtest.py

import torch
from torchvision.datasets import ImageFolder
import torch.utils.data as data
from torchvision import models, transforms
import torch.nn as nn
from PIL import Image, ImageFilter

model = models.vgg16(pretrained=True)
num_ftrs = model.classifier[6].in_features
model.classifier[6] = torch.nn.Linear(num_ftrs,  out_features=3)
model.load_state_dict(torch.load('model_state.pth'))
model.eval()

#画像データをImageFolderを使って取込みする
mean=[0.485, 0.456, 0.406]
std=[0.229, 0.224, 0.225]
new_dir_path = 'predict'

sma_s= 15
sma_m= 30

cs  = mpf.make_mpf_style(base_mpl_style='dark_background',gridcolor="#000000",mavcolors=['#00ff00','#0000ff'], facecolor="#000000")

#STEP4

データの配列を変更している。

学習時は為替データとラベル　 alldata.append({"df":dfspan,"label":dflabel})
予測時は為替データと予測日　alldata.append({"df":dfspan,"date":dfspan.tail(1)})

こうすることで、バックテスト時には予測日でフックして、為替データを取り出せる。
search_df = [x['df'] for x in alldata if x['date'].index == datestr]

datastrで検索して、見つかれば画像を作ってラベルを予想。（バグ対策として、見つからなければダミーの画像を作って予想）

backtest.py

def Imagesrc(datestr):
    
    search_df = [x['df'] for x in alldata if x['date'].index == datestr]
    transform = ImageTransform(mean, std)

    if search_df:
        mpf.plot(search_df[0],savefig= 'predict.png',mav=(sma_s,sma_m),figsize=(2.5,2.5), type='line',             figratio=(12,4),style=cs,axisoff=True,             linecolor='#ff0000',tight_layout=True)
        img = Image.open('predict.png').convert('RGB')
        img_transform = transform(img)
        inputs = img_transform.unsqueeze_(0)

        result = model(inputs)
        idx = torch.argmax(result[0])
        return img_transform,idx.numpy()
    else:
        img=Image.new("RGB", (800, 1280), (255, 255, 255))
        img_transform = transform(img)
        inputs = img_transform.unsqueeze_(0)

        result = model(inputs)
        idx = torch.argmax(result[0])
        return img_transform,idx.numpy()

#バッチから取り出した画像のイメージとラベルを表示する
pic = transforms.ToPILImage(mode='RGB')(img[0])
plt.imshow(pic)
print("Label is ",label)

df=df.set_axis(['Open', 'High', 'Low', 'Close', 'Volume'], axis=1)

#STEP5

realdfは、バックテスト用のデータフレーム。signal列が予測ラベルになる。

backtest.py

import matplotlib.pyplot as plt
import tqdm

start = dt.date(2021,4,1)
end = dt.date(2022,2,3)

realdf = datard.DataReader(symbol,'av-daily',start,end, api_key=apikey)
realdf=realdf.set_axis(['Open', 'High', 'Low', 'Close', 'Volume'], axis=1)
realdf.index = pd.DatetimeIndex(realdf.index)

signal= []
for index, row in tqdm.tqdm(realdf.iterrows()):
    img,label = Imagesrc(index)
    signal.append(label)

realdf['signal']=signal

STEP6

バックテストのメイン。

backtest.py

from backtesting import Backtest, Strategy
from backtesting.lib import SignalStrategy, TrailingStrategy

def SIGNAL():
    return realdf.signal

class VGG16EVA(Strategy):
    stls=0.02
    tkpf=0.01
    
    def init(self):
        self.signal1=self.I(SIGNAL)
        pass
    
    def next(self): #チャートデータの行ごとに呼び出される
        super().next()
        stls = self.stls
        tkpf = self.tkpf
        for trade in self.trades: 
            if trade.is_long: 
                trade.sl = max(trade.sl or -numpy.inf, 
                            self.data.Close[-1] - stls) 
            else: 
                 trade.sl = min(trade.sl or numpy.inf, 
                            self.data.Close[-1] + stls) 
        if self.signal1==0 and len(self.trades)==0:
            sl1= self.data.Close[-1] - stls
            tp1= self.data.Close[-1] + tkpf
            self.buy(sl=sl1,tp=tp1) # 買い
        elif self.signal1==1  and len(self.trades)==0:
            sl1= self.data.Close[-1] + stls
            tp1= self.data.Close[-1] - tkpf
            self.sell(sl=sl1,tp=tp1)# 売り
    
bt = Backtest(
    realdf, # チャートデータ
    VGG16EVA, # 売買戦略
    cash=100000, # 最初の所持金
    commission=0.00495, # 取引手数料
    margin=1.0, # レバレッジ倍率の逆数（0.5で2倍レバレッジ）
    trade_on_close=True, # True：現在の終値で取引，False：次の時間の始値で取引
)

output = bt.run() # バックテスト実行
print(output) # 実行結果(データ)
bt.plot() # 実行結果（グラフ）