Bybitの約定履歴から一秒足をつくる
Photo by moren hsu
黒枝といいます。
Bybitでダウンロードできる一日分の約定履歴から、一秒ローソク足を作るコードを書きました。
せっかく約定履歴から作っているので、ついでにvolume_bとvolume_sというキーで、売りと買いのvolumeを分けて保存しています。何か簡単なフットプリント的なことが出来るかもしれません。
Bybitの約定履歴はこちらから。
使い方は、以下のコードと約定履歴を同じディレクトリに配置して、次のように引数を与えて起動してください。
通貨ペアと日付のフォーマットは、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()