Bybitの約定履歴から一秒足をつくる

Photo by moren hsu

黒枝といいます。

Bybitでダウンロードできる一日分の約定履歴から、一秒ローソク足を作るコードを書きました。

せっかく約定履歴から作っているので、ついでにvolume_bとvolume_sというキーで、売りと買いのvolumeを分けて保存しています。何か簡単なフットプリント的なことが出来るかもしれません。

Bybitの約定履歴はこちらから。

Directory listing for /trading/

使い方は、以下のコードと約定履歴を同じディレクトリに配置して、次のように引数を与えて起動してください。

通貨ペアと日付のフォーマットは、Bybitでダウンロートできるcsvのファイル名と同じです。

ファイル名の例：BTCUSD2021-05-01.csv

 one_sec_candles_gen.py BTCUSD 2021-05-24

あるいは、コード内で定数になっているPAIRとDATEの値を直接書き換えて頂いても大丈夫です。

出来上がるファイル名の例：1s-candles-2021-05-24-BTCUSD.csv

#!/usr/bin/python3
# coding: utf-8

from datetime import timedelta
from datetime import date
from datetime import datetime

import pathlib
import csv
import sys

current_dir = str(pathlib.Path(__file__).parent.absolute())
field_names = ['time', 'open', 'high', 'low', 'close', 'volume_b', 'volume_s']

PAIR = "BTCUSD"
DATE = "2021-05-24"

if len(sys.argv) >= 2:
   PAIR = sys.argv[1]
   DATE = sys.argv[2]


def init_file(_date):
   with open(current_dir + "/1s-candles-{}-{}.csv".format(DATE, PAIR), mode='w', newline="") as csv_file:
       csv_file.write('')
       writer = csv.DictWriter(csv_file, fieldnames=field_names)
       writer.writeheader()


def append_data(data, _date):
   with open(current_dir + "/1s-candles-{}-{}.csv".format(DATE, PAIR), mode='a', newline="") as csv_file:
       writer = csv.DictWriter(csv_file, fieldnames=field_names)
       writer.writerows([row for row in data])


def filling_row(start_time, close_price):
   return {
       'time': start_time.isoformat(),
       'open': close_price,
       'high': close_price,
       'low': close_price,
       'close': close_price,
       'volume_b': 0,
       'volume_s': 0
   }


def within_a_sec(start_time, item_timestamp):
   return start_time + timedelta(seconds=1) > datetime.fromtimestamp(item_timestamp)


def no_trade_for_a_sec(start_time, item_timestamp):
   return start_time + timedelta(seconds=2) < datetime.fromtimestamp(item_timestamp)


def create_ohlcv(pair="BTCUSD", _date="2021-05-21"):
   with open("{}/{}{}.csv".format(current_dir, pair, _date), mode='r', newline="") as f:
       reader = csv.reader(f)

       # 0-time 2-side 3-size 4-price.
       items = list(reader)[::-1]  # reverse items since the raw items are from new to old.
       items = items[:-1]  # exclude the header line.

       start = date.fromisoformat(_date) - timedelta(days=1)  # say data of 5th is from 4th to 5th, thus deduct 1 day.
       start = datetime(start.year, start.month, start.day) + timedelta(hours=20)  # data is from 20:00.

       results = list()

       prices = list()
       row = dict()
       volume_b = 0
       volume_s = 0

       close_price = float(items[0][4])
       if not within_a_sec(start, float(items[0][0])):
           # no trade within the first 1 sec of the day.
           close_price = 0

       for item in items:
           if within_a_sec(start, float(item[0])):
               prices.append(float(item[4]))
               if item[2] == "Buy":
                   volume_b += float(item[3])
               else:
                   volume_s += float(item[3])
           else:
               # 1 sec period breaks.

               if no_trade_for_a_sec(start, float(item[0])):
                   # fill in no trade period.
                   start += timedelta(seconds=1)
                   end = datetime.fromtimestamp(float(item[0]))
                   while True:
                       _row = filling_row(start, close_price)
                       results.append(_row)
                       if start >= end:
                           break
                       start += timedelta(seconds=1)

               start += timedelta(seconds=1)
               row['time'] = start.isoformat()
               row['volume_b'] = volume_b
               row['volume_s'] = volume_s
               if len(prices) > 0:
                   row['open'] = prices[0]
                   row['high'] = max(prices)
                   row['low'] = min(prices)
                   row['close'] = prices[-1]
                   close_price = prices[-1]  # keep the closed price in case no trade done in the next 1 seco.
               else:
                   # nobody traded.
                   row['open'] = close_price
                   row['high'] = close_price
                   row['low'] = close_price
                   row['close'] = close_price

               results.append(row)

               # reset for the next row.
               prices = list()
               row = dict()
               volume_b = 0
               volume_s = 0
       return results


def main():
   data = create_ohlcv(pair=PAIR, _date=DATE)
   init_file(DATE)
   append_data(data, DATE)
   return print("1s-candles-{}-{}.csv is generated.".format(DATE, PAIR))


if __name__ == '__main__':
   main()