やったこと

Websocket で約定情報や板情報などをリアルタイム配信してくれているが、どの程度の遅延が発生するのか調べた。

動機

こちらのツイート。

計測1

期間

2018/07/09 14:00〜2018/07/09 14:20

結果

データ件数 22447件
最大遅延 5.110284[sec]
最小遅延 0.165391[sec]
平均遅延 0.8038816492[sec]

計測2

期間

2018/07/09 17:29〜2018/07/09 18:52

結果

データ件数 99999件 (※スプレッドシートの仕様につき)
最大遅延 3.711575[sec]
最小遅延 0.165403[sec]
平均遅延 0.6548657885[sec]

所感

5秒の遅延は1分足のOHLCデータを作る上で影響が大きいかもしれない。バックテスト用にHistoricalデータを作りたいというだけなら良いが、1分足データをベースに動くbotを作るような場合には注意が必要そう。

ただし、ヒストグラムを見たところ、概ね1秒以内に約定情報が配信されており、極端な遅延は(計測期間内では)そこまで多くないように見受けられる。

今回は計測期間が短いため、より詳細に分析する場合には数日規模のデータを集める必要があるだろう。また、取引所の状態(BUSY など)も合わせて取得した方が良い。

データはこちら

コード

「もっとこうするといいよ」、「計測方法間違ってない？」などあれば教えて欲しい。

import json
import websocket
import datetime
from time import sleep
from logging import getLogger,INFO,StreamHandler
logger = getLogger(__name__)
handler = StreamHandler()
handler.setLevel(INFO)
logger.setLevel(INFO)
logger.addHandler(handler)

def get_exec_date(d):
  exec_date = d["exec_date"].replace('T', ' ')[:-1]
  return datetime.datetime(int(exec_date[:4]), int(exec_date[5:7]), int(exec_date[8:10]), int(exec_date[11:13]), int(exec_date[14:16]), int(exec_date[17:19]), int(exec_date[20:-1])) + datetime.timedelta(hours=9)

"""
This program calls Bitflyer real time API JSON-RPC2.0 over Websocket
"""
class RealtimeAPI(object):
  def __init__(self, url, channel):
    self.url = url
    self.channel = channel

    #Define Websocket
    self.ws = websocket.WebSocketApp(self.url, header=None, on_open=self.on_open, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close)
    websocket.enableTrace(True)

  def run(self):
    #ws has loop. To break this press ctrl + c to occur Keyboard Interruption Exception.
    self.ws.run_forever()
    logger.info('Web Socket process ended.')

  """
  Below are callback functions of websocket.
  """
  # when we get message
  def on_message(self, ws, message):
    output = json.loads(message)['params']
    now = datetime.datetime.now()
    for d in output["message"]:
      exec_date = get_exec_date(d)
      delta = now - exec_date
      side = d["side"]
      price = d["price"]
      size = d["size"]
      id = d["id"]
      logger.info("{}, {}, {}, {}, {}, {}, {}".format(now, exec_date, delta.seconds + delta.microseconds / 1000000, side, price, size, id))

  # when error occurs
  def on_error(self, ws, error):
    logger.error(error)

  # when websocket closed.
  def on_close(self, ws):
    logger.info('disconnected streaming server')

  # when websocket opened.
  def on_open(self, ws):
    logger.info('connected streaming server')
    output_json = json.dumps(
      {'method' : 'subscribe',
      'params' : {'channel' : self.channel}
      }
    )
    ws.send(output_json)

if __name__ == '__main__':
  #API endpoint
  url = 'wss://ws.lightstream.bitflyer.com/json-rpc'
  channel = 'lightning_executions_FX_BTC_JPY' # 約定
  json_rpc = RealtimeAPI(url=url, channel=channel)
  #ctrl + cで終了
  json_rpc.run()

CSV に出力

$ python get_realtime_api.py 2> measure_wss_delay.csv

コードベース

API ドキュメント

おわりに

有料(¥100)にしてるけど、これで内容は全部です。募金してくれる人がいれば、ジュース代としていただけると嬉しい。コードは、インデントくずれが起きたりするようなので、コピペ時には注意してください。

マガジン

コメント用note(未購入者向け)

干し芋