9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

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

93.24

/freqtrade/data/converter/trade_converter.py

"""
Functions to convert data from one format to another
"""
import logging
from pathlib import Path
from typing import Dict, List

import pandas as pd
from pandas import DataFrame, to_datetime

from freqtrade.configuration import TimeRange
from freqtrade.constants import (DEFAULT_DATAFRAME_COLUMNS, DEFAULT_TRADES_COLUMNS, TRADES_DTYPES,
                                 Config, TradeList)
from freqtrade.enums import CandleType, TradingMode
from freqtrade.exceptions import OperationalException


logger = logging.getLogger(__name__)


def trades_df_remove_duplicates(trades: pd.DataFrame) -> pd.DataFrame:
    """
    Removes duplicates from the trades DataFrame.
    Uses pandas.DataFrame.drop_duplicates to remove duplicates based on the 'timestamp' column.
    :param trades: DataFrame with the columns constants.DEFAULT_TRADES_COLUMNS
    :return: DataFrame with duplicates removed based on the 'timestamp' column
    """
    return trades.drop_duplicates(subset=['timestamp', 'id'])


def trades_dict_to_list(trades: List[Dict]) -> TradeList:
    """
    Convert fetch_trades result into a List (to be more memory efficient).
    :param trades: List of trades, as returned by ccxt.fetch_trades.
    :return: List of Lists, with constants.DEFAULT_TRADES_COLUMNS as columns
    """
    return [[t[col] for col in DEFAULT_TRADES_COLUMNS] for t in trades]


def trades_convert_types(trades: DataFrame) -> DataFrame:
    """
    Convert Trades dtypes and add 'date' column
    """
    trades = trades.astype(TRADES_DTYPES)
    trades['date'] = to_datetime(trades['timestamp'], unit='ms', utc=True)
    return trades


def trades_list_to_df(trades: TradeList, convert: bool = True):
    """
    convert trades list to dataframe
    :param trades: List of Lists with constants.DEFAULT_TRADES_COLUMNS as columns
    """
    if not trades:
        df = DataFrame(columns=DEFAULT_TRADES_COLUMNS)
    else:
        df = DataFrame(trades, columns=DEFAULT_TRADES_COLUMNS)

    if convert:
        df = trades_convert_types(df)

    return df


def trades_to_ohlcv(trades: DataFrame, timeframe: str) -> DataFrame:
    """
    Converts trades list to OHLCV list
    :param trades: List of trades, as returned by ccxt.fetch_trades.
    :param timeframe: Timeframe to resample data to
    :return: OHLCV Dataframe.
    :raises: ValueError if no trades are provided
    """
    from freqtrade.exchange import timeframe_to_resample_freq
    if trades.empty:
        raise ValueError('Trade-list empty.')
    df = trades.set_index('date', drop=True)
    resample_interval = timeframe_to_resample_freq(timeframe)
    df_new = df['price'].resample(resample_interval).ohlc()
    df_new['volume'] = df['amount'].resample(resample_interval).sum()
    df_new['date'] = df_new.index
    # Drop 0 volume rows
    df_new = df_new.dropna()
    return df_new.loc[:, DEFAULT_DATAFRAME_COLUMNS]


def convert_trades_to_ohlcv(
    pairs: List[str],
    timeframes: List[str],
    datadir: Path,
    timerange: TimeRange,
    erase: bool,
    data_format_ohlcv: str,
    data_format_trades: str,
    candle_type: CandleType,
) -> None:
    """
    Convert stored trades data to ohlcv data
    """
    from freqtrade.data.history import get_datahandler
    data_handler_trades = get_datahandler(datadir, data_format=data_format_trades)
    data_handler_ohlcv = get_datahandler(datadir, data_format=data_format_ohlcv)

    logger.info(f"About to convert pairs: '{', '.join(pairs)}', "
                f"intervals: '{', '.join(timeframes)}' to {datadir}")
    trading_mode = TradingMode.FUTURES if candle_type != CandleType.SPOT else TradingMode.SPOT
    for pair in pairs:
        trades = data_handler_trades.trades_load(pair, trading_mode)
        for timeframe in timeframes:
            if erase:
                if data_handler_ohlcv.ohlcv_purge(pair, timeframe, candle_type=candle_type):
                    logger.info(f'Deleting existing data for pair {pair}, interval {timeframe}.')
            try:
                ohlcv = trades_to_ohlcv(trades, timeframe)
                # Store ohlcv
                data_handler_ohlcv.ohlcv_store(pair, timeframe, data=ohlcv, candle_type=candle_type)
            except ValueError:
                logger.warning(f'Could not convert {pair} to OHLCV.')


def convert_trades_format(config: Config, convert_from: str, convert_to: str, erase: bool):
    """
    Convert trades from one format to another format.
    :param config: Config dictionary
    :param convert_from: Source format
    :param convert_to: Target format
    :param erase: Erase source data (does not apply if source and target format are identical)
    """
    if convert_from == 'kraken_csv':
        if config['exchange']['name'] != 'kraken':
            raise OperationalException(
                'Converting from csv is only supported for kraken.'
                'Please refer to the documentation for details about this special mode.'
            )
        from freqtrade.data.converter.trade_converter_kraken import import_kraken_trades_from_csv
        import_kraken_trades_from_csv(config, convert_to)
        return

    from freqtrade.data.history import get_datahandler
    src = get_datahandler(config['datadir'], convert_from)
    trg = get_datahandler(config['datadir'], convert_to)

    if 'pairs' not in config:
        config['pairs'] = src.trades_get_pairs(config['datadir'])
    logger.info(f"Converting trades for {config['pairs']}")
    trading_mode: TradingMode = config.get('trading_mode', TradingMode.SPOT)
    for pair in config['pairs']:
        data = src.trades_load(pair, trading_mode)
        logger.info(f"Converting {len(data)} trades for {pair}")
        trg.trades_store(pair, data, trading_mode)

        if erase and convert_from != convert_to:
            logger.info(f"Deleting source Trade data for {pair}.")
            src.trades_purge(pair, trading_mode)

1	"""
2	Functions to convert data from one format to another
3	"""
4	import logging	1✔
5	from pathlib import Path	1✔
6	from typing import Dict, List	1✔
7
8	import pandas as pd	1✔
9	from pandas import DataFrame, to_datetime	1✔
10
11	from freqtrade.configuration import TimeRange	1✔
12	from freqtrade.constants import (DEFAULT_DATAFRAME_COLUMNS, DEFAULT_TRADES_COLUMNS, TRADES_DTYPES,	1✔
13	Config, TradeList)
14	from freqtrade.enums import CandleType, TradingMode	1✔
15	from freqtrade.exceptions import OperationalException	1✔
16
17
18	logger = logging.getLogger(__name__)	1✔
19
20
21	def trades_df_remove_duplicates(trades: pd.DataFrame) -> pd.DataFrame:	1✔
22	"""
23	Removes duplicates from the trades DataFrame.
24	Uses pandas.DataFrame.drop_duplicates to remove duplicates based on the 'timestamp' column.
25	:param trades: DataFrame with the columns constants.DEFAULT_TRADES_COLUMNS
26	:return: DataFrame with duplicates removed based on the 'timestamp' column
27	"""
28	return trades.drop_duplicates(subset=['timestamp', 'id'])	1✔
29
30
31	def trades_dict_to_list(trades: List[Dict]) -> TradeList:	1✔
32	"""
33	Convert fetch_trades result into a List (to be more memory efficient).
34	:param trades: List of trades, as returned by ccxt.fetch_trades.
35	:return: List of Lists, with constants.DEFAULT_TRADES_COLUMNS as columns
36	"""
37	return [[t[col] for col in DEFAULT_TRADES_COLUMNS] for t in trades]	1✔
38
39
40	def trades_convert_types(trades: DataFrame) -> DataFrame:	1✔
41	"""
42	Convert Trades dtypes and add 'date' column
43	"""
44	trades = trades.astype(TRADES_DTYPES)	1✔
45	trades['date'] = to_datetime(trades['timestamp'], unit='ms', utc=True)	1✔
46	return trades	1✔
47
48
49	def trades_list_to_df(trades: TradeList, convert: bool = True):	1✔
50	"""
51	convert trades list to dataframe
52	:param trades: List of Lists with constants.DEFAULT_TRADES_COLUMNS as columns
53	"""
54	if not trades:	1✔
55	df = DataFrame(columns=DEFAULT_TRADES_COLUMNS)	1✔
56	else:
57	df = DataFrame(trades, columns=DEFAULT_TRADES_COLUMNS)	1✔
58
59	if convert:	1✔
60	df = trades_convert_types(df)	1✔
61
62	return df	1✔
63
64
65	def trades_to_ohlcv(trades: DataFrame, timeframe: str) -> DataFrame:	1✔
66	"""
67	Converts trades list to OHLCV list
68	:param trades: List of trades, as returned by ccxt.fetch_trades.
69	:param timeframe: Timeframe to resample data to
70	:return: OHLCV Dataframe.
71	:raises: ValueError if no trades are provided
72	"""
73	from freqtrade.exchange import timeframe_to_resample_freq	1✔
74	if trades.empty:	1✔
75	raise ValueError('Trade-list empty.')	1✔
76	df = trades.set_index('date', drop=True)	1✔
77	resample_interval = timeframe_to_resample_freq(timeframe)	1✔
78	df_new = df['price'].resample(resample_interval).ohlc()	1✔
79	df_new['volume'] = df['amount'].resample(resample_interval).sum()	1✔
80	df_new['date'] = df_new.index	1✔
81	# Drop 0 volume rows
82	df_new = df_new.dropna()	1✔
83	return df_new.loc[:, DEFAULT_DATAFRAME_COLUMNS]	1✔
84
85
86	def convert_trades_to_ohlcv(	1✔
87	pairs: List[str],
88	timeframes: List[str],
89	datadir: Path,
90	timerange: TimeRange,
91	erase: bool,
92	data_format_ohlcv: str,
93	data_format_trades: str,
94	candle_type: CandleType,
95	) -> None:
96	"""
97	Convert stored trades data to ohlcv data
98	"""
99	from freqtrade.data.history import get_datahandler	1✔
100	data_handler_trades = get_datahandler(datadir, data_format=data_format_trades)	1✔
101	data_handler_ohlcv = get_datahandler(datadir, data_format=data_format_ohlcv)	1✔
102
103	logger.info(f"About to convert pairs: '{', '.join(pairs)}', "	1✔
104	f"intervals: '{', '.join(timeframes)}' to {datadir}")
105	trading_mode = TradingMode.FUTURES if candle_type != CandleType.SPOT else TradingMode.SPOT	1✔
106	for pair in pairs:	1✔
107	trades = data_handler_trades.trades_load(pair, trading_mode)	1✔
108	for timeframe in timeframes:	1✔
109	if erase:	1✔
110	if data_handler_ohlcv.ohlcv_purge(pair, timeframe, candle_type=candle_type):	1✔
111	logger.info(f'Deleting existing data for pair {pair}, interval {timeframe}.')	1✔
112	try:	1✔
113	ohlcv = trades_to_ohlcv(trades, timeframe)	1✔
114	# Store ohlcv
115	data_handler_ohlcv.ohlcv_store(pair, timeframe, data=ohlcv, candle_type=candle_type)	1✔
116	except ValueError:	1✔
117	logger.warning(f'Could not convert {pair} to OHLCV.')	1✔
118
119
120	def convert_trades_format(config: Config, convert_from: str, convert_to: str, erase: bool):	1✔
121	"""
122	Convert trades from one format to another format.
123	:param config: Config dictionary
124	:param convert_from: Source format
125	:param convert_to: Target format
126	:param erase: Erase source data (does not apply if source and target format are identical)
127	"""
128	if convert_from == 'kraken_csv':	1✔
129	if config['exchange']['name'] != 'kraken':	×
130	raise OperationalException(	×
131	'Converting from csv is only supported for kraken.'
132	'Please refer to the documentation for details about this special mode.'
133	)
134	from freqtrade.data.converter.trade_converter_kraken import import_kraken_trades_from_csv	×
135	import_kraken_trades_from_csv(config, convert_to)	×
136	return	×
137
138	from freqtrade.data.history import get_datahandler	1✔
139	src = get_datahandler(config['datadir'], convert_from)	1✔
140	trg = get_datahandler(config['datadir'], convert_to)	1✔
141
142	if 'pairs' not in config:	1✔
143	config['pairs'] = src.trades_get_pairs(config['datadir'])	1✔
144	logger.info(f"Converting trades for {config['pairs']}")	1✔
145	trading_mode: TradingMode = config.get('trading_mode', TradingMode.SPOT)	1✔
146	for pair in config['pairs']:	1✔
147	data = src.trades_load(pair, trading_mode)	1✔
148	logger.info(f"Converting {len(data)} trades for {pair}")	1✔
149	trg.trades_store(pair, data, trading_mode)	1✔
150
151	if erase and convert_from != convert_to:	1✔
152	logger.info(f"Deleting source Trade data for {pair}.")	1✔
153	src.trades_purge(pair, trading_mode)	1✔

freqtrade / freqtrade / 9394559170

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