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

merge leverage/margin

Browse Source
This commit is contained in:
மனோஜ்குமார் பழனிச்சாமி
2022-04-04 09:54:08 +05:30
parent 9ceb7fbf4f 69491c1430
commit b646d8ba0e
185 changed files with 61237 additions and 4606 deletions
Download Patch File Download Diff File Expand all files Collapse all files

517
freqtrade/strategy/interface.py
View File

@@ -13,7 +13,8 @@ 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 (CandleType, ExitCheckTuple, ExitType, SignalDirection, SignalTagType,
SignalType, TradingMode)
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
@@ -28,23 +29,7 @@ from freqtrade.wallets import Wallets
logger = logging.getLogger(__name__)
CUSTOM_SELL_MAX_LENGTH = 64
class SellCheckTuple:
"""
NamedTuple for Sell type + reason
"""
sell_type: SellType
sell_reason: str = ''
def __init__(self, sell_type: SellType, sell_reason: str = ''):
self.sell_type = sell_type
self.sell_reason = sell_reason or sell_type.value
@property
def sell_flag(self):
return self.sell_type != SellType.NONE
CUSTOM_EXIT_MAX_LENGTH = 64
class IStrategy(ABC, HyperStrategyMixin):
@@ -61,7 +46,8 @@ class IStrategy(ABC, HyperStrategyMixin):
# Default to version 2
# Version 1 is the initial interface without metadata dict
# Version 2 populate_* include metadata dict
INTERFACE_VERSION: int = 2
# Version 3 - First version with short and leverage support
INTERFACE_VERSION: int = 3
_populate_fun_len: int = 0
_buy_fun_len: int = 0
@@ -80,13 +66,16 @@ class IStrategy(ABC, HyperStrategyMixin):
trailing_only_offset_is_reached = False
use_custom_stoploss: bool = False
# Can this strategy go short?
can_short: bool = False
# associated timeframe
timeframe: str
# Optional order types
order_types: Dict = {
'buy': 'limit',
'sell': 'limit',
'entry': 'limit',
'exit': 'limit',
'stoploss': 'limit',
'stoploss_on_exchange': False,
'stoploss_on_exchange_interval': 60,
@@ -94,8 +83,8 @@ class IStrategy(ABC, HyperStrategyMixin):
# Optional time in force
order_time_in_force: Dict = {
'buy': 'gtc',
'sell': 'gtc',
'entry': 'gtc',
'exit': 'gtc',
}
# run "populate_indicators" only for new candle
@@ -157,31 +146,41 @@ class IStrategy(ABC, HyperStrategyMixin):
if timeframe_to_minutes(informative_data.timeframe) < strategy_timeframe_minutes:
raise OperationalException('Informative timeframe must be equal or higher than '
'strategy timeframe!')
if not informative_data.candle_type:
informative_data.candle_type = config['candle_type_def']
self._ft_informative.append((informative_data, cls_method))
@abstractmethod
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Populate indicators that will be used in the Buy and Sell strategy
Populate indicators that will be used in the Buy, Sell, Short, Exit_short strategy
: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
"""
return dataframe
@abstractmethod
def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the buy signal for the given dataframe
DEPRECATED - please migrate to populate_entry_trend
:param dataframe: DataFrame
:param metadata: Additional information, like the currently traded pair
:return: DataFrame with buy column
"""
return dataframe
@abstractmethod
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the entry signal for the given dataframe
:param dataframe: DataFrame
:param metadata: Additional information, like the currently traded pair
:return: DataFrame with entry columns populated
"""
return self.populate_buy_trend(dataframe, metadata)
def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
DEPRECATED - please migrate to populate_exit_trend
Based on TA indicators, populates the sell signal for the given dataframe
:param dataframe: DataFrame
:param metadata: Additional information, like the currently traded pair
@@ -189,6 +188,15 @@ class IStrategy(ABC, HyperStrategyMixin):
"""
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the exit signal for the given dataframe
:param dataframe: DataFrame
:param metadata: Additional information, like the currently traded pair
:return: DataFrame with exit columns populated
"""
return self.populate_sell_trend(dataframe, metadata)
def bot_loop_start(self, **kwargs) -> None:
"""
Called at the start of the bot iteration (one loop).
@@ -201,9 +209,16 @@ class IStrategy(ABC, HyperStrategyMixin):
def check_buy_timeout(self, pair: str, trade: Trade, order: dict,
current_time: datetime, **kwargs) -> bool:
"""
Check buy timeout function callback.
This method can be used to override the buy-timeout.
It is called whenever a limit buy order has been created,
DEPRECATED: Please use `check_entry_timeout` instead.
"""
return False
def check_entry_timeout(self, pair: str, trade: Trade, order: dict,
current_time: datetime, **kwargs) -> bool:
"""
Check entry timeout function callback.
This method can be used to override the enter-timeout.
It is called whenever a limit entry order has been created,
and is not yet fully filled.
Configuration options in `unfilledtimeout` will be verified before this,
so ensure to set these timeouts high enough.
@@ -214,17 +229,25 @@ class IStrategy(ABC, HyperStrategyMixin):
:param order: Order dictionary as returned from CCXT.
:param current_time: datetime object, containing the current datetime
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
:return bool: When True is returned, then the buy-order is cancelled.
:return bool: When True is returned, then the entry order is cancelled.
"""
return False
return self.check_buy_timeout(
pair=pair, trade=trade, order=order, current_time=current_time)
def check_sell_timeout(self, pair: str, trade: Trade, order: dict,
current_time: datetime, **kwargs) -> bool:
"""
DEPRECATED: Please use `check_exit_timeout` instead.
"""
return False
def check_exit_timeout(self, pair: str, trade: Trade, order: dict,
current_time: datetime, **kwargs) -> bool:
"""
Check sell timeout function callback.
This method can be used to override the sell-timeout.
It is called whenever a limit sell order has been created,
and is not yet fully filled.
This method can be used to override the exit-timeout.
It is called whenever a (long) limit sell order or (short) limit buy
has been created, and is not yet fully filled.
Configuration options in `unfilledtimeout` will be verified before this,
so ensure to set these timeouts high enough.
@@ -234,15 +257,16 @@ class IStrategy(ABC, HyperStrategyMixin):
:param order: Order dictionary as returned from CCXT.
:param current_time: datetime object, containing the current datetime
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
:return bool: When True is returned, then the sell-order is cancelled.
:return bool: When True is returned, then the (long)sell/(short)buy-order is cancelled.
"""
return False
return self.check_sell_timeout(
pair=pair, trade=trade, order=order, current_time=current_time)
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, entry_tag: Optional[str],
**kwargs) -> bool:
side: str, **kwargs) -> bool:
"""
Called right before placing a buy order.
Called right before placing a entry order.
Timing for this function is critical, so avoid doing heavy computations or
network requests in this method.
@@ -250,13 +274,14 @@ class IStrategy(ABC, HyperStrategyMixin):
When not implemented by a strategy, returns True (always confirming).
:param pair: Pair that's about to be bought.
:param pair: Pair that's about to be bought/shorted.
:param order_type: Order type (as configured in order_types). usually limit or market.
:param amount: Amount in target (quote) currency that's going to be traded.
:param rate: Rate that's going to be used when using limit orders
:param time_in_force: Time in force. Defaults to GTC (Good-til-cancelled).
:param current_time: datetime object, containing the current datetime
:param entry_tag: Optional entry_tag (buy_tag) if provided with the buy signal.
:param side: 'long' or 'short' - indicating the direction of the proposed trade
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
:return bool: When True is returned, then the buy-order is placed on the exchange.
False aborts the process
@@ -264,10 +289,10 @@ class IStrategy(ABC, HyperStrategyMixin):
return True
def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float,
rate: float, time_in_force: str, sell_reason: str,
rate: float, time_in_force: str, exit_reason: str,
current_time: datetime, **kwargs) -> bool:
"""
Called right before placing a regular sell order.
Called right before placing a regular exit order.
Timing for this function is critical, so avoid doing heavy computations or
network requests in this method.
@@ -275,18 +300,18 @@ class IStrategy(ABC, HyperStrategyMixin):
When not implemented by a strategy, returns True (always confirming).
:param pair: Pair that's about to be sold.
:param pair: Pair for trade that's about to be exited.
:param trade: trade object.
:param order_type: Order type (as configured in order_types). usually limit or market.
:param amount: Amount in quote currency.
:param rate: Rate that's going to be used when using limit orders
:param time_in_force: Time in force. Defaults to GTC (Good-til-cancelled).
:param sell_reason: Sell reason.
:param exit_reason: Exit reason.
Can be any of ['roi', 'stop_loss', 'stoploss_on_exchange', 'trailing_stop_loss',
'sell_signal', 'force_sell', 'emergency_sell']
:param current_time: datetime object, containing the current datetime
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
:return bool: When True is returned, then the sell-order is placed on the exchange.
:return bool: When True, then the sell-order/exit_short-order is placed on the exchange.
False aborts the process
"""
return True
@@ -306,7 +331,7 @@ class IStrategy(ABC, HyperStrategyMixin):
:param pair: Pair that's currently analyzed
:param trade: trade object.
:param current_time: datetime object, containing the current datetime
:param current_rate: Rate, calculated based on pricing settings in ask_strategy.
:param current_rate: Rate, calculated based on pricing settings in exit_pricing.
:param current_profit: Current profit (as ratio), calculated based on current_rate.
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
:return float: New stoploss value, relative to the current_rate
@@ -324,7 +349,7 @@ class IStrategy(ABC, HyperStrategyMixin):
:param pair: Pair that's currently analyzed
:param current_time: datetime object, containing the current datetime
:param proposed_rate: Rate, calculated based on pricing settings in ask_strategy.
:param proposed_rate: Rate, calculated based on pricing settings in exit_pricing.
:param entry_tag: Optional entry_tag (buy_tag) if provided with the buy signal.
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
:return float: New entry price value if provided
@@ -344,7 +369,7 @@ class IStrategy(ABC, HyperStrategyMixin):
:param pair: Pair that's currently analyzed
:param trade: trade object.
:param current_time: datetime object, containing the current datetime
:param proposed_rate: Rate, calculated based on pricing settings in ask_strategy.
:param proposed_rate: Rate, calculated based on pricing settings in exit_pricing.
:param current_profit: Current profit (as ratio), calculated based on current_rate.
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
:return float: New exit price value if provided
@@ -354,41 +379,66 @@ class IStrategy(ABC, HyperStrategyMixin):
def custom_sell(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
current_profit: float, **kwargs) -> Optional[Union[str, bool]]:
"""
Custom sell signal logic indicating that specified position should be sold. Returning a
string or True from this method is equal to setting sell signal on a candle at specified
time. This method is not called when sell signal is set.
DEPRECATED - please use custom_exit instead.
Custom exit signal logic indicating that specified position should be sold. Returning a
string or True from this method is equal to setting exit signal on a candle at specified
time. This method is not called when exit signal is set.
This method should be overridden to create sell signals that depend on trade parameters. For
example you could implement a sell relative to the candle when the trade was opened,
This method should be overridden to create exit signals that depend on trade parameters. For
example you could implement an exit relative to the candle when the trade was opened,
or a custom 1:2 risk-reward ROI.
Custom sell reason max length is 64. Exceeding characters will be removed.
Custom exit reason max length is 64. Exceeding characters will be removed.
:param pair: Pair that's currently analyzed
:param trade: trade object.
:param current_time: datetime object, containing the current datetime
:param current_rate: Rate, calculated based on pricing settings in ask_strategy.
:param current_rate: Rate, calculated based on pricing settings in exit_pricing.
:param current_profit: Current profit (as ratio), calculated based on current_rate.
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
:return: To execute sell, return a string with custom sell reason or True. Otherwise return
:return: To execute exit, return a string with custom exit reason or True. Otherwise return
None or False.
"""
return None
def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
current_profit: float, **kwargs) -> Optional[Union[str, bool]]:
"""
Custom exit signal logic indicating that specified position should be sold. Returning a
string or True from this method is equal to setting exit signal on a candle at specified
time. This method is not called when exit signal is set.
This method should be overridden to create exit signals that depend on trade parameters. For
example you could implement an exit relative to the candle when the trade was opened,
or a custom 1:2 risk-reward ROI.
Custom exit reason max length is 64. Exceeding characters will be removed.
:param pair: Pair that's currently analyzed
:param trade: trade object.
:param current_time: datetime object, containing the current datetime
:param current_rate: Rate, calculated based on pricing settings in exit_pricing.
:param current_profit: Current profit (as ratio), calculated based on current_rate.
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
:return: To execute exit, return a string with custom exit reason or True. Otherwise return
None or False.
"""
return self.custom_sell(pair, trade, current_time, current_rate, current_profit, **kwargs)
def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
proposed_stake: float, min_stake: float, max_stake: float,
entry_tag: Optional[str], **kwargs) -> float:
entry_tag: Optional[str], side: str, **kwargs) -> float:
"""
Customize stake size for each new trade. This method is not called when edge module is
enabled.
Customize stake size for each new trade.
: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 current_rate: Rate, calculated based on pricing settings in exit_pricing.
:param proposed_stake: A stake amount proposed by the bot.
:param min_stake: Minimal stake size allowed by exchange.
:param max_stake: Balance available for trading.
:param entry_tag: Optional entry_tag (buy_tag) if provided with the buy signal.
:param side: 'long' or 'short' - indicating the direction of the proposed trade
:return: A stake size, which is between min_stake and max_stake.
"""
return proposed_stake
@@ -416,6 +466,22 @@ class IStrategy(ABC, HyperStrategyMixin):
"""
return None
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 only called in futures mode.
: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 exit_pricing.
: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
def informative_pairs(self) -> ListPairsWithTimeframes:
"""
Define additional, informative pair/interval combinations to be cached from the exchange.
@@ -444,16 +510,28 @@ class IStrategy(ABC, HyperStrategyMixin):
Internal method which gathers all informative pairs (user or automatically defined).
"""
informative_pairs = self.informative_pairs()
# Compatibility code for 2 tuple informative pairs
informative_pairs = [
(p[0], p[1], CandleType.from_string(p[2]) if len(
p) > 2 and p[2] != '' else self.config.get('candle_type_def', CandleType.SPOT))
for p in informative_pairs]
for inf_data, _ in self._ft_informative:
# Get default candle type if not provided explicitly.
candle_type = (inf_data.candle_type if inf_data.candle_type
else self.config.get('candle_type_def', CandleType.SPOT))
if inf_data.asset:
pair_tf = (_format_pair_name(self.config, inf_data.asset), inf_data.timeframe)
pair_tf = (
_format_pair_name(self.config, inf_data.asset),
inf_data.timeframe,
candle_type,
)
informative_pairs.append(pair_tf)
else:
if not self.dp:
raise OperationalException('@informative decorator with unspecified asset '
'requires DataProvider instance.')
for pair in self.dp.current_whitelist():
informative_pairs.append((pair, inf_data.timeframe))
informative_pairs.append((pair, inf_data.timeframe, candle_type))
return list(set(informative_pairs))
def get_strategy_name(self) -> str:
@@ -498,7 +576,7 @@ class IStrategy(ABC, HyperStrategyMixin):
Checks if a pair is currently locked
The 2nd, optional parameter ensures that locks are applied until the new candle arrives,
and not stop at 14:00:00 - while the next candle arrives at 14:00:02 leaving a gap
of 2 seconds for a buy to happen on an old signal.
of 2 seconds for an entry order to happen on an old signal.
:param pair: "Pair to check"
:param candle_date: Date of the last candle. Optional, defaults to current date
:returns: locking state of the pair in question.
@@ -514,15 +592,15 @@ class IStrategy(ABC, HyperStrategyMixin):
def analyze_ticker(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Parses the given candle (OHLCV) data and returns a populated DataFrame
add several TA indicators and buy signal to it
add several TA indicators and entry order signal to it
:param dataframe: Dataframe containing data from exchange
:param metadata: Metadata dictionary with additional data (e.g. 'pair')
:return: DataFrame of candle (OHLCV) data with indicator data and signals added
"""
logger.debug("TA Analysis Launched")
dataframe = self.advise_indicators(dataframe, metadata)
dataframe = self.advise_buy(dataframe, metadata)
dataframe = self.advise_sell(dataframe, metadata)
dataframe = self.advise_entry(dataframe, metadata)
dataframe = self.advise_exit(dataframe, metadata)
return dataframe
def _analyze_ticker_internal(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
@@ -544,13 +622,17 @@ class IStrategy(ABC, HyperStrategyMixin):
dataframe = self.analyze_ticker(dataframe, metadata)
self._last_candle_seen_per_pair[pair] = dataframe.iloc[-1]['date']
if self.dp:
self.dp._set_cached_df(pair, self.timeframe, dataframe)
self.dp._set_cached_df(
pair, self.timeframe, dataframe,
candle_type=self.config.get('candle_type_def', CandleType.SPOT))
else:
logger.debug("Skipping TA Analysis for already analyzed candle")
dataframe['buy'] = 0
dataframe['sell'] = 0
dataframe['buy_tag'] = None
dataframe['exit_tag'] = None
dataframe[SignalType.ENTER_LONG.value] = 0
dataframe[SignalType.EXIT_LONG.value] = 0
dataframe[SignalType.ENTER_SHORT.value] = 0
dataframe[SignalType.EXIT_SHORT.value] = 0
dataframe[SignalTagType.ENTER_TAG.value] = None
dataframe[SignalTagType.EXIT_TAG.value] = None
# Other Defs in strategy that want to be called every loop here
# twitter_sell = self.watch_twitter_feed(dataframe, metadata)
@@ -567,7 +649,9 @@ class IStrategy(ABC, HyperStrategyMixin):
"""
if not self.dp:
raise OperationalException("DataProvider not found.")
dataframe = self.dp.ohlcv(pair, self.timeframe)
dataframe = self.dp.ohlcv(
pair, self.timeframe, candle_type=self.config.get('candle_type_def', CandleType.SPOT)
)
if not isinstance(dataframe, DataFrame) or dataframe.empty:
logger.warning('Empty candle (OHLCV) data for pair %s', pair)
return
@@ -609,8 +693,8 @@ class IStrategy(ABC, HyperStrategyMixin):
message = ""
if dataframe is None:
message = "No dataframe returned (return statement missing?)."
elif 'buy' not in dataframe:
message = "Buy column not set."
elif 'enter_long' not in dataframe:
message = "enter_long/buy column not set."
elif df_len != len(dataframe):
message = message_template.format("length")
elif df_close != dataframe["close"].iloc[-1]:
@@ -623,23 +707,24 @@ 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], Optional[str]]:
dataframe: DataFrame,
) -> Tuple[Optional[DataFrame], Optional[arrow.Arrow]]:
"""
Calculates current signal based based on the buy / sell columns of the dataframe.
Used by Bot to get the signal to buy or sell
Calculates current signal based based on the entry order or exit order
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: (Buy, Sell) A bool-tuple indicating buy/sell 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, None
return None, None
latest_date = dataframe['date'].max()
latest = dataframe.loc[dataframe['date'] == latest_date].iloc[-1]
@@ -654,49 +739,124 @@ 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, None
return None, None
return latest, latest_date
buy = latest[SignalType.BUY.value] == 1
def get_exit_signal(
self,
pair: str,
timeframe: str,
dataframe: DataFrame,
is_short: bool = None
) -> Tuple[bool, bool, Optional[str]]:
"""
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, None
sell = False
if SignalType.SELL.value in latest:
sell = 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
buy_tag = latest.get(SignalTagType.BUY_TAG.value, None)
else:
enter = latest[SignalType.ENTER_LONG.value] == 1
exit_ = latest.get(SignalType.EXIT_LONG.value, 0) == 1
exit_tag = latest.get(SignalTagType.EXIT_TAG.value, None)
# Tags can be None, which does not resolve to False.
buy_tag = buy_tag if isinstance(buy_tag, str) else None
exit_tag = exit_tag if isinstance(exit_tag, str) else None
logger.debug('trigger: %s (pair=%s) buy=%s sell=%s',
latest['date'], pair, str(buy), str(sell))
timeframe_seconds = timeframe_to_seconds(timeframe)
if self.ignore_expired_candle(latest_date=latest_date,
current_time=datetime.now(timezone.utc),
timeframe_seconds=timeframe_seconds,
buy=buy):
return False, sell, buy_tag, exit_tag
return buy, sell, buy_tag, exit_tag
logger.debug(f"exit-trigger: {latest['date']} (pair={pair}) "
f"enter={enter} exit={exit_}")
def ignore_expired_candle(self, latest_date: datetime, current_time: datetime,
timeframe_seconds: int, buy: bool):
if self.ignore_buying_expired_candle_after and buy:
return enter, exit_, exit_tag
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.ENTER_TAG.value, None)
if (self.config.get('trading_mode', TradingMode.SPOT) != TradingMode.SPOT
and self.can_short
and enter_short == 1 and not any([exit_short, enter_long])):
enter_signal = SignalDirection.SHORT
enter_tag_value = latest.get(SignalTagType.ENTER_TAG.value, None)
enter_tag_value = enter_tag_value if isinstance(enter_tag_value, str) else None
timeframe_seconds = timeframe_to_seconds(timeframe)
if self.ignore_expired_candle(
latest_date=latest_date.datetime,
current_time=datetime.now(timezone.utc),
timeframe_seconds=timeframe_seconds,
enter=bool(enter_signal)
):
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,
latest_date: datetime,
current_time: datetime,
timeframe_seconds: int,
enter: bool
):
if self.ignore_buying_expired_candle_after and enter:
time_delta = current_time - (latest_date + timedelta(seconds=timeframe_seconds))
return time_delta.total_seconds() > self.ignore_buying_expired_candle_after
else:
return False
def should_sell(self, trade: Trade, rate: float, current_time: datetime, buy: bool,
sell: bool, low: float = None, high: float = None,
force_stoploss: float = 0) -> SellCheckTuple:
def should_exit(self, trade: Trade, rate: float, current_time: datetime, *,
enter: bool, exit_: bool,
low: float = None, high: float = None,
force_stoploss: float = 0) -> ExitCheckTuple:
"""
This function evaluates if one of the conditions required to trigger a sell
has been reached, which can either be a stop-loss, ROI or sell-signal.
:param low: Only used during backtesting to simulate stoploss
:param high: Only used during backtesting, to simulate ROI
This function evaluates if one of the conditions required to trigger an exit order
has been reached, which can either be a stop-loss, ROI or exit-signal.
:param low: Only used during backtesting to simulate (long)stoploss/(short)ROI
:param high: Only used during backtesting, to simulate (short)stoploss/(long)ROI
:param force_stoploss: Externally provided stoploss
:return: True if trade should be sold, False otherwise
:return: True if trade should be exited, False otherwise
"""
current_rate = rate
current_profit = trade.calc_profit_ratio(current_rate)
@@ -708,15 +868,15 @@ class IStrategy(ABC, HyperStrategyMixin):
force_stoploss=force_stoploss, low=low, high=high)
# Set current rate to high for backtesting sell
current_rate = high or rate
current_rate = (low if trade.is_short else high) or rate
current_profit = trade.calc_profit_ratio(current_rate)
# if buy 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)
# if enter signal and ignore_roi is set, we don't need to evaluate min_roi.
roi_reached = (not (enter and self.ignore_roi_if_buy_signal)
and self.min_roi_reached(trade=trade, current_profit=current_profit,
current_time=current_time))
sell_signal = SellType.NONE
exit_signal = ExitType.NONE
custom_reason = ''
# use provided rate in backtesting, not high/low.
current_rate = rate
@@ -725,54 +885,54 @@ 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:
sell_signal = SellType.SELL_SIGNAL
elif self.use_sell_signal and not enter:
if exit_:
exit_signal = ExitType.SELL_SIGNAL
else:
custom_reason = strategy_safe_wrapper(self.custom_sell, default_retval=False)(
trade_type = "exit_short" if trade.is_short else "sell"
custom_reason = strategy_safe_wrapper(self.custom_exit, default_retval=False)(
pair=trade.pair, trade=trade, current_time=current_time,
current_rate=current_rate, current_profit=current_profit)
if custom_reason:
sell_signal = SellType.CUSTOM_SELL
exit_signal = ExitType.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.')
custom_reason = custom_reason[:CUSTOM_SELL_MAX_LENGTH]
if len(custom_reason) > CUSTOM_EXIT_MAX_LENGTH:
logger.warning(f'Custom {trade_type} reason returned from '
f'custom_exit is too long and was trimmed'
f'to {CUSTOM_EXIT_MAX_LENGTH} characters.')
custom_reason = custom_reason[:CUSTOM_EXIT_MAX_LENGTH]
else:
custom_reason = None
if sell_signal in (SellType.CUSTOM_SELL, SellType.SELL_SIGNAL):
if exit_signal in (ExitType.CUSTOM_SELL, ExitType.SELL_SIGNAL):
logger.debug(f"{trade.pair} - Sell signal received. "
f"sell_type=SellType.{sell_signal.name}" +
f"exit_type=ExitType.{exit_signal.name}" +
(f", custom_reason={custom_reason}" if custom_reason else ""))
return SellCheckTuple(sell_type=sell_signal, sell_reason=custom_reason)
return ExitCheckTuple(exit_type=exit_signal, exit_reason=custom_reason)
# Start evaluations
# Sequence:
# Sell-signal
# Exit-signal
# ROI (if not stoploss)
# Stoploss
if roi_reached and stoplossflag.sell_type != SellType.STOP_LOSS:
logger.debug(f"{trade.pair} - Required profit reached. sell_type=SellType.ROI")
return SellCheckTuple(sell_type=SellType.ROI)
if roi_reached and stoplossflag.exit_type != ExitType.STOP_LOSS:
logger.debug(f"{trade.pair} - Required profit reached. exit_type=ExitType.ROI")
return ExitCheckTuple(exit_type=ExitType.ROI)
if stoplossflag.sell_flag:
if stoplossflag.exit_flag:
logger.debug(f"{trade.pair} - Stoploss hit. sell_type={stoplossflag.sell_type}")
logger.debug(f"{trade.pair} - Stoploss hit. exit_type={stoplossflag.exit_type}")
return stoplossflag
# This one is noisy, commented out...
# logger.debug(f"{trade.pair} - No sell signal.")
return SellCheckTuple(sell_type=SellType.NONE)
# logger.debug(f"{trade.pair} - No exit signal.")
return ExitCheckTuple(exit_type=ExitType.NONE)
def stop_loss_reached(self, current_rate: float, trade: Trade,
current_time: datetime, current_profit: float,
force_stoploss: float, low: float = None,
high: float = None) -> SellCheckTuple:
high: float = None) -> ExitCheckTuple:
"""
Based on current profit of the trade and configured (trailing) stoploss,
decides to sell or not
decides to exit or not
:param current_profit: current profit as ratio
:param low: Low value of this candle, only set in backtesting
:param high: High value of this candle, only set in backtesting
@@ -782,7 +942,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,
@@ -795,45 +960,56 @@ class IStrategy(ABC, HyperStrategyMixin):
else:
logger.warning("CustomStoploss function did not return valid stoploss")
if self.trailing_stop and trade.stop_loss < (low or current_rate):
sl_lower_long = (trade.stop_loss < (low or current_rate) and not trade.is_short)
sl_higher_short = (trade.stop_loss > (high or current_rate) and trade.is_short)
if self.trailing_stop and (sl_lower_long or sl_higher_short):
# trailing stoploss handling
sl_offset = self.trailing_stop_positive_offset
# Make sure current_profit is calculated using high for backtesting.
high_profit = current_profit if not high else trade.calc_profit_ratio(high)
bound = low if trade.is_short else high
bound_profit = current_profit if not bound else trade.calc_profit_ratio(bound)
# Don't update stoploss if trailing_only_offset_is_reached is true.
if not (self.trailing_only_offset_is_reached and high_profit < sl_offset):
if not (self.trailing_only_offset_is_reached and bound_profit < sl_offset):
# Specific handling for trailing_stop_positive
if self.trailing_stop_positive is not None and high_profit > sl_offset:
if self.trailing_stop_positive is not None and bound_profit > sl_offset:
stop_loss_value = self.trailing_stop_positive
logger.debug(f"{trade.pair} - Using positive stoploss: {stop_loss_value} "
f"offset: {sl_offset:.4g} profit: {current_profit:.2%}")
trade.adjust_stop_loss(high or current_rate, stop_loss_value)
trade.adjust_stop_loss(bound or current_rate, stop_loss_value)
sl_higher_long = (trade.stop_loss >= (low or current_rate) and not trade.is_short)
sl_lower_short = (trade.stop_loss <= (high or current_rate) and trade.is_short)
# evaluate if the stoploss was hit if stoploss is not on exchange
# in Dry-Run, this handles stoploss logic as well, as the logic will not be different to
# regular stoploss handling.
if ((trade.stop_loss >= (low or current_rate)) and
if ((sl_higher_long or sl_lower_short) and
(not self.order_types.get('stoploss_on_exchange') or self.config['dry_run'])):
sell_type = SellType.STOP_LOSS
exit_type = ExitType.STOP_LOSS
# If initial stoploss is not the same as current one then it is trailing.
if trade.initial_stop_loss != trade.stop_loss:
sell_type = SellType.TRAILING_STOP_LOSS
exit_type = ExitType.TRAILING_STOP_LOSS
logger.debug(
f"{trade.pair} - HIT STOP: current price at {(low or current_rate):.6f}, "
f"{trade.pair} - HIT STOP: current price at "
f"{((high if trade.is_short else low) or current_rate):.6f}, "
f"stoploss is {trade.stop_loss:.6f}, "
f"initial stoploss was at {trade.initial_stop_loss:.6f}, "
f"trade opened at {trade.open_rate:.6f}")
new_stoploss = (
trade.stop_loss + trade.initial_stop_loss
if trade.is_short else
trade.stop_loss - trade.initial_stop_loss
)
logger.debug(f"{trade.pair} - Trailing stop saved "
f"{trade.stop_loss - trade.initial_stop_loss:.6f}")
f"{new_stoploss:.6f}")
return SellCheckTuple(sell_type=sell_type)
return ExitCheckTuple(exit_type=exit_type)
return SellCheckTuple(sell_type=SellType.NONE)
return ExitCheckTuple(exit_type=ExitType.NONE)
def min_roi_reached_entry(self, trade_dur: int) -> Tuple[Optional[int], Optional[float]]:
"""
@@ -863,22 +1039,24 @@ class IStrategy(ABC, HyperStrategyMixin):
else:
return current_profit > roi
def ft_check_timed_out(self, side: str, trade: LocalTrade, order: Order,
def ft_check_timed_out(self, trade: LocalTrade, order: Order,
current_time: datetime) -> bool:
"""
FT Internal method.
Check if timeout is active, and if the order is still open and timed out
"""
side = 'entry' if order.ft_order_side == trade.enter_side else 'exit'
timeout = self.config.get('unfilledtimeout', {}).get(side)
if timeout is not None:
timeout_unit = self.config.get('unfilledtimeout', {}).get('unit', 'minutes')
timeout_kwargs = {timeout_unit: -timeout}
timeout_threshold = current_time + timedelta(**timeout_kwargs)
timedout = (order.status == 'open' and order.side == side
and order.order_date_utc < timeout_threshold)
timedout = (order.status == 'open' and order.order_date_utc < timeout_threshold)
if timedout:
return True
time_method = self.check_sell_timeout if order.side == 'sell' else self.check_buy_timeout
time_method = (self.check_exit_timeout if order.side == trade.exit_side
else self.check_entry_timeout)
return strategy_safe_wrapper(time_method,
default_retval=False)(
@@ -888,7 +1066,7 @@ class IStrategy(ABC, HyperStrategyMixin):
def advise_all_indicators(self, data: Dict[str, DataFrame]) -> Dict[str, DataFrame]:
"""
Populates indicators for given candle (OHLCV) data (for multiple pairs)
Does not run advise_buy or advise_sell!
Does not run advise_entry or advise_exit!
Used by optimize operations only, not during dry / live runs.
Using .copy() to get a fresh copy of the dataframe for every strategy run.
Also copy on output to avoid PerformanceWarnings pandas 1.3.0 started to show.
@@ -900,7 +1078,7 @@ class IStrategy(ABC, HyperStrategyMixin):
def advise_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Populate indicators that will be used in the Buy and Sell strategy
Populate indicators that will be used in the Buy, Sell, short, exit_short strategy
This method should not be overridden.
:param dataframe: Dataframe with data from the exchange
:param metadata: Additional information, like the currently traded pair
@@ -920,37 +1098,46 @@ class IStrategy(ABC, HyperStrategyMixin):
else:
return self.populate_indicators(dataframe, metadata)
def advise_buy(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
def advise_entry(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the buy signal for the given dataframe
Based on TA indicators, populates the entry order signal for the given dataframe
This method should not be overridden.
:param dataframe: DataFrame
:param metadata: Additional information dictionary, with details like the
currently traded pair
:return: DataFrame with buy column
"""
logger.debug(f"Populating buy signals for pair {metadata.get('pair')}.")
logger.debug(f"Populating enter signals for pair {metadata.get('pair')}.")
if self._buy_fun_len == 2:
warnings.warn("deprecated - check out the Sample strategy to see "
"the current function headers!", DeprecationWarning)
return self.populate_buy_trend(dataframe) # type: ignore
df = self.populate_buy_trend(dataframe) # type: ignore
else:
return self.populate_buy_trend(dataframe, metadata)
df = self.populate_entry_trend(dataframe, metadata)
if 'enter_long' not in df.columns:
df = df.rename({'buy': 'enter_long', 'buy_tag': 'enter_tag'}, axis='columns')
def advise_sell(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
return df
def advise_exit(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the sell signal for the given dataframe
Based on TA indicators, populates the exit order signal for the given dataframe
This method should not be overridden.
:param dataframe: DataFrame
:param metadata: Additional information dictionary, with details like the
currently traded pair
:return: DataFrame with sell column
"""
logger.debug(f"Populating sell signals for pair {metadata.get('pair')}.")
logger.debug(f"Populating exit signals for pair {metadata.get('pair')}.")
if self._sell_fun_len == 2:
warnings.warn("deprecated - check out the Sample strategy to see "
"the current function headers!", DeprecationWarning)
return self.populate_sell_trend(dataframe) # type: ignore
df = self.populate_sell_trend(dataframe) # type: ignore
else:
return self.populate_sell_trend(dataframe, metadata)
df = self.populate_exit_trend(dataframe, metadata)
if 'exit_long' not in df.columns:
df = df.rename({'sell': 'exit_long'}, axis='columns')
return df