adelorenzo/stable
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'lev-strat' into lev-freqtradebot

Browse Source
This commit is contained in:
Sam Germain 2021-09-08 00:00:53 -06:00
commit e13b0414d8
22 changed files with 1129 additions and 208 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
freqtrade/edge/edge_positioning.py
View File

@ -159,7 +159,8 @@ class Edge:
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)..')
headers = ['date', 'buy', 'open', 'close', 'sell', 'high', 'low']
# TODO-lev: Should edge support shorts? needs to be investigated further...
headers = ['date', 'open', 'high', 'low', 'close', 'enter_long', 'exit_long']
trades: list = []
for pair, pair_data in preprocessed.items():
@ -168,7 +169,12 @@ class Edge:
pair_data = pair_data.reset_index(drop=True)
df_analyzed = self.strategy.advise_sell(
self.strategy.advise_buy(pair_data, {'pair': pair}), {'pair': pair})[headers].copy()
dataframe=self.strategy.advise_buy(
dataframe=pair_data,
metadata={'pair': pair}
),
metadata={'pair': pair}
)[headers].copy()
trades += self._find_trades_for_stoploss_range(df_analyzed, pair, self._stoploss_range)
@ -382,8 +388,8 @@ class Edge:
return final
def _find_trades_for_stoploss_range(self, df, pair, stoploss_range):
buy_column = df['buy'].values
sell_column = df['sell'].values
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

2
freqtrade/enums/__init__.py
View File

@ -4,6 +4,6 @@ from freqtrade.enums.collateral import Collateral
from freqtrade.enums.rpcmessagetype import RPCMessageType
from freqtrade.enums.runmode import NON_UTIL_MODES, OPTIMIZE_MODES, TRADING_MODES, RunMode
from freqtrade.enums.selltype import SellType
from freqtrade.enums.signaltype import SignalTagType, SignalType
from freqtrade.enums.signaltype import SignalDirection, SignalTagType, SignalType
from freqtrade.enums.state import State
from freqtrade.enums.tradingmode import TradingMode

12
freqtrade/enums/signaltype.py
View File

@ -5,8 +5,10 @@ class SignalType(Enum):
"""
Enum to distinguish between buy and sell signals
"""
BUY = "buy"
SELL = "sell"
ENTER_LONG = "enter_long"
EXIT_LONG = "exit_long"
ENTER_SHORT = "enter_short"
EXIT_SHORT = "exit_short"
class SignalTagType(Enum):
@ -14,3 +16,9 @@ class SignalTagType(Enum):
Enum for signal columns
"""
BUY_TAG = "buy_tag"
SHORT_TAG = "short_tag"
class SignalDirection(Enum):
LONG = 'long'
SHORT = 'short'

46
freqtrade/freqtradebot.py
View File

@ -420,24 +420,25 @@ class FreqtradeBot(LoggingMixin):
return False
# running get_signal on historical data fetched
(buy, sell, buy_tag) = self.strategy.get_signal(
pair,
self.strategy.timeframe,
analyzed_df
(side, enter_tag) = self.strategy.get_entry_signal(
pair, self.strategy.timeframe, analyzed_df
)
if buy and not sell:
if side:
stake_amount = self.wallets.get_trade_stake_amount(pair, self.edge)
bid_check_dom = self.config.get('bid_strategy', {}).get('check_depth_of_market', {})
if ((bid_check_dom.get('enabled', False)) and
(bid_check_dom.get('bids_to_ask_delta', 0) > 0)):
# TODO-lev: Does the below need to be adjusted for shorts?
if self._check_depth_of_market_buy(pair, bid_check_dom):
return self.execute_entry(pair, stake_amount, buy_tag=buy_tag)
# TODO-lev: pass in "enter" as side.
return self.execute_entry(pair, stake_amount, enter_tag=enter_tag)
else:
return False
return self.execute_entry(pair, stake_amount, buy_tag=buy_tag)
return self.execute_entry(pair, stake_amount, enter_tag=enter_tag)
else:
return False
@ -466,7 +467,7 @@ class FreqtradeBot(LoggingMixin):
return False
def execute_entry(self, pair: str, stake_amount: float, price: Optional[float] = None,
forcebuy: bool = False, buy_tag: Optional[str] = None) -> bool:
forcebuy: bool = False, enter_tag: Optional[str] = None) -> bool:
"""
Executes a limit buy for the given pair
:param pair: pair for which we want to create a LIMIT_BUY
@ -575,7 +576,8 @@ class FreqtradeBot(LoggingMixin):
exchange=self.exchange.id,
open_order_id=order_id,
strategy=self.strategy.get_strategy_name(),
buy_tag=buy_tag,
# TODO-lev: compatibility layer for buy_tag (!)
buy_tag=enter_tag,
timeframe=timeframe_to_minutes(self.config['timeframe'])
)
trade.orders.append(order_obj)
@ -699,22 +701,22 @@ class FreqtradeBot(LoggingMixin):
logger.debug('Handling %s ...', trade)
(buy, sell) = (False, False)
(enter, exit_) = (False, False)
if (self.config.get('use_sell_signal', True) or
self.config.get('ignore_roi_if_buy_signal', False)):
analyzed_df, _ = self.dataprovider.get_analyzed_dataframe(trade.pair,
self.strategy.timeframe)
(buy, sell, _) = self.strategy.get_signal(
(enter, exit_) = self.strategy.get_exit_signal(
trade.pair,
self.strategy.timeframe,
analyzed_df
analyzed_df, is_short=trade.is_short
)
logger.debug('checking sell')
# TODO-lev: side should depend on trade side.
sell_rate = self.exchange.get_rate(trade.pair, refresh=True, side="sell")
if self._check_and_execute_sell(trade, sell_rate, buy, sell):
if self._check_and_execute_exit(trade, sell_rate, enter, exit_):
return True
logger.debug('Found no sell signal for %s.', trade)
@ -855,19 +857,19 @@ class FreqtradeBot(LoggingMixin):
logger.warning(f"Could not create trailing stoploss order "
f"for pair {trade.pair}.")
def _check_and_execute_sell(self, trade: Trade, sell_rate: float,
buy: bool, sell: bool) -> bool:
def _check_and_execute_exit(self, trade: Trade, sell_rate: float,
enter: bool, exit_: bool) -> bool:
"""
Check and execute sell
Check and execute trade exit
"""
should_sell = self.strategy.should_sell(
trade, sell_rate, datetime.now(timezone.utc), buy, sell,
should_exit: SellCheckTuple = self.strategy.should_exit(
trade, sell_rate, datetime.now(timezone.utc), enter=enter, exit_=exit_,
force_stoploss=self.edge.stoploss(trade.pair) if self.edge else 0
)
if should_sell.sell_flag:
logger.info(f'Executing Sell for {trade.pair}. Reason: {should_sell.sell_type}')
self.execute_trade_exit(trade, sell_rate, should_sell)
if should_exit.sell_flag:
logger.info(f'Exit for {trade.pair} detected. Reason: {should_exit.sell_type}')
self.execute_trade_exit(trade, sell_rate, should_exit)
return True
return False

91
freqtrade/optimize/backtesting.py
View File

@ -37,13 +37,16 @@ logger = logging.getLogger(__name__)
# Indexes for backtest tuples
DATE_IDX = 0
BUY_IDX = 1
OPEN_IDX = 2
CLOSE_IDX = 3
SELL_IDX = 4
LOW_IDX = 5
HIGH_IDX = 6
BUY_TAG_IDX = 7
OPEN_IDX = 1
HIGH_IDX = 2
LOW_IDX = 3
CLOSE_IDX = 4
LONG_IDX = 5
ELONG_IDX = 6 # Exit long
SHORT_IDX = 7
ESHORT_IDX = 8 # Exit short
BUY_TAG_IDX = 9
SHORT_TAG_IDX = 10
class Backtesting:
@ -65,8 +68,8 @@ class Backtesting:
remove_credentials(self.config)
self.strategylist: List[IStrategy] = []
self.all_results: Dict[str, Dict] = {}
self.exchange = ExchangeResolver.load_exchange(self.config['exchange']['name'], self.config)
self._exchange_name = self.config['exchange']['name']
self.exchange = ExchangeResolver.load_exchange(self._exchange_name, self.config)
self.dataprovider = DataProvider(self.config, None)
if self.config.get('strategy_list', None):
@ -246,7 +249,8 @@ class Backtesting:
"""
# Every change to this headers list must evaluate further usages of the resulting tuple
# and eventually change the constants for indexes at the top
headers = ['date', 'buy', 'open', 'close', 'sell', 'low', 'high', 'buy_tag']
headers = ['date', 'open', 'high', 'low', 'close', 'enter_long', 'exit_long',
'enter_short', 'exit_short', 'long_tag', 'short_tag']
data: Dict = {}
self.progress.init_step(BacktestState.CONVERT, len(processed))
@ -254,10 +258,18 @@ class Backtesting:
for pair, pair_data in processed.items():
self.check_abort()
self.progress.increment()
if not pair_data.empty:
pair_data.loc[:, 'buy'] = 0 # cleanup if buy_signal is exist
pair_data.loc[:, 'sell'] = 0 # cleanup if sell_signal is exist
pair_data.loc[:, 'buy_tag'] = None # cleanup if buy_tag is exist
# Cleanup from prior runs
# TODO-lev: The below is not 100% compatible with the interface compatibility layer
if 'enter_long' in pair_data.columns:
pair_data.loc[:, 'enter_long'] = 0
pair_data.loc[:, 'enter_short'] = 0
if 'exit_long' in pair_data.columns:
pair_data.loc[:, 'exit_long'] = 0
pair_data.loc[:, 'exit_short'] = 0
pair_data.loc[:, 'long_tag'] = None
pair_data.loc[:, 'short_tag'] = None
df_analyzed = self.strategy.advise_sell(
self.strategy.advise_buy(pair_data, {'pair': pair}),
@ -268,9 +280,11 @@ class Backtesting:
startup_candles=self.required_startup)
# To avoid using data from future, we use buy/sell signals shifted
# from the previous candle
df_analyzed.loc[:, 'buy'] = df_analyzed.loc[:, 'buy'].shift(1)
df_analyzed.loc[:, 'sell'] = df_analyzed.loc[:, 'sell'].shift(1)
df_analyzed.loc[:, 'buy_tag'] = df_analyzed.loc[:, 'buy_tag'].shift(1)
df_analyzed.loc[:, 'enter_long'] = df_analyzed.loc[:, 'enter_long'].shift(1)
df_analyzed.loc[:, 'enter_short'] = df_analyzed.loc[:, 'enter_short'].shift(1)
df_analyzed.loc[:, 'exit_long'] = df_analyzed.loc[:, 'exit_long'].shift(1)
df_analyzed.loc[:, 'exit_short'] = df_analyzed.loc[:, 'exit_short'].shift(1)
df_analyzed.loc[:, 'long_tag'] = df_analyzed.loc[:, 'long_tag'].shift(1)
# Update dataprovider cache
self.dataprovider._set_cached_df(pair, self.timeframe, df_analyzed)
@ -351,10 +365,13 @@ class Backtesting:
def _get_sell_trade_entry_for_candle(self, trade: LocalTrade,
sell_row: Tuple) -> Optional[LocalTrade]:
sell_candle_time = sell_row[DATE_IDX].to_pydatetime()
sell = self.strategy.should_sell(trade, sell_row[OPEN_IDX], # type: ignore
sell_candle_time, sell_row[BUY_IDX],
sell_row[SELL_IDX],
low=sell_row[LOW_IDX], high=sell_row[HIGH_IDX])
enter = sell_row[SHORT_IDX] if trade.is_short else sell_row[LONG_IDX]
exit_ = sell_row[ESHORT_IDX] if trade.is_short else sell_row[ELONG_IDX]
sell = self.strategy.should_exit(
trade, sell_row[OPEN_IDX], sell_candle_time, # type: ignore
enter=enter, exit_=exit_,
low=sell_row[LOW_IDX], high=sell_row[HIGH_IDX]
)
if sell.sell_flag:
trade.close_date = sell_candle_time
@ -390,9 +407,12 @@ class Backtesting:
if len(detail_data) == 0:
# Fall back to "regular" data if no detail data was found for this candle
return self._get_sell_trade_entry_for_candle(trade, sell_row)
detail_data['buy'] = sell_row[BUY_IDX]
detail_data['sell'] = sell_row[SELL_IDX]
headers = ['date', 'buy', 'open', 'close', 'sell', 'low', 'high']
detail_data['enter_long'] = sell_row[LONG_IDX]
detail_data['exit_long'] = sell_row[ELONG_IDX]
detail_data['enter_short'] = sell_row[SHORT_IDX]
detail_data['exit_short'] = sell_row[ESHORT_IDX]
headers = ['date', 'open', 'high', 'low', 'close', 'enter_long', 'exit_long',
'enter_short', 'exit_short']
for det_row in detail_data[headers].values.tolist():
res = self._get_sell_trade_entry_for_candle(trade, det_row)
if res:
@ -403,7 +423,7 @@ class Backtesting:
else:
return self._get_sell_trade_entry_for_candle(trade, sell_row)
def _enter_trade(self, pair: str, row: List) -> Optional[LocalTrade]:
def _enter_trade(self, pair: str, row: List, direction: str) -> Optional[LocalTrade]:
try:
stake_amount = self.wallets.get_trade_stake_amount(pair, None)
except DependencyException:
@ -442,7 +462,8 @@ class Backtesting:
fee_close=self.fee,
is_open=True,
buy_tag=row[BUY_TAG_IDX] if has_buy_tag else None,
exchange='backtesting',
exchange=self._exchange_name,
is_short=(direction == 'short'),
)
return trade
return None
@ -476,6 +497,20 @@ class Backtesting:
self.rejected_trades += 1
return False
def check_for_trade_entry(self, row) -> Optional[str]:
enter_long = row[LONG_IDX] == 1
exit_long = row[ELONG_IDX] == 1
enter_short = row[SHORT_IDX] == 1
exit_short = row[ESHORT_IDX] == 1
if enter_long == 1 and not any([exit_long, enter_short]):
# Long
return 'long'
if enter_short == 1 and not any([exit_short, enter_long]):
# Short
return 'short'
return None
def backtest(self, processed: Dict,
start_date: datetime, end_date: datetime,
max_open_trades: int = 0, position_stacking: bool = False,
@ -538,15 +573,15 @@ class Backtesting:
# without positionstacking, we can only have one open trade per pair.
# max_open_trades must be respected
# don't open on the last row
trade_dir = self.check_for_trade_entry(row)
if (
(position_stacking or len(open_trades[pair]) == 0)
and self.trade_slot_available(max_open_trades, open_trade_count_start)
and tmp != end_date
and row[BUY_IDX] == 1
and row[SELL_IDX] != 1
and trade_dir is not None
and not PairLocks.is_pair_locked(pair, row[DATE_IDX])
):
trade = self._enter_trade(pair, row)
trade = self._enter_trade(pair, row, trade_dir)
if trade:
# TODO: hacky workaround to avoid opening > max_open_trades
# This emulates previous behaviour - not sure if this is correct

9
freqtrade/plot/plotting.py
View File

@ -386,8 +386,9 @@ def generate_candlestick_graph(pair: str, data: pd.DataFrame, trades: pd.DataFra
)
fig.add_trace(candles, 1, 1)
if 'buy' in data.columns:
df_buy = data[data['buy'] == 1]
# TODO-lev: Needs short equivalent
if 'enter_long' in data.columns:
df_buy = data[data['enter_long'] == 1]
if len(df_buy) > 0:
buys = go.Scatter(
x=df_buy.date,
@ -405,8 +406,8 @@ def generate_candlestick_graph(pair: str, data: pd.DataFrame, trades: pd.DataFra
else:
logger.warning("No buy-signals found.")
if 'sell' in data.columns:
df_sell = data[data['sell'] == 1]
if 'exit_long' in data.columns:
df_sell = data[data['exit_long'] == 1]
if len(df_sell) > 0:
sells = go.Scatter(
x=df_sell.date,

1
freqtrade/resolvers/hyperopt_resolver.py
View File

@ -51,6 +51,7 @@ class HyperOptResolver(IResolver):
if not hasattr(hyperopt, 'populate_sell_trend'):
logger.info("Hyperopt class does not provide populate_sell_trend() method. "
"Using populate_sell_trend from the strategy.")
# TODO-lev: Short equivelents?
return hyperopt

166
freqtrade/strategy/interface.py
View File

@ -13,7 +13,7 @@ from pandas import DataFrame
from freqtrade.constants import ListPairsWithTimeframes
from freqtrade.data.dataprovider import DataProvider
from freqtrade.enums import SellType, SignalTagType, SignalType
from freqtrade.enums import SellType, SignalDirection, SignalTagType, SignalType
from freqtrade.exceptions import OperationalException, StrategyError
from freqtrade.exchange import timeframe_to_minutes, timeframe_to_seconds
from freqtrade.exchange.exchange import timeframe_to_next_date
@ -166,7 +166,7 @@ class IStrategy(ABC, HyperStrategyMixin):
def check_buy_timeout(self, pair: str, trade: Trade, order: dict, **kwargs) -> bool:
"""
Check buy enter timeout function callback.
Check buy timeout function callback.
This method can be used to override the enter-timeout.
It is called whenever a limit buy/short order has been created,
and is not yet fully filled.
@ -209,6 +209,7 @@ class IStrategy(ABC, HyperStrategyMixin):
"""
pass
# TODO-lev: add side
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, **kwargs) -> bool:
"""
@ -343,6 +344,7 @@ class IStrategy(ABC, HyperStrategyMixin):
"""
return None
# TODO-lev: add side
def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
proposed_stake: float, min_stake: float, max_stake: float,
**kwargs) -> float:
@ -461,7 +463,10 @@ class IStrategy(ABC, HyperStrategyMixin):
logger.debug("Skipping TA Analysis for already analyzed candle")
dataframe['buy'] = 0
dataframe['sell'] = 0
dataframe['enter_short'] = 0
dataframe['exit_short'] = 0
dataframe['buy_tag'] = None
dataframe['short_tag'] = None
# Other Defs in strategy that want to be called every loop here
# twitter_sell = self.watch_twitter_feed(dataframe, metadata)
@ -521,6 +526,7 @@ class IStrategy(ABC, HyperStrategyMixin):
if dataframe is None:
message = "No dataframe returned (return statement missing?)."
elif 'buy' not in dataframe:
# TODO-lev: Something?
message = "Buy column not set."
elif df_len != len(dataframe):
message = message_template.format("length")
@ -534,25 +540,22 @@ class IStrategy(ABC, HyperStrategyMixin):
else:
raise StrategyError(message)
def get_signal(
def get_latest_candle(
self,
pair: str,
timeframe: str,
dataframe: DataFrame
) -> Tuple[bool, bool, Optional[str]]:
dataframe: DataFrame,
) -> Tuple[Optional[DataFrame], Optional[arrow.Arrow]]:
"""
Calculates current signal based based on the buy/short or sell/exit_short
columns of the dataframe.
Used by Bot to get the signal to buy, sell, short, or exit_short
Get the latest candle. Used only during real mode
:param pair: pair in format ANT/BTC
:param timeframe: timeframe to use
:param dataframe: Analyzed dataframe to get signal from.
:return: (Buy, Sell)/(Short, Exit_short) A bool-tuple indicating
(buy/sell)/(short/exit_short) signal
:return: (None, None) or (Dataframe, latest_date) - corresponding to the last candle
"""
if not isinstance(dataframe, DataFrame) or dataframe.empty:
logger.warning(f'Empty candle (OHLCV) data for pair {pair}')
return False, False, None
return None, None
latest_date = dataframe['date'].max()
latest = dataframe.loc[dataframe['date'] == latest_date].iloc[-1]
@ -567,27 +570,89 @@ class IStrategy(ABC, HyperStrategyMixin):
'Outdated history for pair %s. Last tick is %s minutes old',
pair, int((arrow.utcnow() - latest_date).total_seconds() // 60)
)
return False, False, None
return None, None
return latest, latest_date
enter = latest[SignalType.BUY.value] == 1
def get_exit_signal(
self,
pair: str,
timeframe: str,
dataframe: DataFrame,
is_short: bool = None
) -> Tuple[bool, bool]:
"""
Calculates current exit signal based based on the buy/short or sell/exit_short
columns of the dataframe.
Used by Bot to get the signal to exit.
depending on is_short, looks at "short" or "long" columns.
:param pair: pair in format ANT/BTC
:param timeframe: timeframe to use
:param dataframe: Analyzed dataframe to get signal from.
:param is_short: Indicating existing trade direction.
:return: (enter, exit) A bool-tuple with enter / exit values.
"""
latest, latest_date = self.get_latest_candle(pair, timeframe, dataframe)
if latest is None:
return False, False
exit = False
if SignalType.SELL.value in latest:
exit = latest[SignalType.SELL.value] == 1
if is_short:
enter = latest.get(SignalType.ENTER_SHORT.value, 0) == 1
exit_ = latest.get(SignalType.EXIT_SHORT.value, 0) == 1
else:
enter = latest[SignalType.ENTER_LONG.value] == 1
exit_ = latest.get(SignalType.EXIT_LONG.value, 0) == 1
buy_tag = latest.get(SignalTagType.BUY_TAG.value, None)
logger.debug(f"exit-trigger: {latest['date']} (pair={pair}) "
f"enter={enter} exit={exit_}")
return enter, exit_
def get_entry_signal(
self,
pair: str,
timeframe: str,
dataframe: DataFrame,
) -> Tuple[Optional[SignalDirection], Optional[str]]:
"""
Calculates current entry signal based based on the buy/short or sell/exit_short
columns of the dataframe.
Used by Bot to get the signal to buy, sell, short, or exit_short
:param pair: pair in format ANT/BTC
:param timeframe: timeframe to use
:param dataframe: Analyzed dataframe to get signal from.
:return: (SignalDirection, entry_tag)
"""
latest, latest_date = self.get_latest_candle(pair, timeframe, dataframe)
if latest is None or latest_date is None:
return None, None
enter_long = latest[SignalType.ENTER_LONG.value] == 1
exit_long = latest.get(SignalType.EXIT_LONG.value, 0) == 1
enter_short = latest.get(SignalType.ENTER_SHORT.value, 0) == 1
exit_short = latest.get(SignalType.EXIT_SHORT.value, 0) == 1
enter_signal: Optional[SignalDirection] = None
enter_tag_value: Optional[str] = None
if enter_long == 1 and not any([exit_long, enter_short]):
enter_signal = SignalDirection.LONG
enter_tag_value = latest.get(SignalTagType.BUY_TAG.value, None)
if enter_short == 1 and not any([exit_short, enter_long]):
enter_signal = SignalDirection.SHORT
enter_tag_value = latest.get(SignalTagType.SHORT_TAG.value, None)
logger.debug('trigger: %s (pair=%s) buy=%s sell=%s',
latest['date'], pair, str(enter), str(exit))
timeframe_seconds = timeframe_to_seconds(timeframe)
if self.ignore_expired_candle(
latest_date=latest_date,
latest_date=latest_date.datetime,
current_time=datetime.now(timezone.utc),
timeframe_seconds=timeframe_seconds,
enter=enter
enter=bool(enter_signal)
):
return False, exit, buy_tag
return enter, exit, buy_tag
return None, enter_tag_value
logger.debug(f"entry trigger: {latest['date']} (pair={pair}) "
f"enter={enter_long} enter_tag_value={enter_tag_value}")
return enter_signal, enter_tag_value
def ignore_expired_candle(
self,
@ -602,8 +667,9 @@ class IStrategy(ABC, HyperStrategyMixin):
else:
return False
def should_sell(self, trade: Trade, rate: float, date: datetime, buy: bool,
sell: bool, low: float = None, high: float = None,
def should_exit(self, trade: Trade, rate: float, date: datetime, *,
enter: bool, exit_: bool,
low: float = None, high: float = None,
force_stoploss: float = 0) -> SellCheckTuple:
"""
This function evaluates if one of the conditions required to trigger a sell/exit_short
@ -613,6 +679,7 @@ class IStrategy(ABC, HyperStrategyMixin):
:param force_stoploss: Externally provided stoploss
:return: True if trade should be exited, False otherwise
"""
current_rate = rate
current_profit = trade.calc_profit_ratio(current_rate)
@ -627,7 +694,7 @@ class IStrategy(ABC, HyperStrategyMixin):
current_profit = trade.calc_profit_ratio(current_rate)
# if enter signal and ignore_roi is set, we don't need to evaluate min_roi.
roi_reached = (not (buy and self.ignore_roi_if_buy_signal)
roi_reached = (not (enter and self.ignore_roi_if_buy_signal)
and self.min_roi_reached(trade=trade, current_profit=current_profit,
current_time=date))
@ -640,10 +707,11 @@ class IStrategy(ABC, HyperStrategyMixin):
if (self.sell_profit_only and current_profit <= self.sell_profit_offset):
# sell_profit_only and profit doesn't reach the offset - ignore sell signal
pass
elif self.use_sell_signal and not buy:
if sell:
elif self.use_sell_signal and not enter:
if exit_:
sell_signal = SellType.SELL_SIGNAL
else:
trade_type = "exit_short" if trade.is_short else "sell"
custom_reason = strategy_safe_wrapper(self.custom_sell, default_retval=False)(
pair=trade.pair, trade=trade, current_time=date, current_rate=current_rate,
current_profit=current_profit)
@ -651,9 +719,9 @@ class IStrategy(ABC, HyperStrategyMixin):
sell_signal = SellType.CUSTOM_SELL
if isinstance(custom_reason, str):
if len(custom_reason) > CUSTOM_SELL_MAX_LENGTH:
logger.warning(f'Custom sell reason returned from custom_sell is too '
f'long and was trimmed to {CUSTOM_SELL_MAX_LENGTH} '
f'characters.')
logger.warning(f'Custom {trade_type} reason returned from '
f'custom_{trade_type} is too long and was trimmed'
f'to {CUSTOM_SELL_MAX_LENGTH} characters.')
custom_reason = custom_reason[:CUSTOM_SELL_MAX_LENGTH]
else:
custom_reason = None
@ -699,7 +767,12 @@ class IStrategy(ABC, HyperStrategyMixin):
# Initiate stoploss with open_rate. Does nothing if stoploss is already set.
trade.adjust_stop_loss(trade.open_rate, stop_loss_value, initial=True)
if self.use_custom_stoploss and trade.stop_loss < (low or current_rate):
dir_correct = (trade.stop_loss < (low or current_rate)
if not trade.is_short else
trade.stop_loss > (high or current_rate)
)
if self.use_custom_stoploss and dir_correct:
stop_loss_value = strategy_safe_wrapper(self.custom_stoploss, default_retval=None
)(pair=trade.pair, trade=trade,
current_time=current_time,
@ -717,6 +790,7 @@ class IStrategy(ABC, HyperStrategyMixin):
sl_offset = self.trailing_stop_positive_offset
# Make sure current_profit is calculated using high for backtesting.
# TODO-lev: Check this function - high / low usage must be inversed for short trades!
high_profit = current_profit if not high else trade.calc_profit_ratio(high)
# Don't update stoploss if trailing_only_offset_is_reached is true.
@ -825,7 +899,11 @@ class IStrategy(ABC, HyperStrategyMixin):
"the current function headers!", DeprecationWarning)
return self.populate_buy_trend(dataframe) # type: ignore
else:
return self.populate_buy_trend(dataframe, metadata)
df = self.populate_buy_trend(dataframe, metadata)
if 'enter_long' not in df.columns:
df = df.rename({'buy': 'enter_long', 'buy_tag': 'long_tag'}, axis='columns')
return df
def advise_sell(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
@ -843,4 +921,24 @@ class IStrategy(ABC, HyperStrategyMixin):
"the current function headers!", DeprecationWarning)
return self.populate_sell_trend(dataframe) # type: ignore
else:
return self.populate_sell_trend(dataframe, metadata)
df = self.populate_sell_trend(dataframe, metadata)
if 'exit_long' not in df.columns:
df = df.rename({'sell': 'exit_long'}, axis='columns')
return df
def leverage(self, pair: str, current_time: datetime, current_rate: float,
proposed_leverage: float, max_leverage: float, side: str,
**kwargs) -> float:
"""
Customize leverage for each new trade. This method is not called when edge module is
enabled.
:param pair: Pair that's currently analyzed
:param current_time: datetime object, containing the current datetime
:param current_rate: Rate, calculated based on pricing settings in ask_strategy.
:param proposed_leverage: A leverage proposed by the bot.
:param max_leverage: Max leverage allowed on this pair
:param side: 'long' or 'short' - indicating the direction of the proposed trade
:return: A leverage amount, which is between 1.0 and max_leverage.
"""
return 1.0

16
freqtrade/strategy/strategy_helper.py
View File

@ -1,5 +1,6 @@
import pandas as pd
from freqtrade.exceptions import OperationalException
from freqtrade.exchange import timeframe_to_minutes
@ -58,7 +59,11 @@ def merge_informative_pair(dataframe: pd.DataFrame, informative: pd.DataFrame,
return dataframe
def stoploss_from_open(open_relative_stop: float, current_profit: float) -> float:
def stoploss_from_open(
open_relative_stop: float,
current_profit: float,
for_short: bool = False
) -> float:
"""
Given the current profit, and a desired stop loss value relative to the open price,
@ -79,7 +84,16 @@ def stoploss_from_open(open_relative_stop: float, current_profit: float) -> floa
if current_profit == -1:
return 1
if for_short is True:
# TODO-lev: How would this be calculated for short
raise OperationalException(
"Freqtrade hasn't figured out how to calculated stoploss on shorts")
# stoploss = 1-((1+open_relative_stop)/(1+current_profit))
else:
stoploss = 1-((1+open_relative_stop)/(1+current_profit))
# negative stoploss values indicate the requested stop price is higher than the current price
if for_short:
return min(stoploss, 0.0)
else:
return max(stoploss, 0.0)

379
freqtrade/templates/sample_short_strategy.py Normal file
View File

@ -0,0 +1,379 @@
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
# --- Do not remove these libs ---
import numpy as np # noqa
import pandas as pd # noqa
from pandas import DataFrame
from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter,
IStrategy, IntParameter)
# --------------------------------
# Add your lib to import here
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib
# This class is a sample. Feel free to customize it.
class SampleStrategy(IStrategy):
"""
This is a sample strategy to inspire you.
More information in https://www.freqtrade.io/en/latest/strategy-customization/
You can:
:return: a Dataframe with all mandatory indicators for the strategies
- Rename the class name (Do not forget to update class_name)
- Add any methods you want to build your strategy
- Add any lib you need to build your strategy
You must keep:
- the lib in the section "Do not remove these libs"
- the methods: populate_indicators, populate_buy_trend, populate_sell_trend
You should keep:
- timeframe, minimal_roi, stoploss, trailing_*
"""
# Strategy interface version - allow new iterations of the strategy interface.
# Check the documentation or the Sample strategy to get the latest version.
INTERFACE_VERSION = 2
# Minimal ROI designed for the strategy.
# This attribute will be overridden if the config file contains "minimal_roi".
minimal_roi = {
"60": 0.01,
"30": 0.02,
"0": 0.04
}
# Optimal stoploss designed for the strategy.
# This attribute will be overridden if the config file contains "stoploss".
stoploss = -0.10
# Trailing stoploss
trailing_stop = False
# trailing_only_offset_is_reached = False
# trailing_stop_positive = 0.01
# trailing_stop_positive_offset = 0.0 # Disabled / not configured
# Hyperoptable parameters
short_rsi = IntParameter(low=51, high=100, default=70, space='sell', optimize=True, load=True)
exit_short_rsi = IntParameter(low=1, high=50, default=30, space='buy', optimize=True, load=True)
# Optimal timeframe for the strategy.
timeframe = '5m'
# Run "populate_indicators()" only for new candle.
process_only_new_candles = False
# These values can be overridden in the "ask_strategy" section in the config.
use_sell_signal = True
sell_profit_only = False
ignore_roi_if_buy_signal = False
# Number of candles the strategy requires before producing valid signals
startup_candle_count: int = 30
# Optional order type mapping.
order_types = {
'buy': 'limit',
'sell': 'limit',
'stoploss': 'market',
'stoploss_on_exchange': False
}
# Optional order time in force.
order_time_in_force = {
'buy': 'gtc',
'sell': 'gtc'
}
plot_config = {
'main_plot': {
'tema': {},
'sar': {'color': 'white'},
},
'subplots': {
"MACD": {
'macd': {'color': 'blue'},
'macdsignal': {'color': 'orange'},
},
"RSI": {
'rsi': {'color': 'red'},
}
}
}
def informative_pairs(self):
"""
Define additional, informative pair/interval combinations to be cached from the exchange.
These pair/interval combinations are non-tradeable, unless they are part
of the whitelist as well.
For more information, please consult the documentation
:return: List of tuples in the format (pair, interval)
Sample: return [("ETH/USDT", "5m"),
("BTC/USDT", "15m"),
]
"""
return []
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Adds several different TA indicators to the given DataFrame
Performance Note: For the best performance be frugal on the number of indicators
you are using. Let uncomment only the indicator you are using in your strategies
or your hyperopt configuration, otherwise you will waste your memory and CPU usage.
:param dataframe: Dataframe with data from the exchange
:param metadata: Additional information, like the currently traded pair
:return: a Dataframe with all mandatory indicators for the strategies
"""
# Momentum Indicators
# ------------------------------------
# ADX
dataframe['adx'] = ta.ADX(dataframe)
# # Plus Directional Indicator / Movement
# dataframe['plus_dm'] = ta.PLUS_DM(dataframe)
# dataframe['plus_di'] = ta.PLUS_DI(dataframe)
# # Minus Directional Indicator / Movement
# dataframe['minus_dm'] = ta.MINUS_DM(dataframe)
# dataframe['minus_di'] = ta.MINUS_DI(dataframe)
# # Aroon, Aroon Oscillator
# aroon = ta.AROON(dataframe)
# dataframe['aroonup'] = aroon['aroonup']
# dataframe['aroondown'] = aroon['aroondown']
# dataframe['aroonosc'] = ta.AROONOSC(dataframe)
# # Awesome Oscillator
# dataframe['ao'] = qtpylib.awesome_oscillator(dataframe)
# # Keltner Channel
# keltner = qtpylib.keltner_channel(dataframe)
# dataframe["kc_upperband"] = keltner["upper"]
# dataframe["kc_lowerband"] = keltner["lower"]
# dataframe["kc_middleband"] = keltner["mid"]
# dataframe["kc_percent"] = (
# (dataframe["close"] - dataframe["kc_lowerband"]) /
# (dataframe["kc_upperband"] - dataframe["kc_lowerband"])
# )
# dataframe["kc_width"] = (
# (dataframe["kc_upperband"] - dataframe["kc_lowerband"]) / dataframe["kc_middleband"]
# )
# # Ultimate Oscillator
# dataframe['uo'] = ta.ULTOSC(dataframe)
# # Commodity Channel Index: values [Oversold:-100, Overbought:100]
# dataframe['cci'] = ta.CCI(dataframe)
# RSI
dataframe['rsi'] = ta.RSI(dataframe)
# # Inverse Fisher transform on RSI: values [-1.0, 1.0] (https://goo.gl/2JGGoy)
# rsi = 0.1 * (dataframe['rsi'] - 50)
# dataframe['fisher_rsi'] = (np.exp(2 * rsi) - 1) / (np.exp(2 * rsi) + 1)
# # Inverse Fisher transform on RSI normalized: values [0.0, 100.0] (https://goo.gl/2JGGoy)
# dataframe['fisher_rsi_norma'] = 50 * (dataframe['fisher_rsi'] + 1)
# # Stochastic Slow
# stoch = ta.STOCH(dataframe)
# dataframe['slowd'] = stoch['slowd']
# dataframe['slowk'] = stoch['slowk']
# Stochastic Fast
stoch_fast = ta.STOCHF(dataframe)
dataframe['fastd'] = stoch_fast['fastd']
dataframe['fastk'] = stoch_fast['fastk']
# # Stochastic RSI
# Please read https://github.com/freqtrade/freqtrade/issues/2961 before using this.
# STOCHRSI is NOT aligned with tradingview, which may result in non-expected results.
# stoch_rsi = ta.STOCHRSI(dataframe)
# dataframe['fastd_rsi'] = stoch_rsi['fastd']
# dataframe['fastk_rsi'] = stoch_rsi['fastk']
# MACD
macd = ta.MACD(dataframe)
dataframe['macd'] = macd['macd']
dataframe['macdsignal'] = macd['macdsignal']
dataframe['macdhist'] = macd['macdhist']
# MFI
dataframe['mfi'] = ta.MFI(dataframe)
# # ROC
# dataframe['roc'] = ta.ROC(dataframe)
# Overlap Studies
# ------------------------------------
# Bollinger Bands
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2)
dataframe['bb_lowerband'] = bollinger['lower']
dataframe['bb_middleband'] = bollinger['mid']
dataframe['bb_upperband'] = bollinger['upper']
dataframe["bb_percent"] = (
(dataframe["close"] - dataframe["bb_lowerband"]) /
(dataframe["bb_upperband"] - dataframe["bb_lowerband"])
)
dataframe["bb_width"] = (
(dataframe["bb_upperband"] - dataframe["bb_lowerband"]) / dataframe["bb_middleband"]
)
# Bollinger Bands - Weighted (EMA based instead of SMA)
# weighted_bollinger = qtpylib.weighted_bollinger_bands(
# qtpylib.typical_price(dataframe), window=20, stds=2
# )
# dataframe["wbb_upperband"] = weighted_bollinger["upper"]
# dataframe["wbb_lowerband"] = weighted_bollinger["lower"]
# dataframe["wbb_middleband"] = weighted_bollinger["mid"]
# dataframe["wbb_percent"] = (
# (dataframe["close"] - dataframe["wbb_lowerband"]) /
# (dataframe["wbb_upperband"] - dataframe["wbb_lowerband"])
# )
# dataframe["wbb_width"] = (
# (dataframe["wbb_upperband"] - dataframe["wbb_lowerband"]) /
# dataframe["wbb_middleband"]
# )
# # EMA - Exponential Moving Average
# dataframe['ema3'] = ta.EMA(dataframe, timeperiod=3)
# dataframe['ema5'] = ta.EMA(dataframe, timeperiod=5)
# dataframe['ema10'] = ta.EMA(dataframe, timeperiod=10)
# dataframe['ema21'] = ta.EMA(dataframe, timeperiod=21)
# dataframe['ema50'] = ta.EMA(dataframe, timeperiod=50)
# dataframe['ema100'] = ta.EMA(dataframe, timeperiod=100)
# # SMA - Simple Moving Average
# dataframe['sma3'] = ta.SMA(dataframe, timeperiod=3)
# dataframe['sma5'] = ta.SMA(dataframe, timeperiod=5)
# dataframe['sma10'] = ta.SMA(dataframe, timeperiod=10)
# dataframe['sma21'] = ta.SMA(dataframe, timeperiod=21)
# dataframe['sma50'] = ta.SMA(dataframe, timeperiod=50)
# dataframe['sma100'] = ta.SMA(dataframe, timeperiod=100)
# Parabolic SAR
dataframe['sar'] = ta.SAR(dataframe)
# TEMA - Triple Exponential Moving Average
dataframe['tema'] = ta.TEMA(dataframe, timeperiod=9)
# Cycle Indicator
# ------------------------------------
# Hilbert Transform Indicator - SineWave
hilbert = ta.HT_SINE(dataframe)
dataframe['htsine'] = hilbert['sine']
dataframe['htleadsine'] = hilbert['leadsine']
# Pattern Recognition - Bullish candlestick patterns
# ------------------------------------
# # Hammer: values [0, 100]
# dataframe['CDLHAMMER'] = ta.CDLHAMMER(dataframe)
# # Inverted Hammer: values [0, 100]
# dataframe['CDLINVERTEDHAMMER'] = ta.CDLINVERTEDHAMMER(dataframe)
# # Dragonfly Doji: values [0, 100]
# dataframe['CDLDRAGONFLYDOJI'] = ta.CDLDRAGONFLYDOJI(dataframe)
# # Piercing Line: values [0, 100]
# dataframe['CDLPIERCING'] = ta.CDLPIERCING(dataframe) # values [0, 100]
# # Morningstar: values [0, 100]
# dataframe['CDLMORNINGSTAR'] = ta.CDLMORNINGSTAR(dataframe) # values [0, 100]
# # Three White Soldiers: values [0, 100]
# dataframe['CDL3WHITESOLDIERS'] = ta.CDL3WHITESOLDIERS(dataframe) # values [0, 100]
# Pattern Recognition - Bearish candlestick patterns
# ------------------------------------
# # Hanging Man: values [0, 100]
# dataframe['CDLHANGINGMAN'] = ta.CDLHANGINGMAN(dataframe)
# # Shooting Star: values [0, 100]
# dataframe['CDLSHOOTINGSTAR'] = ta.CDLSHOOTINGSTAR(dataframe)
# # Gravestone Doji: values [0, 100]
# dataframe['CDLGRAVESTONEDOJI'] = ta.CDLGRAVESTONEDOJI(dataframe)
# # Dark Cloud Cover: values [0, 100]
# dataframe['CDLDARKCLOUDCOVER'] = ta.CDLDARKCLOUDCOVER(dataframe)
# # Evening Doji Star: values [0, 100]
# dataframe['CDLEVENINGDOJISTAR'] = ta.CDLEVENINGDOJISTAR(dataframe)
# # Evening Star: values [0, 100]
# dataframe['CDLEVENINGSTAR'] = ta.CDLEVENINGSTAR(dataframe)
# Pattern Recognition - Bullish/Bearish candlestick patterns
# ------------------------------------
# # Three Line Strike: values [0, -100, 100]
# dataframe['CDL3LINESTRIKE'] = ta.CDL3LINESTRIKE(dataframe)
# # Spinning Top: values [0, -100, 100]
# dataframe['CDLSPINNINGTOP'] = ta.CDLSPINNINGTOP(dataframe) # values [0, -100, 100]
# # Engulfing: values [0, -100, 100]
# dataframe['CDLENGULFING'] = ta.CDLENGULFING(dataframe) # values [0, -100, 100]
# # Harami: values [0, -100, 100]
# dataframe['CDLHARAMI'] = ta.CDLHARAMI(dataframe) # values [0, -100, 100]
# # Three Outside Up/Down: values [0, -100, 100]
# dataframe['CDL3OUTSIDE'] = ta.CDL3OUTSIDE(dataframe) # values [0, -100, 100]
# # Three Inside Up/Down: values [0, -100, 100]
# dataframe['CDL3INSIDE'] = ta.CDL3INSIDE(dataframe) # values [0, -100, 100]
# # Chart type
# # ------------------------------------
# # Heikin Ashi Strategy
# heikinashi = qtpylib.heikinashi(dataframe)
# dataframe['ha_open'] = heikinashi['open']
# dataframe['ha_close'] = heikinashi['close']
# dataframe['ha_high'] = heikinashi['high']
# dataframe['ha_low'] = heikinashi['low']
# Retrieve best bid and best ask from the orderbook
# ------------------------------------
"""
# first check if dataprovider is available
if self.dp:
if self.dp.runmode.value in ('live', 'dry_run'):
ob = self.dp.orderbook(metadata['pair'], 1)
dataframe['best_bid'] = ob['bids'][0][0]
dataframe['best_ask'] = ob['asks'][0][0]
"""
return dataframe
def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the buy signal for the given dataframe
:param dataframe: DataFrame populated with indicators
:param metadata: Additional information, like the currently traded pair
:return: DataFrame with buy column
"""
dataframe.loc[
(
# Signal: RSI crosses above 70
(qtpylib.crossed_above(dataframe['rsi'], self.short_rsi.value)) &
(dataframe['tema'] > dataframe['bb_middleband']) & # Guard: tema above BB middle
(dataframe['tema'] < dataframe['tema'].shift(1)) & # Guard: tema is falling
(dataframe['volume'] > 0) # Make sure Volume is not 0
),
'enter_short'] = 1
return dataframe
def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the sell signal for the given dataframe
:param dataframe: DataFrame populated with indicators
:param metadata: Additional information, like the currently traded pair
:return: DataFrame with sell column
"""
dataframe.loc[
(
# Signal: RSI crosses above 30
(qtpylib.crossed_above(dataframe['rsi'], self.exit_short_rsi.value)) &
# Guard: tema below BB middle
(dataframe['tema'] <= dataframe['bb_middleband']) &
(dataframe['tema'] > dataframe['tema'].shift(1)) & # Guard: tema is raising
(dataframe['volume'] > 0) # Make sure Volume is not 0
),
'exit_short'] = 1
return dataframe

28
tests/conftest.py
View File

@ -6,6 +6,7 @@ from copy import deepcopy
from datetime import datetime, timedelta
from functools import reduce
from pathlib import Path
from typing import Optional
from unittest.mock import MagicMock, Mock, PropertyMock
import arrow
@ -18,6 +19,7 @@ from freqtrade.commands import Arguments
from freqtrade.data.converter import ohlcv_to_dataframe
from freqtrade.edge import Edge, PairInfo
from freqtrade.enums import RunMode
from freqtrade.enums.signaltype import SignalDirection
from freqtrade.exchange import Exchange
from freqtrade.freqtradebot import FreqtradeBot
from freqtrade.persistence import LocalTrade, Trade, init_db
@ -182,13 +184,35 @@ def get_patched_worker(mocker, config) -> Worker:
return Worker(args=None, config=config)
def patch_get_signal(freqtrade: FreqtradeBot, value=(True, False, None)) -> None:
def patch_get_signal(freqtrade: FreqtradeBot, enter_long=True, exit_long=False,
enter_short=False, exit_short=False, enter_tag: Optional[str] = None) -> None:
"""
:param mocker: mocker to patch IStrategy class
:param value: which value IStrategy.get_signal() must return
(buy, sell, buy_tag)
:return: None
"""
freqtrade.strategy.get_signal = lambda e, s, x: value
# returns (Signal-direction, signaname)
def patched_get_entry_signal(*args, **kwargs):
direction = None
if enter_long and not any([exit_long, enter_short]):
direction = SignalDirection.LONG
if enter_short and not any([exit_short, enter_long]):
direction = SignalDirection.SHORT
return direction, enter_tag
freqtrade.strategy.get_entry_signal = patched_get_entry_signal
def patched_get_exit_signal(pair, timeframe, dataframe, is_short):
if is_short:
return enter_short, exit_short
else:
return enter_long, exit_long
# returns (enter, exit)
freqtrade.strategy.get_exit_signal = patched_get_exit_signal
freqtrade.exchange.refresh_latest_ohlcv = lambda p: None

14
tests/optimize/__init__.py
View File

@ -44,14 +44,20 @@ def _get_frame_time_from_offset(offset):
def _build_backtest_dataframe(data):
columns = ['date', 'open', 'high', 'low', 'close', 'volume', 'buy', 'sell']
columns = columns + ['buy_tag'] if len(data[0]) == 9 else columns
columns = ['date', 'open', 'high', 'low', 'close', 'volume', 'enter_long', 'exit_long',
'enter_short', 'exit_short']
if len(data[0]) == 8:
# No short columns
data = [d + [0, 0] for d in data]
columns = columns + ['long_tag'] if len(data[0]) == 11 else columns
frame = DataFrame.from_records(data, columns=columns)
frame['date'] = frame['date'].apply(_get_frame_time_from_offset)
# Ensure floats are in place
for column in ['open', 'high', 'low', 'close', 'volume']:
frame[column] = frame[column].astype('float64')
if 'buy_tag' not in columns:
frame['buy_tag'] = None
if 'long_tag' not in columns:
frame['long_tag'] = None
if 'short_tag' not in columns:
frame['short_tag'] = None
return frame

271
tests/optimize/hyperopts/short_hyperopt.py Normal file
View File

@ -0,0 +1,271 @@
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
from functools import reduce
from typing import Any, Callable, Dict, List
import talib.abstract as ta
from pandas import DataFrame
from skopt.space import Categorical, Dimension, Integer
import freqtrade.vendor.qtpylib.indicators as qtpylib
from freqtrade.optimize.hyperopt_interface import IHyperOpt
class DefaultHyperOpt(IHyperOpt):
"""
Default hyperopt provided by the Freqtrade bot.
You can override it with your own Hyperopt
"""
@staticmethod
def populate_indicators(dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Add several indicators needed for buy and sell strategies defined below.
"""
# ADX
dataframe['adx'] = ta.ADX(dataframe)
# MACD
macd = ta.MACD(dataframe)
dataframe['macd'] = macd['macd']
dataframe['macdsignal'] = macd['macdsignal']
# MFI
dataframe['mfi'] = ta.MFI(dataframe)
# RSI
dataframe['rsi'] = ta.RSI(dataframe)
# Stochastic Fast
stoch_fast = ta.STOCHF(dataframe)
dataframe['fastd'] = stoch_fast['fastd']
# Minus-DI
dataframe['minus_di'] = ta.MINUS_DI(dataframe)
# Bollinger bands
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2)
dataframe['bb_lowerband'] = bollinger['lower']
dataframe['bb_upperband'] = bollinger['upper']
# SAR
dataframe['sar'] = ta.SAR(dataframe)
return dataframe
@staticmethod
def buy_strategy_generator(params: Dict[str, Any]) -> Callable:
"""
Define the buy strategy parameters to be used by Hyperopt.
"""
def populate_buy_trend(dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Buy strategy Hyperopt will build and use.
"""
long_conditions = []
short_conditions = []
# GUARDS AND TRENDS
if 'mfi-enabled' in params and params['mfi-enabled']:
long_conditions.append(dataframe['mfi'] < params['mfi-value'])
short_conditions.append(dataframe['mfi'] > params['short-mfi-value'])
if 'fastd-enabled' in params and params['fastd-enabled']:
long_conditions.append(dataframe['fastd'] < params['fastd-value'])
short_conditions.append(dataframe['fastd'] > params['short-fastd-value'])
if 'adx-enabled' in params and params['adx-enabled']:
long_conditions.append(dataframe['adx'] > params['adx-value'])
short_conditions.append(dataframe['adx'] < params['short-adx-value'])
if 'rsi-enabled' in params and params['rsi-enabled']:
long_conditions.append(dataframe['rsi'] < params['rsi-value'])
short_conditions.append(dataframe['rsi'] > params['short-rsi-value'])
# TRIGGERS
if 'trigger' in params:
if params['trigger'] == 'boll':
long_conditions.append(dataframe['close'] < dataframe['bb_lowerband'])
short_conditions.append(dataframe['close'] > dataframe['bb_upperband'])
if params['trigger'] == 'macd_cross_signal':
long_conditions.append(qtpylib.crossed_above(
dataframe['macd'],
dataframe['macdsignal']
))
short_conditions.append(qtpylib.crossed_below(
dataframe['macd'],
dataframe['macdsignal']
))
if params['trigger'] == 'sar_reversal':
long_conditions.append(qtpylib.crossed_above(
dataframe['close'],
dataframe['sar']
))
short_conditions.append(qtpylib.crossed_below(
dataframe['close'],
dataframe['sar']
))
if long_conditions:
dataframe.loc[
reduce(lambda x, y: x & y, long_conditions),
'buy'] = 1
if short_conditions:
dataframe.loc[
reduce(lambda x, y: x & y, short_conditions),
'enter_short'] = 1
return dataframe
return populate_buy_trend
@staticmethod
def indicator_space() -> List[Dimension]:
"""
Define your Hyperopt space for searching buy strategy parameters.
"""
return [
Integer(10, 25, name='mfi-value'),
Integer(15, 45, name='fastd-value'),
Integer(20, 50, name='adx-value'),
Integer(20, 40, name='rsi-value'),
Integer(75, 90, name='short-mfi-value'),
Integer(55, 85, name='short-fastd-value'),
Integer(50, 80, name='short-adx-value'),
Integer(60, 80, name='short-rsi-value'),
Categorical([True, False], name='mfi-enabled'),
Categorical([True, False], name='fastd-enabled'),
Categorical([True, False], name='adx-enabled'),
Categorical([True, False], name='rsi-enabled'),
Categorical(['boll', 'macd_cross_signal', 'sar_reversal'], name='trigger')
]
@staticmethod
def sell_strategy_generator(params: Dict[str, Any]) -> Callable:
"""
Define the sell strategy parameters to be used by Hyperopt.
"""
def populate_sell_trend(dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Sell strategy Hyperopt will build and use.
"""
exit_long_conditions = []
exit_short_conditions = []
# GUARDS AND TRENDS
if 'sell-mfi-enabled' in params and params['sell-mfi-enabled']:
exit_long_conditions.append(dataframe['mfi'] > params['sell-mfi-value'])
exit_short_conditions.append(dataframe['mfi'] < params['exit-short-mfi-value'])
if 'sell-fastd-enabled' in params and params['sell-fastd-enabled']:
exit_long_conditions.append(dataframe['fastd'] > params['sell-fastd-value'])
exit_short_conditions.append(dataframe['fastd'] < params['exit-short-fastd-value'])
if 'sell-adx-enabled' in params and params['sell-adx-enabled']:
exit_long_conditions.append(dataframe['adx'] < params['sell-adx-value'])
exit_short_conditions.append(dataframe['adx'] > params['exit-short-adx-value'])
if 'sell-rsi-enabled' in params and params['sell-rsi-enabled']:
exit_long_conditions.append(dataframe['rsi'] > params['sell-rsi-value'])
exit_short_conditions.append(dataframe['rsi'] < params['exit-short-rsi-value'])
# TRIGGERS
if 'sell-trigger' in params:
if params['sell-trigger'] == 'sell-boll':
exit_long_conditions.append(dataframe['close'] > dataframe['bb_upperband'])
exit_short_conditions.append(dataframe['close'] < dataframe['bb_lowerband'])
if params['sell-trigger'] == 'sell-macd_cross_signal':
exit_long_conditions.append(qtpylib.crossed_above(
dataframe['macdsignal'],
dataframe['macd']
))
exit_short_conditions.append(qtpylib.crossed_below(
dataframe['macdsignal'],
dataframe['macd']
))
if params['sell-trigger'] == 'sell-sar_reversal':
exit_long_conditions.append(qtpylib.crossed_above(
dataframe['sar'],
dataframe['close']
))
exit_short_conditions.append(qtpylib.crossed_below(
dataframe['sar'],
dataframe['close']
))
if exit_long_conditions:
dataframe.loc[
reduce(lambda x, y: x & y, exit_long_conditions),
'sell'] = 1
if exit_short_conditions:
dataframe.loc[
reduce(lambda x, y: x & y, exit_short_conditions),
'exit-short'] = 1
return dataframe
return populate_sell_trend
@staticmethod
def sell_indicator_space() -> List[Dimension]:
"""
Define your Hyperopt space for searching sell strategy parameters.
"""
return [
Integer(75, 100, name='sell-mfi-value'),
Integer(50, 100, name='sell-fastd-value'),
Integer(50, 100, name='sell-adx-value'),
Integer(60, 100, name='sell-rsi-value'),
Integer(1, 25, name='exit-short-mfi-value'),
Integer(1, 50, name='exit-short-fastd-value'),
Integer(1, 50, name='exit-short-adx-value'),
Integer(1, 40, name='exit-short-rsi-value'),
Categorical([True, False], name='sell-mfi-enabled'),
Categorical([True, False], name='sell-fastd-enabled'),
Categorical([True, False], name='sell-adx-enabled'),
Categorical([True, False], name='sell-rsi-enabled'),
Categorical(['sell-boll',
'sell-macd_cross_signal',
'sell-sar_reversal'],
name='sell-trigger')
]
def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators. Should be a copy of same method from strategy.
Must align to populate_indicators in this file.
Only used when --spaces does not include buy space.
"""
dataframe.loc[
(
(dataframe['close'] < dataframe['bb_lowerband']) &
(dataframe['mfi'] < 16) &
(dataframe['adx'] > 25) &
(dataframe['rsi'] < 21)
),
'buy'] = 1
dataframe.loc[
(
(dataframe['close'] > dataframe['bb_upperband']) &
(dataframe['mfi'] < 84) &
(dataframe['adx'] > 75) &
(dataframe['rsi'] < 79)
),
'enter_short'] = 1
return dataframe
def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators. Should be a copy of same method from strategy.
Must align to populate_indicators in this file.
Only used when --spaces does not include sell space.
"""
dataframe.loc[
(
(qtpylib.crossed_above(
dataframe['macdsignal'], dataframe['macd']
)) &
(dataframe['fastd'] > 54)
),
'sell'] = 1
dataframe.loc[
(
(qtpylib.crossed_below(
dataframe['macdsignal'], dataframe['macd']
)) &
(dataframe['fastd'] < 46)
),
'exit_short'] = 1
return dataframe

13
tests/optimize/test_backtest_detail.py
View File

@ -519,12 +519,12 @@ tc32 = BTContainer(data=[
# Test 33: trailing_stop should be triggered immediately on trade open candle.
# stop-loss: 1%, ROI: 10% (should not apply)
tc33 = BTContainer(data=[
# D O H L C V B S BT
[0, 5000, 5050, 4950, 5000, 6172, 1, 0, 'buy_signal_01'],
[1, 5000, 5500, 5000, 4900, 6172, 0, 0, None], # enter trade (signal on last candle) and stop
[2, 4900, 5250, 4500, 5100, 6172, 0, 0, None],
[3, 5100, 5100, 4650, 4750, 6172, 0, 0, None],
[4, 4750, 4950, 4350, 4750, 6172, 0, 0, None]],
# D O H L C V EL XL ES Xs BT
[0, 5000, 5050, 4950, 5000, 6172, 1, 0, 0, 0, 'buy_signal_01'],
[1, 5000, 5500, 5000, 4900, 6172, 0, 0, 0, 0, None], # enter trade and stop
[2, 4900, 5250, 4500, 5100, 6172, 0, 0, 0, 0, None],
[3, 5100, 5100, 4650, 4750, 6172, 0, 0, 0, 0, None],
[4, 4750, 4950, 4350, 4750, 6172, 0, 0, 0, 0, None]],
stop_loss=-0.01, roi={"0": 0.10}, profit_perc=-0.01, trailing_stop=True,
trailing_only_offset_is_reached=True, trailing_stop_positive_offset=0.02,
trailing_stop_positive=0.01, use_custom_stoploss=True,
@ -571,6 +571,7 @@ TESTS = [
tc31,
tc32,
tc33,
# TODO-lev: Add tests for short here
]

86
tests/optimize/test_backtesting.py
View File

@ -123,12 +123,14 @@ def _trend(signals, buy_value, sell_value):
n = len(signals['low'])
buy = np.zeros(n)
sell = np.zeros(n)
for i in range(0, len(signals['buy'])):
for i in range(0, len(signals['date'])):
if random.random() > 0.5: # Both buy and sell signals at same timeframe
buy[i] = buy_value
sell[i] = sell_value
signals['buy'] = buy
signals['sell'] = sell
signals['enter_long'] = buy
signals['exit_long'] = sell
signals['enter_short'] = 0
signals['exit_short'] = 0
return signals
@ -143,8 +145,10 @@ def _trend_alternate(dataframe=None, metadata=None):
buy[i] = 1
else:
sell[i] = 1
signals['buy'] = buy
signals['sell'] = sell
signals['enter_long'] = buy
signals['exit_long'] = sell
signals['enter_short'] = 0
signals['exit_short'] = 0
return dataframe
@ -508,41 +512,47 @@ def test_backtest__enter_trade(default_conf, fee, mocker) -> None:
0.0012, # High
'', # Buy Signal Name
]
trade = backtesting._enter_trade(pair, row=row)
trade = backtesting._enter_trade(pair, row=row, direction='long')
assert isinstance(trade, LocalTrade)
assert trade.stake_amount == 495
# Fake 2 trades, so there's not enough amount for the next trade left.
LocalTrade.trades_open.append(trade)
LocalTrade.trades_open.append(trade)
trade = backtesting._enter_trade(pair, row=row)
trade = backtesting._enter_trade(pair, row=row, direction='long')
assert trade is None
LocalTrade.trades_open.pop()
trade = backtesting._enter_trade(pair, row=row)
trade = backtesting._enter_trade(pair, row=row, direction='long')
assert trade is not None
backtesting.strategy.custom_stake_amount = lambda **kwargs: 123.5
trade = backtesting._enter_trade(pair, row=row)
trade = backtesting._enter_trade(pair, row=row, direction='long')
assert trade
assert trade.stake_amount == 123.5
# In case of error - use proposed stake
backtesting.strategy.custom_stake_amount = lambda **kwargs: 20 / 0
trade = backtesting._enter_trade(pair, row=row)
trade = backtesting._enter_trade(pair, row=row, direction='long')
assert trade
assert trade.stake_amount == 495
assert trade.is_short is False
trade = backtesting._enter_trade(pair, row=row, direction='short')
assert trade
assert trade.stake_amount == 495
assert trade.is_short is True
# Stake-amount too high!
mocker.patch("freqtrade.exchange.Exchange.get_min_pair_stake_amount", return_value=600.0)
trade = backtesting._enter_trade(pair, row=row)
trade = backtesting._enter_trade(pair, row=row, direction='long')
assert trade is None
# Stake-amount throwing error
mocker.patch("freqtrade.wallets.Wallets.get_trade_stake_amount",
side_effect=DependencyException)
trade = backtesting._enter_trade(pair, row=row)
trade = backtesting._enter_trade(pair, row=row, direction='long')
assert trade is None
backtesting.cleanup()
@ -560,47 +570,54 @@ def test_backtest__get_sell_trade_entry(default_conf, fee, mocker) -> None:
pair = 'UNITTEST/BTC'
row = [
pd.Timestamp(year=2020, month=1, day=1, hour=4, minute=55, tzinfo=timezone.utc),
1, # Buy
200, # Open
201, # Close
0, # Sell
195, # Low
201.5, # High
'', # Buy Signal Name
195, # Low
201, # Close
1, # enter_long
0, # exit_long
0, # enter_short
0, # exit_hsort
'', # Long Signal Name
'', # Short Signal Name
]
trade = backtesting._enter_trade(pair, row=row)
trade = backtesting._enter_trade(pair, row=row, direction='long')
assert isinstance(trade, LocalTrade)
row_sell = [
pd.Timestamp(year=2020, month=1, day=1, hour=5, minute=0, tzinfo=timezone.utc),
0, # Buy
200, # Open
201, # Close
0, # Sell
195, # Low
210.5, # High
'', # Buy Signal Name
195, # Low
201, # Close
0, # enter_long
0, # exit_long
0, # enter_short
0, # exit_short
'', # long Signal Name
'', # Short Signal Name
]
row_detail = pd.DataFrame(
[
[
pd.Timestamp(year=2020, month=1, day=1, hour=5, minute=0, tzinfo=timezone.utc),
1, 200, 199, 0, 197, 200.1, '',
200, 200.1, 197, 199, 1, 0, 0, 0, '', '',
], [
pd.Timestamp(year=2020, month=1, day=1, hour=5, minute=1, tzinfo=timezone.utc),
0, 199, 199.5, 0, 199, 199.7, '',
199, 199.7, 199, 199.5, 0, 0, 0, 0, '', ''
], [
pd.Timestamp(year=2020, month=1, day=1, hour=5, minute=2, tzinfo=timezone.utc),
0, 199.5, 200.5, 0, 199, 200.8, '',
199.5, 200.8, 199, 200.9, 0, 0, 0, 0, '', ''
], [
pd.Timestamp(year=2020, month=1, day=1, hour=5, minute=3, tzinfo=timezone.utc),
0, 200.5, 210.5, 0, 193, 210.5, '', # ROI sell (?)
200.5, 210.5, 193, 210.5, 0, 0, 0, 0, '', '' # ROI sell (?)
], [
pd.Timestamp(year=2020, month=1, day=1, hour=5, minute=4, tzinfo=timezone.utc),
0, 200, 199, 0, 193, 200.1, '',
200, 200.1, 193, 199, 0, 0, 0, 0, '', ''
],
], columns=["date", "buy", "open", "close", "sell", "low", "high", "buy_tag"]
], columns=['date', 'open', 'high', 'low', 'close', 'enter_long', 'exit_long',
'enter_short', 'exit_short', 'long_tag', 'short_tag']
)
# No data available.
@ -610,11 +627,12 @@ def test_backtest__get_sell_trade_entry(default_conf, fee, mocker) -> None:
assert res.close_date_utc == datetime(2020, 1, 1, 5, 0, tzinfo=timezone.utc)
# Enter new trade
trade = backtesting._enter_trade(pair, row=row)
trade = backtesting._enter_trade(pair, row=row, direction='long')
assert isinstance(trade, LocalTrade)
# Assign empty ... no result.
backtesting.detail_data[pair] = pd.DataFrame(
[], columns=["date", "buy", "open", "close", "sell", "low", "high", "buy_tag"])
[], columns=['date', 'open', 'high', 'low', 'close', 'enter_long', 'exit_long',
'enter_short', 'exit_short', 'long_tag', 'short_tag'])
res = backtesting._get_sell_trade_entry(trade, row)
assert res is None
@ -857,8 +875,10 @@ def test_backtest_multi_pair(default_conf, fee, mocker, tres, pair, testdatadir)
multi = 20
else:
multi = 18
dataframe['buy'] = np.where(dataframe.index % multi == 0, 1, 0)
dataframe['sell'] = np.where((dataframe.index + multi - 2) % multi == 0, 1, 0)
dataframe['enter_long'] = np.where(dataframe.index % multi == 0, 1, 0)
dataframe['exit_long'] = np.where((dataframe.index + multi - 2) % multi == 0, 1, 0)
dataframe['enter_short'] = 0
dataframe['exit_short'] = 0
return dataframe
mocker.patch("freqtrade.exchange.Exchange.get_min_pair_stake_amount", return_value=0.00001)

11
tests/optimize/test_hyperopt.py
View File

@ -25,6 +25,9 @@ from tests.conftest import (get_args, log_has, log_has_re, patch_exchange,
from .hyperopts.hyperopt_test_sep_file import HyperoptTestSepFile
# TODO-lev: This file
def test_setup_hyperopt_configuration_without_arguments(mocker, default_conf, caplog) -> None:
patched_configuration_load_config_file(mocker, default_conf)
@ -448,6 +451,10 @@ def test_buy_strategy_generator(hyperopt, testdatadir) -> None:
'fastd-value': 20,
'mfi-value': 20,
'rsi-value': 20,
'short-adx-value': 80,
'short-fastd-value': 80,
'short-mfi-value': 80,
'short-rsi-value': 80,
'adx-enabled': True,
'fastd-enabled': True,
'mfi-enabled': True,
@ -473,6 +480,10 @@ def test_sell_strategy_generator(hyperopt, testdatadir) -> None:
'sell-fastd-value': 75,
'sell-mfi-value': 80,
'sell-rsi-value': 20,
'exit-short-adx-value': 80,
'exit-short-fastd-value': 25,
'exit-short-mfi-value': 20,
'exit-short-rsi-value': 80,
'sell-adx-enabled': True,
'sell-fastd-enabled': True,
'sell-mfi-enabled': True,

1
tests/strategy/test_default_strategy.py
View File

@ -42,5 +42,6 @@ def test_strategy_test_v2(result, fee):
rate=20000, time_in_force='gtc', sell_reason='roi',
current_time=datetime.utcnow()) is True
# TODO-lev: Test for shorts?
assert strategy.custom_stoploss(pair='ETH/BTC', trade=trade, current_time=datetime.now(),
current_rate=20_000, current_profit=0.05) == strategy.stoploss

97
tests/strategy/test_interface.py
View File

@ -1,4 +1,5 @@
# pragma pylint: disable=missing-docstring, C0103
from freqtrade.enums.signaltype import SignalDirection
import logging
from datetime import datetime, timedelta, timezone
from pathlib import Path
@ -30,28 +31,56 @@ _STRATEGY = StrategyTestV2(config={})
_STRATEGY.dp = DataProvider({}, None, None)
def test_returns_latest_signal(mocker, default_conf, ohlcv_history):
def test_returns_latest_signal(ohlcv_history):
ohlcv_history.loc[1, 'date'] = arrow.utcnow()
# 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, 'sell'] = 1
mocked_history['enter_long'] = 0
mocked_history['exit_long'] = 0
mocked_history['enter_short'] = 0
mocked_history['exit_short'] = 0
mocked_history.loc[1, 'exit_long'] = 1
assert _STRATEGY.get_signal('ETH/BTC', '5m', mocked_history) == (False, True, None)
mocked_history.loc[1, 'sell'] = 0
mocked_history.loc[1, 'buy'] = 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)
assert _STRATEGY.get_exit_signal('ETH/BTC', '5m', mocked_history, True) == (False, False)
mocked_history.loc[1, 'exit_long'] = 0
mocked_history.loc[1, 'enter_long'] = 1
assert _STRATEGY.get_signal('ETH/BTC', '5m', mocked_history) == (True, False, None)
mocked_history.loc[1, 'sell'] = 0
mocked_history.loc[1, 'buy'] = 0
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)
assert _STRATEGY.get_exit_signal('ETH/BTC', '5m', mocked_history, True) == (False, False)
mocked_history.loc[1, 'exit_long'] = 0
mocked_history.loc[1, 'enter_long'] = 0
assert _STRATEGY.get_signal('ETH/BTC', '5m', mocked_history) == (False, False, None)
mocked_history.loc[1, 'sell'] = 0
mocked_history.loc[1, 'buy'] = 1
assert _STRATEGY.get_entry_signal('ETH/BTC', '5m', mocked_history) == (None, None)
assert _STRATEGY.get_exit_signal('ETH/BTC', '5m', mocked_history) == (False, False)
assert _STRATEGY.get_exit_signal('ETH/BTC', '5m', mocked_history, True) == (False, False)
mocked_history.loc[1, 'exit_long'] = 0
mocked_history.loc[1, 'enter_long'] = 1
mocked_history.loc[1, 'buy_tag'] = 'buy_signal_01'
assert _STRATEGY.get_signal('ETH/BTC', '5m', mocked_history) == (True, False, '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)
assert _STRATEGY.get_exit_signal('ETH/BTC', '5m', mocked_history, True) == (False, False)
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
assert _STRATEGY.get_entry_signal(
'ETH/BTC', '5m', mocked_history) == (SignalDirection.SHORT, None)
assert _STRATEGY.get_exit_signal('ETH/BTC', '5m', mocked_history) == (False, False)
assert _STRATEGY.get_exit_signal('ETH/BTC', '5m', mocked_history, True) == (True, False)
mocked_history.loc[1, 'enter_short'] = 0
mocked_history.loc[1, 'exit_short'] = 1
assert _STRATEGY.get_entry_signal(
'ETH/BTC', '5m', mocked_history) == (None, None)
assert _STRATEGY.get_exit_signal('ETH/BTC', '5m', mocked_history) == (False, False)
assert _STRATEGY.get_exit_signal('ETH/BTC', '5m', mocked_history, True) == (False, True)
def test_analyze_pair_empty(default_conf, mocker, caplog, ohlcv_history):
@ -67,18 +96,18 @@ def test_analyze_pair_empty(default_conf, mocker, caplog, ohlcv_history):
assert log_has('Empty dataframe for pair ETH/BTC', caplog)
def test_get_signal_empty(default_conf, mocker, caplog):
assert (False, False, None) == _STRATEGY.get_signal(
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 (False, False, None) == _STRATEGY.get_signal('bar', default_conf['timeframe'], None)
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 (False, False, None) == _STRATEGY.get_signal(
assert (None, None) == _STRATEGY.get_latest_candle(
'baz',
default_conf['timeframe'],
DataFrame([])
@ -86,7 +115,7 @@ def test_get_signal_empty(default_conf, mocker, caplog):
assert log_has('Empty candle (OHLCV) data for pair baz', caplog)
def test_get_signal_exception_valueerror(default_conf, mocker, caplog, ohlcv_history):
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(
@ -111,14 +140,14 @@ def test_get_signal_old_dataframe(default_conf, mocker, caplog, ohlcv_history):
ohlcv_history.loc[1, 'date'] = arrow.utcnow().shift(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
mocked_history['exit_long'] = 0
mocked_history['enter_long'] = 0
mocked_history.loc[1, 'enter_long'] = 1
caplog.set_level(logging.INFO)
mocker.patch.object(_STRATEGY, 'assert_df')
assert (False, False, None) == _STRATEGY.get_signal(
assert (None, None) == _STRATEGY.get_latest_candle(
'xyz',
default_conf['timeframe'],
mocked_history
@ -134,13 +163,13 @@ def test_get_signal_no_sell_column(default_conf, mocker, caplog, ohlcv_history):
mocked_history = ohlcv_history.copy()
# Intentionally don't set sell column
# mocked_history['sell'] = 0
mocked_history['buy'] = 0
mocked_history.loc[1, 'buy'] = 1
mocked_history['enter_long'] = 0
mocked_history.loc[1, 'enter_long'] = 1
caplog.set_level(logging.INFO)
mocker.patch.object(_STRATEGY, 'assert_df')
assert (True, False, None) == _STRATEGY.get_signal(
assert (SignalDirection.LONG, None) == _STRATEGY.get_entry_signal(
'xyz',
default_conf['timeframe'],
mocked_history
@ -452,27 +481,35 @@ def test_custom_sell(default_conf, fee, caplog) -> None:
)
now = arrow.utcnow().datetime
res = strategy.should_sell(trade, 1, now, False, False, None, None, 0)
res = strategy.should_exit(trade, 1, now,
enter=False, exit_=False,
low=None, high=None)
assert res.sell_flag is False
assert res.sell_type == SellType.NONE
strategy.custom_sell = MagicMock(return_value=True)
res = strategy.should_sell(trade, 1, now, False, False, None, None, 0)
res = strategy.should_exit(trade, 1, now,
enter=False, exit_=False,
low=None, high=None)
assert res.sell_flag is True
assert res.sell_type == SellType.CUSTOM_SELL
assert res.sell_reason == 'custom_sell'
strategy.custom_sell = MagicMock(return_value='hello world')
res = strategy.should_sell(trade, 1, now, False, False, None, None, 0)
res = strategy.should_exit(trade, 1, now,
enter=False, exit_=False,
low=None, high=None)
assert res.sell_type == SellType.CUSTOM_SELL
assert res.sell_flag is True
assert res.sell_reason == 'hello world'
caplog.clear()
strategy.custom_sell = MagicMock(return_value='h' * 100)
res = strategy.should_sell(trade, 1, now, False, False, None, None, 0)
res = strategy.should_exit(trade, 1, now,
enter=False, exit_=False,
low=None, high=None)
assert res.sell_type == SellType.CUSTOM_SELL
assert res.sell_flag is True
assert res.sell_reason == 'h' * 64

15
tests/strategy/test_strategy_loading.py
View File

@ -118,10 +118,12 @@ def test_strategy(result, default_conf):
assert 'adx' in df_indicators
dataframe = strategy.advise_buy(df_indicators, metadata=metadata)
assert 'buy' in dataframe.columns
assert 'buy' not in dataframe.columns
assert 'enter_long' in dataframe.columns
dataframe = strategy.advise_sell(df_indicators, metadata=metadata)
assert 'sell' in dataframe.columns
assert 'sell' not in dataframe.columns
assert 'exit_long' in dataframe.columns
def test_strategy_override_minimal_roi(caplog, default_conf):
@ -394,7 +396,7 @@ def test_call_deprecated_function(result, monkeypatch, default_conf, caplog):
caplog)
def test_strategy_interface_versioning(result, monkeypatch, default_conf):
def test_strategy_interface_versioning(result, default_conf):
default_conf.update({'strategy': 'StrategyTestV2'})
strategy = StrategyResolver.load_strategy(default_conf)
metadata = {'pair': 'ETH/BTC'}
@ -411,8 +413,11 @@ def test_strategy_interface_versioning(result, monkeypatch, default_conf):
enterdf = strategy.advise_buy(result, metadata=metadata)
assert isinstance(enterdf, DataFrame)
assert 'buy' in enterdf.columns
assert 'buy' not in enterdf.columns
assert 'enter_long' in enterdf.columns
exitdf = strategy.advise_sell(result, metadata=metadata)
assert isinstance(exitdf, DataFrame)
assert 'sell' in exitdf
assert 'sell' not in exitdf
assert 'exit_long' in exitdf

47
tests/test_freqtradebot.py
View File

@ -256,7 +256,7 @@ def test_edge_overrides_stoploss(limit_buy_order, fee, caplog, mocker, edge_conf
# stoploss shoud be hit
assert freqtrade.handle_trade(trade) is True
assert log_has('Executing Sell for NEO/BTC. Reason: stop_loss', caplog)
assert log_has('Exit for NEO/BTC detected. Reason: stop_loss', caplog)
assert trade.sell_reason == SellType.STOP_LOSS.value
@ -542,7 +542,7 @@ def test_create_trade_no_signal(default_conf, fee, mocker) -> None:
)
default_conf['stake_amount'] = 10
freqtrade = FreqtradeBot(default_conf)
patch_get_signal(freqtrade, value=(False, False, None))
patch_get_signal(freqtrade, enter_long=False)
Trade.query = MagicMock()
Trade.query.filter = MagicMock()
@ -763,9 +763,10 @@ def test_process_informative_pairs_added(default_conf, ticker, mocker) -> None:
refresh_latest_ohlcv=refresh_mock,
)
inf_pairs = MagicMock(return_value=[("BTC/ETH", '1m'), ("ETH/USDT", "1h")])
mocker.patch(
'freqtrade.strategy.interface.IStrategy.get_signal',
return_value=(False, False, '')
mocker.patch.multiple(
'freqtrade.strategy.interface.IStrategy',
get_exit_signal=MagicMock(return_value=(False, False)),
get_entry_signal=MagicMock(return_value=(None, None))
)
mocker.patch('time.sleep', return_value=None)
@ -1944,7 +1945,7 @@ def test_handle_trade(default_conf, limit_buy_order, limit_sell_order_open, limi
assert trade.is_open is True
freqtrade.wallets.update()
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
assert freqtrade.handle_trade(trade) is True
assert trade.open_order_id == limit_sell_order['id']
@ -1972,7 +1973,7 @@ def test_handle_overlapping_signals(default_conf, ticker, limit_buy_order_open,
)
freqtrade = FreqtradeBot(default_conf)
patch_get_signal(freqtrade, value=(True, True, None))
patch_get_signal(freqtrade, enter_long=True, exit_long=True)
freqtrade.strategy.min_roi_reached = MagicMock(return_value=False)
freqtrade.enter_positions()
@ -1991,7 +1992,7 @@ def test_handle_overlapping_signals(default_conf, ticker, limit_buy_order_open,
assert trades[0].is_open is True
# Buy and Sell are not triggering, so doing nothing ...
patch_get_signal(freqtrade, value=(False, False, None))
patch_get_signal(freqtrade, enter_long=False)
assert freqtrade.handle_trade(trades[0]) is False
trades = Trade.query.all()
nb_trades = len(trades)
@ -1999,7 +2000,7 @@ def test_handle_overlapping_signals(default_conf, ticker, limit_buy_order_open,
assert trades[0].is_open is True
# Buy and Sell are triggering, so doing nothing ...
patch_get_signal(freqtrade, value=(True, True, None))
patch_get_signal(freqtrade, enter_long=True, exit_long=True)
assert freqtrade.handle_trade(trades[0]) is False
trades = Trade.query.all()
nb_trades = len(trades)
@ -2007,7 +2008,7 @@ def test_handle_overlapping_signals(default_conf, ticker, limit_buy_order_open,
assert trades[0].is_open is True
# Sell is triggering, guess what : we are Selling!
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
trades = Trade.query.all()
assert freqtrade.handle_trade(trades[0]) is True
@ -2041,7 +2042,7 @@ def test_handle_trade_roi(default_conf, ticker, limit_buy_order_open,
# we might just want to check if we are in a sell condition without
# executing
# if ROI is reached we must sell
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
assert freqtrade.handle_trade(trade)
assert log_has("ETH/BTC - Required profit reached. sell_type=SellType.ROI",
caplog)
@ -2071,10 +2072,10 @@ def test_handle_trade_use_sell_signal(default_conf, ticker, limit_buy_order_open
trade = Trade.query.first()
trade.is_open = True
patch_get_signal(freqtrade, value=(False, False, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=False)
assert not freqtrade.handle_trade(trade)
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
assert freqtrade.handle_trade(trade)
assert log_has("ETH/BTC - Sell signal received. sell_type=SellType.SELL_SIGNAL",
caplog)
@ -3196,7 +3197,7 @@ def test_sell_profit_only_enable_profit(default_conf, limit_buy_order, limit_buy
trade = Trade.query.first()
trade.update(limit_buy_order)
freqtrade.wallets.update()
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
assert freqtrade.handle_trade(trade) is False
freqtrade.strategy.sell_profit_offset = 0.0
@ -3234,7 +3235,7 @@ def test_sell_profit_only_disable_profit(default_conf, limit_buy_order, limit_bu
trade = Trade.query.first()
trade.update(limit_buy_order)
freqtrade.wallets.update()
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
assert freqtrade.handle_trade(trade) is True
assert trade.sell_reason == SellType.SELL_SIGNAL.value
@ -3268,7 +3269,7 @@ def test_sell_profit_only_enable_loss(default_conf, limit_buy_order, limit_buy_o
trade = Trade.query.first()
trade.update(limit_buy_order)
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
assert freqtrade.handle_trade(trade) is False
@ -3303,7 +3304,7 @@ def test_sell_profit_only_disable_loss(default_conf, limit_buy_order, limit_buy_
trade = Trade.query.first()
trade.update(limit_buy_order)
freqtrade.wallets.update()
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
assert freqtrade.handle_trade(trade) is True
assert trade.sell_reason == SellType.SELL_SIGNAL.value
@ -3335,7 +3336,7 @@ def test_sell_not_enough_balance(default_conf, limit_buy_order, limit_buy_order_
trade = Trade.query.first()
amnt = trade.amount
trade.update(limit_buy_order)
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
mocker.patch('freqtrade.wallets.Wallets.get_free', MagicMock(return_value=trade.amount * 0.985))
assert freqtrade.handle_trade(trade) is True
@ -3460,11 +3461,11 @@ def test_ignore_roi_if_buy_signal(default_conf, limit_buy_order, limit_buy_order
trade = Trade.query.first()
trade.update(limit_buy_order)
freqtrade.wallets.update()
patch_get_signal(freqtrade, value=(True, True, None))
patch_get_signal(freqtrade, enter_long=True, exit_long=True)
assert freqtrade.handle_trade(trade) is False
# Test if buy-signal is absent (should sell due to roi = true)
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
assert freqtrade.handle_trade(trade) is True
assert trade.sell_reason == SellType.ROI.value
@ -3745,11 +3746,11 @@ def test_disable_ignore_roi_if_buy_signal(default_conf, limit_buy_order, limit_b
trade = Trade.query.first()
trade.update(limit_buy_order)
# Sell due to min_roi_reached
patch_get_signal(freqtrade, value=(True, True, None))
patch_get_signal(freqtrade, enter_long=True, exit_long=True)
assert freqtrade.handle_trade(trade) is True
# Test if buy-signal is absent
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
assert freqtrade.handle_trade(trade) is True
assert trade.sell_reason == SellType.SELL_SIGNAL.value
@ -4297,7 +4298,7 @@ def test_order_book_ask_strategy(default_conf, limit_buy_order_open, limit_buy_o
freqtrade.wallets.update()
assert trade.is_open is True
patch_get_signal(freqtrade, value=(False, True, None))
patch_get_signal(freqtrade, enter_long=False, exit_long=True)
assert freqtrade.handle_trade(trade) is True
assert trade.close_rate_requested == order_book_l2.return_value['asks'][0][0]

4
tests/test_integration.py
View File

@ -72,7 +72,7 @@ def test_may_execute_exit_stoploss_on_exchange_multi(default_conf, ticker, fee,
create_stoploss_order=MagicMock(return_value=True),
_notify_sell=MagicMock(),
)
mocker.patch("freqtrade.strategy.interface.IStrategy.should_sell", should_sell_mock)
mocker.patch("freqtrade.strategy.interface.IStrategy.should_exit", should_sell_mock)
wallets_mock = mocker.patch("freqtrade.wallets.Wallets.update", MagicMock())
mocker.patch("freqtrade.wallets.Wallets.get_free", MagicMock(return_value=1000))
@ -163,7 +163,7 @@ def test_forcebuy_last_unlimited(default_conf, ticker, fee, limit_buy_order, moc
SellCheckTuple(sell_type=SellType.NONE),
SellCheckTuple(sell_type=SellType.NONE)]
)
mocker.patch("freqtrade.strategy.interface.IStrategy.should_sell", should_sell_mock)
mocker.patch("freqtrade.strategy.interface.IStrategy.should_exit", should_sell_mock)
freqtrade = get_patched_freqtradebot(mocker, default_conf)
rpc = RPC(freqtrade)

8
tests/test_plotting.py
View File

@ -201,8 +201,8 @@ def test_generate_candlestick_graph_no_signals_no_trades(default_conf, mocker, t
timerange = TimeRange(None, 'line', 0, -1000)
data = history.load_pair_history(pair=pair, timeframe='1m',
datadir=testdatadir, timerange=timerange)
data['buy'] = 0
data['sell'] = 0
data['enter_long'] = 0
data['exit_long'] = 0
indicators1 = []
indicators2 = []
@ -261,12 +261,12 @@ def test_generate_candlestick_graph_no_trades(default_conf, mocker, testdatadir)
buy = find_trace_in_fig_data(figure.data, "buy")
assert isinstance(buy, go.Scatter)
# All buy-signals should be plotted
assert int(data.buy.sum()) == len(buy.x)
assert int(data['enter_long'].sum()) == len(buy.x)
sell = find_trace_in_fig_data(figure.data, "sell")
assert isinstance(sell, go.Scatter)
# All buy-signals should be plotted
assert int(data.sell.sum()) == len(sell.x)
assert int(data['exit_long'].sum()) == len(sell.x)
assert find_trace_in_fig_data(figure.data, "Bollinger Band")