日次で価格データをDBに自動保存してもらおう

どうもみなさん、ガチホしていますか？

4/12にBTCの価格がわずか1時間程度の間に10万円以上値上がりしたのは記憶に新しいと思います。
そんな中、私は2017年の12/8の朝8時ぐらいにCoinCheckのチャートを見ながら、
ものの数分で40万近く値上がりしていく姿を見て、
"いよいよBTCも終わりか"と悟っていたことを思い出しました。

★　　　　　　★　　　　　　★

さて、私はこのごろずっと"Cryptowatchのデータ少なすぎ、バックテストの意味ないじゃん！"
と思っていまして、自動でDBにデータを保存してくれるスクリプトを作成してみました。

エンジニアでないとなかなかDBを触ることもないと思いますので、
DBなんてわからないよ！という方にも取り組んでいただけるように説明させていただきます。

手順

そもそもDBとは？

簡単に言えば、データを保存しておくストレージというところでしょうか。
そんなこと知ってるよ(ﾟДﾟ)ｺﾞﾙｧ!という方はいったん落ち着いてくださいw

CryptowatchのAPIで価格データを取得するときも実はDBからデータを取得しているのですが、
一定期間しか価格データを取得することができません。
それに対して、自分のローカルもしくはAWS上にDBを構築してしまえば、
好きなように保存したデータを使い倒すことができてしまうのです！
下のような図をイメージしていただけるとわかりやすいのではないのでしょうか。

すでに私がこのブログに投稿しているbotもそうなんですが、
①のAPIでデータを取得しただけでは、データは保存されません。
メモリに一時的に保存されているだけで、プログラムが停止すればデータは消えます。
一方、➁の処理まで実行すればプログラムを停止してもPCの電源を切ってもデータは残っています。

➁のデータを登録する部分が今回のメインテーマとなります。
DBにも、Oracle、SQLServer、MySQL、Hadoop...etcとさまざまなものがあるのですが、
個人でbot制作をするレベルであれば特に気にする必要はないと考えているため、
自作のアプリケーションと相性の良いものを使っていただければと思います。

今回はPostgresSQLを使って説明したいと思います。

DBの環境構築をしよう

DB自体の環境構築については、すでにまとめてくださっている記事も多いので
引用させていただく形で説明したいと思います。

qiita.com

今回紹介するプログラムの都合上、PostgreSQL9.5以上のバージョンを対象としています。
上記リンクの方法に従って、PostgreSQL9.5以上のバージョンで環境構築をお願いいたします。

環境構築が終わりましたら、デフォルトで「pgAdmin3」という
GUIのクライアントツールが入っていると思いますのでそちらを使って操作していきましょう。
「C:\Program Files\PostgreSQL\9.5\bin」の配下の"pgAdmin3.exe"というファイルを実行します。

DBを使ってみよう

1.DBの構造

さて、実際にDBをつかってみましょうか。
DBにはテーブルという表領域を確保しなければなりません。
このテーブルの中にさまざまなデータを登録していきます。

テーブルはそれぞれ属性の異なるデータを保持していくのが定石です。
例えば、上図のように5分足のデータと1時間足のデータは分けてテーブルに持ちます。

テーブルでは日付をPK(特定の行を一意に特定できる制約)を設定しなければなりません。
もし仮に、5分足と1時間足の価格データを同じテーブルに持ってしまうと、
"2018-04-13 15:00:00"の行が2つあるのに、1時間足なのか5分足なのか識別できなくなるからです。

テーブルを作るというのは、
図にあるような"日付","始値","高値","安値","終値","取引高"という項目(カラム)を設定したり、
"日付"がPKですよー。という定義を作ってあげることを言います。

2.空のテーブルを作成する

説明がかなりしんどいので、スライドで解説させてくださいw

DDLと呼ばれるSQLを実行するとデータが空っぽのテーブルが作成されます。
このSQLを実行すると暗黙的にテーブルが保存されます。

※ちなみにSQLとはDBにあるデータを取り出したり更新・登録・削除などする言語のことです。
SQLは上記の④で表示しているウインドウに張り付けて実行していきます。

DDL："T_BTC_FX_5MIN_BITFLYER"(5分足データ専用のテーブルになります)

CREATE TABLE public."T_BTC_FX_5MIN_BITFLYER"
(
  "DATA_DATE" timestamp without time zone NOT NULL,
  "OPEN" integer, -- 始値
  "HIGH" integer, -- 高値
  "LOW" integer, -- 安値
  "CLOSE" integer, -- 終値
  "VOLUME" double precision, -- 出来高
  CONSTRAINT "PK_T_BTC_FX_5MIN_BITFLYER" PRIMARY KEY ("DATA_DATE")
)
WITH (
  OIDS=FALSE
);
ALTER TABLE public."T_BTC_FX_5MIN_BITFLYER"
  OWNER TO postgres;
COMMENT ON COLUMN public."T_BTC_FX_5MIN_BITFLYER"."OPEN" IS '始値';
COMMENT ON COLUMN public."T_BTC_FX_5MIN_BITFLYER"."HIGH" IS '高値';
COMMENT ON COLUMN public."T_BTC_FX_5MIN_BITFLYER"."LOW" IS '安値';
COMMENT ON COLUMN public."T_BTC_FX_5MIN_BITFLYER"."CLOSE" IS '終値';
COMMENT ON COLUMN public."T_BTC_FX_5MIN_BITFLYER"."VOLUME" IS '出来高';

3.データを入れてみる、取得してみる、削除してみる

A.登録(INSERT文)

INSERT INTO "T_BTC_FX_5MIN_BITFLYER" 
       ("DATA_DATE", "OPEN", "HIGH", "LOW", "CLOSE", "VOLUME") 
VALUES ('2018-04-18 21:55:00', 70000, 800, 30, 20, 10)

B.参照(SELECT文)

select *
from "T_BTC_FX_5MIN_BITFLYER"
where "DATA_DATE" = '2018-04-18 21:55:00'

C.削除(DELETE文)

delete
from "T_BTC_FX_5MIN_BITFLYER"
where "DATA_DATE" = '2018-04-18 21:55:00'

サンプルのSQLを掲載しておきます。
A.を実行して、B.を実行すると結果が返ってきます。
C.を実行するとA.で登録したデータは消えます。

SQLの文法についてはいろいろ調べてみてください！(適当)

また、A.やC.のようなテーブルのデータを登録・削除する処理のあとには
"commit"というSQLを単独で実行します。
人間の目にはデータが入っているように見えますが、
実際にはメモリに一時的に保存されているだけで保存されていません。
"commit"をしなければDBを停止した時にデータが消えます。

また、"rollback"というSQLもあり、これは前回"commit"した時点までデータの状態を戻します。
※なお、DBの設定状態によってはオートコミットというものがあり、
"commit"を実行しなくてもデータが保存されてしまいますので事前に確認して下さい！
私のPostgreSQLのDBはオートコミットになっていました。

pythonでDBにデータをインポートしよう

さあ、お待たせしました。
いよいよpythonのプログラムを使ってデータをDBに登録する作業です。
以下にコードを掲載します。

OneFileGetBitFlyerTicker.py

import psycopg2
import datetime as dt
import requests
import json
import time
from datetime import timedelta as td



# 価格取得
def get_ticker(candle_span,data_span):

    # 時間足の指定
    candle_span=[candle_span]

    # 現在時刻から指定した時間間隔の時刻を取得
    endDate = dt.datetime.now()
    startDate = endDate + td(hours=data_span)

    # 時刻データのフォーマット変換
    startTimestamp = startDate.timestamp()
    endTimestamp = endDate.timestamp()

    # 価格データのAPIリクエスト@cryptowatch
    query = {"periods": candle_span, "after": str(int(startTimestamp)), "before": str(int(endTimestamp))}

    # 成功するまでリトライ
    while True:
        try:
            res = json.loads(requests.get("https://api.cryptowat.ch/markets/bitflyer/btcfxjpy/ohlc", params=query).text)["result"]
        except:
            # 失敗した場合sleep
            print('価格の取得に失敗したためリトライ処理を実行します。')
            time.sleep(15)
        else:
            # 成功した場合ループを抜ける
            break
    else:
        print('価格の取得処理においてリトライに失敗したので異常終了します。')

    # ローソク足のデータを入れる配列
    data = []
    for i in candle_span:
        row = res[i]
        for column in row[::-1]:
            # dataへローソク足データを追加
            if column[4] != 0:
                column = column[0:6]
                data.append(column)

    return data[::-1]



def connection_db(candleStick):

    # DBに接続する
    cnn = psycopg2.connect('dbname=XXXX host=localhost port=XXXX user=XXXX password=XXXX')
    cur = cnn.cursor()

    try:

        for row in candleStick:
            data_date = dt.datetime.fromtimestamp(row[0])
            open = row[1]
            high = row[2]
            low = row[3]
            close = row[4]
            volume = row[5]
            sql=(data_date, open, high, low, close, volume, open, high, low, close, volume)
            cur.execute(u"""INSERT INTO
                             "T_BTC_FX_5MIN_BITFLYER"
                             ("DATA_DATE", "OPEN", "HIGH", "LOW", "CLOSE", "VOLUME")
                             VALUES (%s, %s, %s, %s, %s, %s)
                             ON CONFLICT ON CONSTRAINT "PK_T_BTC_FX_5MIN_BITFLYER"
                             DO UPDATE SET
                                 "OPEN"   = %s
                               , "HIGH"   = %s
                               , "LOW"    = %s
                               , "CLOSE"  = %s
                               , "VOLUME" = %s """ , sql)

    except (psycopg2.OperationalError) as e:
        print (e)
    else:
        cnn.commit()
    finally:
        cur.close()
        cnn.close()


if __name__ == '__main__':

    # 価格の取得
    candleStick=get_ticker("300",-100)

    # DBへの保存
    connection_db(candleStick)

簡単に処理の解説だけします。
以下はメイン処理です。

if __name__ == '__main__':

    # 価格の取得
    candleStick=get_ticker("300",-100)

    # DBへの保存
    connection_db(candleStick)

get_tickerというファンクションでCryptoWatchのAPIを叩いて価格を取得していますが、
引数で渡している”300”というのは5分足のデータ、-100というのは100時間分のデータを
現在時刻からさかのぼって、指定して取得しています。

価格取得処理はいろいろなbotソースコードにも載っているので説明は割愛します。

続いてDBの登録処理です。

def connection_db(candleStick):

    # DBに接続する
    cnn = psycopg2.connect('dbname=XXXX host=localhost port=XXXX user=XXXX password=XXXX')
    cur = cnn.cursor()

    try:

        for row in candleStick:
            data_date = dt.datetime.fromtimestamp(row[0])
            open = row[1]
            high = row[2]
            low = row[3]
            close = row[4]
            volume = row[5]
            sql=(data_date, open, high, low, close, volume, open, high, low, close, volume)
            cur.execute(u"""INSERT INTO
                             "T_BTC_FX_5MIN_BITFLYER"
                             ("DATA_DATE", "OPEN", "HIGH", "LOW", "CLOSE", "VOLUME")
                             VALUES (%s, %s, %s, %s, %s, %s)
                             ON CONFLICT ON CONSTRAINT "PK_T_BTC_FX_5MIN_BITFLYER"
                             DO UPDATE SET
                                 "OPEN"   = %s
                               , "HIGH"   = %s
                               , "LOW"    = %s
                               , "CLOSE"  = %s
                               , "VOLUME" = %s """ , sql)

    except (psycopg2.OperationalError) as e:
        print (e)
    else:
        cnn.commit()
    finally:
        cur.close()
        cnn.close()

今回はpsycopg2というライブラリを使ってDBに接続しています。
他にもライブラリはあるのですが、postgreSQLに関して言えばインストール数が多く
ポピュラーだそうなので使用しています。

cnnという変数に接続情報を文字列で渡していますのでXXXXを書き換えてください。
localhostというのは利用者が現在操作している手元のコンピュータや端末を表します。
リモート環境にDBがある場合はIPを指定してください。

DB接続処理でのポイントは3つあります。

1. try-except-else-finallyで処理しよう

SQL以外のプログラムを介してDBとやり取りするときの定石なんですが、
接続が切れたり、処理に失敗したときの例外処理を丁寧に処理しなければなりません。
処理に成功したらelse:のブロックに入って登録したレコードをコミット(保存)していきます。

また、処理の成功・失敗に関わらずclose()を使って、接続を切断してあげなければなりません。
接続を切断しないと次にPGを実行しても、前のコネクションが次々とたまり続けて
DBが高負荷な状態になり、停止します。

2. バインドパラメータで変数を渡そう

INSERT文の中で”%s”と記載してありますがなんでしょう？
ここに実際に登録するデータが入るのですが「sql」という変数で渡している
パラメータの順にセットしています。

"pythonの文字列SQLを生成しているのなら+を使って変数と文字列結合すればいいじゃん！"
という意見があると思いますが、カッコいいだけでこのような書き方にしているわけではありません。
今回はあまり関係ないですが、SQLインジェクションというハッキング手法がありますので
これを機に覚えてみてください。
qiita.com

3. UPSERT文で処理する

INSERT INTO
             "T_BTC_FX_5MIN_BITFLYER"
             ("DATA_DATE", "OPEN", "HIGH", "LOW", "CLOSE", "VOLUME")
             VALUES (%s, %s, %s, %s, %s, %s)
             ON CONFLICT ON CONSTRAINT "PK_T_BTC_FX_5MIN_BITFLYER"
             DO UPDATE SET
                "OPEN"   = %s
              , "HIGH"   = %s
              , "LOW"    = %s
              , "CLOSE"  = %s
              , "VOLUME" = %s

実行SQLの部分を見てみましょう。
この構文はPostgreSQL9.5から採用されているUPSERT文というものです。
INSERT文を基本的に実行しますが、PK違反(一意制約違反)が発生した場合は
すでにDBに登録されているPKに合致する行の各項目に対してUPDATEをするという処理です。
OracleだとMerge文になります。

CryptowatchのAPIから取得するデータを調整して、同じデータを登録しないように制御するのは
かなり根気がいりますので、SQLで制御したほうが遥かに楽です！

インポートを自動化しよう

さぁ、いよいよ最後になります。
登録できるPGはあるものの、結局人間が1日1回とか手動で実行するの？(´・ω・｀)
そんなことはありません。プログラムに自動でやってもらいましょう！
Windowsのタスクスケジューラというアプリを使います。Cortanaさんに聞いてみましょう。

詳しい使い方は以下のリンクにありますので、こちらからどうぞ。
www.atmarkit.co.jp
このアプリで設定したファイルを時間になったら定期実行してくれます。

Windowsのタスクスケジューラで実行するファイルは以下のように作りました。
テキストファイルを開いて、以下の内容を張り付けて、
拡張子「.bat」で保存すれば完成です。
"db_import_job.bat"のファイルは"OneFileGetBitFlyerTicker.py"と同じディレクトリに配置してください。
今回はCドライブの直下に配置しましたので、下記のようなコードになっています。

db_import_job.bat

python C:\OneFileGetBitFlyerTicker.py

では最後にタスクスケジューラが正常終了したらpgAdmin3から結果を見てみましょう。

やったね^^

さて、今回はDBの基本からつらつらと書いてみましたが、いかがでしたでしょうか。
教養科目としてプログラムを習っていて、ソースを書ける学生さんはいても、
DBの取り扱いまで含めると数が少ないと思います。
なので、

”これからbot作成を通してITスキルを身に着けたい”

そんな風に思っている学生さんなどの皆様に読んでいただけたら幸いです。