【Pandas入門】データ補間で為替データを増やしてみた♬

データ補間については通常の時系列データについて、以前取り扱った。
今回は、それをバックテスト目的に日足データを補間することにより、6時間、4時間、1時間、そして10分データを作成してその精度を見た。

やったこと

・ラグランジュ補間公式
・Pandasデータの補間
・6時間～10分データ作成
・精度を見る

・ラグランジュ補間公式

【参考】
・【補間】線形補間から二次補間、ラグランジュ補間で補間する♬
まず、(x0,y0),(x1,y1)を通る1次補間関数は、以下の関数で計算できる。

def interpolation(x0,y0,x1,y1,x):
    dn = (x0-x1)
    return y0*(x-x1)/dn + y1*(x0-x)/dn
#Lagrange interpolation
# y0*(x-x1)/(x0-x1)+y1*(x-x0)/(x1-x0)

3点(x0,y0),(x1,y1),(x2,y2)を通る2次補間関数は以下のように計算できる。

def interpolation2(x0,y0,x1,y1,x2,y2,x):
    dn1 = (x0-x1)*(x0-x2)
    dn2 = (x1-x2)*(x1-x0)
    dn3 = (x2-x0)*(x2-x1)
    return y0*(x-x1)*(x-x2)/dn1+y1*(x-x2)*(x-x0)/dn2+y2*(x-x0)*(x-x1)/dn3

これらを組み合わせて、端点処理を上記の2点を通る関数とし、その他は3点を通る関数を利用すると、2点間を10点補間する関数は以下のとおりでした。

m=10
sigxm=np.zeros(m*pitch-(m-1))
sigxm[0]=y[0]
sigxm[m*pitch-m]=y[pitch-1]
for i in range(1,m*pitch-m,1):
    if i%m==0:
        sigxm[i]=y[int(i/m)]
    if i > m*pitch-(2*m+1):
        sigxm[i] = interpolation(int(i/m),y[int(i/m)],int(i/m)+1,y[int(i/m)+1],int(i/m)+(i%m)/m)
    else:
        sigxm[i] = interpolation2(int(i/m),y[int(i/m)],int(i/m)+1,y[int(i/m)+1],int(i/m)+2,y[int(i/m)+2],int(i/m)+(i%m)/m)

前回は、n次補間も可能だと言っているが、実際の補間では2次補間を組み合わせた上記の関数で滑らかな関数が得られているので、今回も上記関数を採用する。
今回は、この関数の変数yをPandasデータに置き換えることにより、為替や株データの補間データを作成する。

・Pandasデータの補間

結論から記載すると1次補間、2次補間関数は変わらない。
そして、m分割点の計算は以下の関数で計算できる。
まず、為替データをpandas.DataReaderで読み込む。

df=DataReader.get_data_yahoo("{}".format("JPY=X"),start,end)

そして、以下のようにyの代わりにdf["Close"]に置き換えて順次計算すると、Pandasデータについても計算できる。

m=24
sigxm=np.zeros(m*pitch-(m-1))
sigxm[0]=df["Close"][0]
sigxm[m*pitch-m]=df["Close"][pitch-1]
for i in range(1,m*pitch-m,1):
    if i%m==0:
        sigxm[i]=df["Close"][int(i/m)]
    if i > m*pitch-(2*m+1):
        sigxm[i] = interpolation(int(i/m),df["Close"][int(i/m)],int(i/m)+1,df["Close"][int(i/m)+1],int(i/m)+(i%m)/m)
    else:
        sigxm[i] = interpolation2(int(i/m),df["Close"][int(i/m)],int(i/m)+1,df["Close"][int(i/m)+1],int(i/m)+2,df["Close"][int(i/m)+2],int(i/m)+(i%m)/m)

dfsigxm = pd.DataFrame()
dfsigxm["sig"] = sigxm
date_df=dfsigxm['sig'].index.tolist()
plot_fig(date_df,dfsigxm,m,end)

そして、上記の関数でmを変化させて計算されたdfsigxmを以下のplot_fig(data_df,dfsig,m,end)関数でプロットする。

def plot_fig(data_df,dfsig,m,end):
    fig, (ax1,ax2) = plt.subplots(2,1,figsize=(1.6180 * 8, 4*2),dpi=100)
    ax1.plot(date_df[0:],dfsig["sig"][0:])
    ax2.plot(date_df[0:32*m],dfsig["sig"][0:32*m])
    ax1.grid()
    ax2.grid()
    plt.pause(1)
    plt.savefig("./fx/{}_{}_{}_{}_.png".format(m,"interpolate","JPN=X",end))
    plt.close()

・6時間～10分データ作成

m=4だと6時間のデータ、m=6で4時間、m=24で1時間、そしてm=24*6で10分データを計算できる。
結果は以下のとおりとなった。
m=1はまんまのグラフなので、端点が残っているが、それ以外は下の拡大図含めてほとんど変化ないが、プロット密度が増加し、よく見ると横軸の数字が増加しているのが分かる。

・精度を見る

簡単なのは、実際の10分足などと比較することであるが、どうもそれを取得できるサイトは無いようだ。

まとめ

・日足為替データをpandasで取得して、補間することにより、6時間、4時間、1時間、そして10分データを作成した
・滑らかなデータ列が作成できた

・残念ながら実際のデータとの比較検証は出来なかった

全体コード

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import pandas_datareader.data as DataReader
import datetime as dt

def plot_fig(data_df,dfsig,m,end):
    fig, (ax1,ax2) = plt.subplots(2,1,figsize=(1.6180 * 8, 4*2),dpi=100)
    ax1.plot(date_df[0:],dfsig["sig"][0:],"o-")
    ax2.plot(date_df[0:32*m],dfsig["sig"][0:32*m],"o-")
    ax1.grid()
    ax2.grid()
    plt.pause(1)
    plt.savefig("./fx/{}_{}_{}_{}_.png".format(m,"interpolate","JPN=X",end))
    plt.close()

def interpolation(x0,y0,x1,y1,x):
    return y0 + (y1 - y0) * (x - x0) / (x1 - x0)

def interpolation2(x0,y0,x1,y1,x2,y2,x):
    dn1 = (x0-x1)*(x0-x2)
    dn2 = (x1-x2)*(x1-x0)
    dn3 = (x2-x0)*(x2-x1)
    return y0*(x-x1)*(x-x2)/dn1+y1*(x-x2)*(x-x0)/dn2+y2*(x-x0)*(x-x1)/dn3

def calc_interpolate(df,pitch,m):
    sigxm=np.zeros(m*pitch-(m-1))
    sigxm[0]=df["Close"][0]
    sigxm[m*pitch-m]=df["Close"][pitch-1]
    for i in range(1,m*pitch-m,1):
        if i%m==0:
            sigxm[i]=df["Close"][int(i/m)]
        if i > m*pitch-(2*m+1):
            sigxm[i] = interpolation(int(i/m),df["Close"][int(i/m)],int(i/m)+1,df["Close"][int(i/m)+1],int(i/m)+(i%m)/m)
        else:
            sigxm[i] = interpolation2(int(i/m),df["Close"][int(i/m)],int(i/m)+1,df["Close"][int(i/m)+1],int(i/m)+2,df["Close"][int(i/m)+2],int(i/m)+(i%m)/m)
    return sigxm        

start = dt.date(2020,1,1)
end = dt.date(2020,6,15)
df=DataReader.get_data_yahoo("{}".format("JPY=X"),start,end)

m=1
dfsigxm = pd.DataFrame()
dfsigxm["sig"] = df["Close"]
print(dfsigxm)
date_df=dfsigxm['sig'].index.tolist()
plot_fig(date_df,dfsigxm,m,end)

m=4
pitch = len(df)
sigx2=calc_interpolate(df,pitch,m)
dfsigx2 = pd.DataFrame()
dfsigx2["sig"] = sigx2
print(dfsigx2)
date_df=dfsigx2['sig'].index.tolist()
plot_fig(date_df,dfsigx2,m,end)

m=6
sigxm=calc_interpolate(df,pitch,m)
dfsigxm = pd.DataFrame()
dfsigxm["sig"] = sigxm
print(dfsigxm)
date_df=dfsigxm['sig'].index.tolist()
plot_fig(date_df,dfsigxm,m,end)

m=24
sigxm=calc_interpolate(df,pitch,m)
dfsigxm = pd.DataFrame()
dfsigxm["sig"] = sigxm
print(dfsigxm)
date_df=dfsigxm['sig'].index.tolist()
plot_fig(date_df,dfsigxm,m,end)

m=24*6
sigxm=calc_interpolate(df,pitch,m)
dfsigxm = pd.DataFrame()
dfsigxm["sig"] = sigxm
print(dfsigxm)
date_df=dfsigxm['sig'].index.tolist()
plot_fig(date_df,dfsigxm,m,end)