Pythonでティッカー情報をデータベースへの格納に挑戦してみた

データベースとfor文　

仮想通貨取引所のAPIから取得したティッカー情報は膨大だ。それを加工するためには、ティッカー情報をデータベースに入れる方が処理速度が早い。

そして、暗号資産の銘柄ごとにデータベースへ格納するためには、for文を使って繰り返すのが効率的だ。

今回から、少しづつレベルが上がるけど、打倒資本家を目指して手を動かそう。

for文と配列

今回データベースを作るので、以前に書いたコードを大幅に書き直さないといけない。前に書いていたコードはこちら。

# -*- coding: utf-8 -*- #
"""
We crush capitalism.
"""

import pybitflyer
import csv
import time

"""
初期設定
"""
Start_time = time.perf_counter()

while True:
   print(Start_time)
   BTC_tick = pybitflyer.API().ticker(product_code="BTC_JPY")
   print("BTC_timestamp : " + str(BTC_tick["timestamp"]))
   print("BTC_ltp : " + str(BTC_tick["ltp"]))
   print("BTC_best_bid : " + str(BTC_tick["best_bid"]))
   print("BTC_best_ask : " + str(BTC_tick["best_ask"]))

   ETH_tick = pybitflyer.API().ticker(product_code="ETH_JPY")
   print("ETH_timestamp : " + str(ETH_tick["timestamp"]))
   print("ETH_ltp : " + str(ETH_tick["ltp"]))
   print("ETH_best_bid : " + str(ETH_tick["best_bid"]))
   print("ETH_best_ask : " + str(ETH_tick["best_ask"]))

   with open('furture.csv','a',newline='') as files:
      writer = csv.writer(files)
      writer.writerow(['TIME',str(ETH_tick["timestamp"]),'ltp',str(ETH_tick["ltp"]),'bid',(ETH_tick["best_bid"]),'ask',str(ETH_tick["best_ask"])])
   time.sleep((10 + Start_time ) - time.perf_counter())
   Start_time = Start_time + 10

今回は、暗号資産の銘柄ごとに繰り返しデータを抽出するコードを書こう。そうすれば、取引所がAPIで使える銘柄を増やしたとしても対応できる。
まずは、for文のおさらいだ。

pythonのfor文では、変数の後ろに「in」が付くことと、オブジェクトの後ろに「:」が付くことに注意しましょう。

まず、初期設定の下に次のコードを入れる。この、ビットコインとイーサリアムのシンボルであるBTCとETHをTicker_symbolという引数に配列として代入している。

Ticker_symbol = ["BTC", "ETH"]

プログラムでいう配列は、データのまとまりのことを言うけど、Pythonの配列にはリストとタプルがある。今回は、リストを使うので[ ]で囲っているけど、タプルの場合は( )で囲う。

Pythonで配列を扱う際にはリストを使います。タプルも配列を扱うことができますが、タプルは要素を変更できないという違いがあります。

次は、以前書いたwhileからwithの間のコードを一気に書き直してしまおう。
for文の意味はこうだ。引数Ticker_symbolのリストの数だけ繰り返し実行し、引数の中にあるBTCやETHという文字列をSYMBOLという引数に代入するという命令だ。

   for SYMBOL in Ticker_symbol:
      """
      ティッカーからデータ取得
      """

一回、引数SYMBOLに代入されれば、BTCやETHを別々にコードに書かなくてもfor文で繰り返し実行してくれる。
次のコードは、3文字のシンボルに_JPYという文字列を加えることで、BTC_JPYかETH_JPYという新しい引数を作っている。

CODE = SYMBOL + "_JPY"

今回は、以前記事に書いた拡張モジュールpybitflyerを使って、bitflyer APIの全11情報を引数にしている。これは、データベース化するための前準備だ。
ちなみに、初期設定に暗号資産のシンボルコードを追加すれば、さらに繰り返し処理ができる。

      tick = pybitflyer.API().ticker(product_code=CODE)
      timestamp = tick["timestamp"]
      ltp = tick["ltp"]
      ask = tick["best_ask"]
      bid = tick["best_bid"]
      ask_size = tick["best_ask_size"]
      bid_size = tick["best_bid_size"]
      ask_depth = tick["total_ask_depth"]
      bid_depth = tick["total_ask_depth"]
      volume_product = tick["volume_by_product"]
      volume = tick["volume"]

データベースSQLite3を使う

本当は、pandasを使いたかったんだけど、携帯電話アプリPythonista3のlaunch_stashではエラーが出てインストールできなかった。だから、Python3内臓のSQLite3を使うことにした。

SQLiteを一言でいうと、オープンソースで軽量のRDBMS（データベース管理システム）です。SQLと名前が似ているため、データベース操作言語の一種と思う方もいますが、SQLiteはデータベースです。
データベースには、OracleをはじめMicrosoft SQL Serverなどの商用データベースが存在します。また、SQLiteと同じくオープンソースのMySQLやPostgreSQLも有名です。

いつものように、拡張モジュールとしてSQLite3をコードの初期設定の上に書く。

import sqlite3

次に、SQLite3のデータベースである拡張子.dbというファイルに接続する。SQLite3の命令文connectは、ファイルがない時に自動で新しいファイルを作ってくれる。

conn = sqlite3.connect('dbファイルパス')

指定したDBファイルが存在しない場合は自動で作成してくれるため、高い可搬性を実現することができます。また、特別な名前である :memory: を使うとRAM上にデータベース、いわゆるインメモリ（若しくはオンメモリ）で作ることもできます。

コードで書くとこうなる。Data_Baseという引数に、furtureというデータベースに接続するという命令だ。次に、coursor()の命令も代入する。おまじないみたいだけど、これをやらないとエラーになる。

"""
初期設定
"""
Data_Base = sqlite3.connect("furture.db")
Data_cur = Data_Base.cursor()

次は、データベースのテーブルの作成だ。Excelの中にあるデータ項目を入れるシートだと思ってくれればいいかな。次は、while文の下に入れるデータベースに情報を加える方法だ。

データベースにはさまざまな種類のデータが格納されています。そのデータはバラバラではなく、データの種類ごとに規則正しく保存されているのが特徴です。このデータベースの種類ごとの単位を「テーブル」と呼びます。
データベースのテーブルはExcelで例えられることが多く、Excelのシートがテーブルを表します。データベースは複数のテーブルを保持し、テーブルごとにデータを管理しています。

今回は、ティッカーの11情報全てをデータベースにするから、11項目作る必要がある。
CREATE TABLE IF NOT EXISTS furtureというのは、sqlite3でfurtureというテーブルがなければ新しく作りますという内容の命令文だ。
次は、データベースに書き込む命令。最後は、データベースを閉じる命令だ。

Data_Base.cursor().execute('CREATE TABLE IF NOT EXISTS furture(symbol text, timestamp datatime, lpt real, ask real, bid real, ask_size real, bid_size real, ask_depth real, bid_depth real, volume_product real, volume real)')
Data_Base.commit()
Data_Base.close()

書き込む命令と閉じる命令は、面倒だけどデータが変更された都度やらないといけないらしい。テーブルの準備ができたら、データベースにデータを差し込むための準備をする。

データを挿入するSQLは以下の形をしています。
insert into テーブル名(列名、列名、…) values(値、値、…)
insert文を実行した後は、commitメソッドを実行する必要があります。テーブルへの挿入がcommitによりテーブルに反映されます。commitメソッドを実行しないとテーブルへの挿入は行われません。

最初の二行は、データベースを開いて、命令を出せる状態にしている。三行目は、11個の格納箱を作り、データベースへと情報を格納する命令だ。そして、最後が書き込みの処理になる。

Data_Base = sqlite3.connect("furture.db")
Data_cur = Data_Base.cursor()
Data_cur.execute('INSERT INTO furture values (?,?,?,?,?,?,?,?,?,?,?)',[SYMBOL, timestamp, ltp, ask, bid, ask_size, bid_size, ask_depth, bid_depth, volume_product, volume])
Data_Base.commit()

正しくデータベースにデータが格納したか確認するために、データを抽出して画面にprintしてみよう。

この記事では、Pythonのsqlite3を使ってテーブルのデータの抽出(SELECT)を行う処理について説明しています。
具体的には単純なSELECT文、条件を絞り込んだSELECT、テーブル同士をJOINした場合のSELECTについてのサンプルソースを使いながら解説していきたいと思います。

SELECT * FROM furtureの命令文で全てのデータを抽出する。そして、先ほど説明したfor文を使いデータを出力してみた。そして、close()でデータベースを閉じている。

"""
実行結果の表示
"""
      Data_cur.execute("SELECT * FROM furture")
      for Row in Data_cur:
         print(str(Row[0]) + "," + str(Row[1]) + "," + str(Row[2]))
      Data_Base.close()

試しに出力すると何故かデータが重複していた。色々なサイトを調べてもどうしてこうなったのかわからなかった。

試しにデータベースの中身を見るアプリで確認してみると、データは重複していなかった。print出力が間違っていたようだ。

今回は、csvファイルに11データ分を出力しているから、csvを作るコードも修正した。念のため、csvデータの重複も確認したけど、こちらも問題なかった。

今回出来上がったコードはこちら。
次回は、データベースから情報を取り出すことに挑戦しよう。

# -*- coding: utf-8 -*- #
"""
We crush capitalism.
"""

import pybitflyer
import csv
import time
import sqlite3
"""
初期設定
"""
Data_Base = sqlite3.connect("furture.db")
Data_cur = Data_Base.cursor()
Data_cur.execute('CREATE TABLE IF NOT EXISTS furture(symbol text, timestamp datatime, lpt real, ask real, bid real, ask_size real, bid_size real, ask_depth real, bid_depth real, volume_product real, volume real)')
Data_Base.commit()
Data_Base.close()

Start_time = time.perf_counter()
Ticker_symbol = ["BTC", "ETH"]

while True:
   print(Start_time)
   print(time.perf_counter())
   """
   ティッカー操作処理
   """
   for SYMBOL in Ticker_symbol:
      """
      ティッカーからデータ取得
      """
      CODE = SYMBOL + "_JPY"
      tick = pybitflyer.API().ticker(product_code=CODE)
      timestamp = tick["timestamp"]
      ltp = tick["ltp"]
      ask = tick["best_ask"]
      bid = tick["best_bid"]
      ask_size = tick["best_ask_size"]
      bid_size = tick["best_bid_size"]
      ask_depth = tick["total_ask_depth"]
      bid_depth = tick["total_ask_depth"]
      volume_product = tick["volume_by_product"]
      volume = tick["volume"]
      """
      データベースへ格納
      """
      Data_Base = sqlite3.connect("furture.db")
      Data_cur = Data_Base.cursor()
      Data_cur.execute('INSERT INTO furture values (?,?,?,?,?,?,?,?,?,?,?)',[SYMBOL, timestamp, ltp, ask, bid, ask_size, bid_size, ask_depth, bid_depth, volume_product, volume])
      Data_Base.commit()
      """
      実行結果の表示
      """
      Data_cur.execute("SELECT * FROM furture")
      for Row in Data_cur:
         print(str(Row[0]) + "," + str(Row[1]) + "," + str(Row[2]))
      Data_Base.close()
      """
      CSVへの書き込む
      """
      with open('furture.csv','a',newline='') as files:
         writer = csv.writer(files)
         writer.writerow([SYMBOL,str(timestamp),str(ltp),str(ask),str(bid),str(ask_size),str(bid_size),str(ask_depth),str(bid_depth),str(volume_product),str(volume)])
   time.sleep((10 + Start_time ) - time.perf_counter())
   Start_time = Start_time + 10

目次

資本主義社会を、誰も傷つけずに修正させる方法を考える｜中島充｜note

次の記事

SQLite3を使ってデータベース化したティッカー情報抽出に挑戦してみた｜中島充｜note

前の記事

チャート分析のため見えない情報の検討に挑戦してみた｜中島充｜note