#!/usr/bin/env python3
# coding: utf-8
# @Author: ArthurBernard
# @Email: arthur.bernard.92@gmail.com
# @Date: 2019-02-13 18:25:01
# @Last modified by: ArthurBernard
# @Last modified time: 2019-09-03 22:05:04
""" Objects to download historical data from Kraken exchange.
.. currentmodule:: dccd.histo_dl.kraken
.. autoclass:: FromKraken
: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 time
import warnings
from typing import Any
# Import third party packages
# Import local packages
from dccd.histo_dl.exchange import ImportDataCryptoCurrencies
__all__ = ['FromKraken']
[docs]
class FromKraken(ImportDataCryptoCurrencies):
""" Class to import crypto-currencies data from the Kraken 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'``. The resulting Kraken pair uses
the X/Z prefix scheme (e.g. ``'USD'`` → ``'XXBTZUSD'``); use
:meth:`format_pair` to inspect the resolved symbol.
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, FromBybit, FromOKX
Notes
-----
See Kraken API documentation [1]_ for more details on parameters.
References
----------
.. [1] https://www.kraken.com/features/api
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}/kraken/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'):
ImportDataCryptoCurrencies.__init__(
self, path, crypto, span, 'Kraken', fiat=fiat, form=form, tz=tz
)
self.pair = self.format_pair(crypto, fiat)
def _import_data(
self, start: int | str = 'last', end: int | str | None = None
) -> list[dict[str, Any]]:
if end is not None:
warnings.warn(
"The Kraken OHLC API does not support an end date — the 'end' "
"parameter is ignored and data is always fetched up to now.",
UserWarning,
stacklevel=2,
)
self.start, self.end = self._set_time(start, int(time.time()))
param = {
'pair': self.pair,
'interval': int(self.span / 60),
'since': self.start - self.span
}
r = self._fetch('https://api.kraken.com/0/public/OHLC', param)
text = r.json()['result'][self.pair]
data = [{
'date': float(e[0]),
'open': float(e[1]),
'high': float(e[2]),
'low': float(e[3]),
'close': float(e[4]),
'weightedAverage': float(e[5]),
'volume': float(e[6]),
'quoteVolume': float(e[6]) * float(e[5])
} for e in text]
return data
def _import_trades(self, start: int, end: int) -> list[dict[str, Any]]:
r = self._fetch(
'https://api.kraken.com/0/public/Trades',
{'pair': self.pair, 'since': start},
)
trades = r.json()['result'][self.pair]
return [{
'tid': None,
'timestamp': float(e[2]),
'price': float(e[0]),
'amount': float(e[1]),
'type': 'buy' if e[3] == 'b' else 'sell',
} for e in trades if float(e[2]) <= end]
def _import_orderbook(self, depth: int = 50) -> list[dict[str, Any]]:
r = self._fetch(
'https://api.kraken.com/0/public/Depth',
{'pair': self.pair, 'count': depth},
)
book = r.json()['result'][self.pair]
result = []
for bid in book['bids']:
result.append({'side': 'bid', 'price': str(bid[0]), 'amount': float(bid[1]), 'count': None})
for ask in book['asks']:
result.append({'side': 'ask', 'price': str(ask[0]), 'amount': float(ask[1]), 'count': None})
return result
[docs]
def import_data(
self, start: int | str = 'last', end: int | str | None = None
) -> ImportDataCryptoCurrencies:
""" Download data from Kraken since a specific time until now.
Parameters
----------
start : int or str
Timestamp of the first observation as a Unix timestamp (int) or a
date string ``'yyyy-mm-dd hh:mm:ss'``.
end : int, str or None
Ignored. The Kraken OHLC API does not support a custom end date
and always returns data up to the current time. Passing a non-None
value raises a :class:`UserWarning`.
Returns
-------
self : FromKraken
Data sorted and cleaned in a data frame, accessible via
:meth:`get_data`.
"""
data = self._import_data(start=start, end=end)
return self._sort_data(data)
