# pragma pylint: disable=missing-docstring, C0103
import logging
import math
from datetime import UTC, datetime, timedelta
from pathlib import Path
from unittest.mock import MagicMock

import pytest
from pandas import DataFrame, concat

from freqtrade.configuration import TimeRange
from freqtrade.constants import CUSTOM_TAG_MAX_LENGTH
from freqtrade.data.dataprovider import DataProvider
from freqtrade.data.history import load_data
from freqtrade.enums import ExitCheckTuple, ExitType, SignalDirection
from freqtrade.exceptions import DependencyException, OperationalException, StrategyError
from freqtrade.persistence import PairLocks, Trade
from freqtrade.resolvers import StrategyResolver
from freqtrade.strategy.hyper import detect_all_parameters
from freqtrade.strategy.parameters import (
    IntParameter,
)
from freqtrade.strategy.strategy_validation import StrategyResultValidator
from freqtrade.util import dt_now
from tests.conftest import CURRENT_TEST_STRATEGY, TRADE_SIDES, log_has, log_has_re

from .strats.strategy_test_v3 import StrategyTestV3


# Avoid to reinit the same object again and again
_STRATEGY = StrategyTestV3(config={})
_STRATEGY.dp = DataProvider({}, None, None)


def test_returns_latest_signal(ohlcv_history):
    ohlcv_history.loc[1, "date"] = dt_now()
    # Take a copy to correctly modify the call
    mocked_history = ohlcv_history.copy()
    mocked_history["enter_long"] = 0
    mocked_history["exit_long"] = 0
    mocked_history["enter_short"] = 0
    mocked_history["exit_short"] = 0
    # Set tags in lines that don't matter to test nan in the sell line
    mocked_history.loc[0, "enter_tag"] = "wrong_line"
    mocked_history.loc[0, "exit_tag"] = "wrong_line"
    mocked_history.loc[1, "exit_long"] = 1

    assert _STRATEGY.get_entry_signal("ETH/BTC", "5m", mocked_history) == (None, None)
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history) == (False, True, None)
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history, True) == (False, False, None)
    mocked_history.loc[1, "exit_long"] = 0
    mocked_history.loc[1, "enter_long"] = 1

    assert _STRATEGY.get_entry_signal("ETH/BTC", "5m", mocked_history) == (
        SignalDirection.LONG,
        None,
    )
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history) == (True, False, None)
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history, True) == (False, False, None)
    mocked_history.loc[1, "exit_long"] = 0
    mocked_history.loc[1, "enter_long"] = 0

    assert _STRATEGY.get_entry_signal("ETH/BTC", "5m", mocked_history) == (None, None)
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history) == (False, False, None)
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history, True) == (False, False, None)
    mocked_history.loc[1, "exit_long"] = 0
    mocked_history.loc[1, "enter_long"] = 1
    mocked_history.loc[1, "enter_tag"] = "buy_signal_01"

    assert _STRATEGY.get_entry_signal("ETH/BTC", "5m", mocked_history) == (
        SignalDirection.LONG,
        "buy_signal_01",
    )
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history) == (True, False, None)
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history, True) == (False, False, None)

    mocked_history.loc[1, "exit_long"] = 0
    mocked_history.loc[1, "enter_long"] = 0
    mocked_history.loc[1, "enter_short"] = 1
    mocked_history.loc[1, "exit_short"] = 0
    mocked_history.loc[1, "enter_tag"] = "sell_signal_01"

    # Don't provide short signal while in spot mode
    assert _STRATEGY.get_entry_signal("ETH/BTC", "5m", mocked_history) == (None, None)

    _STRATEGY.config["trading_mode"] = "futures"
    # Short signal gets ignored as can_short is not set.
    assert _STRATEGY.get_entry_signal("ETH/BTC", "5m", mocked_history) == (None, None)

    _STRATEGY.can_short = True

    assert _STRATEGY.get_entry_signal("ETH/BTC", "5m", mocked_history) == (
        SignalDirection.SHORT,
        "sell_signal_01",
    )
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history) == (False, False, None)
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history, True) == (True, False, None)

    mocked_history.loc[1, "enter_short"] = 0
    mocked_history.loc[1, "exit_short"] = 1
    mocked_history.loc[1, "exit_tag"] = "sell_signal_02"
    assert _STRATEGY.get_entry_signal("ETH/BTC", "5m", mocked_history) == (None, None)
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history) == (
        False,
        False,
        "sell_signal_02",
    )
    assert _STRATEGY.get_exit_signal("ETH/BTC", "5m", mocked_history, True) == (
        False,
        True,
        "sell_signal_02",
    )

    _STRATEGY.can_short = False
    _STRATEGY.config["trading_mode"] = "spot"


def test_analyze_pair_empty(mocker, caplog, ohlcv_history):
    mocker.patch.object(_STRATEGY.dp, "ohlcv", return_value=ohlcv_history)
    mocker.patch.object(_STRATEGY, "_analyze_ticker_internal", return_value=DataFrame([]))
    mocker.patch("freqtrade.strategy.interface.StrategyResultValidator.assert_df")

    _STRATEGY.analyze_pair("ETH/BTC")

    assert log_has("Empty dataframe for pair ETH/BTC", caplog)


def test_get_signal_empty(default_conf, caplog):
    assert (None, None) == _STRATEGY.get_latest_candle(
        "foo", default_conf["timeframe"], DataFrame()
    )
    assert log_has("Empty candle (OHLCV) data for pair foo", caplog)
    caplog.clear()

    assert (None, None) == _STRATEGY.get_latest_candle("bar", default_conf["timeframe"], None)
    assert log_has("Empty candle (OHLCV) data for pair bar", caplog)
    caplog.clear()

    assert (None, None) == _STRATEGY.get_latest_candle(
        "baz", default_conf["timeframe"], DataFrame([])
    )
    assert log_has("Empty candle (OHLCV) data for pair baz", caplog)


def test_get_signal_exception_valueerror(mocker, caplog, ohlcv_history):
    caplog.set_level(logging.INFO)
    mocker.patch.object(_STRATEGY.dp, "ohlcv", return_value=ohlcv_history)
    mocker.patch.object(_STRATEGY, "_analyze_ticker_internal", side_effect=ValueError("xyz"))
    _STRATEGY.analyze_pair("foo")
    assert log_has_re(r"Strategy caused the following exception: ValueError\('xyz'\).*", caplog)
    caplog.clear()

    mocker.patch.object(
        _STRATEGY, "analyze_ticker", side_effect=Exception("invalid ticker history ")
    )
    _STRATEGY.analyze_pair("foo")
    assert log_has_re(r"Strategy caused the following exception: ValueError\('xyz'\).*", caplog)


def test_get_signal_old_dataframe(default_conf, mocker, caplog, ohlcv_history):
    # default_conf defines a 5m interval. we check interval * 2 + 5m
    # this is necessary as the last candle is removed (partial candles) by default
    ohlcv_history.loc[1, "date"] = dt_now() - timedelta(minutes=16)
    # Take a copy to correctly modify the call
    mocked_history = ohlcv_history.copy()
    mocked_history["exit_long"] = 0
    mocked_history["enter_long"] = 0
    mocked_history.loc[1, "enter_long"] = 1

    caplog.set_level(logging.INFO)
    mocker.patch("freqtrade.strategy.interface.StrategyResultValidator.assert_df")

    assert (None, None) == _STRATEGY.get_latest_candle(
        "xyz", default_conf["timeframe"], mocked_history
    )
    assert log_has("Outdated history for pair xyz. Last tick is 16 minutes old", caplog)


def test_get_signal_no_sell_column(default_conf, mocker, caplog, ohlcv_history):
    # default_conf defines a 5m interval. we check interval * 2 + 5m
    # this is necessary as the last candle is removed (partial candles) by default
    ohlcv_history.loc[1, "date"] = dt_now()
    # Take a copy to correctly modify the call
    mocked_history = ohlcv_history.copy()
    # Intentionally don't set sell column
    # mocked_history['sell'] = 0
    mocked_history["enter_long"] = 0
    mocked_history.loc[1, "enter_long"] = 1

    caplog.set_level(logging.INFO)
    mocker.patch(
        "freqtrade.strategy.interface.StrategyResultValidator.assert_df",
    )

    assert (SignalDirection.LONG, None) == _STRATEGY.get_entry_signal(
        "xyz", default_conf["timeframe"], mocked_history
    )


def test_ignore_expired_candle(default_conf):
    strategy = StrategyResolver.load_strategy(default_conf)
    strategy.ignore_buying_expired_candle_after = 60

    latest_date = datetime(2020, 12, 30, 7, 0, 0, tzinfo=UTC)
    # Add 1 candle length as the "latest date" defines candle open.
    current_time = latest_date + timedelta(seconds=80 + 300)

    assert (
        strategy.ignore_expired_candle(
            latest_date=latest_date, current_time=current_time, timeframe_seconds=300, enter=True
        )
        is True
    )

    current_time = latest_date + timedelta(seconds=30 + 300)

    assert (
        strategy.ignore_expired_candle(
            latest_date=latest_date, current_time=current_time, timeframe_seconds=300, enter=True
        )
        is not True
    )


def test_assert_df_raise(mocker, caplog, ohlcv_history):
    ohlcv_history.loc[1, "date"] = dt_now() - timedelta(minutes=16)
    # Take a copy to correctly modify the call
    mocked_history = ohlcv_history.copy()
    mocked_history["sell"] = 0
    mocked_history["buy"] = 0
    mocked_history.loc[1, "buy"] = 1

    caplog.set_level(logging.INFO)
    mocker.patch.object(_STRATEGY.dp, "ohlcv", return_value=ohlcv_history)
    mocker.patch.object(_STRATEGY.dp, "get_analyzed_dataframe", return_value=(mocked_history, 0))
    mocker.patch(
        "freqtrade.strategy.interface.StrategyResultValidator.assert_df",
        side_effect=StrategyError("Dataframe returned..."),
    )
    _STRATEGY.analyze_pair("xyz")
    assert log_has(
        "Unable to analyze candle (OHLCV) data for pair xyz: Dataframe returned...", caplog
    )


def test_assert_df(ohlcv_history, caplog):
    df_len = len(ohlcv_history) - 1
    ohlcv_history.loc[:, "enter_long"] = 0
    ohlcv_history.loc[:, "exit_long"] = 0
    validator = StrategyResultValidator(ohlcv_history, warn_only=False)
    # Ensure it's running when passed correctly
    validator.assert_df(ohlcv_history)

    with pytest.raises(StrategyError, match=r"Dataframe returned from strategy.*length\."):
        validator.assert_df(concat([ohlcv_history, ohlcv_history]))

    with pytest.raises(
        StrategyError, match=r"Dataframe returned from strategy.*last close price\."
    ):
        df = ohlcv_history.copy()
        df.loc[df_len, "close"] += 0.01
        validator.assert_df(df)

    with pytest.raises(StrategyError, match=r"Dataframe returned from strategy.*last date\."):
        df = ohlcv_history.copy()
        df.loc[df_len, "date"] = ohlcv_history.loc[0, "date"] - timedelta(days=1)
        validator.assert_df(df)

    with pytest.raises(
        StrategyError, match=r"No dataframe returned \(return statement missing\?\)."
    ):
        validator.assert_df(None)

    validator = StrategyResultValidator(ohlcv_history, warn_only=True)

    caplog.clear()
    df = ohlcv_history.copy()
    df.loc[df_len, "date"] = ohlcv_history.loc[0, "date"] - timedelta(days=1)

    validator.assert_df(df)
    assert log_has_re(r"Dataframe returned from strategy.*last date\.", caplog)
    # reset to avoid problems in other tests due to test leakage


def test_advise_all_indicators(default_conf, testdatadir) -> None:
    strategy = StrategyResolver.load_strategy(default_conf)

    timerange = TimeRange.parse_timerange("1510694220-1510700340")
    data = load_data(testdatadir, "1m", ["UNITTEST/BTC"], timerange=timerange, fill_up_missing=True)
    processed = strategy.advise_all_indicators(data)
    assert len(processed["UNITTEST/BTC"]) == 103


def test_freqai_not_initialized(default_conf) -> None:
    strategy = StrategyResolver.load_strategy(default_conf)
    strategy.ft_bot_start()
    with pytest.raises(OperationalException, match=r"freqAI is not enabled\."):
        strategy.freqai.start()


def test_advise_all_indicators_copy(mocker, default_conf, testdatadir) -> None:
    strategy = StrategyResolver.load_strategy(default_conf)
    aimock = mocker.patch("freqtrade.strategy.interface.IStrategy.advise_indicators")
    timerange = TimeRange.parse_timerange("1510694220-1510700340")
    data = load_data(testdatadir, "1m", ["UNITTEST/BTC"], timerange=timerange, fill_up_missing=True)
    strategy.advise_all_indicators(data)
    assert aimock.call_count == 1
    # Ensure that a copy of the dataframe is passed to advice_indicators
    assert aimock.call_args_list[0][0][0] is not data


def test_min_roi_reached(default_conf, fee) -> None:
    # Use list to confirm sequence does not matter
    min_roi_list = [{20: 0.05, 55: 0.01, 0: 0.1}, {0: 0.1, 20: 0.05, 55: 0.01}]
    for roi in min_roi_list:
        strategy = StrategyResolver.load_strategy(default_conf)
        strategy.minimal_roi = roi
        trade = Trade(
            pair="ETH/BTC",
            stake_amount=0.001,
            amount=5,
            open_date=dt_now() - timedelta(hours=1),
            fee_open=fee.return_value,
            fee_close=fee.return_value,
            exchange="binance",
            open_rate=1,
        )

        assert not strategy.min_roi_reached(trade, 0.02, dt_now() - timedelta(minutes=56))
        assert strategy.min_roi_reached(trade, 0.12, dt_now() - timedelta(minutes=56))

        assert not strategy.min_roi_reached(trade, 0.04, dt_now() - timedelta(minutes=39))
        assert strategy.min_roi_reached(trade, 0.06, dt_now() - timedelta(minutes=39))

        assert not strategy.min_roi_reached(trade, -0.01, dt_now() - timedelta(minutes=1))
        assert strategy.min_roi_reached(trade, 0.02, dt_now() - timedelta(minutes=1))


def test_min_roi_reached2(default_conf, fee) -> None:
    # test with ROI raising after last interval
    min_roi_list = [
        {20: 0.07, 30: 0.05, 55: 0.30, 0: 0.1},
        {0: 0.1, 20: 0.07, 30: 0.05, 55: 0.30},
    ]
    for roi in min_roi_list:
        strategy = StrategyResolver.load_strategy(default_conf)
        strategy.minimal_roi = roi
        trade = Trade(
            pair="ETH/BTC",
            stake_amount=0.001,
            amount=5,
            open_date=dt_now() - timedelta(hours=1),
            fee_open=fee.return_value,
            fee_close=fee.return_value,
            exchange="binance",
            open_rate=1,
        )

        assert not strategy.min_roi_reached(trade, 0.02, dt_now() - timedelta(minutes=56))
        assert strategy.min_roi_reached(trade, 0.12, dt_now() - timedelta(minutes=56))

        assert not strategy.min_roi_reached(trade, 0.04, dt_now() - timedelta(minutes=39))
        assert strategy.min_roi_reached(trade, 0.071, dt_now() - timedelta(minutes=39))

        assert not strategy.min_roi_reached(trade, 0.04, dt_now() - timedelta(minutes=26))
        assert strategy.min_roi_reached(trade, 0.06, dt_now() - timedelta(minutes=26))

        # Should not trigger with 20% profit since after 55 minutes only 30% is active.
        assert not strategy.min_roi_reached(trade, 0.20, dt_now() - timedelta(minutes=2))
        assert strategy.min_roi_reached(trade, 0.31, dt_now() - timedelta(minutes=2))


def test_min_roi_reached3(default_conf, fee) -> None:
    # test for issue #1948
    min_roi = {
        20: 0.07,
        30: 0.05,
        55: 0.30,
    }
    strategy = StrategyResolver.load_strategy(default_conf)
    strategy.minimal_roi = min_roi
    trade = Trade(
        pair="ETH/BTC",
        stake_amount=0.001,
        amount=5,
        open_date=dt_now() - timedelta(hours=1),
        fee_open=fee.return_value,
        fee_close=fee.return_value,
        exchange="binance",
        open_rate=1,
    )

    assert not strategy.min_roi_reached(trade, 0.02, dt_now() - timedelta(minutes=56))
    assert not strategy.min_roi_reached(trade, 0.12, dt_now() - timedelta(minutes=56))

    assert not strategy.min_roi_reached(trade, 0.04, dt_now() - timedelta(minutes=39))
    assert strategy.min_roi_reached(trade, 0.071, dt_now() - timedelta(minutes=39))

    assert not strategy.min_roi_reached(trade, 0.04, dt_now() - timedelta(minutes=26))
    assert strategy.min_roi_reached(trade, 0.06, dt_now() - timedelta(minutes=26))

    # Should not trigger with 20% profit since after 55 minutes only 30% is active.
    assert not strategy.min_roi_reached(trade, 0.20, dt_now() - timedelta(minutes=2))
    assert strategy.min_roi_reached(trade, 0.31, dt_now() - timedelta(minutes=2))


def test_min_roi_reached_custom_roi(default_conf, fee) -> None:
    strategy = StrategyResolver.load_strategy(default_conf)
    # Move traditional ROI out of the way
    strategy.minimal_roi = {0: 2000}
    strategy.use_custom_roi = True

    def custom_roi(*args, trade: Trade, current_time: datetime, **kwargs):
        trade_dur = int((current_time.timestamp() - trade.open_date_utc.timestamp()) // 60)
        # Profit is reduced after 30 minutes.
        if trade.pair == "XRP/BTC":
            return 0.2
        if trade_dur > 30:
            return 0.05
        return 0.1

    strategy.custom_roi = MagicMock(side_effect=custom_roi)

    trade = Trade(
        pair="ETH/BTC",
        stake_amount=0.001,
        amount=5,
        open_date=dt_now() - timedelta(hours=1),
        fee_open=fee.return_value,
        fee_close=fee.return_value,
        exchange="binance",
        open_rate=1,
    )

    assert not strategy.min_roi_reached(trade, 0.02, dt_now() - timedelta(minutes=56))
    assert strategy.custom_roi.call_count == 1
    assert strategy.min_roi_reached(trade, 0.12, dt_now() - timedelta(minutes=56))

    # after 30 minutes, the profit is reduced to 5%
    assert strategy.min_roi_reached(trade, 0.12, dt_now() - timedelta(minutes=29))
    assert strategy.min_roi_reached(trade, 0.06, dt_now() - timedelta(minutes=29))
    assert strategy.min_roi_reached(trade, 0.051, dt_now() - timedelta(minutes=29))
    # Comparison to exactly 5% should not trigger
    assert not strategy.min_roi_reached(trade, 0.05, dt_now() - timedelta(minutes=29))

    # XRP/BTC has a custom roi of 20%.

    trade.pair = "XRP/BTC"
    assert not strategy.min_roi_reached(trade, 0.12, dt_now() - timedelta(minutes=56))
    assert not strategy.min_roi_reached(trade, 0.12, dt_now() - timedelta(minutes=1))
    # XRP/BTC is not time related
    assert strategy.min_roi_reached(trade, 0.201, dt_now() - timedelta(minutes=1))
    assert strategy.min_roi_reached(trade, 0.201, dt_now() - timedelta(minutes=56))

    assert strategy.custom_roi.call_count == 10


@pytest.mark.parametrize(
    "profit,adjusted,expected,liq,trailing,custom,profit2,adjusted2,expected2,custom_stop",
    [
        # Profit, adjusted stoploss(absolute), profit for 2nd call, enable trailing,
        #   enable custom stoploss, expected after 1st call, expected after 2nd call
        (0.2, 0.9, ExitType.NONE, None, False, False, 0.3, 0.9, ExitType.NONE, None),
        (0.2, 0.9, ExitType.NONE, None, False, False, -0.2, 0.9, ExitType.STOP_LOSS, None),
        (0.2, 0.9, ExitType.NONE, 0.92, False, False, -0.09, 0.9, ExitType.LIQUIDATION, None),
        (
            0.2,
            1.14,
            ExitType.NONE,
            None,
            True,
            False,
            0.05,
            1.14,
            ExitType.TRAILING_STOP_LOSS,
            None,
        ),
        (0.01, 0.96, ExitType.NONE, None, True, False, 0.05, 0.998, ExitType.NONE, None),
        (
            0.05,
            0.998,
            ExitType.NONE,
            None,
            True,
            False,
            -0.01,
            0.998,
            ExitType.TRAILING_STOP_LOSS,
            None,
        ),
        # Default custom case - trails with 10%
        (0.05, 0.945, ExitType.NONE, None, False, True, -0.02, 0.945, ExitType.NONE, None),
        (
            0.05,
            0.945,
            ExitType.NONE,
            None,
            False,
            True,
            -0.06,
            0.945,
            ExitType.TRAILING_STOP_LOSS,
            None,
        ),
        (
            0.05,
            0.998,
            ExitType.NONE,
            None,
            False,
            True,
            -0.06,
            0.998,
            ExitType.TRAILING_STOP_LOSS,
            lambda **kwargs: -0.05,
        ),
        (
            0.05,
            0.998,
            ExitType.NONE,
            None,
            False,
            True,
            0.09,
            1.036,
            ExitType.NONE,
            lambda **kwargs: -0.05,
        ),
        (
            0.05,
            0.945,
            ExitType.NONE,
            None,
            False,
            True,
            0.09,
            0.981,
            ExitType.NONE,
            lambda current_profit, **kwargs: (
                -0.1 if current_profit < 0.6 else -(current_profit * 2)
            ),
        ),
        # Error case - static stoploss in place
        (
            0.05,
            0.9,
            ExitType.NONE,
            None,
            False,
            True,
            0.09,
            0.9,
            ExitType.NONE,
            lambda **kwargs: None,
        ),
        # Error case - Returning inf.
        (
            0.05,
            0.9,
            ExitType.NONE,
            None,
            False,
            True,
            0.09,
            0.9,
            ExitType.NONE,
            lambda **kwargs: math.inf,
        ),
    ],
)
def test_ft_stoploss_reached(
    default_conf,
    fee,
    profit,
    adjusted,
    expected,
    liq,
    trailing,
    custom,
    profit2,
    adjusted2,
    expected2,
    custom_stop,
) -> None:
    strategy = StrategyResolver.load_strategy(default_conf)
    trade = Trade(
        pair="ETH/BTC",
        stake_amount=0.01,
        amount=1,
        open_date=dt_now() - timedelta(hours=1),
        fee_open=fee.return_value,
        fee_close=fee.return_value,
        exchange="binance",
        open_rate=1,
        liquidation_price=liq,
        price_precision=4,
        precision_mode=2,
        precision_mode_price=2,
    )
    trade.adjust_min_max_rates(trade.open_rate, trade.open_rate)
    strategy.trailing_stop = trailing
    strategy.trailing_stop_positive = -0.05
    strategy.use_custom_stoploss = custom
    original_stopvalue = strategy.custom_stoploss
    if custom_stop:
        strategy.custom_stoploss = custom_stop

    now = dt_now()
    current_rate = trade.open_rate * (1 + profit)
    sl_flag = strategy.ft_stoploss_reached(
        current_rate=current_rate,
        trade=trade,
        current_time=now,
        current_profit=profit,
        force_stoploss=0,
        high=None,
    )
    assert isinstance(sl_flag, ExitCheckTuple)
    assert sl_flag.exit_type == expected
    if expected == ExitType.NONE:
        assert sl_flag.exit_flag is False
    else:
        assert sl_flag.exit_flag is True
    assert round(trade.stop_loss, 3) == adjusted
    current_rate2 = trade.open_rate * (1 + profit2)

    sl_flag = strategy.ft_stoploss_reached(
        current_rate=current_rate2,
        trade=trade,
        current_time=now,
        current_profit=profit2,
        force_stoploss=0,
        high=None,
    )
    assert sl_flag.exit_type == expected2
    if expected2 == ExitType.NONE:
        assert sl_flag.exit_flag is False
    else:
        assert sl_flag.exit_flag is True
    assert round(trade.stop_loss, 3) == adjusted2

    strategy.custom_stoploss = original_stopvalue


def test_custom_exit(default_conf, fee, caplog) -> None:
    strategy = StrategyResolver.load_strategy(default_conf)
    trade = Trade(
        pair="ETH/BTC",
        stake_amount=0.01,
        amount=1,
        open_date=dt_now() - timedelta(hours=1),
        fee_open=fee.return_value,
        fee_close=fee.return_value,
        exchange="binance",
        open_rate=1,
        leverage=1.0,
    )

    now = dt_now()
    res = strategy.should_exit(trade, 1, now, enter=False, exit_=False, low=None, high=None)

    assert res == []

    strategy.custom_exit = MagicMock(return_value=True)
    res = strategy.should_exit(trade, 1, now, enter=False, exit_=False, low=None, high=None)
    assert res[0].exit_flag is True
    assert res[0].exit_type == ExitType.CUSTOM_EXIT
    assert res[0].exit_reason == "custom_exit"

    strategy.custom_exit = MagicMock(return_value="hello world")

    res = strategy.should_exit(trade, 1, now, enter=False, exit_=False, low=None, high=None)
    assert res[0].exit_type == ExitType.CUSTOM_EXIT
    assert res[0].exit_flag is True
    assert res[0].exit_reason == "hello world"

    caplog.clear()
    strategy.custom_exit = MagicMock(return_value="h" * CUSTOM_TAG_MAX_LENGTH * 2)
    res = strategy.should_exit(trade, 1, now, enter=False, exit_=False, low=None, high=None)
    assert res[0].exit_type == ExitType.CUSTOM_EXIT
    assert res[0].exit_flag is True
    assert res[0].exit_reason == "h" * (CUSTOM_TAG_MAX_LENGTH)
    assert log_has_re("Custom exit reason returned from custom_exit is too long.*", caplog)


def test_should_sell(default_conf, fee) -> None:
    strategy = StrategyResolver.load_strategy(default_conf)
    trade = Trade(
        pair="ETH/BTC",
        stake_amount=0.01,
        amount=1,
        open_date=dt_now() - timedelta(hours=1),
        fee_open=fee.return_value,
        fee_close=fee.return_value,
        exchange="binance",
        open_rate=1,
        leverage=1.0,
    )
    now = dt_now()
    res = strategy.should_exit(trade, 1, now, enter=False, exit_=False, low=None, high=None)

    assert res == []
    strategy.min_roi_reached = MagicMock(return_value=True)

    res = strategy.should_exit(trade, 1, now, enter=False, exit_=False, low=None, high=None)
    assert len(res) == 1
    assert res == [ExitCheckTuple(exit_type=ExitType.ROI)]

    strategy.min_roi_reached = MagicMock(return_value=True)
    strategy.ft_stoploss_reached = MagicMock(
        return_value=ExitCheckTuple(exit_type=ExitType.STOP_LOSS)
    )

    res = strategy.should_exit(trade, 1, now, enter=False, exit_=False, low=None, high=None)
    assert len(res) == 2
    assert res == [
        ExitCheckTuple(exit_type=ExitType.STOP_LOSS),
        ExitCheckTuple(exit_type=ExitType.ROI),
    ]

    strategy.custom_exit = MagicMock(return_value="hello world")
    # custom-exit and exit-signal is first
    res = strategy.should_exit(trade, 1, now, enter=False, exit_=False, low=None, high=None)
    assert len(res) == 3
    assert res == [
        ExitCheckTuple(exit_type=ExitType.CUSTOM_EXIT, exit_reason="hello world"),
        ExitCheckTuple(exit_type=ExitType.STOP_LOSS),
        ExitCheckTuple(exit_type=ExitType.ROI),
    ]

    strategy.ft_stoploss_reached = MagicMock(
        return_value=ExitCheckTuple(exit_type=ExitType.TRAILING_STOP_LOSS)
    )
    # Regular exit signal
    res = strategy.should_exit(trade, 1, now, enter=False, exit_=True, low=None, high=None)
    assert len(res) == 3
    assert res == [
        ExitCheckTuple(exit_type=ExitType.EXIT_SIGNAL),
        ExitCheckTuple(exit_type=ExitType.ROI),
        ExitCheckTuple(exit_type=ExitType.TRAILING_STOP_LOSS),
    ]

    # Regular exit signal, no ROI
    strategy.min_roi_reached = MagicMock(return_value=False)
    res = strategy.should_exit(trade, 1, now, enter=False, exit_=True, low=None, high=None)
    assert len(res) == 2
    assert res == [
        ExitCheckTuple(exit_type=ExitType.EXIT_SIGNAL),
        ExitCheckTuple(exit_type=ExitType.TRAILING_STOP_LOSS),
    ]


@pytest.mark.parametrize("side", TRADE_SIDES)
def test_leverage_callback(default_conf, side) -> None:
    default_conf["strategy"] = "StrategyTestV2"
    strategy = StrategyResolver.load_strategy(default_conf)

    assert (
        strategy.leverage(
            pair="XRP/USDT",
            current_time=datetime.now(UTC),
            current_rate=2.2,
            proposed_leverage=1.0,
            max_leverage=5.0,
            side=side,
            entry_tag=None,
        )
        == 1
    )

    default_conf["strategy"] = CURRENT_TEST_STRATEGY
    strategy = StrategyResolver.load_strategy(default_conf)
    assert (
        strategy.leverage(
            pair="XRP/USDT",
            current_time=datetime.now(UTC),
            current_rate=2.2,
            proposed_leverage=1.0,
            max_leverage=5.0,
            side=side,
            entry_tag="entry_tag_test",
        )
        == 3
    )


def test_analyze_ticker_default(ohlcv_history, mocker, caplog) -> None:
    caplog.set_level(logging.DEBUG)
    ind_mock = MagicMock(side_effect=lambda x, meta: x)
    entry_mock = MagicMock(side_effect=lambda x, meta: x)
    exit_mock = MagicMock(side_effect=lambda x, meta: x)
    mocker.patch.multiple(
        "freqtrade.strategy.interface.IStrategy",
        advise_indicators=ind_mock,
        advise_entry=entry_mock,
        advise_exit=exit_mock,
    )
    strategy = StrategyTestV3({})
    strategy.analyze_ticker(ohlcv_history, {"pair": "ETH/BTC"})
    assert ind_mock.call_count == 1
    assert entry_mock.call_count == 1
    assert entry_mock.call_count == 1

    assert log_has("TA Analysis Launched", caplog)
    assert not log_has("Skipping TA Analysis for already analyzed candle", caplog)
    caplog.clear()

    strategy.analyze_ticker(ohlcv_history, {"pair": "ETH/BTC"})
    # No analysis happens as process_only_new_candles is true
    assert ind_mock.call_count == 2
    assert entry_mock.call_count == 2
    assert entry_mock.call_count == 2
    assert log_has("TA Analysis Launched", caplog)
    assert not log_has("Skipping TA Analysis for already analyzed candle", caplog)


def test__analyze_ticker_internal_skip_analyze(ohlcv_history, mocker, caplog) -> None:
    caplog.set_level(logging.DEBUG)
    ind_mock = MagicMock(side_effect=lambda x, meta: x)
    entry_mock = MagicMock(side_effect=lambda x, meta: x)
    exit_mock = MagicMock(side_effect=lambda x, meta: x)
    mocker.patch.multiple(
        "freqtrade.strategy.interface.IStrategy",
        advise_indicators=ind_mock,
        advise_entry=entry_mock,
        advise_exit=exit_mock,
    )
    strategy = StrategyTestV3({})
    strategy.dp = DataProvider({}, None, None)
    strategy.process_only_new_candles = True

    ret = strategy._analyze_ticker_internal(ohlcv_history, {"pair": "ETH/BTC"})
    assert "high" in ret.columns
    assert "low" in ret.columns
    assert "close" in ret.columns
    assert isinstance(ret, DataFrame)
    assert ind_mock.call_count == 1
    assert entry_mock.call_count == 1
    assert entry_mock.call_count == 1
    assert log_has("TA Analysis Launched", caplog)
    assert not log_has("Skipping TA Analysis for already analyzed candle", caplog)
    caplog.clear()

    ret = strategy._analyze_ticker_internal(ohlcv_history, {"pair": "ETH/BTC"})
    # No analysis happens as process_only_new_candles is true
    assert ind_mock.call_count == 1
    assert entry_mock.call_count == 1
    assert entry_mock.call_count == 1
    # only skipped analyze adds buy and sell columns, otherwise it's all mocked
    assert "enter_long" in ret.columns
    assert "exit_long" in ret.columns
    assert ret["enter_long"].sum() == 0
    assert ret["exit_long"].sum() == 0
    assert not log_has("TA Analysis Launched", caplog)
    assert log_has("Skipping TA Analysis for already analyzed candle", caplog)


@pytest.mark.usefixtures("init_persistence")
def test_is_pair_locked(default_conf):
    PairLocks.timeframe = default_conf["timeframe"]
    PairLocks.use_db = True
    strategy = StrategyResolver.load_strategy(default_conf)
    # No lock should be present
    assert len(PairLocks.get_pair_locks(None)) == 0

    pair = "ETH/BTC"
    assert not strategy.is_pair_locked(pair)
    strategy.lock_pair(pair, dt_now() + timedelta(minutes=4))
    # ETH/BTC locked for 4 minutes
    assert strategy.is_pair_locked(pair)

    # XRP/BTC should not be locked now
    pair = "XRP/BTC"
    assert not strategy.is_pair_locked(pair)

    # Unlocking a pair that's not locked should not raise an error
    strategy.unlock_pair(pair)

    # Unlock original pair
    pair = "ETH/BTC"
    strategy.unlock_pair(pair)
    assert not strategy.is_pair_locked(pair)

    # Lock with reason
    reason = "TestLockR"
    strategy.lock_pair(pair, dt_now() + timedelta(minutes=4), reason)
    assert strategy.is_pair_locked(pair)
    strategy.unlock_reason(reason)
    assert not strategy.is_pair_locked(pair)

    pair = "BTC/USDT"
    # Lock until 14:30
    lock_time = datetime(2020, 5, 1, 14, 30, 0, tzinfo=UTC)
    # Subtract 2 seconds, as locking rounds up to the next candle.
    strategy.lock_pair(pair, lock_time - timedelta(seconds=2))

    assert not strategy.is_pair_locked(pair)
    # latest candle is from 14:20, lock goes to 14:30
    assert strategy.is_pair_locked(pair, candle_date=lock_time + timedelta(minutes=-10))
    assert strategy.is_pair_locked(pair, candle_date=lock_time + timedelta(minutes=-50))

    # latest candle is from 14:25 (lock should be lifted)
    # Since this is the "new candle" available at 14:30
    assert not strategy.is_pair_locked(pair, candle_date=lock_time + timedelta(minutes=-4))

    # Should not be locked after time expired
    assert not strategy.is_pair_locked(pair, candle_date=lock_time + timedelta(minutes=10))

    # Change timeframe to 15m
    strategy.timeframe = "15m"
    # Candle from 14:14 - lock goes until 14:30
    assert strategy.is_pair_locked(pair, candle_date=lock_time + timedelta(minutes=-16))
    assert strategy.is_pair_locked(pair, candle_date=lock_time + timedelta(minutes=-15, seconds=-2))
    # Candle from 14:15 - lock goes until 14:30
    assert not strategy.is_pair_locked(pair, candle_date=lock_time + timedelta(minutes=-15))


def test_is_informative_pairs_callback(default_conf):
    default_conf.update({"strategy": "StrategyTestV2"})
    strategy = StrategyResolver.load_strategy(default_conf)
    # Should return empty
    # Uses fallback to base implementation
    assert [] == strategy.gather_informative_pairs()


def test_auto_hyperopt_interface(default_conf):
    default_conf.update({"strategy": "HyperoptableStrategyV2"})
    PairLocks.timeframe = default_conf["timeframe"]
    strategy = StrategyResolver.load_strategy(default_conf)
    strategy.ft_bot_start()
    assert list(strategy.enumerate_parameters("deadBeef")) == []

    assert strategy.buy_rsi.value == strategy.buy_params["buy_rsi"]
    # PlusDI is NOT in the buy-params, so default should be used
    assert strategy.buy_plusdi.value == 0.5
    assert strategy.sell_rsi.value == strategy.sell_params["sell_rsi"]

    assert repr(strategy.sell_rsi) == "IntParameter(74)"

    # Parameter is disabled - so value from sell_param dict will NOT be used.
    assert strategy.sell_minusdi.value == 0.5
    all_params = detect_all_parameters(strategy.__class__)
    assert isinstance(all_params, dict)
    # Only one buy param at class level
    assert len(all_params["buy"]) == 1
    # Running detect params at instance level reveals both parameters.
    params_inst = detect_all_parameters(strategy)
    assert len(params_inst["buy"]) == 2
    assert len(params_inst["sell"]) == 2

    strategy.__class__.sell_rsi = IntParameter([0, 10], default=5, space="buy")

    spaces = detect_all_parameters(strategy.__class__)
    assert "buy" in spaces
    assert spaces["buy"]["sell_rsi"] == strategy.sell_rsi
    del strategy.__class__.sell_rsi

    strategy.__class__.exit22_rsi = IntParameter([0, 10], default=5)

    with pytest.raises(
        DependencyException, match=r"Cannot determine parameter space for exit22_rsi\."
    ):
        detect_all_parameters(strategy.__class__)

    # Invalid parameter space
    strategy.__class__.exit22_rsi = IntParameter([0, 10], default=5, space="all")
    with pytest.raises(
        DependencyException, match=r"'all' is not a valid space\. Parameter: exit22_rsi\."
    ):
        detect_all_parameters(strategy.__class__)

    strategy.__class__.exit22_rsi = IntParameter([0, 10], default=5, space="hello:world:22")
    with pytest.raises(
        DependencyException,
        match=r"'hello:world:22' is not a valid space\. Parameter: exit22_rsi\.",
    ):
        detect_all_parameters(strategy.__class__)
    del strategy.__class__.exit22_rsi

    # Valid exit parameter
    strategy.__class__.exit_rsi = IntParameter([0, 10], default=5)
    strategy.__class__.enter_rsi = IntParameter([0, 10], default=5)
    spaces = detect_all_parameters(strategy.__class__)
    assert "exit" in spaces
    assert "enter" in spaces
    del strategy.__class__.exit_rsi
    del strategy.__class__.enter_rsi


def test_auto_hyperopt_interface_loadparams(default_conf, mocker, caplog):
    default_conf.update({"strategy": "HyperoptableStrategy"})
    del default_conf["stoploss"]
    del default_conf["minimal_roi"]
    mocker.patch.object(Path, "is_file", MagicMock(return_value=True))
    mocker.patch.object(Path, "open")
    expected_result = {
        "strategy_name": "HyperoptableStrategy",
        "params": {
            "stoploss": {
                "stoploss": -0.05,
            },
            "roi": {"0": 0.2, "1200": 0.01},
        },
    }
    mocker.patch("freqtrade.strategy.hyper.HyperoptTools.load_params", return_value=expected_result)
    PairLocks.timeframe = default_conf["timeframe"]
    strategy = StrategyResolver.load_strategy(default_conf)
    assert strategy.stoploss == -0.05
    assert strategy.minimal_roi == {0: 0.2, 1200: 0.01}

    expected_result = {
        "strategy_name": "HyperoptableStrategy_No",
        "params": {
            "stoploss": {
                "stoploss": -0.05,
            },
            "roi": {"0": 0.2, "1200": 0.01},
        },
    }

    mocker.patch("freqtrade.strategy.hyper.HyperoptTools.load_params", return_value=expected_result)
    with pytest.raises(OperationalException, match=r"Invalid parameter file provided\."):
        StrategyResolver.load_strategy(default_conf)

    mocker.patch(
        "freqtrade.strategy.hyper.HyperoptTools.load_params", MagicMock(side_effect=ValueError())
    )

    StrategyResolver.load_strategy(default_conf)
    assert log_has("Invalid parameter file format.", caplog)


@pytest.mark.parametrize(
    "function,raises",
    [
        ("populate_entry_trend", False),
        ("advise_entry", False),
        ("populate_exit_trend", False),
        ("advise_exit", False),
    ],
)
def test_pandas_warning_direct(ohlcv_history, function, raises, recwarn):
    df = _STRATEGY.populate_indicators(ohlcv_history, {"pair": "ETH/BTC"})
    if raises:
        assert len(recwarn) == 1
        # https://github.com/pandas-dev/pandas/issues/56503
        # Fixed in 2.2.x
        getattr(_STRATEGY, function)(df, {"pair": "ETH/BTC"})
    else:
        assert len(recwarn) == 0, f"warnings: {', '.join(recwarn.list)}"

        getattr(_STRATEGY, function)(df, {"pair": "ETH/BTC"})


def test_pandas_warning_through_analyze_pair(ohlcv_history, mocker, recwarn):
    mocker.patch.object(_STRATEGY.dp, "ohlcv", return_value=ohlcv_history)
    _STRATEGY.analyze_pair("ETH/BTC")
    assert len(recwarn) == 0, f"warnings: {', '.join(recwarn.list)}"
