freqtrade_origin/freqtrade/edge/edge_positioning.py

# 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, Dict, List, 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