内容

　外国為替市場(FOREX)のチャート予想に取り組んでいます。今回は画像生成系のタスクを使って、過去の為替チャートから未来の予想チャートを描くことをやってみました。最終的には流行であるPix2Pixを使って、何とかそれらしいチャートを描くところまで来ました。

コンセプト

　これまではメタトレードのインジケータを使ってトレードしましたが、やはりこれは難しいということで断念しました。自身のブログではチャートを山に見立てて、地形図を為替相場予想に使ってみようと思ったのが始まりです。詳細は下記の記事をお読みください。　http://1969681.blog66.fc2.com/blog-entry-637.html

手法

　アルゴリズムは完全にGITHUBのリポジトリを使っています。
https://github.com/junyanz/pytorch-CycleGAN-and-pix2pix

やっていることは単に教師データの作成だけです（ただし、こういうコンセプトは、やはりオリジナルではないかと思います）
　以下にポイントを書きます。

チャートはpandas_datareaderを使用　　https://pandas-datareader.readthedocs.io/en/latest/remote_data.html
データサーバーはAlpha Vantage　　　https://pandas-datareader.readthedocs.io/en/latest/remote_data.html#remote-data-alphavantage

　df = data.DataReader(symbol1,'av-daily',start,end, api_key=apikey)

時系列データを画像タスクに変換するために、ベクトルの外積を使って二次元マッピング
　https://deepage.net/features/numpy-outer.html
データは９６日分を一塊にして、Ａ画像、Ｂ画像を生成。32日分の重複個所を作って、連続性を確保した。
Pix2PixへのINPUTは、64x128　→　4倍へ拡張　→　256x512
画像はここに置く　→　pytorch-CycleGAN-and-pix2pix\datasets\facades\train

PIX2PIX

PIX2PIXの実行コマンドは、サイトからの転機
エポック数は、2つのパラメータで与える
GPUは二枚を使用
反転はしないに設定すること
バッチサイズは３２

batch.sh

python train.py --dataroot ./datasets/facades --name facades_pix2pix --model pix2pix --direction AtoB --batch_size 32 --gpu_ids 0,1 --no_flip

ちなみに、引数を間違えても、下記のようなメッセージで教えてくれる。

batch.sh

usage: train.py [-h] --dataroot DATAROOT [--name NAME] [--use_wandb] [--gpu_ids GPU_IDS] [--checkpoints_dir CHECKPOINTS_DIR]
                [--model MODEL] [--input_nc INPUT_NC] [--output_nc OUTPUT_NC] [--ngf NGF] [--ndf NDF] [--netD NETD] [--netG NETG]
                [--n_layers_D N_LAYERS_D] [--norm NORM] [--init_type INIT_TYPE] [--init_gain INIT_GAIN] [--no_dropout]
                [--dataset_mode DATASET_MODE] [--direction DIRECTION] [--serial_batches] [--num_threads NUM_THREADS]
                [--batch_size BATCH_SIZE] [--load_size LOAD_SIZE] [--crop_size CROP_SIZE] [--max_dataset_size MAX_DATASET_SIZE]
                [--preprocess PREPROCESS] [--no_flip] [--display_winsize DISPLAY_WINSIZE] [--epoch EPOCH] [--load_iter LOAD_ITER]
                [--verbose] [--suffix SUFFIX] [--display_freq DISPLAY_FREQ] [--display_ncols DISPLAY_NCOLS] [--display_id DISPLAY_ID]
                [--display_server DISPLAY_SERVER] [--display_env DISPLAY_ENV] [--display_port DISPLAY_PORT]
                [--update_html_freq UPDATE_HTML_FREQ] [--print_freq PRINT_FREQ] [--no_html] [--save_latest_freq SAVE_LATEST_FREQ]
                [--save_epoch_freq SAVE_EPOCH_FREQ] [--save_by_iter] [--continue_train] [--epoch_count EPOCH_COUNT] [--phase PHASE]
                [--n_epochs N_EPOCHS] [--n_epochs_decay N_EPOCHS_DECAY] [--beta1 BETA1] [--lr LR] [--gan_mode GAN_MODE]
                [--pool_size POOL_SIZE] [--lr_policy LR_POLICY] [--lr_decay_iters LR_DECAY_ITERS] [--lambda_L1 LAMBDA_L1]

うまくいけば、下記のメッセージで評価が進む。

batch.sh

----------------- Options ---------------
               batch_size: 32                                   [default: 1]
                    beta1: 0.5
          checkpoints_dir: ./checkpoints
           continue_train: False
                crop_size: 256
                 dataroot: ./datasets/facades                   [default: None]
             dataset_mode: aligned
                direction: AtoB
              display_env: main
             display_freq: 400
               display_id: 0                                    [default: 1]
            display_ncols: 4
             display_port: 8097
           display_server: http://localhost
          display_winsize: 256
                    epoch: latest
              epoch_count: 1
                 gan_mode: vanilla
                  gpu_ids: 0,1                                  [default: 0]
                init_gain: 0.02
                init_type: normal
                 input_nc: 3
                  isTrain: True                                 [default: None]
                lambda_L1: 100.0
                load_iter: 0                                    [default: 0]
                load_size: 286
                       lr: 0.0002
           lr_decay_iters: 50
                lr_policy: linear
         max_dataset_size: inf
                    model: pix2pix                              [default: cycle_gan]
                 n_epochs: 400                                  [default: 100]
           n_epochs_decay: 400                                  [default: 100]
               n_layers_D: 3
                     name: facades_pix2pix                      [default: experiment_name]
                      ndf: 64
                     netD: basic
                     netG: unet_256
                      ngf: 64
               no_dropout: False
                  no_flip: False
                  no_html: False
                     norm: batch
              num_threads: 4
                output_nc: 3
                    phase: train
                pool_size: 0
               preprocess: resize_and_crop
               print_freq: 100
             save_by_iter: False
          save_epoch_freq: 5
         save_latest_freq: 5000
           serial_batches: False
                   suffix:
         update_html_freq: 1000
                use_wandb: False
                  verbose: False
----------------- End -------------------
dataset [AlignedDataset] was created

Trainデータの作成

解析期間　WSIZE　96日　-->　a(n=32)+b(n=32)+c(n=32)　　TRAIN時に　Aはa,b、Bはb,cを使う　　
テンソル　０：終値　１：高値－終値　２：終値－安値
正規化　MAX-MIN　（A,B別々に正規化）　

main.py

import pandas as pd
import datetime as dt
from pandas_datareader import data
import mplfinance as mpf
import torch
from torchvision.datasets import ImageFolder
from torchvision import models, transforms
import torch.nn as nn
import numpy as np
import pdb


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

apikey = '取得したAPIキー'
symbol1 = 'EURUSD'

start = dt.date(2001,8,1)
end = dt.date(2022,2,20)
df1 = data.DataReader(symbol1,'av-daily',start,end, api_key=apikey)

wsize=96 #サンプル期間

df1.index = pd.DatetimeIndex(df1.index)

alldata =[]

for time in range(len(df1)-wsize):
    try:
        dfspan1 = df1[time:time+wsize]
        alldata.append({"df":dfspan1,"date":dfspan1.tail(1)})
    except:
        pass
    
from PIL import Image
import matplotlib.pyplot as plt

def min_max(x, axis=None):
    min = x.min(axis=axis, keepdims=True)
    max = x.max(axis=axis, keepdims=True)
    result = (x-min)/(max-min)
    return result

def imagemake(dfspan1):
    a = np.array([x for x in dfspan1['df']['close']])
    b = np.array([x for x in dfspan1['df']['high']])
    c = np.array([x for x in dfspan1['df']['low']])
    
    m = np.outer(a,a).astype(np.float32)
    n = np.outer(b-a,b-a).astype(np.float32)
    o = np.outer(a-c,a-c).astype(np.float32)
    
    m1 = min_max(m[0:64,0:64])
    m2 = min_max(m[32:96,32:96])
    n1 = min_max(n[0:64,0:64])
    n2 = min_max(n[32:96,32:96])
    o1 = min_max(o[0:64,0:64])
    o2 = min_max(o[32:96,32:96])
    
    te1 = np.stack([m1,n1,o1])
    te2 = np.stack([m2,n2,o2])
    te3=np.concatenate([te1,te2], 2)
    
    te4=np.kron(te3, np.ones((4,4)))
    tmp = torch.from_numpy(te4).clone()
    return  transforms.ToPILImage(mode='RGB')(tmp)   

for item in alldata:
    img=imagemake(item)
    fn = item['date'].index.astype(str).values.tolist()
    fname = "datasets/facades/train/" + fn[0] + ".png"
    img.save(fname)

Pix2Pix予測¶

テストフォルダのデータで予想

batch.sh

python test.py --dataroot ./datasets/facades --name facades_pix2pix --model pix2pix --direction AtoB

main.py

import torch
from torchvision import transforms
from torchvision.datasets import ImageFolder
import os
from PIL import Image
import matplotlib.pyplot as plt
%matplotlib inline
import numpy as np
import pdb

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

path = os.getcwd()

new_dir_path = "results/facades_pix2pix/test_latest/images"
                
img=[]
image={}
for imageName in os.listdir(new_dir_path):
    inputPath = os.path.join(path, new_dir_path,imageName)
    if "fake_B" in  imageName : image['fakeB']=inputPath
    if "real_A" in  imageName: image['realA']=inputPath
    if "real_B" in  imageName: image['realB']=inputPath
    if len(image)==3:
        img.append(image)
        image={}
import re 
import datetime

def getprice(img,transform):
    match = re.search(r'\d{4}-\d{2}-\d{2}', img['fakeB'])
    date = datetime.datetime.strptime(match.group(), '%Y-%m-%d').date()
    fake = transform(Image.open(img['fakeB'])) 
    real = transform(Image.open(img['realB']))
    fake=fake.numpy()
    real=real.numpy()
    
    real1 = min_max(np.diag(real[0,:,:]))  
    plt.plot(real1)
 
    fake1 = min_max(np.sqrt(np.diag(fake[0,:,:])))
    fake2 = min_max(fake[0,0,:])
    fake3 = min_max(fake[0,:,0])
    
    plt.plot((fake1+fake2+fake3)/3)   
   
    plt.title(date)
    plt.show()
transform = transforms.PILToTensor()
for item in img:
    price = getprice(item,transform)

未来予測

教師データの作成

変更点は以下のみ
te3=np.concatenate([te2,te2], 2)#real = fake = te2で与える

OANDAの日足データを使用するように変更した
　EURUSD.oj5k1440.csv

main.py

import pandas as pd
import datetime as dt
from pandas_datareader import data
import mplfinance as mpf
import torch
from torchvision.datasets import ImageFolder
from torchvision import models, transforms
import torch.nn as nn
import numpy as np
import pdb


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

apikey = '取得したAPIキー'
symbol1 = 'EURUSD'

start = dt.date(2021,1,1)
end = dt.date(2021,1,2)
df1 = data.DataReader(symbol1,'av-daily',start,end, api_key=apikey)

df = pd.read_csv(r'C://Users//User//Desktop//EURUSD.oj5k1440.csv', index_col=0)
df=df.drop(df.columns[0], axis=1)
df.columns=df1.columns
df_new=pd.concat([df1,df.tail(100)])
print(df_new)

wsize=96 #サンプル期間

df_new.index = pd.DatetimeIndex(df_new.index)

alldata =[]

for time in range(len(df_new)-wsize+1):
    try:
        dfspan1 = df_new[time:time+wsize]
        alldata.append({"df":dfspan1,"date":dfspan1.tail(1)})
    except:
        pass
    
from PIL import Image
import matplotlib.pyplot as plt

def min_max(x, axis=None):
    min = x.min(axis=axis, keepdims=True)
    max = x.max(axis=axis, keepdims=True)
    result = (x-min)/(max-min)
    return result

def imagemake(dfspan1):
    a = np.array([x for x in dfspan1['df']['close']])
    b = np.array([x for x in dfspan1['df']['high']])
    c = np.array([x for x in dfspan1['df']['low']])
    
    m = np.outer(a,a).astype(np.float32)
    n = np.outer(b-a,b-a).astype(np.float32)
    o = np.outer(a-c,a-c).astype(np.float32)
    
    m1 = min_max(m[0:64,0:64])
    m2 = min_max(m[32:96,32:96])
    n1 = min_max(n[0:64,0:64])
    n2 = min_max(n[32:96,32:96])
    o1 = min_max(o[0:64,0:64])
    o2 = min_max(o[32:96,32:96])
    
    #te1 = np.stack([m1,n1,o1])
    te2 = np.stack([m2,n2,o2])

    te3=np.concatenate([te2,te2], 2)#real = fake = te2で与える
    
    te4=np.kron(te3, np.ones((4,4)))
    tmp = torch.from_numpy(te4).clone()
    return  transforms.ToPILImage(mode='RGB')(tmp)   


for item in alldata:
    img=imagemake(item)
    fn = item['date'].index.astype(str).values.tolist()
    fname = "datasets/facades/test/" + fn[0] + "sk.png"
    img.save(fname)

後処理

テストを実施　→　予想チャートを描く

batch.sh

python test.py --dataroot ./datasets/facades --name facades_pix2pix --model pix2pix --direction AtoB

main.py

import torch
from torchvision import transforms
from torchvision.datasets import ImageFolder
import os
from PIL import Image
import matplotlib.pyplot as plt
%matplotlib inline
import numpy as np
import pdb

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


path = os.getcwd()

new_dir_path = "results/facades_pix2pix/test_latest/images"
                
img=[]
image={}
for imageName in os.listdir(new_dir_path):
    inputPath = os.path.join(path, new_dir_path,imageName)
    if "fake_B" in  imageName : image['fakeB']=inputPath
    if "real_A" in  imageName: image['realA']=inputPath
    if "real_B" in  imageName: image['realB']=inputPath
    if len(image)==3:
        img.append(image)
        image={}
import re 
import datetime
import pdb

def getprice(img,transform):
    match = re.search(r'\d{4}-\d{2}-\d{2}', img['fakeB'])
    date = datetime.datetime.strptime(match.group(), '%Y-%m-%d').date()
    fake = transform(Image.open(img['fakeB'])) 
    real = transform(Image.open(img['realA']))
    fake=fake.numpy()
    real=real.numpy()
            
    plt.figure(figsize=(10,4))
    plt.subplot(1,2,1)
    plt.title(date) 
    
    real1 = np.sqrt(np.diag(real[0,:,:]))#対角成分
    real1_avg = np.mean(real1[128:256])
    print(real1_avg)
    real2= real1-real1_avg
    plt.plot(real2,label="real") 

    fake1 = np.sqrt(np.diag(fake[0,:,:]))#対角成分
    fake1_avg = np.mean(fake1[0:128])
    print(fake1_avg)
    fake2= fake1-fake1_avg
    fake3=np.append(real2[0:128],fake2)
    plt.plot(fake3,label="fake")
    
    plt.legend() 
    plt.tight_layout()
    plt.show()
transform = transforms.PILToTensor()
for item in img:
    price = getprice(item,transform)