stable/freqtrade/tests/edge/test_edge.py

from freqtrade.tests.conftest import get_patched_exchange
from freqtrade.edge import Edge
from pandas import DataFrame, to_datetime
from freqtrade.strategy.interface import SellType
import arrow
import numpy as np
import math

from unittest.mock import MagicMock


# Cases to be tested:
# 1) Open trade should be removed from the end
# 2) Two complete trades within dataframe (with sell hit for all)
# 3) Entered, sl 1%, candle drops 8% => Trade closed, 1% loss
# 4) Entered, sl 3%, candle drops 4%, recovers to 1% => Trade closed, 3% loss
# 5) Stoploss and sell are hit. should sell on stoploss
####################################################################

ticker_start_time = arrow.get(2018, 10, 3)
ticker_interval_in_minute = 60
_ohlc = {'date': 0, 'buy': 1, 'open': 2, 'high': 3, 'low': 4, 'close': 5, 'sell': 6, 'volume': 7}


def test_adjust(mocker, default_conf):
    exchange = get_patched_exchange(mocker, default_conf)
    edge = Edge(default_conf, exchange)
    mocker.patch('freqtrade.edge.Edge._cached_pairs', mocker.PropertyMock(
        return_value={
            'E/F': Edge._pair_info(-0.01, 0.66, 3.71, 0.50, 1.71),
            'C/D': Edge._pair_info(-0.01, 0.66, 3.71, 0.50, 1.71),
            'N/O': Edge._pair_info(-0.01, 0.66, 3.71, 0.50, 1.71)
        }
    ))

    pairs = ['A/B', 'C/D', 'E/F', 'G/H']
    assert(edge.adjust(pairs) == ['E/F', 'C/D'])


def test_stoploss(mocker, default_conf):
    exchange = get_patched_exchange(mocker, default_conf)
    edge = Edge(default_conf, exchange)
    mocker.patch('freqtrade.edge.Edge._cached_pairs', mocker.PropertyMock(
        return_value={
            'E/F': Edge._pair_info(-0.01, 0.66, 3.71, 0.50, 1.71),
            'C/D': Edge._pair_info(-0.01, 0.66, 3.71, 0.50, 1.71),
            'N/O': Edge._pair_info(-0.01, 0.66, 3.71, 0.50, 1.71)
        }
    ))

    assert edge.stoploss('E/F') == -0.01


def _validate_ohlc(buy_ohlc_sell_matrice):
    for index, ohlc in enumerate(buy_ohlc_sell_matrice):
        # if not high < open < low or not high < close < low
        if not ohlc[3] >= ohlc[2] >= ohlc[4] or not ohlc[3] >= ohlc[5] >= ohlc[4]:
            raise Exception('Line ' + str(index + 1) + ' of ohlc has invalid values!')
    return True


def _build_dataframe(buy_ohlc_sell_matrice):
    _validate_ohlc(buy_ohlc_sell_matrice)
    tickers = []
    for ohlc in buy_ohlc_sell_matrice:
        ticker = {
            'date': ticker_start_time.shift(
                minutes=(
                    ohlc[0] *
                    ticker_interval_in_minute)).timestamp *
            1000,
            'buy': ohlc[1],
            'open': ohlc[2],
            'high': ohlc[3],
            'low': ohlc[4],
            'close': ohlc[5],
            'sell': ohlc[6]}
        tickers.append(ticker)

    frame = DataFrame(tickers)
    frame['date'] = to_datetime(frame['date'],
                                unit='ms',
                                utc=True,
                                infer_datetime_format=True)

    return frame


def _time_on_candle(number):
    return np.datetime64(ticker_start_time.shift(
        minutes=(number * ticker_interval_in_minute)).timestamp * 1000, 'ms')


def test_edge_heartbeat_calculate(mocker, default_conf):
    exchange = get_patched_exchange(mocker, default_conf)
    edge = Edge(default_conf, exchange)
    heartbeat = default_conf['edge']['process_throttle_secs']

    # should not recalculate if heartbeat not reached
    edge._last_updated = arrow.utcnow().timestamp - heartbeat + 1

    assert edge.calculate() is False


def mocked_load_data(datadir, pairs=[], ticker_interval='0m', refresh_pairs=False,
                     timerange=None, exchange=None):
    hz = 0.1
    base = 0.001

    ETHBTC = [
        [
            ticker_start_time.shift(minutes=(x * ticker_interval_in_minute)).timestamp * 1000,
            math.sin(x * hz) / 1000 + base,
            math.sin(x * hz) / 1000 + base + 0.0001,
            math.sin(x * hz) / 1000 + base - 0.0001,
            math.sin(x * hz) / 1000 + base,
            123.45
        ] for x in range(0, 500)]

    hz = 0.2
    base = 0.002
    LTCBTC = [
        [
            ticker_start_time.shift(minutes=(x * ticker_interval_in_minute)).timestamp * 1000,
            math.sin(x * hz) / 1000 + base,
            math.sin(x * hz) / 1000 + base + 0.0001,
            math.sin(x * hz) / 1000 + base - 0.0001,
            math.sin(x * hz) / 1000 + base,
            123.45
        ] for x in range(0, 500)]

    pairdata = {'NEO/BTC': ETHBTC, 'LTC/BTC': LTCBTC}
    return pairdata


def test_edge_process_downloaded_data(mocker, default_conf):
    default_conf['datadir'] = None
    exchange = get_patched_exchange(mocker, default_conf)
    mocker.patch('freqtrade.exchange.Exchange.get_fee', MagicMock(return_value=0.001))
    mocker.patch('freqtrade.optimize.load_data', mocked_load_data)
    mocker.patch('freqtrade.exchange.Exchange.refresh_tickers', MagicMock())
    edge = Edge(default_conf, exchange)

    assert edge.calculate()
    assert len(edge._cached_pairs) == 2
    assert edge._last_updated <= arrow.utcnow().timestamp + 2


def test_process_expectancy(mocker, default_conf):
    default_conf['edge']['min_trade_number'] = 2
    exchange = get_patched_exchange(mocker, default_conf)

    def get_fee():
        return 0.001

    exchange.get_fee = get_fee
    edge = Edge(default_conf, exchange)

    trades = [
        {'pair': 'TEST/BTC',
         'stoploss': -0.9,
         'profit_percent': '',
         'profit_abs': '',
         'open_time': np.datetime64('2018-10-03T00:05:00.000000000'),
         'close_time': np.datetime64('2018-10-03T00:10:00.000000000'),
         'open_index': 1,
         'close_index': 1,
         'trade_duration': '',
         'open_rate': 17,
         'close_rate': 17,
         'exit_type': 'sell_signal'},

        {'pair': 'TEST/BTC',
         'stoploss': -0.9,
         'profit_percent': '',
         'profit_abs': '',
         'open_time': np.datetime64('2018-10-03T00:20:00.000000000'),
         'close_time': np.datetime64('2018-10-03T00:25:00.000000000'),
         'open_index': 4,
         'close_index': 4,
         'trade_duration': '',
         'open_rate': 20,
         'close_rate': 20,
         'exit_type': 'sell_signal'},

        {'pair': 'TEST/BTC',
         'stoploss': -0.9,
         'profit_percent': '',
         'profit_abs': '',
         'open_time': np.datetime64('2018-10-03T00:30:00.000000000'),
         'close_time': np.datetime64('2018-10-03T00:40:00.000000000'),
         'open_index': 6,
         'close_index': 7,
         'trade_duration': '',
         'open_rate': 26,
         'close_rate': 34,
         'exit_type': 'sell_signal'}
    ]

    trades_df = DataFrame(trades)
    trades_df = edge._fill_calculable_fields(trades_df)
    final = edge._process_expectancy(trades_df)
    assert len(final) == 1

    assert 'TEST/BTC' in final
    assert final['TEST/BTC'].stoploss == -0.9
    assert round(final['TEST/BTC'].winrate, 10) == 0.3333333333
    assert round(final['TEST/BTC'].risk_reward_ratio, 10) == 306.5384615384
    assert round(final['TEST/BTC'].required_risk_reward, 10) == 2.0
    assert round(final['TEST/BTC'].expectancy, 10) == 101.5128205128

# 1) Open trade should be removed from the end


def test_case_1(mocker, default_conf):
    exchange = get_patched_exchange(mocker, default_conf)
    edge = Edge(default_conf, exchange)

    stoploss = -0.99  # we don't want stoploss to be hit in this test
    ticker = [
        # D=Date, B=Buy,  O=Open,  H=High,  L=Low,  C=Close, S=Sell
        # D, B,  O,  H,  L,  C, S
        [3, 1, 12, 25, 11, 20, 0],  # ->
        [4, 0, 20, 30, 19, 25, 1],  # -> should enter the trade
    ]

    ticker_df = _build_dataframe(ticker)
    trades = edge._find_trades_for_stoploss_range(ticker_df, 'TEST/BTC', [stoploss])

    # No trade should be found
    assert len(trades) == 0


# 2) Two complete trades within dataframe (with sell hit for all)
def test_case_2(mocker, default_conf):
    exchange = get_patched_exchange(mocker, default_conf)
    edge = Edge(default_conf, exchange)

    stoploss = -0.99  # we don't want stoploss to be hit in this test
    ticker = [
        # D=Date, B=Buy,  O=Open,  H=High,  L=Low,  C=Close, S=Sell
        # D, B,  O,  H,  L,  C, S
        [0, 1, 15, 20, 12, 17, 0],  # -> no action
        [1, 0, 17, 18, 13, 14, 1],  # -> should enter the trade as B signal recieved on last candle
        [2, 0, 14, 15, 11, 12, 0],  # -> exit the trade as the sell signal recieved on last candle
        [3, 1, 12, 25, 11, 20, 0],  # -> no action
        [4, 0, 20, 30, 19, 25, 0],  # -> should enter the trade
        [5, 0, 25, 27, 22, 26, 1],  # -> no action
        [6, 0, 26, 36, 25, 35, 0],  # -> should sell
    ]

    ticker_df = _build_dataframe(ticker)
    trades = edge._find_trades_for_stoploss_range(ticker_df, 'TEST/BTC', [stoploss])

    # Two trades must have occured
    assert len(trades) == 2

    # First trade check
    assert trades[0]['open_time'] == _time_on_candle(1)
    assert trades[0]['close_time'] == _time_on_candle(2)
    assert trades[0]['open_rate'] == ticker[1][_ohlc['open']]
    assert trades[0]['close_rate'] == ticker[2][_ohlc['open']]
    assert trades[0]['exit_type'] == SellType.SELL_SIGNAL
    ##############################################################

    # Second trade check
    assert trades[1]['open_time'] == _time_on_candle(4)
    assert trades[1]['close_time'] == _time_on_candle(6)
    assert trades[1]['open_rate'] == ticker[4][_ohlc['open']]
    assert trades[1]['close_rate'] == ticker[6][_ohlc['open']]
    assert trades[1]['exit_type'] == SellType.SELL_SIGNAL
    ##############################################################


# 3) Entered, sl 1%, candle drops 8% => Trade closed, 1% loss
def test_case_3(mocker, default_conf):
    exchange = get_patched_exchange(mocker, default_conf)
    edge = Edge(default_conf, exchange)

    stoploss = -0.01  # we don't want stoploss to be hit in this test
    ticker = [
        # D=Date, B=Buy,  O=Open,  H=High,  L=Low,  C=Close, S=Sell
        # D, B,  O,  H,  L,  C, S
        [0, 1, 15, 20, 12, 17, 0],  # -> no action
        [1, 0, 14, 15, 11, 12, 0],  # -> enter to trade, stoploss hit
        [2, 1, 12, 25, 11, 20, 0],  # -> no action
    ]

    ticker_df = _build_dataframe(ticker)
    trades = edge._find_trades_for_stoploss_range(ticker_df, 'TEST/BTC', [stoploss])

    # Two trades must have occured
    assert len(trades) == 1

    # First trade check
    assert trades[0]['open_time'] == _time_on_candle(1)
    assert trades[0]['close_time'] == _time_on_candle(1)
    assert trades[0]['open_rate'] == ticker[1][_ohlc['open']]
    assert trades[0]['close_rate'] == (stoploss + 1) * trades[0]['open_rate']
    assert trades[0]['exit_type'] == SellType.STOP_LOSS
    ##############################################################


# 4) Entered, sl 3 %, candle drops 4%, recovers to 1 % = > Trade closed, 3 % loss
def test_case_4(mocker, default_conf):
    exchange = get_patched_exchange(mocker, default_conf)
    edge = Edge(default_conf, exchange)

    stoploss = -0.03  # we don't want stoploss to be hit in this test
    ticker = [
        # D=Date, B=Buy,  O=Open,  H=High,  L=Low,  C=Close, S=Sell
        # D, B,  O,  H,  L,   C,    S
        [0, 1, 15, 20, 12, 17, 0],  # -> no action
        [1, 0, 17, 22, 16.90, 17, 0],  # -> enter to trade
        [2, 0, 16, 17, 14.4, 15.5, 0],  # -> stoploss hit
        [3, 0, 17, 25, 16.9, 22, 0],  # -> no action
    ]

    ticker_df = _build_dataframe(ticker)
    trades = edge._find_trades_for_stoploss_range(ticker_df, 'TEST/BTC', [stoploss])

    # Two trades must have occured
    assert len(trades) == 1

    # First trade check
    assert trades[0]['open_time'] == _time_on_candle(1)
    assert trades[0]['close_time'] == _time_on_candle(2)
    assert trades[0]['open_rate'] == ticker[1][_ohlc['open']]
    assert trades[0]['close_rate'] == (stoploss + 1) * trades[0]['open_rate']
    assert trades[0]['exit_type'] == SellType.STOP_LOSS
    ##############################################################


# 5) Stoploss and sell are hit. should sell on stoploss
def test_case_5(mocker, default_conf):
    exchange = get_patched_exchange(mocker, default_conf)
    edge = Edge(default_conf, exchange)

    exchange = get_patched_exchange(mocker, default_conf)
    edge = Edge(default_conf, exchange)

    stoploss = -0.03  # we don't want stoploss to be hit in this test
    ticker = [
        # D=Date, B=Buy,  O=Open,  H=High,  L=Low,  C=Close, S=Sell
        # D, B,  O,  H,  L,   C,    S
        [0, 1, 15, 20, 12, 17, 0],  # -> no action
        [1, 0, 17, 22, 16.90, 17, 0],  # -> enter to trade
        [2, 0, 16, 17, 14.4, 15.5, 1],  # -> stoploss hit and also sell signal
        [3, 0, 17, 25, 16.9, 22, 0],  # -> no action
    ]

    ticker_df = _build_dataframe(ticker)
    trades = edge._find_trades_for_stoploss_range(ticker_df, 'TEST/BTC', [stoploss])

    # Two trades must have occured
    assert len(trades) == 1

    # First trade check
    assert trades[0]['open_time'] == _time_on_candle(1)
    assert trades[0]['close_time'] == _time_on_candle(2)
    assert trades[0]['open_rate'] == ticker[1][_ohlc['open']]
    assert trades[0]['close_rate'] == (stoploss + 1) * trades[0]['open_rate']
    assert trades[0]['exit_type'] == SellType.STOP_LOSS
    ##############################################################