PYTHON test_trade_custom_data.py

from copy import deepcopy
from unittest.mock import MagicMock

import pytest

from freqtrade.data.history.history_utils import get_timerange
from freqtrade.optimize.backtesting import Backtesting
from freqtrade.persistence import Trade, disable_database_use, enable_database_use
from freqtrade.persistence.custom_data import CustomDataWrapper
from tests.conftest import (
    EXMS,
    create_mock_trades_usdt,
    generate_test_data,
    get_patched_freqtradebot,
    patch_exchange,
)


@pytest.mark.usefixtures("init_persistence")
@pytest.mark.parametrize("use_db", [True, False])
def test_trade_custom_data(fee, use_db):
    if not use_db:
        disable_database_use("5m")
    Trade.reset_trades()
    CustomDataWrapper.reset_custom_data()

    create_mock_trades_usdt(fee, use_db=use_db)

    trade1 = Trade.get_trades_proxy()[0]
    if not use_db:
        trade1.id = 1

    assert trade1.get_all_custom_data() == []
    trade1.set_custom_data("test_str", "test_value")
    trade1.set_custom_data("test_int", 1)
    trade1.set_custom_data("test_float", 1.55)
    trade1.set_custom_data("test_bool", True)
    trade1.set_custom_data("test_dict", {"test": "dict"})

    assert len(trade1.get_all_custom_data()) == 5
    assert trade1.get_custom_data("test_str") == "test_value"
    trade1.set_custom_data("test_str", "test_value_updated")
    assert trade1.get_custom_data("test_str") == "test_value_updated"

    assert trade1.get_custom_data("test_int") == 1
    assert isinstance(trade1.get_custom_data("test_int"), int)

    assert trade1.get_custom_data("test_float") == 1.55
    assert isinstance(trade1.get_custom_data("test_float"), float)

    assert trade1.get_custom_data("test_bool") is True
    assert isinstance(trade1.get_custom_data("test_bool"), bool)

    assert trade1.get_custom_data("test_dict") == {"test": "dict"}
    assert isinstance(trade1.get_custom_data("test_dict"), dict)
    if not use_db:
        enable_database_use()


def test_trade_custom_data_strategy_compat(mocker, default_conf_usdt, fee):
    mocker.patch(f"{EXMS}.get_rate", return_value=0.50)
    mocker.patch("freqtrade.freqtradebot.FreqtradeBot.get_real_amount", return_value=None)
    mocker.patch("freqtrade.freqtradebot.FreqtradeBot.handle_cancel_exit", return_value=True)
    default_conf_usdt["minimal_roi"] = {"0": 100}

    freqtrade = get_patched_freqtradebot(mocker, default_conf_usdt)
    create_mock_trades_usdt(fee)

    trade1 = Trade.get_trades_proxy(pair="ADA/USDT")[0]
    trade1.set_custom_data("test_str", "test_value")
    trade1.set_custom_data("test_int", 1)

    def custom_exit(pair, trade, **kwargs):
        if pair == "ADA/USDT":
            custom_val = trade.get_custom_data("test_str")
            custom_val_i = trade.get_custom_data("test_int")

            return f"{custom_val}_{custom_val_i}"

    freqtrade.strategy.custom_exit = custom_exit
    ff_spy = mocker.spy(freqtrade.strategy, "custom_exit")
    trades = Trade.get_open_trades()
    freqtrade.exit_positions(trades)
    Trade.commit()

    trade_after = Trade.get_trades_proxy(pair="ADA/USDT")[0]
    assert trade_after.get_custom_data("test_str") == "test_value"
    assert trade_after.get_custom_data("test_int") == 1
    # 2 trades filled entry, with open exit order
    # 1 trade with filled entry order
    assert ff_spy.call_count == 3

    assert trade_after.exit_reason == "test_value_1"


def test_trade_custom_data_strategy_backtest_compat(mocker, default_conf_usdt, fee):
    mocker.patch(f"{EXMS}.get_fee", fee)
    mocker.patch(f"{EXMS}.get_min_pair_stake_amount", return_value=10)
    mocker.patch(f"{EXMS}.get_max_pair_stake_amount", return_value=float("inf"))
    mocker.patch(f"{EXMS}.get_max_leverage", return_value=10)
    mocker.patch(f"{EXMS}.get_maintenance_ratio_and_amt", return_value=(0.1, 0.1))
    mocker.patch("freqtrade.optimize.backtesting.Backtesting._run_funding_fees")

    patch_exchange(mocker)
    default_conf_usdt.update(
        {
            "stake_amount": 100.0,
            "max_open_trades": 2,
            "dry_run_wallet": 1000.0,
            "strategy": "StrategyTestV3",
            "trading_mode": "futures",
            "margin_mode": "isolated",
            "stoploss": -2,
            "minimal_roi": {"0": 100},
        }
    )
    default_conf_usdt["pairlists"] = [{"method": "StaticPairList", "allow_inactive": True}]
    backtesting = Backtesting(default_conf_usdt)

    df = generate_test_data(default_conf_usdt["timeframe"], 100, "2022-01-01 00:00:00+00:00")

    pair_exp = "XRP/USDT:USDT"

    def custom_exit(pair, trade, **kwargs):
        custom_val = trade.get_custom_data("test_str")
        custom_val_i = trade.get_custom_data("test_int", 0)

        if pair == pair_exp:
            trade.set_custom_data("test_str", "test_value")
            trade.set_custom_data("test_int", custom_val_i + 1)

        if custom_val_i >= 2:
            return f"{custom_val}_{custom_val_i}"

    backtesting._set_strategy(backtesting.strategylist[0])
    processed = backtesting.strategy.advise_all_indicators(
        {
            pair_exp: df,
            "BTC/USDT:USDT": df,
        }
    )

    def fun(dataframe, *args, **kwargs):
        dataframe.loc[dataframe.index == 50, "enter_long"] = 1
        return dataframe

    backtesting.strategy.advise_entry = fun
    backtesting.strategy.leverage = MagicMock(return_value=1)
    backtesting.strategy.custom_exit = custom_exit
    ff_spy = mocker.spy(backtesting.strategy, "custom_exit")

    min_date, max_date = get_timerange(processed)

    result = backtesting.backtest(
        processed=deepcopy(processed),
        start_date=min_date,
        end_date=max_date,
    )
    results = result["results"]
    assert not results.empty
    assert len(results) == 2
    assert results["pair"][0] == pair_exp
    assert results["pair"][1] == "BTC/USDT:USDT"
    assert results["exit_reason"][0] == "test_value_2"
    assert results["exit_reason"][1] == "exit_signal"

    assert ff_spy.call_count == 7
    Backtesting.cleanup()