stable/freqtrade/data/dataprovider.py

120 lines
4.3 KiB
Python

"""
Dataprovider
Responsible to provide data to the bot
including Klines, tickers, historic data
Common Interface for bot and strategy to access data.
"""
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from pandas import DataFrame
from freqtrade.data.history import load_pair_history
from freqtrade.exchange import Exchange
from freqtrade.state import RunMode
logger = logging.getLogger(__name__)
class DataProvider:
def __init__(self, config: dict, exchange: Exchange) -> None:
self._config = config
self._exchange = exchange
def refresh(self,
pairlist: List[Tuple[str, str]],
helping_pairs: List[Tuple[str, str]] = None) -> None:
"""
Refresh data, called with each cycle
"""
if helping_pairs:
self._exchange.refresh_latest_ohlcv(pairlist + helping_pairs)
else:
self._exchange.refresh_latest_ohlcv(pairlist)
@property
def available_pairs(self) -> List[Tuple[str, str]]:
"""
Return a list of tuples containing (pair, timeframe) for which data is currently cached.
Should be whitelist + open trades.
"""
return list(self._exchange._klines.keys())
def ohlcv(self, pair: str, timeframe: str = None, copy: bool = True) -> DataFrame:
"""
Get ohlcv data for the given pair as DataFrame
Please use the `available_pairs` method to verify which pairs are currently cached.
:param pair: pair to get the data for
:param timeframe: Ticker timeframe to get data for
:param copy: copy dataframe before returning if True.
Use False only for read-only operations (where the dataframe is not modified)
"""
if self.runmode in (RunMode.DRY_RUN, RunMode.LIVE):
return self._exchange.klines((pair, timeframe or self._config['ticker_interval']),
copy=copy)
else:
return DataFrame()
def historic_ohlcv(self, pair: str, timeframe: str = None) -> DataFrame:
"""
Get stored historic ohlcv data
:param pair: pair to get the data for
:param timeframe: ticker interval to get data for
"""
return load_pair_history(pair=pair,
timeframe=timeframe or self._config['ticker_interval'],
datadir=Path(self._config['datadir'])
)
def get_pair_dataframe(self, pair: str, timeframe: str = None) -> DataFrame:
"""
Return pair ohlcv data, either live or cached historical -- depending
on the runmode.
:param pair: pair to get the data for
:param timeframe: ticker interval to get data for
:return: Dataframe for this pair
"""
if self.runmode in (RunMode.DRY_RUN, RunMode.LIVE):
# Get live ohlcv data.
data = self.ohlcv(pair=pair, timeframe=timeframe)
else:
# Get historic ohlcv data (cached on disk).
data = self.historic_ohlcv(pair=pair, timeframe=timeframe)
if len(data) == 0:
logger.warning(f"No data found for ({pair}, {timeframe}).")
return data
def market(self, pair: str) -> Optional[Dict[str, Any]]:
"""
Return market data for the pair
:param pair: Pair to get the data for
:return: Market data dict from ccxt or None if market info is not available for the pair
"""
return self._exchange.markets.get(pair)
def ticker(self, pair: str):
"""
Return last ticker data
"""
# TODO: Implement me
pass
def orderbook(self, pair: str, maximum: int) -> Dict[str, List]:
"""
fetch latest orderbook data
:param pair: pair to get the data for
:param maximum: Maximum number of orderbook entries to query
:return: dict including bids/asks with a total of `maximum` entries.
"""
return self._exchange.get_order_book(pair, maximum)
@property
def runmode(self) -> RunMode:
"""
Get runmode of the bot
can be "live", "dry-run", "backtest", "edgecli", "hyperopt" or "other".
"""
return RunMode(self._config.get('runmode', RunMode.OTHER))