From ff06d58aced552b1abba6d5005d523fc15fddd11 Mon Sep 17 00:00:00 2001 From: Matthias Date: Tue, 10 Jun 2025 07:00:38 +0200 Subject: [PATCH] chore: remove edge module and config options --- freqtrade/commands/arguments.py | 2 +- freqtrade/commands/cli_options.py | 7 - freqtrade/configuration/configuration.py | 9 - freqtrade/edge/__init__.py | 1 - freqtrade/edge/edge_positioning.py | 524 ----------------------- 5 files changed, 1 insertion(+), 542 deletions(-) delete mode 100644 freqtrade/edge/__init__.py delete mode 100644 freqtrade/edge/edge_positioning.py diff --git a/freqtrade/commands/arguments.py b/freqtrade/commands/arguments.py index 65bd7c46f..2e3de0a98 100755 --- a/freqtrade/commands/arguments.py +++ b/freqtrade/commands/arguments.py @@ -82,7 +82,7 @@ ARGS_HYPEROPT = [ "early_stop", ] -ARGS_EDGE = [*ARGS_COMMON_OPTIMIZE, "stoploss_range"] +ARGS_EDGE = [*ARGS_COMMON_OPTIMIZE] ARGS_LIST_STRATEGIES = [ "strategy_path", diff --git a/freqtrade/commands/cli_options.py b/freqtrade/commands/cli_options.py index 669f68f5a..45e50b2ce 100755 --- a/freqtrade/commands/cli_options.py +++ b/freqtrade/commands/cli_options.py @@ -240,13 +240,6 @@ AVAILABLE_CLI_OPTIONS = { default=constants.BACKTEST_CACHE_DEFAULT, choices=constants.BACKTEST_CACHE_AGE, ), - # Edge - "stoploss_range": Arg( - "--stoplosses", - help="Defines a range of stoploss values against which edge will assess the strategy. " - 'The format is "min,max,step" (without any space). ' - "Example: `--stoplosses=-0.01,-0.1,-0.001`", - ), # Hyperopt "hyperopt": Arg( "--hyperopt", diff --git a/freqtrade/configuration/configuration.py b/freqtrade/configuration/configuration.py index e1e62e778..b9da05d0c 100644 --- a/freqtrade/configuration/configuration.py +++ b/freqtrade/configuration/configuration.py @@ -2,7 +2,6 @@ This module contains the configuration class """ -import ast import logging import warnings from collections.abc import Callable @@ -314,14 +313,6 @@ class Configuration: ] self._args_to_config_loop(config, configurations) - # Edge section: - if self.args.get("stoploss_range"): - txt_range = ast.literal_eval(self.args["stoploss_range"]) - config["edge"].update({"stoploss_range_min": txt_range[0]}) - config["edge"].update({"stoploss_range_max": txt_range[1]}) - config["edge"].update({"stoploss_range_step": txt_range[2]}) - logger.info("Parameter --stoplosses detected: %s ...", self.args["stoploss_range"]) - # Hyperopt section configurations = [ diff --git a/freqtrade/edge/__init__.py b/freqtrade/edge/__init__.py deleted file mode 100644 index d275a80e3..000000000 --- a/freqtrade/edge/__init__.py +++ /dev/null @@ -1 +0,0 @@ -from .edge_positioning import Edge, PairInfo # noqa: F401 diff --git a/freqtrade/edge/edge_positioning.py b/freqtrade/edge/edge_positioning.py deleted file mode 100644 index 546126513..000000000 --- a/freqtrade/edge/edge_positioning.py +++ /dev/null @@ -1,524 +0,0 @@ -# pragma pylint: disable=W0603 -"""Edge positioning package""" - -import logging -from collections import defaultdict -from copy import deepcopy -from datetime import timedelta -from typing import Any, NamedTuple - -import numpy as np -import utils_find_1st as utf1st -from pandas import DataFrame - -from freqtrade.configuration import TimeRange -from freqtrade.constants import DATETIME_PRINT_FORMAT, UNLIMITED_STAKE_AMOUNT, Config -from freqtrade.data.history import get_timerange, load_data, refresh_data -from freqtrade.enums import CandleType, ExitType, RunMode -from freqtrade.exceptions import OperationalException -from freqtrade.exchange import timeframe_to_seconds -from freqtrade.plugins.pairlist.pairlist_helpers import expand_pairlist -from freqtrade.strategy.interface import IStrategy -from freqtrade.util import dt_now - - -logger = logging.getLogger(__name__) - - -class PairInfo(NamedTuple): - stoploss: float - winrate: float - risk_reward_ratio: float - required_risk_reward: float - expectancy: float - nb_trades: int - avg_trade_duration: float - - -class Edge: - """ - Calculates Win Rate, Risk Reward Ratio, Expectancy - against historical data for a give set of markets and a strategy - it then adjusts stoploss and position size accordingly - and force it into the strategy - Author: https://github.com/mishaker - """ - - _cached_pairs: dict[str, Any] = {} # Keeps a list of pairs - - def __init__(self, config: Config, exchange, strategy) -> None: - self.config = config - self.exchange = exchange - self.strategy: IStrategy = strategy - - self.edge_config = self.config.get("edge", {}) - self._cached_pairs: dict[str, Any] = {} # Keeps a list of pairs - self._final_pairs: list = [] - - # checking max_open_trades. it should be -1 as with Edge - # the number of trades is determined by position size - if self.config["max_open_trades"] != float("inf"): - logger.critical("max_open_trades should be -1 in config !") - - if self.config["stake_amount"] != UNLIMITED_STAKE_AMOUNT: - raise OperationalException("Edge works only with unlimited stake amount") - - self._capital_ratio: float = self.config["tradable_balance_ratio"] - self._allowed_risk: float = self.edge_config.get("allowed_risk") - self._since_number_of_days: int = self.edge_config.get("calculate_since_number_of_days", 14) - self._last_updated: int = 0 # Timestamp of pairs last updated time - self._refresh_pairs = True - - self._stoploss_range_min = float(self.edge_config.get("stoploss_range_min", -0.01)) - self._stoploss_range_max = float(self.edge_config.get("stoploss_range_max", -0.05)) - self._stoploss_range_step = float(self.edge_config.get("stoploss_range_step", -0.001)) - - # calculating stoploss range - self._stoploss_range = np.arange( - self._stoploss_range_min, self._stoploss_range_max, self._stoploss_range_step - ) - - self._timerange: TimeRange = TimeRange.parse_timerange( - f"{(dt_now() - timedelta(days=self._since_number_of_days)).strftime('%Y%m%d')}-" - ) - if config.get("fee"): - self.fee = config["fee"] - else: - try: - self.fee = self.exchange.get_fee( - symbol=expand_pairlist( - self.config["exchange"]["pair_whitelist"], list(self.exchange.markets) - )[0] - ) - except IndexError: - self.fee = None - - def calculate(self, pairs: list[str]) -> bool: - if self.fee is None and pairs: - self.fee = self.exchange.get_fee(pairs[0]) - - heartbeat = self.edge_config.get("process_throttle_secs") - - if (self._last_updated > 0) and ( - self._last_updated + heartbeat > int(dt_now().timestamp()) - ): - return False - - data: dict[str, Any] = {} - logger.info("Using stake_currency: %s ...", self.config["stake_currency"]) - logger.info("Using local backtesting data (using whitelist in given config) ...") - - if self._refresh_pairs: - timerange_startup = deepcopy(self._timerange) - timerange_startup.subtract_start( - timeframe_to_seconds(self.strategy.timeframe) * self.strategy.startup_candle_count - ) - refresh_data( - datadir=self.config["datadir"], - pairs=pairs, - exchange=self.exchange, - timeframe=self.strategy.timeframe, - timerange=timerange_startup, - data_format=self.config["dataformat_ohlcv"], - candle_type=self.config.get("candle_type_def", CandleType.SPOT), - ) - # Download informative pairs too - res = defaultdict(list) - for pair, timeframe, _ in self.strategy.gather_informative_pairs(): - res[timeframe].append(pair) - for timeframe, inf_pairs in res.items(): - timerange_startup = deepcopy(self._timerange) - timerange_startup.subtract_start( - timeframe_to_seconds(timeframe) * self.strategy.startup_candle_count - ) - refresh_data( - datadir=self.config["datadir"], - pairs=inf_pairs, - exchange=self.exchange, - timeframe=timeframe, - timerange=timerange_startup, - data_format=self.config["dataformat_ohlcv"], - candle_type=self.config.get("candle_type_def", CandleType.SPOT), - ) - - data = load_data( - datadir=self.config["datadir"], - pairs=pairs, - timeframe=self.strategy.timeframe, - timerange=self._timerange, - startup_candles=self.strategy.startup_candle_count, - data_format=self.config["dataformat_ohlcv"], - candle_type=self.config.get("candle_type_def", CandleType.SPOT), - ) - - if not data: - # Reinitializing cached pairs - self._cached_pairs = {} - logger.critical("No data found. Edge is stopped ...") - return False - # Fake run-mode to Edge - prior_rm = self.config["runmode"] - self.config["runmode"] = RunMode.EDGE - preprocessed = self.strategy.advise_all_indicators(data) - self.config["runmode"] = prior_rm - - # Print timeframe - min_date, max_date = get_timerange(preprocessed) - logger.info( - f"Measuring data from {min_date.strftime(DATETIME_PRINT_FORMAT)} " - f"up to {max_date.strftime(DATETIME_PRINT_FORMAT)} " - f"({(max_date - min_date).days} days).." - ) - # TODO: Should edge support shorts? needs to be investigated further - # * (add enter_short exit_short) - headers = ["date", "open", "high", "low", "close", "enter_long", "exit_long"] - - trades: list = [] - for pair, pair_data in preprocessed.items(): - # Sorting dataframe by date and reset index - pair_data = pair_data.sort_values(by=["date"]) - pair_data = pair_data.reset_index(drop=True) - - df_analyzed = self.strategy.ft_advise_signals(pair_data, {"pair": pair})[headers].copy() - - trades += self._find_trades_for_stoploss_range(df_analyzed, pair, self._stoploss_range) - - # If no trade found then exit - if len(trades) == 0: - logger.info("No trades found.") - return False - - # Fill missing, calculable columns, profit, duration , abs etc. - trades_df = self._fill_calculable_fields(DataFrame(trades)) - self._cached_pairs = self._process_expectancy(trades_df) - self._last_updated = int(dt_now().timestamp()) - - return True - - def stake_amount( - self, pair: str, free_capital: float, total_capital: float, capital_in_trade: float - ) -> float: - stoploss = self.get_stoploss(pair) - available_capital = (total_capital + capital_in_trade) * self._capital_ratio - allowed_capital_at_risk = available_capital * self._allowed_risk - max_position_size = abs(allowed_capital_at_risk / stoploss) - # Position size must be below available capital. - position_size = min(min(max_position_size, free_capital), available_capital) - if pair in self._cached_pairs: - logger.info( - "winrate: %s, expectancy: %s, position size: %s, pair: %s," - " capital in trade: %s, free capital: %s, total capital: %s," - " stoploss: %s, available capital: %s.", - self._cached_pairs[pair].winrate, - self._cached_pairs[pair].expectancy, - position_size, - pair, - capital_in_trade, - free_capital, - total_capital, - stoploss, - available_capital, - ) - return round(position_size, 15) - - def get_stoploss(self, pair: str) -> float: - if pair in self._cached_pairs: - return self._cached_pairs[pair].stoploss - else: - logger.warning( - f"Tried to access stoploss of non-existing pair {pair}, " - "strategy stoploss is returned instead." - ) - return self.strategy.stoploss - - def adjust(self, pairs: list[str]) -> list: - """ - Filters out and sorts "pairs" according to Edge calculated pairs - """ - final = [] - for pair, info in self._cached_pairs.items(): - if ( - info.expectancy > float(self.edge_config.get("minimum_expectancy", 0.2)) - and info.winrate > float(self.edge_config.get("minimum_winrate", 0.60)) - and pair in pairs - ): - final.append(pair) - - if self._final_pairs != final: - self._final_pairs = final - if self._final_pairs: - logger.info( - "Minimum expectancy and minimum winrate are met only for %s," - " so other pairs are filtered out.", - self._final_pairs, - ) - else: - logger.info( - "Edge removed all pairs as no pair with minimum expectancy " - "and minimum winrate was found !" - ) - - return self._final_pairs - - def accepted_pairs(self) -> list[dict[str, Any]]: - """ - return a list of accepted pairs along with their winrate, expectancy and stoploss - """ - final = [] - for pair, info in self._cached_pairs.items(): - if info.expectancy > float( - self.edge_config.get("minimum_expectancy", 0.2) - ) and info.winrate > float(self.edge_config.get("minimum_winrate", 0.60)): - final.append( - { - "Pair": pair, - "Winrate": info.winrate, - "Expectancy": info.expectancy, - "Stoploss": info.stoploss, - } - ) - return final - - def _fill_calculable_fields(self, result: DataFrame) -> DataFrame: - """ - The result frame contains a number of columns that are calculable - from other columns. These are left blank till all rows are added, - to be populated in single vector calls. - - Columns to be populated are: - - Profit - - trade duration - - profit abs - :param result Dataframe - :return: result Dataframe - """ - # We set stake amount to an arbitrary amount, as it doesn't change the calculation. - # All returned values are relative, they are defined as ratios. - stake = 0.015 - - result["trade_duration"] = result["close_date"] - result["open_date"] - - result["trade_duration"] = result["trade_duration"].map( - lambda x: int(x.total_seconds() / 60) - ) - - # Spends, Takes, Profit, Absolute Profit - - # Buy Price - result["buy_vol"] = stake / result["open_rate"] # How many target are we buying - result["buy_fee"] = stake * self.fee - result["buy_spend"] = stake + result["buy_fee"] # How much we're spending - - # Sell price - result["sell_sum"] = result["buy_vol"] * result["close_rate"] - result["sell_fee"] = result["sell_sum"] * self.fee - result["sell_take"] = result["sell_sum"] - result["sell_fee"] - - # profit_ratio - result["profit_ratio"] = (result["sell_take"] - result["buy_spend"]) / result["buy_spend"] - - # Absolute profit - result["profit_abs"] = result["sell_take"] - result["buy_spend"] - - return result - - def _process_expectancy(self, results: DataFrame) -> dict[str, Any]: - """ - This calculates WinRate, Required Risk Reward, Risk Reward and Expectancy of all pairs - The calculation will be done per pair and per strategy. - """ - # Removing pairs having less than min_trades_number - min_trades_number = self.edge_config.get("min_trade_number", 10) - results = results.groupby(["pair", "stoploss"]).filter(lambda x: len(x) > min_trades_number) - ################################### - - # Removing outliers (Only Pumps) from the dataset - # The method to detect outliers is to calculate standard deviation - # Then every value more than (standard deviation + 2*average) is out (pump) - # - # Removing Pumps - if self.edge_config.get("remove_pumps", False): - results = results[ - results["profit_abs"] - < 2 * results["profit_abs"].std() + results["profit_abs"].mean() - ] - ########################################################################## - - # Removing trades having a duration more than X minutes (set in config) - max_trade_duration = self.edge_config.get("max_trade_duration_minute", 1440) - results = results[results.trade_duration < max_trade_duration] - ####################################################################### - - if results.empty: - return {} - - groupby_aggregator = { - "profit_abs": [ - ("nb_trades", "count"), # number of all trades - ("profit_sum", lambda x: x[x > 0].sum()), # cumulative profit of all winning trades - ("loss_sum", lambda x: abs(x[x < 0].sum())), # cumulative loss of all losing trades - ("nb_win_trades", lambda x: x[x > 0].count()), # number of winning trades - ], - "trade_duration": [("avg_trade_duration", "mean")], - } - - # Group by (pair and stoploss) by applying above aggregator - df = ( - results.groupby(["pair", "stoploss"])[["profit_abs", "trade_duration"]] - .agg(groupby_aggregator) - .reset_index(col_level=1) - ) - - # Dropping level 0 as we don't need it - df.columns = df.columns.droplevel(0) - - # Calculating number of losing trades, average win and average loss - df["nb_loss_trades"] = df["nb_trades"] - df["nb_win_trades"] - df["average_win"] = np.where( - df["nb_win_trades"] == 0, 0.0, df["profit_sum"] / df["nb_win_trades"] - ) - df["average_loss"] = np.where( - df["nb_loss_trades"] == 0, 0.0, df["loss_sum"] / df["nb_loss_trades"] - ) - - # Win rate = number of profitable trades / number of trades - df["winrate"] = df["nb_win_trades"] / df["nb_trades"] - - # risk_reward_ratio = average win / average loss - df["risk_reward_ratio"] = df["average_win"] / df["average_loss"] - - # required_risk_reward = (1 / winrate) - 1 - df["required_risk_reward"] = (1 / df["winrate"]) - 1 - - # expectancy = (risk_reward_ratio * winrate) - (lossrate) - df["expectancy"] = (df["risk_reward_ratio"] * df["winrate"]) - (1 - df["winrate"]) - - # sort by expectancy and stoploss - df = ( - df.sort_values(by=["expectancy", "stoploss"], ascending=False) - .groupby("pair") - .first() - .sort_values(by=["expectancy"], ascending=False) - .reset_index() - ) - - final = {} - for x in df.itertuples(): - final[x.pair] = PairInfo( - x.stoploss, - x.winrate, - x.risk_reward_ratio, - x.required_risk_reward, - x.expectancy, - x.nb_trades, - x.avg_trade_duration, - ) - - # Returning a list of pairs in order of "expectancy" - return final - - def _find_trades_for_stoploss_range(self, df, pair: str, stoploss_range) -> list: - buy_column = df["enter_long"].values - sell_column = df["exit_long"].values - date_column = df["date"].values - ohlc_columns = df[["open", "high", "low", "close"]].values - - result: list = [] - for stoploss in stoploss_range: - result += self._detect_next_stop_or_sell_point( - buy_column, sell_column, date_column, ohlc_columns, round(stoploss, 6), pair - ) - - return result - - def _detect_next_stop_or_sell_point( - self, buy_column, sell_column, date_column, ohlc_columns, stoploss, pair: str - ): - """ - Iterate through ohlc_columns in order to find the next trade - Next trade opens from the first buy signal noticed to - The sell or stoploss signal after it. - It then cuts OHLC, buy_column, sell_column and date_column. - Cut from (the exit trade index) + 1. - - Author: https://github.com/mishaker - """ - - result: list = [] - start_point = 0 - - while True: - open_trade_index = utf1st.find_1st(buy_column, 1, utf1st.cmp_equal) - - # Return empty if we don't find trade entry (i.e. buy==1) or - # we find a buy but at the end of array - if open_trade_index == -1 or open_trade_index == len(buy_column) - 1: - break - else: - # When a buy signal is seen, - # trade opens in reality on the next candle - open_trade_index += 1 - - open_price = ohlc_columns[open_trade_index, 0] - stop_price = open_price * (stoploss + 1) - - # Searching for the index where stoploss is hit - stop_index = utf1st.find_1st( - ohlc_columns[open_trade_index:, 2], stop_price, utf1st.cmp_smaller - ) - - # If we don't find it then we assume stop_index will be far in future (infinite number) - if stop_index == -1: - stop_index = float("inf") - - # Searching for the index where sell is hit - sell_index = utf1st.find_1st(sell_column[open_trade_index:], 1, utf1st.cmp_equal) - - # If we don't find it then we assume sell_index will be far in future (infinite number) - if sell_index == -1: - sell_index = float("inf") - - # Check if we don't find any stop or sell point (in that case trade remains open) - # It is not interesting for Edge to consider it so we simply ignore the trade - # And stop iterating there is no more entry - if stop_index == sell_index == float("inf"): - break - - if stop_index <= sell_index: - exit_index = open_trade_index + stop_index - exit_type = ExitType.STOP_LOSS - exit_price = stop_price - elif stop_index > sell_index: - # If exit is SELL then we exit at the next candle - exit_index = open_trade_index + sell_index + 1 - - # Check if we have the next candle - if len(ohlc_columns) - 1 < exit_index: - break - - exit_type = ExitType.EXIT_SIGNAL - exit_price = ohlc_columns[exit_index, 0] - - trade = { - "pair": pair, - "stoploss": stoploss, - "profit_ratio": "", - "profit_abs": "", - "open_date": date_column[open_trade_index], - "close_date": date_column[exit_index], - "trade_duration": "", - "open_rate": round(open_price, 15), - "close_rate": round(exit_price, 15), - "exit_type": exit_type, - } - - result.append(trade) - - # Giving a view of exit_index till the end of array - buy_column = buy_column[exit_index:] - sell_column = sell_column[exit_index:] - date_column = date_column[exit_index:] - ohlc_columns = ohlc_columns[exit_index:] - start_point += exit_index - - return result