【Python】BitflyerのTickerをSQLiteに突っ込む
おはようございます。
今回は、前回までに取得できるようにした Bitflyer の Ticker情報を、そのままSQLiteに保存してみました。
プログラムは前回のものを流用します。
【Python】pubnub と WebSocket で更にリアルタイムで情報を取得する
スポンサーリンク
新規クラスの追加
Utils/SQLiteUtil.py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | importsqlite3 importlogging fromcontextlibimportclosing classSQLiteUtil: u""" SQLite 操作用クラス """ def__init__(self): """ インスタンス作成時にテーブルを作る """ self.db_name="Ticker.db" self.create_db() defcreate_db(self): u""" データベース、及び必要なテーブルを作成します. """ logging.info("create database") withclosing(sqlite3.connect(self.db_name))asconn: c=conn.cursor() # データクリア sql="DELETE FROM TBL_TICKER" c.execute(sql) # ティッカーテーブル sql="CREATE TABLE IF NOT EXISTS TBL_TICKER (" sql+=" PRODUCT_CODE TEXT" sql+=", TIME_STAMP TEXT" sql+=", TICK_ID TEXT" sql+=", BEST_BID REAL" sql+=", BEST_ASK REAL" sql+=", BEST_BID_SIZE REAL" sql+=", BEST_ASK_SIZE REAL" sql+=", TOTAL_BID_DEPTH REAL" sql+=", TOTAL_ASK_DEPTH REAL" sql+=", LTP REAL" sql+=", VOLUME REAL" sql+=", VOLUME_BY_PRODUCT REAL" sql+=", PRIMARY KEY (TICK_ID)" sql+=")" c.execute(sql) c.close() conn.commit() definsert_data(self,ticker): u""" 渡されたタプルデータを登録します """ withclosing(sqlite3.connect(self.db_name))asconn: c=conn.cursor() # 猫データ sql="INSERT INTO TBL_TICKER VALUES (?,?,?,?,?,?,?,?,?,?,?,?)" c.execute(sql,ticker) c.close() conn.commit() |
プログラムの修正
BfApi.py
ライブラリ使用宣言の追加
1 2 3 4 | fromdateutil importparser fromdatetimeimporttimedelta fromUtils.SQLiteUtil importSQLiteUtil |
pubnubのコールバッククラスで、メッセージ受信時の処理を修正
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | classMySubscriberCallback(SubscribeCallback): """ Pubnub登録のコールバッククラス """ def__init__(self,client): self.cl=client self.sqlite_util=SQLiteUtil() defpresence(self,pubnub,presence): pass # handle incoming presence data defstatus(self,pubnub,status): ifstatus.category==PNStatusCategory.PNUnexpectedDisconnectCategory: pass # This event happens when radio / connectivity is lost elifstatus.category==PNStatusCategory.PNConnectedCategory: # Connect event. You can do stuff like publish, and know you'll get it. # Or just use the connected event to confirm you are subscribed for # UI / internal notifications, etc pass elifstatus.category==PNStatusCategory.PNReconnectedCategory: pass # Happens as part of our regular operation. This event happens when # radio / connectivity is lost, then regained. elifstatus.category==PNStatusCategory.PNDecryptionErrorCategory: pass # Handle message decryption error. Probably client configured to # encrypt messages and on live data feed it received plain text. defmessage(self,pubnub,message): """ 登録したチャンネルからメッセージを受信した際の処理 :param pubnub: :param message: :return: """ # WEBソケットを利用してクライアントに配信 forcinself.cl: d=message.message self.sqlite_util.insert_data(( d["product_code"] ,self.parse_date(d["timestamp"]) ,d["tick_id"] ,d["best_bid"] ,d["best_ask"] ,d["best_bid_size"] ,d["best_ask_size"] ,d["total_bid_depth"] ,d["total_ask_depth"] ,d["ltp"] ,d["volume"] ,d["volume_by_product"] )) c.write_message(message.message) defparse_date(self,iso_date): date_time=parser.parse(iso_date)+timedelta(hours=9) returndate_time.strftime("%Y/%m/%d %H:%M:%S") |
データを追加してみる
起動して一覧の更新をスタートしてみてから、
A5:SQL Mk2 にてデータの確認をしました。
無事にデータが追加されています。
まとめ
単純にpubnubだけ動かすツールを作って、データを収集するだけでもよさそうですね。
ただ、Ticker情報だとデータ量が膨大になるため、1分、5分、10分などで処理やテーブルを分けてしまってもいいかもしれません。
ではでは。