#!/usr/bin/env python3
# coding: utf-8
# @Author: ArthurBernard
# @Email: arthur.bernard.92@gmail.com
# @Date: 2019-02-13 18:26:20
# @Last modified by: ArthurBernard
# @Last modified time: 2019-09-03 22:04:22
""" Objects to download historical data from Binance exchange.
.. currentmodule:: dccd.histo_dl.binance
.. autoclass:: FromBinance
:members: import_data, save, get_data, import_trades, save_trades, import_orderbook, save_orderbook
:show-inheritance:
"""
from __future__ import annotations
# Import built-in packages
import logging
from typing import Any
# Import third-party packages
from dccd.histo_dl.exchange import ImportDataCryptoCurrencies
# Import local packages
from dccd.tools.date_time import binance_interval
__all__ = ['FromBinance']
_logger = logging.getLogger(__name__)
[docs]
class FromBinance(ImportDataCryptoCurrencies):
""" Class to import crypto-currencies data from the Binance 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 ``'USD'``, which is silently coerced to
``'USDT'`` (Binance does not support fiat; only USDT is accepted as
a USD-equivalent). ``'EUR'`` is likewise coerced to ``'USDT'`` with
a warning.
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 (e.g. ``'Europe/Paris'``).
See Also
--------
FromKraken, FromCoinbase, FromBybit, FromOKX
Notes
-----
See Binance API documentation [1]_ for more details on parameters.
References
----------
.. [1] https://github.com/binance-exchange/binance-official-api-docs
Attributes
----------
pair : str
Pair symbol, `crypto + fiat`.
start, end : int
Timestamp to starting and ending download data.
span : int
Number of seconds between observations.
full_path : str
Directory managed by :class:`~dccd.storage.DataStore` —
``{path}/binance/ohlc/{pair}/{span}/``.
Methods
-------
import_data
save
get_data
import_trades
save_trades
import_orderbook
save_orderbook
"""
def __init__(self, path, crypto, span, fiat='USD', form='xlsx', tz='local'):
if fiat in ['EUR', 'USD']:
_logger.warning(
"Binance don't allow fiat currencies. "
"The equivalent of US dollar is Tether USD as USDT."
)
self.fiat = fiat = 'USDT'
ImportDataCryptoCurrencies.__init__(
self, path, crypto, span, 'Binance', 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 = {
'symbol': self.pair,
'startTime': self.start * 1000,
'endTime': self.end * 1000,
'interval': binance_interval(self.span),
'limit': 1000,
}
r = self._fetch('https://api.binance.com/api/v3/klines', param)
text = r.json()
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[7])
} for e in text]
return data
def _import_trades(self, start: int, end: int) -> list[dict[str, Any]]:
param = {
'symbol': self.pair,
'startTime': start * 1000,
'endTime': end * 1000,
'limit': 1000,
}
r = self._fetch('https://api.binance.com/api/v3/aggTrades', param)
return [{
'tid': int(e['a']),
'timestamp': float(e['T']) / 1000,
'price': float(e['p']),
'amount': float(e['q']),
'type': 'sell' if e['m'] else 'buy',
} for e in r.json()]
def _import_orderbook(self, depth: int = 50) -> list[dict[str, Any]]:
r = self._fetch(
'https://api.binance.com/api/v3/depth',
{'symbol': self.pair, 'limit': depth},
)
book = r.json()
result = []
for bid in book['bids']:
result.append({'side': 'bid', 'price': bid[0], 'amount': float(bid[1]), 'count': None})
for ask in book['asks']:
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 Binance for specific time interval.
Parameters
----------
start : int or str
Timestamp of the first observation of you want as int or date
format 'yyyy-mm-dd hh:mm:ss' as string.
end : int or str
Timestamp of the last observation of you want as int or date
format 'yyyy-mm-dd hh:mm:ss' as string.
Returns
-------
data : pl.DataFrame
Data sorted and cleaned in a data frame.
"""
data = self._import_data(start=start, end=end)
return self._sort_data(data)
