#!/usr/bin/env python3
# coding: utf-8
""" Objects to download historical data from Bybit exchange.
.. currentmodule:: dccd.histo_dl.bybit
.. autoclass:: FromBybit
:members: import_data, save, get_data, import_trades, save_trades, import_orderbook, save_orderbook
:show-inheritance:
"""
from __future__ import annotations
# Import built-in packages
from typing import Any
# Import third-party packages
# Import local packages
from dccd.histo_dl.exchange import ImportDataCryptoCurrencies
__all__ = ['FromBybit']
_BYBIT_INTERVALS = {
60: '1', 180: '3', 300: '5', 900: '15', 1800: '30',
3600: '60', 7200: '120', 14400: '240', 21600: '360', 43200: '720',
86400: 'D', 604800: 'W',
}
def bybit_interval(span):
""" Return the Bybit interval string for the given span in seconds.
Parameters
----------
span : int
Interval in seconds.
Returns
-------
str
Bybit interval identifier.
Examples
--------
>>> bybit_interval(3600)
'60'
>>> bybit_interval(86400)
'D'
"""
interval = _BYBIT_INTERVALS.get(span)
if interval is None:
raise ValueError(f"Unsupported Bybit interval: {span}s")
return interval
[docs]
class FromBybit(ImportDataCryptoCurrencies):
""" Class to import crypto-currencies data from the Bybit exchange.
Parameters
----------
path : str
Root directory for data files.
crypto : str
Crypto-currency symbol, e.g. ``'BTC'``.
span : int or str
Candle interval in seconds (minimum 60) or a label such as
``'hourly'`` or ``'1h'``.
fiat : str, optional
Quote currency. Default is ``'USDT'``. The pair is formed by
concatenation (e.g. ``'BTC'`` + ``'USDT'`` → ``'BTCUSDT'``).
form : str, optional
Legacy parameter — ignored. Storage is always Parquet via
:class:`~dccd.storage.DataStore`.
tz : str, optional
Timezone for date parsing: ``'local'`` (default), ``'UTC'``, or any
IANA timezone name.
See Also
--------
FromBinance, FromCoinbase, FromKraken, FromOKX
Notes
-----
Uses the Bybit v5 REST API [1]_.
References
----------
.. [1] https://bybit-exchange.github.io/docs/v5/market/kline
Attributes
----------
pair : str
Pair symbol (e.g. 'BTCUSDT').
start, end : int
Timestamps bounding the download.
span : int
Seconds between observations.
full_path : str
Directory managed by :class:`~dccd.storage.DataStore` —
``{path}/bybit/ohlc/{pair}/{span}/``.
Methods
-------
import_data
save
get_data
import_trades
save_trades
import_orderbook
save_orderbook
"""
def __init__(self, path, crypto, span, fiat='USDT', form='xlsx', tz='local'):
ImportDataCryptoCurrencies.__init__(
self, path, crypto, span, 'Bybit', fiat, form, tz=tz
)
self.pair = self.format_pair(crypto, fiat)
def _import_data(self, start: int | str = 'last', end: int | str = 'now') -> list[dict[str, Any]]:
self.start, self.end = self._set_time(start, end)
param = {
'category': 'spot',
'symbol': self.pair,
'interval': bybit_interval(self.span),
'start': self.start * 1000,
'end': self.end * 1000,
'limit': 1000,
}
r = self._fetch('https://api.bybit.com/v5/market/kline', param)
text = r.json()['result']['list']
text.reverse()
data = [{
'date': float(e[0]) / 1000,
'open': float(e[1]),
'high': float(e[2]),
'low': float(e[3]),
'close': float(e[4]),
'volume': float(e[5]),
'quoteVolume': float(e[6]),
} for e in text]
return data
def _import_trades(self, start: int, end: int) -> list[dict[str, Any]]:
""" Fetch the most recent trades from Bybit (recent data only).
Notes
-----
The Bybit public REST API returns up to 1 000 of the most recent trades
regardless of ``start``/``end``. Historical trades are not available
via this endpoint.
"""
r = self._fetch(
'https://api.bybit.com/v5/market/recent-trade',
{'category': 'spot', 'symbol': self.pair, 'limit': 1000},
)
return [{
'tid': None,
'timestamp': float(e['time']) / 1000,
'price': float(e['price']),
'amount': float(e['size']),
'type': 'buy' if e['side'] == 'Buy' else 'sell',
} for e in r.json()['result']['list']]
def _import_orderbook(self, depth: int = 50) -> list[dict[str, Any]]:
r = self._fetch(
'https://api.bybit.com/v5/market/orderbook',
{'category': 'spot', 'symbol': self.pair, 'limit': depth},
)
book = r.json()['result']
result = []
for bid in book['b']:
result.append({'side': 'bid', 'price': bid[0], 'amount': float(bid[1]), 'count': None})
for ask in book['a']:
result.append({'side': 'ask', 'price': ask[0], 'amount': float(ask[1]), 'count': None})
return result
[docs]
def import_data(self, start: int | str = 'last', end: int | str = 'now') -> ImportDataCryptoCurrencies:
""" Download data from Bybit for a specific time interval.
Parameters
----------
start : int or str
Timestamp of the first observation or date 'yyyy-mm-dd hh:mm:ss'.
end : int or str
Timestamp of the last observation or date 'yyyy-mm-dd hh:mm:ss'.
Returns
-------
data : pl.DataFrame
OHLCV data sorted and cleaned.
"""
data = self._import_data(start=start, end=end)
return self._sort_data(data)
