364 lines
13 KiB
Python
364 lines
13 KiB
Python
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_filter(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', -0.01, 0.66, 3.71, 0.50, 1.71],
|
|
['C/D', -0.01, 0.66, 3.71, 0.50, 1.71],
|
|
['N/O', -0.01, 0.66, 3.71, 0.50, 1.71]
|
|
]
|
|
))
|
|
|
|
pairs = ['A/B', 'C/D', 'E/F', 'G/H']
|
|
assert(edge.filter(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', -0.01, 0.66, 3.71, 0.50, 1.71],
|
|
['C/D', -0.01, 0.66, 3.71, 0.50, 1.71],
|
|
['N/O', -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 final[0][0] == 'TEST/BTC'
|
|
assert final[0][1] == -0.9
|
|
assert round(final[0][2], 10) == 0.3333333333
|
|
assert round(final[0][3], 10) == 306.5384615384
|
|
assert round(final[0][4], 10) == 2.0
|
|
assert round(final[0][5], 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
|
|
##############################################################
|