From 298ce335b922c41f758da3bd99f0dbf1f62ff44a Mon Sep 17 00:00:00 2001 From: Matthias Date: Wed, 4 Dec 2024 19:56:26 +0100 Subject: [PATCH 01/17] chore: eliminate duplicate trades data grouping --- freqtrade/data/converter/orderflow.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/freqtrade/data/converter/orderflow.py b/freqtrade/data/converter/orderflow.py index 0bfd7a05f..52256e7b4 100644 --- a/freqtrade/data/converter/orderflow.py +++ b/freqtrade/data/converter/orderflow.py @@ -91,8 +91,6 @@ def populate_dataframe_with_trades( trades = trades.loc[trades["candle_start"] >= start_date] trades.reset_index(inplace=True, drop=True) - # group trades by candle start - trades_grouped_by_candle_start = trades.groupby("candle_start", group_keys=False) # Create Series to hold complex data trades_series = pd.Series(index=dataframe.index, dtype=object) orderflow_series = pd.Series(index=dataframe.index, dtype=object) @@ -100,7 +98,9 @@ def populate_dataframe_with_trades( stacked_imbalances_bid_series = pd.Series(index=dataframe.index, dtype=object) stacked_imbalances_ask_series = pd.Series(index=dataframe.index, dtype=object) + # group trades by candle start trades_grouped_by_candle_start = trades.groupby("candle_start", group_keys=False) + candle_start: datetime for candle_start, trades_grouped_df in trades_grouped_by_candle_start: is_between = candle_start == dataframe["date"] From 6be5947f69ed1b150836c8512537089e099b8303 Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 5 Dec 2024 06:37:19 +0100 Subject: [PATCH 02/17] chore: Move local Import out of the loop --- freqtrade/data/converter/orderflow.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/freqtrade/data/converter/orderflow.py b/freqtrade/data/converter/orderflow.py index 52256e7b4..364f97987 100644 --- a/freqtrade/data/converter/orderflow.py +++ b/freqtrade/data/converter/orderflow.py @@ -71,6 +71,8 @@ def populate_dataframe_with_trades( :param trades: Trades to populate with :return: Dataframe with trades populated """ + from freqtrade.exchange import timeframe_to_next_date + timeframe = config["timeframe"] config_orderflow = config["orderflow"] @@ -105,8 +107,6 @@ def populate_dataframe_with_trades( for candle_start, trades_grouped_df in trades_grouped_by_candle_start: is_between = candle_start == dataframe["date"] if is_between.any(): - from freqtrade.exchange import timeframe_to_next_date - candle_next = timeframe_to_next_date(timeframe, candle_start) if candle_next not in trades_grouped_by_candle_start.groups: logger.warning( From 0bf0e1808ca11776f8ddf856f1b81b86bf30b8ce Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 5 Dec 2024 06:58:25 +0100 Subject: [PATCH 03/17] chore: add todo for future cleanup --- tests/data/test_converter_orderflow.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/data/test_converter_orderflow.py b/tests/data/test_converter_orderflow.py index 9a337da91..88e21d7ab 100644 --- a/tests/data/test_converter_orderflow.py +++ b/tests/data/test_converter_orderflow.py @@ -37,6 +37,7 @@ def populate_dataframe_with_trades_trades(testdatadir): @pytest.fixture def candles(testdatadir): + # TODO: this fixture isn't really necessary and could be removed return pd.read_json(testdatadir / "orderflow/candles.json").copy() From ff371c43e00e9b11a8174e1a15f97854715b43d7 Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 5 Dec 2024 07:10:20 +0100 Subject: [PATCH 04/17] tests: add test to ensure caching works part of #11008 --- tests/data/test_converter_orderflow.py | 69 ++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/tests/data/test_converter_orderflow.py b/tests/data/test_converter_orderflow.py index 88e21d7ab..4b2dd17d5 100644 --- a/tests/data/test_converter_orderflow.py +++ b/tests/data/test_converter_orderflow.py @@ -8,6 +8,8 @@ from freqtrade.constants import DEFAULT_TRADES_COLUMNS from freqtrade.data.converter import populate_dataframe_with_trades from freqtrade.data.converter.orderflow import trades_to_volumeprofile_with_total_delta_bid_ask from freqtrade.data.converter.trade_converter import trades_list_to_df +from freqtrade.data.dataprovider import DataProvider +from tests.strategy.strats.strategy_test_v3 import StrategyTestV3 BIN_SIZE_SCALE = 0.5 @@ -483,3 +485,70 @@ def test_public_trades_testdata_sanity( "cost", "date", ] + + +def test_analyze_with_orderflow( + default_conf_usdt, + mocker, + populate_dataframe_with_trades_dataframe, + populate_dataframe_with_trades_trades, +): + ohlcv_history = populate_dataframe_with_trades_dataframe + # call without orderflow + strategy = StrategyTestV3(config=default_conf_usdt) + strategy.dp = DataProvider(default_conf_usdt, None, None) + + mocker.patch.object(strategy.dp, "trades", return_value=populate_dataframe_with_trades_trades) + + df = strategy.advise_indicators(ohlcv_history, {"pair:": "ETH/BTC"}) + assert len(df) == len(ohlcv_history) + assert "open" in df.columns + + expected_cols = [ + "trades", + "orderflow", + "imbalances", + "stacked_imbalances_bid", + "stacked_imbalances_ask", + "max_delta", + "min_delta", + "bid", + "ask", + "delta", + "total_trades", + ] + # Not expected to run - shouldn't have added orderflow columns + for col in expected_cols: + assert col not in df.columns, f"Column {col} found in df.columns" + + default_conf_usdt["exchange"]["use_public_trades"] = True + default_conf_usdt["orderflow"] = { + "cache_size": 5, + "max_candles": 5, + "scale": 0.005, + "imbalance_volume": 0, + "imbalance_ratio": 3, + "stacked_imbalance_range": 3, + } + + strategy.config = default_conf_usdt + df1 = strategy.advise_indicators(ohlcv_history, {"pair": "ETH/BTC"}) + assert len(df1) == len(ohlcv_history) + assert "open" in df1.columns + for col in expected_cols: + assert col in df1.columns, f"Column {col} not found in df.columns" + + if col not in ("stacked_imbalances_bid", "stacked_imbalances_ask"): + assert df1[col].count() == 5, f"Column {col} has {df1[col].count()} non-NaN values" + + # Ensure caching works - call the same logic again. + df2 = strategy.advise_indicators(ohlcv_history, {"pair": "ETH/BTC"}) + assert len(df2) == len(ohlcv_history) + assert "open" in df2.columns + for col in expected_cols: + assert col in df2.columns, f"Round2: Column {col} not found in df.columns" + + if col not in ("stacked_imbalances_bid", "stacked_imbalances_ask"): + assert ( + df2[col].count() == 5 + ), f"Round2: Column {col} has {df2[col].count()} non-NaN values" From 82d517fcbb1931e3986564d00daa786b3bf50073 Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 5 Dec 2024 07:12:35 +0100 Subject: [PATCH 05/17] test: improve orderflow test --- tests/data/test_converter_orderflow.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/data/test_converter_orderflow.py b/tests/data/test_converter_orderflow.py index 4b2dd17d5..42398ead9 100644 --- a/tests/data/test_converter_orderflow.py +++ b/tests/data/test_converter_orderflow.py @@ -503,6 +503,7 @@ def test_analyze_with_orderflow( df = strategy.advise_indicators(ohlcv_history, {"pair:": "ETH/BTC"}) assert len(df) == len(ohlcv_history) assert "open" in df.columns + pair = "ETH/BTC" expected_cols = [ "trades", @@ -532,7 +533,8 @@ def test_analyze_with_orderflow( } strategy.config = default_conf_usdt - df1 = strategy.advise_indicators(ohlcv_history, {"pair": "ETH/BTC"}) + # First round - builds cache + df1 = strategy.advise_indicators(ohlcv_history, {"pair": pair}) assert len(df1) == len(ohlcv_history) assert "open" in df1.columns for col in expected_cols: @@ -541,8 +543,10 @@ def test_analyze_with_orderflow( if col not in ("stacked_imbalances_bid", "stacked_imbalances_ask"): assert df1[col].count() == 5, f"Column {col} has {df1[col].count()} non-NaN values" + assert len(strategy._cached_grouped_trades_per_pair[pair]) == 5 + # Ensure caching works - call the same logic again. - df2 = strategy.advise_indicators(ohlcv_history, {"pair": "ETH/BTC"}) + df2 = strategy.advise_indicators(ohlcv_history, {"pair": pair}) assert len(df2) == len(ohlcv_history) assert "open" in df2.columns for col in expected_cols: From 6584f86bce6f8cb4064bcdc8fa8ae74053bd880b Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 5 Dec 2024 18:25:53 +0100 Subject: [PATCH 06/17] test: add Spy to improve test --- tests/data/test_converter_orderflow.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/tests/data/test_converter_orderflow.py b/tests/data/test_converter_orderflow.py index 42398ead9..c73706bed 100644 --- a/tests/data/test_converter_orderflow.py +++ b/tests/data/test_converter_orderflow.py @@ -499,11 +499,15 @@ def test_analyze_with_orderflow( strategy.dp = DataProvider(default_conf_usdt, None, None) mocker.patch.object(strategy.dp, "trades", return_value=populate_dataframe_with_trades_trades) + import freqtrade.data.converter.orderflow as orderflow_module - df = strategy.advise_indicators(ohlcv_history, {"pair:": "ETH/BTC"}) + spy = mocker.spy(orderflow_module, "trades_to_volumeprofile_with_total_delta_bid_ask") + + pair = "ETH/BTC" + df = strategy.advise_indicators(ohlcv_history, {"pair:": pair}) assert len(df) == len(ohlcv_history) assert "open" in df.columns - pair = "ETH/BTC" + assert spy.call_count == 0 expected_cols = [ "trades", @@ -537,6 +541,8 @@ def test_analyze_with_orderflow( df1 = strategy.advise_indicators(ohlcv_history, {"pair": pair}) assert len(df1) == len(ohlcv_history) assert "open" in df1.columns + assert spy.call_count == 5 + for col in expected_cols: assert col in df1.columns, f"Column {col} not found in df.columns" @@ -545,10 +551,16 @@ def test_analyze_with_orderflow( assert len(strategy._cached_grouped_trades_per_pair[pair]) == 5 + lastval_trades = df1.at[len(df1) - 1, "trades"] + assert isinstance(lastval_trades, list) + assert len(lastval_trades) == 122 + + spy.reset_mock() # Ensure caching works - call the same logic again. df2 = strategy.advise_indicators(ohlcv_history, {"pair": pair}) assert len(df2) == len(ohlcv_history) assert "open" in df2.columns + assert spy.call_count == 0 for col in expected_cols: assert col in df2.columns, f"Round2: Column {col} not found in df.columns" From 4879582896408457660393e0f6b75c68a2dde6dc Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 5 Dec 2024 19:31:34 +0100 Subject: [PATCH 07/17] feat: use dataframe.at directly to avoid intermediate series --- freqtrade/data/converter/orderflow.py | 63 ++++++++++----------------- 1 file changed, 23 insertions(+), 40 deletions(-) diff --git a/freqtrade/data/converter/orderflow.py b/freqtrade/data/converter/orderflow.py index 364f97987..a3065c414 100644 --- a/freqtrade/data/converter/orderflow.py +++ b/freqtrade/data/converter/orderflow.py @@ -93,13 +93,6 @@ def populate_dataframe_with_trades( trades = trades.loc[trades["candle_start"] >= start_date] trades.reset_index(inplace=True, drop=True) - # Create Series to hold complex data - trades_series = pd.Series(index=dataframe.index, dtype=object) - orderflow_series = pd.Series(index=dataframe.index, dtype=object) - imbalances_series = pd.Series(index=dataframe.index, dtype=object) - stacked_imbalances_bid_series = pd.Series(index=dataframe.index, dtype=object) - stacked_imbalances_ask_series = pd.Series(index=dataframe.index, dtype=object) - # group trades by candle start trades_grouped_by_candle_start = trades.groupby("candle_start", group_keys=False) @@ -126,37 +119,34 @@ def populate_dataframe_with_trades( ) continue - indices = dataframe.index[is_between].tolist() - # Add trades to each candle - trades_series.loc[indices] = [ - trades_grouped_df.drop(columns=["candle_start", "candle_end"]).to_dict( - orient="records" - ) - ] + # there can only be one row with the same date + index = dataframe.index[is_between][0] + dataframe.at[index, "trades"] = trades_grouped_df.drop( + columns=["candle_start", "candle_end"] + ).to_dict(orient="records") + # Calculate orderflow for each candle orderflow = trades_to_volumeprofile_with_total_delta_bid_ask( trades_grouped_df, scale=config_orderflow["scale"] ) - orderflow_series.loc[indices] = [orderflow.to_dict(orient="index")] + dataframe.at[index, "orderflow"] = orderflow.to_dict(orient="index") + # orderflow_series.loc[[index]] = [orderflow.to_dict(orient="index")] # Calculate imbalances for each candle's orderflow imbalances = trades_orderflow_to_imbalances( orderflow, imbalance_ratio=config_orderflow["imbalance_ratio"], imbalance_volume=config_orderflow["imbalance_volume"], ) - imbalances_series.loc[indices] = [imbalances.to_dict(orient="index")] + dataframe.at[index, "imbalances"] = imbalances.to_dict(orient="index") stacked_imbalance_range = config_orderflow["stacked_imbalance_range"] - stacked_imbalances_bid_series.loc[indices] = [ - stacked_imbalance_bid( - imbalances, stacked_imbalance_range=stacked_imbalance_range - ) - ] - stacked_imbalances_ask_series.loc[indices] = [ - stacked_imbalance_ask( - imbalances, stacked_imbalance_range=stacked_imbalance_range - ) - ] + dataframe.at[index, "stacked_imbalances_bid"] = stacked_imbalance_bid( + imbalances, stacked_imbalance_range=stacked_imbalance_range + ) + + dataframe.at[index, "stacked_imbalances_ask"] = stacked_imbalance_ask( + imbalances, stacked_imbalance_range=stacked_imbalance_range + ) bid = np.where( trades_grouped_df["side"].str.contains("sell"), trades_grouped_df["amount"], 0 @@ -168,15 +158,15 @@ def populate_dataframe_with_trades( deltas_per_trade = ask - bid min_delta = deltas_per_trade.cumsum().min() max_delta = deltas_per_trade.cumsum().max() - dataframe.loc[indices, "max_delta"] = max_delta - dataframe.loc[indices, "min_delta"] = min_delta + dataframe.loc[index, "max_delta"] = max_delta + dataframe.loc[index, "min_delta"] = min_delta - dataframe.loc[indices, "bid"] = bid.sum() - dataframe.loc[indices, "ask"] = ask.sum() - dataframe.loc[indices, "delta"] = ( - dataframe.loc[indices, "ask"] - dataframe.loc[indices, "bid"] + dataframe.loc[index, "bid"] = bid.sum() + dataframe.loc[index, "ask"] = ask.sum() + dataframe.loc[index, "delta"] = ( + dataframe.loc[index, "ask"] - dataframe.loc[index, "bid"] ) - dataframe.loc[indices, "total_trades"] = len(trades_grouped_df) + dataframe.loc[index, "total_trades"] = len(trades_grouped_df) # Cache the result cached_grouped_trades[(candle_start, candle_next)] = dataframe.loc[ @@ -193,13 +183,6 @@ def populate_dataframe_with_trades( logger.debug(f"Found NO candles for trades starting with {candle_start}") logger.debug(f"trades.groups_keys in {time.time() - start_time} seconds") - # Merge the complex data Series back into the DataFrame - dataframe["trades"] = trades_series - dataframe["orderflow"] = orderflow_series - dataframe["imbalances"] = imbalances_series - dataframe["stacked_imbalances_bid"] = stacked_imbalances_bid_series - dataframe["stacked_imbalances_ask"] = stacked_imbalances_ask_series - except Exception as e: logger.exception("Error populating dataframe with trades") raise DependencyException(e) From 48dd86bd9b4c012327318dca72c566451ca6b3ab Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 5 Dec 2024 19:33:49 +0100 Subject: [PATCH 08/17] chore: use `.at` for assignments --- freqtrade/data/converter/orderflow.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/freqtrade/data/converter/orderflow.py b/freqtrade/data/converter/orderflow.py index a3065c414..233923ab9 100644 --- a/freqtrade/data/converter/orderflow.py +++ b/freqtrade/data/converter/orderflow.py @@ -158,15 +158,15 @@ def populate_dataframe_with_trades( deltas_per_trade = ask - bid min_delta = deltas_per_trade.cumsum().min() max_delta = deltas_per_trade.cumsum().max() - dataframe.loc[index, "max_delta"] = max_delta - dataframe.loc[index, "min_delta"] = min_delta + dataframe.at[index, "max_delta"] = max_delta + dataframe.at[index, "min_delta"] = min_delta - dataframe.loc[index, "bid"] = bid.sum() - dataframe.loc[index, "ask"] = ask.sum() - dataframe.loc[index, "delta"] = ( - dataframe.loc[index, "ask"] - dataframe.loc[index, "bid"] + dataframe.at[index, "bid"] = bid.sum() + dataframe.at[index, "ask"] = ask.sum() + dataframe.at[index, "delta"] = ( + dataframe.at[index, "ask"] - dataframe.at[index, "bid"] ) - dataframe.loc[index, "total_trades"] = len(trades_grouped_df) + dataframe.at[index, "total_trades"] = len(trades_grouped_df) # Cache the result cached_grouped_trades[(candle_start, candle_next)] = dataframe.loc[ From 9d07f5dc2e3d662355bdbb533ef4811e72f44a40 Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 5 Dec 2024 19:37:36 +0100 Subject: [PATCH 09/17] chore: reduce orderflow code duplication --- freqtrade/data/converter/orderflow.py | 42 ++++++++++++++++----------- 1 file changed, 25 insertions(+), 17 deletions(-) diff --git a/freqtrade/data/converter/orderflow.py b/freqtrade/data/converter/orderflow.py index 233923ab9..ede5690c0 100644 --- a/freqtrade/data/converter/orderflow.py +++ b/freqtrade/data/converter/orderflow.py @@ -17,6 +17,20 @@ from freqtrade.exceptions import DependencyException logger = logging.getLogger(__name__) +ADDED_COLUMNS = [ + "trades", + "orderflow", + "imbalances", + "stacked_imbalances_bid", + "stacked_imbalances_ask", + "max_delta", + "min_delta", + "bid", + "ask", + "delta", + "total_trades", +] + def _init_dataframe_with_trades_columns(dataframe: pd.DataFrame): """ @@ -24,24 +38,18 @@ def _init_dataframe_with_trades_columns(dataframe: pd.DataFrame): :param dataframe: Dataframe to populate """ # Initialize columns with appropriate dtypes - dataframe["trades"] = np.nan - dataframe["orderflow"] = np.nan - dataframe["imbalances"] = np.nan - dataframe["stacked_imbalances_bid"] = np.nan - dataframe["stacked_imbalances_ask"] = np.nan - dataframe["max_delta"] = np.nan - dataframe["min_delta"] = np.nan - dataframe["bid"] = np.nan - dataframe["ask"] = np.nan - dataframe["delta"] = np.nan - dataframe["total_trades"] = np.nan + for column in ADDED_COLUMNS: + dataframe[column] = np.nan - # Ensure the 'trades' column is of object type - dataframe["trades"] = dataframe["trades"].astype(object) - dataframe["orderflow"] = dataframe["orderflow"].astype(object) - dataframe["imbalances"] = dataframe["imbalances"].astype(object) - dataframe["stacked_imbalances_bid"] = dataframe["stacked_imbalances_bid"].astype(object) - dataframe["stacked_imbalances_ask"] = dataframe["stacked_imbalances_ask"].astype(object) + # Set columns to object type + for column in ( + "trades", + "orderflow", + "imbalances", + "stacked_imbalances_bid", + "stacked_imbalances_ask", + ): + dataframe[column] = dataframe[column].astype(object) def _calculate_ohlcv_candle_start_and_end(df: pd.DataFrame, timeframe: str): From 621dfc136ed6b45bf4c542e281632d2066e3b0fe Mon Sep 17 00:00:00 2001 From: Matthias Date: Fri, 6 Dec 2024 06:38:49 +0100 Subject: [PATCH 10/17] fix: Improved caching closes #11008 --- freqtrade/data/converter/orderflow.py | 54 +++++++++------------------ freqtrade/strategy/interface.py | 8 +--- 2 files changed, 20 insertions(+), 42 deletions(-) diff --git a/freqtrade/data/converter/orderflow.py b/freqtrade/data/converter/orderflow.py index ede5690c0..c3a622522 100644 --- a/freqtrade/data/converter/orderflow.py +++ b/freqtrade/data/converter/orderflow.py @@ -68,18 +68,17 @@ def _calculate_ohlcv_candle_start_and_end(df: pd.DataFrame, timeframe: str): def populate_dataframe_with_trades( - cached_grouped_trades: OrderedDict[tuple[datetime, datetime], pd.DataFrame], + cached_grouped_trades: pd.DataFrame | None, config: Config, dataframe: pd.DataFrame, trades: pd.DataFrame, -) -> tuple[pd.DataFrame, OrderedDict[tuple[datetime, datetime], pd.DataFrame]]: +) -> tuple[pd.DataFrame, pd.DataFrame]: """ Populates a dataframe with trades :param dataframe: Dataframe to populate :param trades: Trades to populate with :return: Dataframe with trades populated """ - from freqtrade.exchange import timeframe_to_next_date timeframe = config["timeframe"] config_orderflow = config["orderflow"] @@ -101,34 +100,27 @@ def populate_dataframe_with_trades( trades = trades.loc[trades["candle_start"] >= start_date] trades.reset_index(inplace=True, drop=True) - # group trades by candle start trades_grouped_by_candle_start = trades.groupby("candle_start", group_keys=False) candle_start: datetime for candle_start, trades_grouped_df in trades_grouped_by_candle_start: is_between = candle_start == dataframe["date"] if is_between.any(): - candle_next = timeframe_to_next_date(timeframe, candle_start) - if candle_next not in trades_grouped_by_candle_start.groups: - logger.warning( - f"candle at {candle_start} with {len(trades_grouped_df)} trades " - f"might be unfinished, because no finished trades at {candle_next}" - ) - - # Use caching mechanism - if (candle_start, candle_next) in cached_grouped_trades: - cache_entry = cached_grouped_trades[(candle_start, candle_next)] - # dataframe.loc[is_between] = cache_entry # doesn't take, so we need workaround: - # Create a dictionary of the column values to be assigned - update_dict = {c: cache_entry[c].iat[0] for c in cache_entry.columns} - # Assign the values using the update_dict - dataframe.loc[is_between, update_dict.keys()] = pd.DataFrame( - [update_dict], index=dataframe.loc[is_between].index - ) - continue - # there can only be one row with the same date index = dataframe.index[is_between][0] + + if ( + cached_grouped_trades is not None + and (candle_start == cached_grouped_trades["date"]).any() + ): + logger.info(f"Using cached orderflow data for {candle_start}") + # Check if the trades are already in the cache + for col in ADDED_COLUMNS: + dataframe.at[index, col] = cached_grouped_trades.loc[ + (cached_grouped_trades["date"] == candle_start), col + ].values + continue + dataframe.at[index, "trades"] = trades_grouped_df.drop( columns=["candle_start", "candle_end"] ).to_dict(orient="records") @@ -176,21 +168,11 @@ def populate_dataframe_with_trades( ) dataframe.at[index, "total_trades"] = len(trades_grouped_df) - # Cache the result - cached_grouped_trades[(candle_start, candle_next)] = dataframe.loc[ - is_between - ].copy() - - # Maintain cache size - if ( - config.get("runmode") in (RunMode.DRY_RUN, RunMode.LIVE) - and len(cached_grouped_trades) > config_orderflow["cache_size"] - ): - cached_grouped_trades.popitem(last=False) - else: - logger.debug(f"Found NO candles for trades starting with {candle_start}") logger.debug(f"trades.groups_keys in {time.time() - start_time} seconds") + # Cache the entire dataframe + cached_grouped_trades = dataframe.tail(config_orderflow["cache_size"]).copy() + except Exception as e: logger.exception("Error populating dataframe with trades") raise DependencyException(e) diff --git a/freqtrade/strategy/interface.py b/freqtrade/strategy/interface.py index 1e415e9fc..1babfdcd6 100644 --- a/freqtrade/strategy/interface.py +++ b/freqtrade/strategy/interface.py @@ -141,9 +141,7 @@ class IStrategy(ABC, HyperStrategyMixin): market_direction: MarketDirection = MarketDirection.NONE # Global cache dictionary - _cached_grouped_trades_per_pair: dict[ - str, OrderedDict[tuple[datetime, datetime], DataFrame] - ] = {} + _cached_grouped_trades_per_pair: dict[str, DataFrame] = {} def __init__(self, config: Config) -> None: self.config = config @@ -1608,9 +1606,7 @@ class IStrategy(ABC, HyperStrategyMixin): config["timeframe"] = self.timeframe pair = metadata["pair"] # TODO: slice trades to size of dataframe for faster backtesting - cached_grouped_trades: OrderedDict[tuple[datetime, datetime], DataFrame] = ( - self._cached_grouped_trades_per_pair.get(pair, OrderedDict()) - ) + cached_grouped_trades: DataFrame | None = self._cached_grouped_trades_per_pair.get(pair) dataframe, cached_grouped_trades = populate_dataframe_with_trades( cached_grouped_trades, config, dataframe, trades ) From 6ab528748ec3e007b30b7268966758c7e2dab76f Mon Sep 17 00:00:00 2001 From: Matthias Date: Fri, 6 Dec 2024 06:40:15 +0100 Subject: [PATCH 11/17] tests: Update orderflow tests --- freqtrade/data/converter/orderflow.py | 2 -- freqtrade/strategy/interface.py | 1 - tests/data/test_converter_orderflow.py | 10 +++------- 3 files changed, 3 insertions(+), 10 deletions(-) diff --git a/freqtrade/data/converter/orderflow.py b/freqtrade/data/converter/orderflow.py index c3a622522..993b5eebc 100644 --- a/freqtrade/data/converter/orderflow.py +++ b/freqtrade/data/converter/orderflow.py @@ -4,14 +4,12 @@ Functions to convert orderflow data from public_trades import logging import time -from collections import OrderedDict from datetime import datetime import numpy as np import pandas as pd from freqtrade.constants import DEFAULT_ORDERFLOW_COLUMNS, Config -from freqtrade.enums import RunMode from freqtrade.exceptions import DependencyException diff --git a/freqtrade/strategy/interface.py b/freqtrade/strategy/interface.py index 1babfdcd6..24de4252c 100644 --- a/freqtrade/strategy/interface.py +++ b/freqtrade/strategy/interface.py @@ -5,7 +5,6 @@ This module defines the interface to apply for strategies import logging from abc import ABC, abstractmethod -from collections import OrderedDict from datetime import datetime, timedelta, timezone from math import isinf, isnan diff --git a/tests/data/test_converter_orderflow.py b/tests/data/test_converter_orderflow.py index c73706bed..ded6d2088 100644 --- a/tests/data/test_converter_orderflow.py +++ b/tests/data/test_converter_orderflow.py @@ -1,5 +1,3 @@ -from collections import OrderedDict - import numpy as np import pandas as pd import pytest @@ -105,7 +103,7 @@ def test_public_trades_mock_populate_dataframe_with_trades__check_orderflow( }, } # Apply the function to populate the data frame with order flow data - df, _ = populate_dataframe_with_trades(OrderedDict(), config, dataframe, trades) + df, _ = populate_dataframe_with_trades(None, config, dataframe, trades) # Extract results from the first row of the DataFrame results = df.iloc[0] t = results["trades"] @@ -246,7 +244,7 @@ def test_public_trades_trades_mock_populate_dataframe_with_trades__check_trades( } # Populate the DataFrame with trades and order flow data - df, _ = populate_dataframe_with_trades(OrderedDict(), config, dataframe, trades) + df, _ = populate_dataframe_with_trades(None, config, dataframe, trades) # --- DataFrame and Trade Data Validation --- @@ -404,9 +402,7 @@ def test_public_trades_config_max_trades( }, } - df, _ = populate_dataframe_with_trades( - OrderedDict(), default_conf | orderflow_config, dataframe, trades - ) + df, _ = populate_dataframe_with_trades(None, default_conf | orderflow_config, dataframe, trades) assert df.delta.count() == 1 From 3137d7cf2cdd39d95782bf7b1298008f3a399b8a Mon Sep 17 00:00:00 2001 From: Matthias Date: Fri, 6 Dec 2024 06:43:01 +0100 Subject: [PATCH 12/17] chore: remove unnecessary config alias --- freqtrade/strategy/interface.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/freqtrade/strategy/interface.py b/freqtrade/strategy/interface.py index 24de4252c..48f77136e 100644 --- a/freqtrade/strategy/interface.py +++ b/freqtrade/strategy/interface.py @@ -1601,13 +1601,11 @@ class IStrategy(ABC, HyperStrategyMixin): if use_public_trades: trades = self.dp.trades(pair=metadata["pair"], copy=False) - config = self.config - config["timeframe"] = self.timeframe pair = metadata["pair"] # TODO: slice trades to size of dataframe for faster backtesting cached_grouped_trades: DataFrame | None = self._cached_grouped_trades_per_pair.get(pair) dataframe, cached_grouped_trades = populate_dataframe_with_trades( - cached_grouped_trades, config, dataframe, trades + cached_grouped_trades, self.config, dataframe, trades ) # dereference old cache From 171157c100da88b137051b8e9991a63089d537f2 Mon Sep 17 00:00:00 2001 From: Matthias Date: Fri, 6 Dec 2024 07:03:24 +0100 Subject: [PATCH 13/17] chore: further simplify orderflow code --- freqtrade/data/converter/orderflow.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/freqtrade/data/converter/orderflow.py b/freqtrade/data/converter/orderflow.py index 993b5eebc..fc016664e 100644 --- a/freqtrade/data/converter/orderflow.py +++ b/freqtrade/data/converter/orderflow.py @@ -154,10 +154,8 @@ def populate_dataframe_with_trades( trades_grouped_df["side"].str.contains("buy"), trades_grouped_df["amount"], 0 ) deltas_per_trade = ask - bid - min_delta = deltas_per_trade.cumsum().min() - max_delta = deltas_per_trade.cumsum().max() - dataframe.at[index, "max_delta"] = max_delta - dataframe.at[index, "min_delta"] = min_delta + dataframe.at[index, "max_delta"] = deltas_per_trade.cumsum().max() + dataframe.at[index, "min_delta"] = deltas_per_trade.cumsum().min() dataframe.at[index, "bid"] = bid.sum() dataframe.at[index, "ask"] = ask.sum() From 6c25feabf26ccdd240dca04fc422f1b06feeedf6 Mon Sep 17 00:00:00 2001 From: Matthias Date: Fri, 6 Dec 2024 20:04:59 +0100 Subject: [PATCH 14/17] tests: assert type of orderflow object --- tests/data/test_converter_orderflow.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/tests/data/test_converter_orderflow.py b/tests/data/test_converter_orderflow.py index ded6d2088..3bf7faf58 100644 --- a/tests/data/test_converter_orderflow.py +++ b/tests/data/test_converter_orderflow.py @@ -551,6 +551,9 @@ def test_analyze_with_orderflow( assert isinstance(lastval_trades, list) assert len(lastval_trades) == 122 + lastval_of = df1.at[len(df1) - 1, "orderflow"] + assert isinstance(lastval_of, dict) + spy.reset_mock() # Ensure caching works - call the same logic again. df2 = strategy.advise_indicators(ohlcv_history, {"pair": pair}) @@ -564,3 +567,10 @@ def test_analyze_with_orderflow( assert ( df2[col].count() == 5 ), f"Round2: Column {col} has {df2[col].count()} non-NaN values" + + lastval_trade2 = df2.at[len(df2) - 1, "trades"] + assert isinstance(lastval_trade2, list) + assert len(lastval_trade2) == 122 + + lastval_of2 = df2.at[len(df2) - 1, "orderflow"] + assert isinstance(lastval_of2, dict) From ab39ac29e80c23d4e2128dedd7b193c88527d606 Mon Sep 17 00:00:00 2001 From: Matthias Date: Fri, 6 Dec 2024 20:05:33 +0100 Subject: [PATCH 15/17] fix: ensure data type is maintained when data comes from cache. --- freqtrade/data/converter/orderflow.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/freqtrade/data/converter/orderflow.py b/freqtrade/data/converter/orderflow.py index fc016664e..877325e5e 100644 --- a/freqtrade/data/converter/orderflow.py +++ b/freqtrade/data/converter/orderflow.py @@ -113,10 +113,11 @@ def populate_dataframe_with_trades( ): logger.info(f"Using cached orderflow data for {candle_start}") # Check if the trades are already in the cache + cache_idx = cached_grouped_trades.index[ + cached_grouped_trades["date"] == candle_start + ][0] for col in ADDED_COLUMNS: - dataframe.at[index, col] = cached_grouped_trades.loc[ - (cached_grouped_trades["date"] == candle_start), col - ].values + dataframe.at[index, col] = cached_grouped_trades.at[cache_idx, col] continue dataframe.at[index, "trades"] = trades_grouped_df.drop( From 267d9333a18287e2d31feb32c6f8b296ae9277dd Mon Sep 17 00:00:00 2001 From: Matthias Date: Fri, 6 Dec 2024 20:09:07 +0100 Subject: [PATCH 16/17] chore: remove pointless, very noisy log message. --- freqtrade/data/converter/orderflow.py | 1 - 1 file changed, 1 deletion(-) diff --git a/freqtrade/data/converter/orderflow.py b/freqtrade/data/converter/orderflow.py index 877325e5e..d82cccdb5 100644 --- a/freqtrade/data/converter/orderflow.py +++ b/freqtrade/data/converter/orderflow.py @@ -111,7 +111,6 @@ def populate_dataframe_with_trades( cached_grouped_trades is not None and (candle_start == cached_grouped_trades["date"]).any() ): - logger.info(f"Using cached orderflow data for {candle_start}") # Check if the trades are already in the cache cache_idx = cached_grouped_trades.index[ cached_grouped_trades["date"] == candle_start From e77ca024d7ee69b1fb01ae31326552d73c999ecc Mon Sep 17 00:00:00 2001 From: Matthias Date: Fri, 6 Dec 2024 20:16:46 +0100 Subject: [PATCH 17/17] chore: improve naming, don't duplicate column list --- freqtrade/data/converter/orderflow.py | 6 +++--- tests/data/test_converter_orderflow.py | 24 +++++++----------------- 2 files changed, 10 insertions(+), 20 deletions(-) diff --git a/freqtrade/data/converter/orderflow.py b/freqtrade/data/converter/orderflow.py index d82cccdb5..777af17dc 100644 --- a/freqtrade/data/converter/orderflow.py +++ b/freqtrade/data/converter/orderflow.py @@ -15,7 +15,7 @@ from freqtrade.exceptions import DependencyException logger = logging.getLogger(__name__) -ADDED_COLUMNS = [ +ORDERFLOW_ADDED_COLUMNS = [ "trades", "orderflow", "imbalances", @@ -36,7 +36,7 @@ def _init_dataframe_with_trades_columns(dataframe: pd.DataFrame): :param dataframe: Dataframe to populate """ # Initialize columns with appropriate dtypes - for column in ADDED_COLUMNS: + for column in ORDERFLOW_ADDED_COLUMNS: dataframe[column] = np.nan # Set columns to object type @@ -115,7 +115,7 @@ def populate_dataframe_with_trades( cache_idx = cached_grouped_trades.index[ cached_grouped_trades["date"] == candle_start ][0] - for col in ADDED_COLUMNS: + for col in ORDERFLOW_ADDED_COLUMNS: dataframe.at[index, col] = cached_grouped_trades.at[cache_idx, col] continue diff --git a/tests/data/test_converter_orderflow.py b/tests/data/test_converter_orderflow.py index 3bf7faf58..96f550020 100644 --- a/tests/data/test_converter_orderflow.py +++ b/tests/data/test_converter_orderflow.py @@ -4,7 +4,10 @@ import pytest from freqtrade.constants import DEFAULT_TRADES_COLUMNS from freqtrade.data.converter import populate_dataframe_with_trades -from freqtrade.data.converter.orderflow import trades_to_volumeprofile_with_total_delta_bid_ask +from freqtrade.data.converter.orderflow import ( + ORDERFLOW_ADDED_COLUMNS, + trades_to_volumeprofile_with_total_delta_bid_ask, +) from freqtrade.data.converter.trade_converter import trades_list_to_df from freqtrade.data.dataprovider import DataProvider from tests.strategy.strats.strategy_test_v3 import StrategyTestV3 @@ -505,21 +508,8 @@ def test_analyze_with_orderflow( assert "open" in df.columns assert spy.call_count == 0 - expected_cols = [ - "trades", - "orderflow", - "imbalances", - "stacked_imbalances_bid", - "stacked_imbalances_ask", - "max_delta", - "min_delta", - "bid", - "ask", - "delta", - "total_trades", - ] # Not expected to run - shouldn't have added orderflow columns - for col in expected_cols: + for col in ORDERFLOW_ADDED_COLUMNS: assert col not in df.columns, f"Column {col} found in df.columns" default_conf_usdt["exchange"]["use_public_trades"] = True @@ -539,7 +529,7 @@ def test_analyze_with_orderflow( assert "open" in df1.columns assert spy.call_count == 5 - for col in expected_cols: + for col in ORDERFLOW_ADDED_COLUMNS: assert col in df1.columns, f"Column {col} not found in df.columns" if col not in ("stacked_imbalances_bid", "stacked_imbalances_ask"): @@ -560,7 +550,7 @@ def test_analyze_with_orderflow( assert len(df2) == len(ohlcv_history) assert "open" in df2.columns assert spy.call_count == 0 - for col in expected_cols: + for col in ORDERFLOW_ADDED_COLUMNS: assert col in df2.columns, f"Round2: Column {col} not found in df.columns" if col not in ("stacked_imbalances_bid", "stacked_imbalances_ask"):