6181253459

Committed 08 Sep 2023 06:04AM UTC coverage: 94.614% (+0.06%) from 94.556%

Build # 6181253459

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Merge pull request #9159 from stash86/fix-adjust

remove old codes when we only can do partial entries

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

19114 of 20202 relevant lines covered (94.61%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.29

/freqtrade/data/history/idatahandler.py

"""
Abstract datahandler interface.
It's subclasses handle and storing data from disk.

"""
import logging
import re
from abc import ABC, abstractmethod
from copy import deepcopy
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Optional, Tuple, Type

from pandas import DataFrame

from freqtrade import misc
from freqtrade.configuration import TimeRange
from freqtrade.constants import DEFAULT_TRADES_COLUMNS, ListPairsWithTimeframes
from freqtrade.data.converter import (clean_ohlcv_dataframe, trades_convert_types,
                                      trades_df_remove_duplicates, trim_dataframe)
from freqtrade.enums import CandleType, TradingMode
from freqtrade.exchange import timeframe_to_seconds


logger = logging.getLogger(__name__)


class IDataHandler(ABC):

    _OHLCV_REGEX = r'^([a-zA-Z_\d-]+)\-(\d+[a-zA-Z]{1,2})\-?([a-zA-Z_]*)?(?=\.)'

    def __init__(self, datadir: Path) -> None:
        self._datadir = datadir

    @classmethod
    def _get_file_extension(cls) -> str:
        """
        Get file extension for this particular datahandler
        """
        raise NotImplementedError()

    @classmethod
    def ohlcv_get_available_data(
            cls, datadir: Path, trading_mode: TradingMode) -> ListPairsWithTimeframes:
        """
        Returns a list of all pairs with ohlcv data available in this datadir
        :param datadir: Directory to search for ohlcv files
        :param trading_mode: trading-mode to be used
        :return: List of Tuples of (pair, timeframe, CandleType)
        """
        if trading_mode == TradingMode.FUTURES:
            datadir = datadir.joinpath('futures')
        _tmp = [
            re.search(
                cls._OHLCV_REGEX, p.name
            ) for p in datadir.glob(f"*.{cls._get_file_extension()}")]
        return [
            (
                cls.rebuild_pair_from_filename(match[1]),
                cls.rebuild_timeframe_from_filename(match[2]),
                CandleType.from_string(match[3])
            ) for match in _tmp if match and len(match.groups()) > 1]

    @classmethod
    def ohlcv_get_pairs(cls, datadir: Path, timeframe: str, candle_type: CandleType) -> List[str]:
        """
        Returns a list of all pairs with ohlcv data available in this datadir
        for the specified timeframe
        :param datadir: Directory to search for ohlcv files
        :param timeframe: Timeframe to search pairs for
        :param candle_type: Any of the enum CandleType (must match trading mode!)
        :return: List of Pairs
        """
        candle = ""
        if candle_type != CandleType.SPOT:
            datadir = datadir.joinpath('futures')
            candle = f"-{candle_type}"
        ext = cls._get_file_extension()
        _tmp = [re.search(r'^(\S+)(?=\-' + timeframe + candle + f'.{ext})', p.name)
                for p in datadir.glob(f"*{timeframe}{candle}.{ext}")]
        # Check if regex found something and only return these results
        return [cls.rebuild_pair_from_filename(match[0]) for match in _tmp if match]

    @abstractmethod
    def ohlcv_store(
            self, pair: str, timeframe: str, data: DataFrame, candle_type: CandleType) -> None:
        """
        Store ohlcv data.
        :param pair: Pair - used to generate filename
        :param timeframe: Timeframe - used to generate filename
        :param data: Dataframe containing OHLCV data
        :param candle_type: Any of the enum CandleType (must match trading mode!)
        :return: None
        """

    def ohlcv_data_min_max(self, pair: str, timeframe: str,
                           candle_type: CandleType) -> Tuple[datetime, datetime]:
        """
        Returns the min and max timestamp for the given pair and timeframe.
        :param pair: Pair to get min/max for
        :param timeframe: Timeframe to get min/max for
        :param candle_type: Any of the enum CandleType (must match trading mode!)
        :return: (min, max)
        """
        data = self._ohlcv_load(pair, timeframe, None, candle_type)
        if data.empty:
            return (
                datetime.fromtimestamp(0, tz=timezone.utc),
                datetime.fromtimestamp(0, tz=timezone.utc)
            )
        return data.iloc[0]['date'].to_pydatetime(), data.iloc[-1]['date'].to_pydatetime()

    @abstractmethod
    def _ohlcv_load(self, pair: str, timeframe: str, timerange: Optional[TimeRange],
                    candle_type: CandleType
                    ) -> DataFrame:
        """
        Internal method used to load data for one pair from disk.
        Implements the loading and conversion to a Pandas dataframe.
        Timerange trimming and dataframe validation happens outside of this method.
        :param pair: Pair to load data
        :param timeframe: Timeframe (e.g. "5m")
        :param timerange: Limit data to be loaded to this timerange.
                        Optionally implemented by subclasses to avoid loading
                        all data where possible.
        :param candle_type: Any of the enum CandleType (must match trading mode!)
        :return: DataFrame with ohlcv data, or empty DataFrame
        """

    def ohlcv_purge(self, pair: str, timeframe: str, candle_type: CandleType) -> bool:
        """
        Remove data for this pair
        :param pair: Delete data for this pair.
        :param timeframe: Timeframe (e.g. "5m")
        :param candle_type: Any of the enum CandleType (must match trading mode!)
        :return: True when deleted, false if file did not exist.
        """
        filename = self._pair_data_filename(self._datadir, pair, timeframe, candle_type)
        if filename.exists():
            filename.unlink()
            return True
        return False

    @abstractmethod
    def ohlcv_append(
        self,
        pair: str,
        timeframe: str,
        data: DataFrame,
        candle_type: CandleType
    ) -> None:
        """
        Append data to existing data structures
        :param pair: Pair
        :param timeframe: Timeframe this ohlcv data is for
        :param data: Data to append.
        :param candle_type: Any of the enum CandleType (must match trading mode!)
        """

    @classmethod
    def trades_get_pairs(cls, datadir: Path) -> List[str]:
        """
        Returns a list of all pairs for which trade data is available in this
        :param datadir: Directory to search for ohlcv files
        :return: List of Pairs
        """
        _ext = cls._get_file_extension()
        _tmp = [re.search(r'^(\S+)(?=\-trades.' + _ext + ')', p.name)
                for p in datadir.glob(f"*trades.{_ext}")]
        # Check if regex found something and only return these results to avoid exceptions.
        return [cls.rebuild_pair_from_filename(match[0]) for match in _tmp if match]

    @abstractmethod
    def _trades_store(self, pair: str, data: DataFrame) -> None:
        """
        Store trades data (list of Dicts) to file
        :param pair: Pair - used for filename
        :param data: Dataframe containing trades
                     column sequence as in DEFAULT_TRADES_COLUMNS
        """

    @abstractmethod
    def trades_append(self, pair: str, data: DataFrame):
        """
        Append data to existing files
        :param pair: Pair - used for filename
        :param data: Dataframe containing trades
                     column sequence as in DEFAULT_TRADES_COLUMNS
        """

    @abstractmethod
    def _trades_load(self, pair: str, timerange: Optional[TimeRange] = None) -> DataFrame:
        """
        Load a pair from file, either .json.gz or .json
        :param pair: Load trades for this pair
        :param timerange: Timerange to load trades for - currently not implemented
        :return: Dataframe containing trades
        """

    def trades_store(self, pair: str, data: DataFrame) -> None:
        """
        Store trades data (list of Dicts) to file
        :param pair: Pair - used for filename
        :param data: Dataframe containing trades
                     column sequence as in DEFAULT_TRADES_COLUMNS
        """
        # Filter on expected columns (will remove the actual date column).
        self._trades_store(pair, data[DEFAULT_TRADES_COLUMNS])

    def trades_purge(self, pair: str) -> bool:
        """
        Remove data for this pair
        :param pair: Delete data for this pair.
        :return: True when deleted, false if file did not exist.
        """
        filename = self._pair_trades_filename(self._datadir, pair)
        if filename.exists():
            filename.unlink()
            return True
        return False

    def trades_load(self, pair: str, timerange: Optional[TimeRange] = None) -> DataFrame:
        """
        Load a pair from file, either .json.gz or .json
        Removes duplicates in the process.
        :param pair: Load trades for this pair
        :param timerange: Timerange to load trades for - currently not implemented
        :return: List of trades
        """
        trades = trades_df_remove_duplicates(self._trades_load(pair, timerange=timerange))

        trades = trades_convert_types(trades)
        return trades

    @classmethod
    def create_dir_if_needed(cls, datadir: Path):
        """
        Creates datadir if necessary
        should only create directories for "futures" mode at the moment.
        """
        if not datadir.parent.is_dir():
            datadir.parent.mkdir()

    @classmethod
    def _pair_data_filename(
        cls,
        datadir: Path,
        pair: str,
        timeframe: str,
        candle_type: CandleType,
        no_timeframe_modify: bool = False
    ) -> Path:
        pair_s = misc.pair_to_filename(pair)
        candle = ""
        if not no_timeframe_modify:
            timeframe = cls.timeframe_to_file(timeframe)

        if candle_type != CandleType.SPOT:
            datadir = datadir.joinpath('futures')
            candle = f"-{candle_type}"
        filename = datadir.joinpath(
            f'{pair_s}-{timeframe}{candle}.{cls._get_file_extension()}')
        return filename

    @classmethod
    def _pair_trades_filename(cls, datadir: Path, pair: str) -> Path:
        pair_s = misc.pair_to_filename(pair)
        filename = datadir.joinpath(f'{pair_s}-trades.{cls._get_file_extension()}')
        return filename

    @staticmethod
    def timeframe_to_file(timeframe: str):
        return timeframe.replace('M', 'Mo')

    @staticmethod
    def rebuild_timeframe_from_filename(timeframe: str) -> str:
        """
        converts timeframe from disk to file
        Replaces mo with M (to avoid problems on case-insensitive filesystems)
        """
        return re.sub('1mo', '1M', timeframe, flags=re.IGNORECASE)

    @staticmethod
    def rebuild_pair_from_filename(pair: str) -> str:
        """
        Rebuild pair name from filename
        Assumes a asset name of max. 7 length to also support BTC-PERP and BTC-PERP:USD names.
        """
        res = re.sub(r'^(([A-Za-z\d]{1,10})|^([A-Za-z\-]{1,6}))(_)', r'\g<1>/', pair, 1)
        res = re.sub('_', ':', res, 1)
        return res

    def ohlcv_load(self, pair, timeframe: str,
                   candle_type: CandleType, *,
                   timerange: Optional[TimeRange] = None,
                   fill_missing: bool = True,
                   drop_incomplete: bool = False,
                   startup_candles: int = 0,
                   warn_no_data: bool = True,
                   ) -> DataFrame:
        """
        Load cached candle (OHLCV) data for the given pair.

        :param pair: Pair to load data for
        :param timeframe: Timeframe (e.g. "5m")
        :param timerange: Limit data to be loaded to this timerange
        :param fill_missing: Fill missing values with "No action"-candles
        :param drop_incomplete: Drop last candle assuming it may be incomplete.
        :param startup_candles: Additional candles to load at the start of the period
        :param warn_no_data: Log a warning message when no data is found
        :param candle_type: Any of the enum CandleType (must match trading mode!)
        :return: DataFrame with ohlcv data, or empty DataFrame
        """
        # Fix startup period
        timerange_startup = deepcopy(timerange)
        if startup_candles > 0 and timerange_startup:
            timerange_startup.subtract_start(timeframe_to_seconds(timeframe) * startup_candles)

        pairdf = self._ohlcv_load(
            pair,
            timeframe,
            timerange=timerange_startup,
            candle_type=candle_type
        )
        if self._check_empty_df(pairdf, pair, timeframe, candle_type, warn_no_data):
            return pairdf
        else:
            enddate = pairdf.iloc[-1]['date']

            if timerange_startup:
                self._validate_pairdata(pair, pairdf, timeframe, candle_type, timerange_startup)
                pairdf = trim_dataframe(pairdf, timerange_startup)
                if self._check_empty_df(pairdf, pair, timeframe, candle_type, warn_no_data, True):
                    return pairdf

            # incomplete candles should only be dropped if we didn't trim the end beforehand.
            pairdf = clean_ohlcv_dataframe(pairdf, timeframe,
                                           pair=pair,
                                           fill_missing=fill_missing,
                                           drop_incomplete=(drop_incomplete and
                                                            enddate == pairdf.iloc[-1]['date']))
            self._check_empty_df(pairdf, pair, timeframe, candle_type, warn_no_data)
            return pairdf

    def _check_empty_df(
            self, pairdf: DataFrame, pair: str, timeframe: str, candle_type: CandleType,
            warn_no_data: bool, warn_price: bool = False) -> bool:
        """
        Warn on empty dataframe
        """
        if pairdf.empty:
            if warn_no_data:
                logger.warning(
                    f"No history for {pair}, {candle_type}, {timeframe} found. "
                    "Use `freqtrade download-data` to download the data"
                )
            return True
        elif warn_price:
            candle_price_gap = 0
            if (candle_type in (CandleType.SPOT, CandleType.FUTURES) and
                    not pairdf.empty
                    and 'close' in pairdf.columns and 'open' in pairdf.columns):
                # Detect gaps between prior close and open
                gaps = ((pairdf['open'] - pairdf['close'].shift(1)) / pairdf['close'].shift(1))
                gaps = gaps.dropna()
                if len(gaps):
                    candle_price_gap = max(abs(gaps))
            if candle_price_gap > 0.1:
                logger.info(f"Price jump in {pair}, {timeframe}, {candle_type} between two candles "
                            f"of {candle_price_gap:.2%} detected.")

        return False

    def _validate_pairdata(self, pair, pairdata: DataFrame, timeframe: str,
                           candle_type: CandleType, timerange: TimeRange):
        """
        Validates pairdata for missing data at start end end and logs warnings.
        :param pairdata: Dataframe to validate
        :param timerange: Timerange specified for start and end dates
        """

        if timerange.starttype == 'date':
            if pairdata.iloc[0]['date'] > timerange.startdt:
                logger.warning(f"{pair}, {candle_type}, {timeframe}, "
                               f"data starts at {pairdata.iloc[0]['date']:%Y-%m-%d %H:%M:%S}")
        if timerange.stoptype == 'date':
            if pairdata.iloc[-1]['date'] < timerange.stopdt:
                logger.warning(f"{pair}, {candle_type}, {timeframe}, "
                               f"data ends at {pairdata.iloc[-1]['date']:%Y-%m-%d %H:%M:%S}")

    def rename_futures_data(
            self, pair: str, new_pair: str, timeframe: str, candle_type: CandleType):
        """
        Temporary method to migrate data from old naming to new naming (BTC/USDT -> BTC/USDT:USDT)
        Only used for binance to support the binance futures naming unification.
        """

        file_old = self._pair_data_filename(self._datadir, pair, timeframe, candle_type)
        file_new = self._pair_data_filename(self._datadir, new_pair, timeframe, candle_type)
        # print(file_old, file_new)
        if file_new.exists():
            logger.warning(f"{file_new} exists already, can't migrate {pair}.")
            return
        file_old.rename(file_new)


def get_datahandlerclass(datatype: str) -> Type[IDataHandler]:
    """
    Get datahandler class.
    Could be done using Resolvers, but since this may be called often and resolvers
    are rather expensive, doing this directly should improve performance.
    :param datatype: datatype to use.
    :return: Datahandler class
    """

    if datatype == 'json':
        from .jsondatahandler import JsonDataHandler
        return JsonDataHandler
    elif datatype == 'jsongz':
        from .jsondatahandler import JsonGzDataHandler
        return JsonGzDataHandler
    elif datatype == 'hdf5':
        from .hdf5datahandler import HDF5DataHandler
        return HDF5DataHandler
    elif datatype == 'feather':
        from .featherdatahandler import FeatherDataHandler
        return FeatherDataHandler
    elif datatype == 'parquet':
        from .parquetdatahandler import ParquetDataHandler
        return ParquetDataHandler
    else:
        raise ValueError(f"No datahandler for datatype {datatype} available.")


def get_datahandler(datadir: Path, data_format: Optional[str] = None,
                    data_handler: Optional[IDataHandler] = None) -> IDataHandler:
    """
    :param datadir: Folder to save data
    :param data_format: dataformat to use
    :param data_handler: returns this datahandler if it exists or initializes a new one
    """

    if not data_handler:
        HandlerClass = get_datahandlerclass(data_format or 'feather')
        data_handler = HandlerClass(datadir)
    return data_handler

1	"""
2	Abstract datahandler interface.
3	It's subclasses handle and storing data from disk.
4
5	"""
6	import logging	1✔
7	import re	1✔
8	from abc import ABC, abstractmethod	1✔
9	from copy import deepcopy	1✔
10	from datetime import datetime, timezone	1✔
11	from pathlib import Path	1✔
12	from typing import List, Optional, Tuple, Type	1✔
13
14	from pandas import DataFrame	1✔
15
16	from freqtrade import misc	1✔
17	from freqtrade.configuration import TimeRange	1✔
18	from freqtrade.constants import DEFAULT_TRADES_COLUMNS, ListPairsWithTimeframes	1✔
19	from freqtrade.data.converter import (clean_ohlcv_dataframe, trades_convert_types,	1✔
20	trades_df_remove_duplicates, trim_dataframe)
21	from freqtrade.enums import CandleType, TradingMode	1✔
22	from freqtrade.exchange import timeframe_to_seconds	1✔
23
24
25	logger = logging.getLogger(__name__)	1✔
26
27
28	class IDataHandler(ABC):	1✔
29
30	_OHLCV_REGEX = r'^([a-zA-Z_\d-]+)\-(\d+[a-zA-Z]{1,2})\-?([a-zA-Z_]*)?(?=\.)'	1✔
31
32	def __init__(self, datadir: Path) -> None:	1✔
33	self._datadir = datadir	1✔
34
35	@classmethod	1✔
36	def _get_file_extension(cls) -> str:	1✔
37	"""
38	Get file extension for this particular datahandler
39	"""
40	raise NotImplementedError()	×
41
42	@classmethod	1✔
43	def ohlcv_get_available_data(	1✔
44	cls, datadir: Path, trading_mode: TradingMode) -> ListPairsWithTimeframes:
45	"""
46	Returns a list of all pairs with ohlcv data available in this datadir
47	:param datadir: Directory to search for ohlcv files
48	:param trading_mode: trading-mode to be used
49	:return: List of Tuples of (pair, timeframe, CandleType)
50	"""
51	if trading_mode == TradingMode.FUTURES:	1✔
52	datadir = datadir.joinpath('futures')	1✔
53	_tmp = [	1✔
54	re.search(
55	cls._OHLCV_REGEX, p.name
56	) for p in datadir.glob(f"*.{cls._get_file_extension()}")]
57	return [	1✔
58	(
59	cls.rebuild_pair_from_filename(match[1]),
60	cls.rebuild_timeframe_from_filename(match[2]),
61	CandleType.from_string(match[3])
62	) for match in _tmp if match and len(match.groups()) > 1]
63
64	@classmethod	1✔
65	def ohlcv_get_pairs(cls, datadir: Path, timeframe: str, candle_type: CandleType) -> List[str]:	1✔
66	"""
67	Returns a list of all pairs with ohlcv data available in this datadir
68	for the specified timeframe
69	:param datadir: Directory to search for ohlcv files
70	:param timeframe: Timeframe to search pairs for
71	:param candle_type: Any of the enum CandleType (must match trading mode!)
72	:return: List of Pairs
73	"""
74	candle = ""	1✔
75	if candle_type != CandleType.SPOT:	1✔
76	datadir = datadir.joinpath('futures')	1✔
77	candle = f"-{candle_type}"	1✔
78	ext = cls._get_file_extension()	1✔
79	_tmp = [re.search(r'^(\S+)(?=\-' + timeframe + candle + f'.{ext})', p.name)	1✔
80	for p in datadir.glob(f"*{timeframe}{candle}.{ext}")]
81	# Check if regex found something and only return these results
82	return [cls.rebuild_pair_from_filename(match[0]) for match in _tmp if match]	1✔
83
84	@abstractmethod	1✔
85	def ohlcv_store(	1✔
86	self, pair: str, timeframe: str, data: DataFrame, candle_type: CandleType) -> None:
87	"""
88	Store ohlcv data.
89	:param pair: Pair - used to generate filename
90	:param timeframe: Timeframe - used to generate filename
91	:param data: Dataframe containing OHLCV data
92	:param candle_type: Any of the enum CandleType (must match trading mode!)
93	:return: None
94	"""
95
96	def ohlcv_data_min_max(self, pair: str, timeframe: str,	1✔
97	candle_type: CandleType) -> Tuple[datetime, datetime]:
98	"""
99	Returns the min and max timestamp for the given pair and timeframe.
100	:param pair: Pair to get min/max for
101	:param timeframe: Timeframe to get min/max for
102	:param candle_type: Any of the enum CandleType (must match trading mode!)
103	:return: (min, max)
104	"""
105	data = self._ohlcv_load(pair, timeframe, None, candle_type)	1✔
106	if data.empty:	1✔
107	return (	1✔
108	datetime.fromtimestamp(0, tz=timezone.utc),
109	datetime.fromtimestamp(0, tz=timezone.utc)
110	)
111	return data.iloc[0]['date'].to_pydatetime(), data.iloc[-1]['date'].to_pydatetime()	1✔
112
113	@abstractmethod	1✔
114	def _ohlcv_load(self, pair: str, timeframe: str, timerange: Optional[TimeRange],	1✔
115	candle_type: CandleType
116	) -> DataFrame:
117	"""
118	Internal method used to load data for one pair from disk.
119	Implements the loading and conversion to a Pandas dataframe.
120	Timerange trimming and dataframe validation happens outside of this method.
121	:param pair: Pair to load data
122	:param timeframe: Timeframe (e.g. "5m")
123	:param timerange: Limit data to be loaded to this timerange.
124	Optionally implemented by subclasses to avoid loading
125	all data where possible.
126	:param candle_type: Any of the enum CandleType (must match trading mode!)
127	:return: DataFrame with ohlcv data, or empty DataFrame
128	"""
129
130	def ohlcv_purge(self, pair: str, timeframe: str, candle_type: CandleType) -> bool:	1✔
131	"""
132	Remove data for this pair
133	:param pair: Delete data for this pair.
134	:param timeframe: Timeframe (e.g. "5m")
135	:param candle_type: Any of the enum CandleType (must match trading mode!)
136	:return: True when deleted, false if file did not exist.
137	"""
138	filename = self._pair_data_filename(self._datadir, pair, timeframe, candle_type)	1✔
139	if filename.exists():	1✔
140	filename.unlink()	1✔
141	return True	1✔
142	return False	1✔
143
144	@abstractmethod	1✔
145	def ohlcv_append(	1✔
146	self,
147	pair: str,
148	timeframe: str,
149	data: DataFrame,
150	candle_type: CandleType
151	) -> None:
152	"""
153	Append data to existing data structures
154	:param pair: Pair
155	:param timeframe: Timeframe this ohlcv data is for
156	:param data: Data to append.
157	:param candle_type: Any of the enum CandleType (must match trading mode!)
158	"""
159
160	@classmethod	1✔
161	def trades_get_pairs(cls, datadir: Path) -> List[str]:	1✔
162	"""
163	Returns a list of all pairs for which trade data is available in this
164	:param datadir: Directory to search for ohlcv files
165	:return: List of Pairs
166	"""
167	_ext = cls._get_file_extension()	1✔
168	_tmp = [re.search(r'^(\S+)(?=\-trades.' + _ext + ')', p.name)	1✔
169	for p in datadir.glob(f"*trades.{_ext}")]
170	# Check if regex found something and only return these results to avoid exceptions.
171	return [cls.rebuild_pair_from_filename(match[0]) for match in _tmp if match]	1✔
172
173	@abstractmethod	1✔
174	def _trades_store(self, pair: str, data: DataFrame) -> None:	1✔
175	"""
176	Store trades data (list of Dicts) to file
177	:param pair: Pair - used for filename
178	:param data: Dataframe containing trades
179	column sequence as in DEFAULT_TRADES_COLUMNS
180	"""
181
182	@abstractmethod	1✔
183	def trades_append(self, pair: str, data: DataFrame):	1✔
184	"""
185	Append data to existing files
186	:param pair: Pair - used for filename
187	:param data: Dataframe containing trades
188	column sequence as in DEFAULT_TRADES_COLUMNS
189	"""
190
191	@abstractmethod	1✔
192	def _trades_load(self, pair: str, timerange: Optional[TimeRange] = None) -> DataFrame:	1✔
193	"""
194	Load a pair from file, either .json.gz or .json
195	:param pair: Load trades for this pair
196	:param timerange: Timerange to load trades for - currently not implemented
197	:return: Dataframe containing trades
198	"""
199
200	def trades_store(self, pair: str, data: DataFrame) -> None:	1✔
201	"""
202	Store trades data (list of Dicts) to file
203	:param pair: Pair - used for filename
204	:param data: Dataframe containing trades
205	column sequence as in DEFAULT_TRADES_COLUMNS
206	"""
207	# Filter on expected columns (will remove the actual date column).
208	self._trades_store(pair, data[DEFAULT_TRADES_COLUMNS])	1✔
209
210	def trades_purge(self, pair: str) -> bool:	1✔
211	"""
212	Remove data for this pair
213	:param pair: Delete data for this pair.
214	:return: True when deleted, false if file did not exist.
215	"""
216	filename = self._pair_trades_filename(self._datadir, pair)	1✔
217	if filename.exists():	1✔
218	filename.unlink()	1✔
219	return True	1✔
220	return False	1✔
221
222	def trades_load(self, pair: str, timerange: Optional[TimeRange] = None) -> DataFrame:	1✔
223	"""
224	Load a pair from file, either .json.gz or .json
225	Removes duplicates in the process.
226	:param pair: Load trades for this pair
227	:param timerange: Timerange to load trades for - currently not implemented
228	:return: List of trades
229	"""
230	trades = trades_df_remove_duplicates(self._trades_load(pair, timerange=timerange))	1✔
231
232	trades = trades_convert_types(trades)	1✔
233	return trades	1✔
234
235	@classmethod	1✔
236	def create_dir_if_needed(cls, datadir: Path):	1✔
237	"""
238	Creates datadir if necessary
239	should only create directories for "futures" mode at the moment.
240	"""
241	if not datadir.parent.is_dir():	1✔
242	datadir.parent.mkdir()	1✔
243
244	@classmethod	1✔
245	def _pair_data_filename(	1✔
246	cls,
247	datadir: Path,
248	pair: str,
249	timeframe: str,
250	candle_type: CandleType,
251	no_timeframe_modify: bool = False
252	) -> Path:
253	pair_s = misc.pair_to_filename(pair)	1✔
254	candle = ""	1✔
255	if not no_timeframe_modify:	1✔
256	timeframe = cls.timeframe_to_file(timeframe)	1✔
257
258	if candle_type != CandleType.SPOT:	1✔
259	datadir = datadir.joinpath('futures')	1✔
260	candle = f"-{candle_type}"	1✔
261	filename = datadir.joinpath(	1✔
262	f'{pair_s}-{timeframe}{candle}.{cls._get_file_extension()}')
263	return filename	1✔
264
265	@classmethod	1✔
266	def _pair_trades_filename(cls, datadir: Path, pair: str) -> Path:	1✔
267	pair_s = misc.pair_to_filename(pair)	1✔
268	filename = datadir.joinpath(f'{pair_s}-trades.{cls._get_file_extension()}')	1✔
269	return filename	1✔
270
271	@staticmethod	1✔
272	def timeframe_to_file(timeframe: str):	1✔
273	return timeframe.replace('M', 'Mo')	1✔
274
275	@staticmethod	1✔
276	def rebuild_timeframe_from_filename(timeframe: str) -> str:	1✔
277	"""
278	converts timeframe from disk to file
279	Replaces mo with M (to avoid problems on case-insensitive filesystems)
280	"""
281	return re.sub('1mo', '1M', timeframe, flags=re.IGNORECASE)	1✔
282
283	@staticmethod	1✔
284	def rebuild_pair_from_filename(pair: str) -> str:	1✔
285	"""
286	Rebuild pair name from filename
287	Assumes a asset name of max. 7 length to also support BTC-PERP and BTC-PERP:USD names.
288	"""
289	res = re.sub(r'^(([A-Za-z\d]{1,10})\|^([A-Za-z\-]{1,6}))(_)', r'\g<1>/', pair, 1)	1✔
290	res = re.sub('_', ':', res, 1)	1✔
291	return res	1✔
292
293	def ohlcv_load(self, pair, timeframe: str,	1✔
294	candle_type: CandleType, *,
295	timerange: Optional[TimeRange] = None,
296	fill_missing: bool = True,
297	drop_incomplete: bool = False,
298	startup_candles: int = 0,
299	warn_no_data: bool = True,
300	) -> DataFrame:
301	"""
302	Load cached candle (OHLCV) data for the given pair.
303
304	:param pair: Pair to load data for
305	:param timeframe: Timeframe (e.g. "5m")
306	:param timerange: Limit data to be loaded to this timerange
307	:param fill_missing: Fill missing values with "No action"-candles
308	:param drop_incomplete: Drop last candle assuming it may be incomplete.
309	:param startup_candles: Additional candles to load at the start of the period
310	:param warn_no_data: Log a warning message when no data is found
311	:param candle_type: Any of the enum CandleType (must match trading mode!)
312	:return: DataFrame with ohlcv data, or empty DataFrame
313	"""
314	# Fix startup period
315	timerange_startup = deepcopy(timerange)	1✔
316	if startup_candles > 0 and timerange_startup:	1✔
317	timerange_startup.subtract_start(timeframe_to_seconds(timeframe) * startup_candles)	1✔
318
319	pairdf = self._ohlcv_load(	1✔
320	pair,
321	timeframe,
322	timerange=timerange_startup,
323	candle_type=candle_type
324	)
325	if self._check_empty_df(pairdf, pair, timeframe, candle_type, warn_no_data):	1✔
326	return pairdf	1✔
327	else:
328	enddate = pairdf.iloc[-1]['date']	1✔
329
330	if timerange_startup:	1✔
331	self._validate_pairdata(pair, pairdf, timeframe, candle_type, timerange_startup)	1✔
332	pairdf = trim_dataframe(pairdf, timerange_startup)	1✔
333	if self._check_empty_df(pairdf, pair, timeframe, candle_type, warn_no_data, True):	1✔
334	return pairdf	1✔
335
336	# incomplete candles should only be dropped if we didn't trim the end beforehand.
337	pairdf = clean_ohlcv_dataframe(pairdf, timeframe,	1✔
338	pair=pair,
339	fill_missing=fill_missing,
340	drop_incomplete=(drop_incomplete and
341	enddate == pairdf.iloc[-1]['date']))
342	self._check_empty_df(pairdf, pair, timeframe, candle_type, warn_no_data)	1✔
343	return pairdf	1✔
344
345	def _check_empty_df(	1✔
346	self, pairdf: DataFrame, pair: str, timeframe: str, candle_type: CandleType,
347	warn_no_data: bool, warn_price: bool = False) -> bool:
348	"""
349	Warn on empty dataframe
350	"""
351	if pairdf.empty:	1✔
352	if warn_no_data:	1✔
353	logger.warning(	1✔
354	f"No history for {pair}, {candle_type}, {timeframe} found. "
355	"Use `freqtrade download-data` to download the data"
356	)
357	return True	1✔
358	elif warn_price:	1✔
359	candle_price_gap = 0	1✔
360	if (candle_type in (CandleType.SPOT, CandleType.FUTURES) and	1✔
361	not pairdf.empty
362	and 'close' in pairdf.columns and 'open' in pairdf.columns):
363	# Detect gaps between prior close and open
364	gaps = ((pairdf['open'] - pairdf['close'].shift(1)) / pairdf['close'].shift(1))	1✔
365	gaps = gaps.dropna()	1✔
366	if len(gaps):	1✔
367	candle_price_gap = max(abs(gaps))	1✔
368	if candle_price_gap > 0.1:	1✔
369	logger.info(f"Price jump in {pair}, {timeframe}, {candle_type} between two candles "	1✔
370	f"of {candle_price_gap:.2%} detected.")
371
372	return False	1✔
373
374	def _validate_pairdata(self, pair, pairdata: DataFrame, timeframe: str,	1✔
375	candle_type: CandleType, timerange: TimeRange):
376	"""
377	Validates pairdata for missing data at start end end and logs warnings.
378	:param pairdata: Dataframe to validate
379	:param timerange: Timerange specified for start and end dates
380	"""
381
382	if timerange.starttype == 'date':	1✔
383	if pairdata.iloc[0]['date'] > timerange.startdt:	1✔
384	logger.warning(f"{pair}, {candle_type}, {timeframe}, "	1✔
385	f"data starts at {pairdata.iloc[0]['date']:%Y-%m-%d %H:%M:%S}")
386	if timerange.stoptype == 'date':	1✔
387	if pairdata.iloc[-1]['date'] < timerange.stopdt:	1✔
388	logger.warning(f"{pair}, {candle_type}, {timeframe}, "	1✔
389	f"data ends at {pairdata.iloc[-1]['date']:%Y-%m-%d %H:%M:%S}")
390
391	def rename_futures_data(	1✔
392	self, pair: str, new_pair: str, timeframe: str, candle_type: CandleType):
393	"""
394	Temporary method to migrate data from old naming to new naming (BTC/USDT -> BTC/USDT:USDT)
395	Only used for binance to support the binance futures naming unification.
396	"""
397
398	file_old = self._pair_data_filename(self._datadir, pair, timeframe, candle_type)	1✔
399	file_new = self._pair_data_filename(self._datadir, new_pair, timeframe, candle_type)	1✔
400	# print(file_old, file_new)
401	if file_new.exists():	1✔
402	logger.warning(f"{file_new} exists already, can't migrate {pair}.")	×
403	return	×
404	file_old.rename(file_new)	1✔
405
406
407	def get_datahandlerclass(datatype: str) -> Type[IDataHandler]:	1✔
408	"""
409	Get datahandler class.
410	Could be done using Resolvers, but since this may be called often and resolvers
411	are rather expensive, doing this directly should improve performance.
412	:param datatype: datatype to use.
413	:return: Datahandler class
414	"""
415
416	if datatype == 'json':	1✔
417	from .jsondatahandler import JsonDataHandler	1✔
418	return JsonDataHandler	1✔
419	elif datatype == 'jsongz':	1✔
420	from .jsondatahandler import JsonGzDataHandler	1✔
421	return JsonGzDataHandler	1✔
422	elif datatype == 'hdf5':	1✔
423	from .hdf5datahandler import HDF5DataHandler	1✔
424	return HDF5DataHandler	1✔
425	elif datatype == 'feather':	1✔
426	from .featherdatahandler import FeatherDataHandler	1✔
427	return FeatherDataHandler	1✔
428	elif datatype == 'parquet':	1✔
429	from .parquetdatahandler import ParquetDataHandler	1✔
430	return ParquetDataHandler	1✔
431	else:
432	raise ValueError(f"No datahandler for datatype {datatype} available.")	1✔
433
434
435	def get_datahandler(datadir: Path, data_format: Optional[str] = None,	1✔
436	data_handler: Optional[IDataHandler] = None) -> IDataHandler:
437	"""
438	:param datadir: Folder to save data
439	:param data_format: dataformat to use
440	:param data_handler: returns this datahandler if it exists or initializes a new one
441	"""
442
443	if not data_handler:	1✔
444	HandlerClass = get_datahandlerclass(data_format or 'feather')	1✔
445	data_handler = HandlerClass(datadir)	1✔
446	return data_handler	1✔

freqtrade / freqtrade / 6181253459

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous