stable/freqtrade/strategy/interface.py

883 lines
40 KiB
Python
Raw Normal View History

2018-01-28 05:26:57 +00:00
"""
IStrategy interface
This module defines the interface to apply for strategies
"""
import logging
import warnings
from abc import ABC, abstractmethod
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional, Tuple, Union
2018-03-17 21:44:47 +00:00
import arrow
2018-01-15 08:35:11 +00:00
from pandas import DataFrame
from freqtrade.constants import ListPairsWithTimeframes
2018-12-26 13:32:17 +00:00
from freqtrade.data.dataprovider import DataProvider
2021-07-21 13:05:35 +00:00
from freqtrade.enums import SellType, SignalTagType, SignalType
2020-08-24 09:44:32 +00:00
from freqtrade.exceptions import OperationalException, StrategyError
2021-01-05 13:49:35 +00:00
from freqtrade.exchange import timeframe_to_minutes, timeframe_to_seconds
2020-08-24 09:44:32 +00:00
from freqtrade.exchange.exchange import timeframe_to_next_date
2020-10-25 09:54:30 +00:00
from freqtrade.persistence import PairLocks, Trade
from freqtrade.strategy.hyper import HyperStrategyMixin
2021-09-12 15:26:41 +00:00
from freqtrade.strategy.informative_decorator import (InformativeData, PopulateIndicators,
_create_and_merge_informative_pair,
_format_pair_name)
from freqtrade.strategy.strategy_wrapper import strategy_safe_wrapper
2018-12-26 13:32:17 +00:00
from freqtrade.wallets import Wallets
2020-09-28 17:39:41 +00:00
logger = logging.getLogger(__name__)
CUSTOM_SELL_MAX_LENGTH = 64
class SellCheckTuple(object):
2018-07-12 20:21:52 +00:00
"""
NamedTuple for Sell type + reason
"""
sell_type: SellType
2021-04-26 07:42:24 +00:00
sell_reason: str = ''
2021-04-26 07:42:24 +00:00
def __init__(self, sell_type: SellType, sell_reason: str = ''):
self.sell_type = sell_type
self.sell_reason = sell_reason or sell_type.value
2018-07-12 20:21:52 +00:00
@property
def sell_flag(self):
return self.sell_type != SellType.NONE
2018-07-12 20:21:52 +00:00
class IStrategy(ABC, HyperStrategyMixin):
2018-01-28 05:26:57 +00:00
"""
Interface for freqtrade strategies
Defines the mandatory structure must follow any custom strategies
Attributes you can use:
minimal_roi -> Dict: Minimal ROI designed for the strategy
stoploss -> float: optimal stoploss designed for the strategy
timeframe -> str: value of the timeframe (ticker interval) to use with the strategy
2018-01-28 05:26:57 +00:00
"""
2019-08-26 17:44:33 +00:00
# Strategy interface version
# Default to version 2
# Version 1 is the initial interface without metadata dict
2019-08-26 17:44:33 +00:00
# Version 2 populate_* include metadata dict
INTERFACE_VERSION: int = 2
2018-01-15 08:35:11 +00:00
_populate_fun_len: int = 0
_buy_fun_len: int = 0
_sell_fun_len: int = 0
2021-06-29 05:07:34 +00:00
_ft_params_from_file: Dict = {}
# associated minimal roi
2018-05-31 19:59:22 +00:00
minimal_roi: Dict
# associated stoploss
2018-05-31 19:59:22 +00:00
stoploss: float
2019-01-05 06:10:25 +00:00
# trailing stoploss
trailing_stop: bool = False
trailing_stop_positive: Optional[float] = None
2019-10-11 06:55:31 +00:00
trailing_stop_positive_offset: float = 0.0
2019-03-12 14:43:53 +00:00
trailing_only_offset_is_reached = False
2020-12-20 10:12:22 +00:00
use_custom_stoploss: bool = False
2019-01-05 06:10:25 +00:00
2020-06-02 07:50:56 +00:00
# associated timeframe
ticker_interval: str # DEPRECATED
2020-06-02 07:50:56 +00:00
timeframe: str
2018-05-31 19:59:22 +00:00
2018-11-15 05:58:24 +00:00
# Optional order types
order_types: Dict = {
'buy': 'limit',
'sell': 'limit',
'stoploss': 'limit',
'stoploss_on_exchange': False,
'stoploss_on_exchange_interval': 60,
2018-11-15 05:58:24 +00:00
}
2018-11-25 21:02:59 +00:00
# Optional time in force
order_time_in_force: Dict = {
'buy': 'gtc',
'sell': 'gtc',
}
2018-08-09 17:24:00 +00:00
# run "populate_indicators" only for new candle
process_only_new_candles: bool = False
2018-08-09 17:24:00 +00:00
use_sell_signal: bool
sell_profit_only: bool
sell_profit_offset: float
ignore_roi_if_buy_signal: bool
# Number of seconds after which the candle will no longer result in a buy on expired candles
ignore_buying_expired_candle_after: int = 0
# Disable checking the dataframe (converts the error into a warning message)
disable_dataframe_checks: bool = False
# Count of candles the strategy requires before producing valid signals
startup_candle_count: int = 0
# Protections
protections: List = []
2018-12-26 13:32:17 +00:00
# Class level variables (intentional) containing
# the dataprovider (dp) (access to other candles, historic data, ...)
# and wallets - access to the current balance.
2021-09-12 15:26:41 +00:00
dp: Optional[DataProvider]
wallets: Optional[Wallets] = None
# Filled from configuration
stake_currency: str
2020-10-02 04:41:28 +00:00
# container variable for strategy source code
__source__: str = ''
2018-12-26 13:32:17 +00:00
# Definition of plot_config. See plotting documentation for more details.
plot_config: Dict = {}
2020-01-04 10:14:00 +00:00
def __init__(self, config: dict) -> None:
self.config = config
# Dict to determine if analysis is necessary
self._last_candle_seen_per_pair: Dict[str, datetime] = {}
super().__init__(config)
# Gather informative pairs from @informative-decorated methods.
self._ft_informative: List[Tuple[InformativeData, PopulateIndicators]] = []
for attr_name in dir(self.__class__):
cls_method = getattr(self.__class__, attr_name)
if not callable(cls_method):
continue
informative_data_list = getattr(cls_method, '_ft_informative', None)
if not isinstance(informative_data_list, list):
# Type check is required because mocker would return a mock object that evaluates to
# True, confusing this code.
continue
2021-09-05 06:54:05 +00:00
strategy_timeframe_minutes = timeframe_to_minutes(self.timeframe)
for informative_data in informative_data_list:
if timeframe_to_minutes(informative_data.timeframe) < strategy_timeframe_minutes:
2021-09-05 06:54:05 +00:00
raise OperationalException('Informative timeframe must be equal or higher than '
'strategy timeframe!')
self._ft_informative.append((informative_data, cls_method))
@abstractmethod
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
2018-01-15 08:35:11 +00:00
"""
Populate indicators that will be used in the Buy and Sell strategy
:param dataframe: DataFrame with data from the exchange
:param metadata: Additional information, like the currently traded pair
2018-01-15 08:35:11 +00:00
:return: a Dataframe with all mandatory indicators for the strategies
"""
return dataframe
2018-01-15 08:35:11 +00:00
@abstractmethod
def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
2018-01-15 08:35:11 +00:00
"""
Based on TA indicators, populates the buy signal for the given dataframe
:param dataframe: DataFrame
:param metadata: Additional information, like the currently traded pair
2018-01-15 08:35:11 +00:00
:return: DataFrame with buy column
"""
return dataframe
2018-01-15 08:35:11 +00:00
@abstractmethod
def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
2018-01-15 08:35:11 +00:00
"""
Based on TA indicators, populates the sell signal for the given dataframe
:param dataframe: DataFrame
:param metadata: Additional information, like the currently traded pair
2018-03-25 18:24:56 +00:00
:return: DataFrame with sell column
2018-01-15 08:35:11 +00:00
"""
return dataframe
2018-07-12 18:38:14 +00:00
def check_buy_timeout(self, pair: str, trade: Trade, order: dict, **kwargs) -> bool:
2020-02-06 19:30:17 +00:00
"""
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,
and is not yet fully filled.
Configuration options in `unfilledtimeout` will be verified before this,
so ensure to set these timeouts high enough.
When not implemented by a strategy, this simply returns False.
:param pair: Pair the trade is for
:param trade: trade object.
:param order: Order dictionary as returned from CCXT.
: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 False
def check_sell_timeout(self, pair: str, trade: Trade, order: dict, **kwargs) -> bool:
2020-02-21 19:27:13 +00:00
"""
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.
Configuration options in `unfilledtimeout` will be verified before this,
so ensure to set these timeouts high enough.
When not implemented by a strategy, this simply returns False.
:param pair: Pair the trade is for
:param trade: trade object.
:param order: Order dictionary as returned from CCXT.
: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 False
2020-06-14 04:52:11 +00:00
def bot_loop_start(self, **kwargs) -> None:
"""
Called at the start of the bot iteration (one loop).
Might be used to perform pair-independent tasks
(e.g. gather some remote resource for comparison)
2020-06-14 04:52:11 +00:00
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
"""
pass
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, **kwargs) -> bool:
"""
Called right before placing a buy order.
Timing for this function is critical, so avoid doing heavy computations or
2020-06-14 08:49:15 +00:00
network requests in this method.
For full documentation please go to https://www.freqtrade.io/en/latest/strategy-advanced/
When not implemented by a strategy, returns True (always confirming).
:param pair: Pair that's about to be bought.
: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 **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
"""
return True
2020-06-18 17:46:03 +00:00
def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float,
rate: float, time_in_force: str, sell_reason: str,
current_time: datetime, **kwargs) -> bool:
"""
Called right before placing a regular sell order.
Timing for this function is critical, so avoid doing heavy computations or
2020-06-14 08:49:15 +00:00
network requests in this method.
For full documentation please go to https://www.freqtrade.io/en/latest/strategy-advanced/
When not implemented by a strategy, returns True (always confirming).
:param pair: Pair that's about to be sold.
: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.
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.
False aborts the process
"""
return True
2020-12-20 10:17:50 +00:00
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
current_profit: float, **kwargs) -> float:
2020-12-09 06:45:41 +00:00
"""
2020-12-19 12:18:06 +00:00
Custom stoploss logic, returning the new distance relative to current_rate (as ratio).
e.g. returning -0.05 would create a stoploss 5% below current_rate.
2020-12-09 06:45:41 +00:00
The custom stoploss can never be below self.stoploss, which serves as a hard maximum loss.
For full documentation please go to https://www.freqtrade.io/en/latest/strategy-advanced/
When not implemented by a strategy, returns the initial stoploss value
2020-12-20 10:12:22 +00:00
Only called when use_custom_stoploss is set to True.
2020-12-09 06:45:41 +00:00
:param pair: Pair that's currently analyzed
2020-12-09 06:45:41 +00:00
:param trade: trade object.
2020-12-19 10:58:42 +00:00
:param current_time: datetime object, containing the current datetime
:param current_rate: Rate, calculated based on pricing settings in ask_strategy.
:param current_profit: Current profit (as ratio), calculated based on current_rate.
2020-12-09 06:45:41 +00:00
:param **kwargs: Ensure to keep this here so updates to this won't break your strategy.
2021-06-25 13:45:49 +00:00
:return float: New stoploss value, relative to the current_rate
2020-12-09 06:45:41 +00:00
"""
return self.stoploss
def custom_entry_price(self, pair: str, current_time: datetime, proposed_rate: float,
**kwargs) -> float:
"""
Custom entry price logic, returning the new entry price.
For full documentation please go to https://www.freqtrade.io/en/latest/strategy-advanced/
When not implemented by a strategy, returns None, orderbook is used to set entry price
: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 **kwargs: Ensure to keep this here so updates to this won't break your strategy.
:return float: New entry price value if provided
"""
return proposed_rate
def custom_exit_price(self, pair: str, trade: Trade,
current_time: datetime, proposed_rate: float,
current_profit: float, **kwargs) -> float:
"""
Custom exit price logic, returning the new exit price.
For full documentation please go to https://www.freqtrade.io/en/latest/strategy-advanced/
When not implemented by a strategy, returns None, orderbook is used to set exit price
: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 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
"""
return proposed_rate
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.
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,
or a custom 1:2 risk-reward ROI.
Custom sell 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_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
None or False.
"""
return None
def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
proposed_stake: float, min_stake: float, max_stake: float,
**kwargs) -> float:
"""
Customize stake size 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_stake: A stake amount proposed by the bot.
:param min_stake: Minimal stake size allowed by exchange.
:param max_stake: Balance available for trading.
:return: A stake size, which is between min_stake and max_stake.
"""
return proposed_stake
def informative_pairs(self) -> ListPairsWithTimeframes:
2019-01-21 19:22:27 +00:00
"""
Define additional, informative pair/interval combinations to be cached from the exchange.
2021-06-25 13:45:49 +00:00
These pair/interval combinations are non-tradable, unless they are part
2019-01-21 19:22:27 +00:00
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 []
2020-06-13 05:09:44 +00:00
###
# END - Intended to be overridden by strategy
###
def gather_informative_pairs(self) -> ListPairsWithTimeframes:
"""
Internal method which gathers all informative pairs (user or automatically defined).
"""
informative_pairs = self.informative_pairs()
for inf_data, _ in self._ft_informative:
if inf_data.asset:
pair_tf = (_format_pair_name(self.config, inf_data.asset), inf_data.timeframe)
informative_pairs.append(pair_tf)
else:
2021-09-12 15:26:41 +00:00
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))
return list(set(informative_pairs))
2018-07-12 18:38:14 +00:00
def get_strategy_name(self) -> str:
"""
Returns strategy class name
"""
2018-07-19 17:41:42 +00:00
return self.__class__.__name__
2020-10-17 09:28:34 +00:00
def lock_pair(self, pair: str, until: datetime, reason: str = None) -> None:
2019-08-12 14:29:09 +00:00
"""
Locks pair until a given timestamp happens.
Locked pairs are not analyzed, and are prevented from opening new trades.
2019-12-22 08:46:00 +00:00
Locks can only count up (allowing users to lock pairs for a longer period of time).
To remove a lock from a pair, use `unlock_pair()`
2019-08-12 17:50:22 +00:00
:param pair: Pair to lock
:param until: datetime in UTC until the pair should be blocked from opening new trades.
Needs to be timezone aware `datetime.now(timezone.utc)`
2020-10-21 17:35:57 +00:00
:param reason: Optional string explaining why the pair was locked.
2019-08-12 14:29:09 +00:00
"""
2020-10-25 09:54:30 +00:00
PairLocks.lock_pair(pair, until, reason)
2019-12-22 08:46:00 +00:00
2020-02-02 04:00:40 +00:00
def unlock_pair(self, pair: str) -> None:
2019-12-22 08:46:00 +00:00
"""
Unlocks a pair previously locked using lock_pair.
Not used by freqtrade itself, but intended to be used if users lock pairs
manually from within the strategy, to allow an easy way to unlock pairs.
:param pair: Unlock pair to allow trading again
"""
2020-10-25 09:54:30 +00:00
PairLocks.unlock_pair(pair, datetime.now(timezone.utc))
2019-08-12 17:50:22 +00:00
2020-08-24 09:09:09 +00:00
def is_pair_locked(self, pair: str, candle_date: datetime = None) -> bool:
2019-08-12 17:50:22 +00:00
"""
Checks if a pair is currently locked
2020-08-24 15:18:57 +00:00
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.
2021-06-25 17:13:31 +00:00
:param pair: "Pair to check"
2020-08-24 15:18:57 +00:00
:param candle_date: Date of the last candle. Optional, defaults to current date
2020-08-24 15:31:00 +00:00
:returns: locking state of the pair in question.
2019-08-12 17:50:22 +00:00
"""
2020-10-17 09:28:34 +00:00
2020-08-24 09:09:09 +00:00
if not candle_date:
2020-10-17 09:28:34 +00:00
# Simple call ...
return PairLocks.is_pair_locked(pair)
2020-08-24 09:09:09 +00:00
else:
lock_time = timeframe_to_next_date(self.timeframe, candle_date)
2020-10-25 09:54:30 +00:00
return PairLocks.is_pair_locked(pair, lock_time)
2019-08-12 14:29:09 +00:00
2018-12-11 18:47:48 +00:00
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
: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)
return dataframe
def _analyze_ticker_internal(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
WARNING: Used internally only, may skip analysis if `process_only_new_candles` is set.
: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
"""
pair = str(metadata.get('pair'))
2018-09-01 17:53:49 +00:00
# Test if seen this pair and last candle before.
2018-12-13 18:43:17 +00:00
# always run if process_only_new_candles is set to false
if (not self.process_only_new_candles or
2018-09-01 17:50:45 +00:00
self._last_candle_seen_per_pair.get(pair, None) != dataframe.iloc[-1]['date']):
# Defs that only make change on new candle data.
dataframe = self.analyze_ticker(dataframe, metadata)
2018-09-01 17:50:45 +00:00
self._last_candle_seen_per_pair[pair] = dataframe.iloc[-1]['date']
2020-06-15 12:08:57 +00:00
if self.dp:
self.dp._set_cached_df(pair, self.timeframe, dataframe)
else:
2019-02-13 09:42:39 +00:00
logger.debug("Skipping TA Analysis for already analyzed candle")
2018-08-09 17:53:47 +00:00
dataframe['buy'] = 0
dataframe['sell'] = 0
2021-07-23 04:42:43 +00:00
dataframe['buy_tag'] = None
dataframe['exit_tag'] = None
# Other Defs in strategy that want to be called every loop here
# twitter_sell = self.watch_twitter_feed(dataframe, metadata)
2019-02-10 18:02:53 +00:00
logger.debug("Loop Analysis Launched")
return dataframe
def analyze_pair(self, pair: str) -> None:
"""
Fetch data for this pair from dataprovider and analyze.
Stores the dataframe into the dataprovider.
The analyzed dataframe is then accessible via `dp.get_analyzed_dataframe()`.
:param pair: Pair to analyze.
"""
2020-06-15 12:08:57 +00:00
if not self.dp:
raise OperationalException("DataProvider not found.")
dataframe = self.dp.ohlcv(pair, self.timeframe)
if not isinstance(dataframe, DataFrame) or dataframe.empty:
logger.warning('Empty candle (OHLCV) data for pair %s', pair)
return
try:
df_len, df_close, df_date = self.preserve_df(dataframe)
dataframe = strategy_safe_wrapper(
self._analyze_ticker_internal, message=""
)(dataframe, {'pair': pair})
self.assert_df(dataframe, df_len, df_close, df_date)
except StrategyError as error:
logger.warning(f"Unable to analyze candle (OHLCV) data for pair {pair}: {error}")
return
if dataframe.empty:
logger.warning('Empty dataframe for pair %s', pair)
return
def analyze(self, pairs: List[str]) -> None:
"""
Analyze all pairs using analyze_pair().
:param pairs: List of pairs to analyze
"""
for pair in pairs:
self.analyze_pair(pair)
@staticmethod
2020-03-26 06:05:30 +00:00
def preserve_df(dataframe: DataFrame) -> Tuple[int, float, datetime]:
""" keep some data for dataframes """
2020-03-26 06:05:30 +00:00
return len(dataframe), dataframe["close"].iloc[-1], dataframe["date"].iloc[-1]
def assert_df(self, dataframe: DataFrame, df_len: int, df_close: float, df_date: datetime):
"""
Ensure dataframe (length, last candle) was not modified, and has all elements we need.
"""
message_template = "Dataframe returned from strategy has mismatching {}."
2020-03-26 06:05:30 +00:00
message = ""
if dataframe is None:
message = "No dataframe returned (return statement missing?)."
elif 'buy' not in dataframe:
message = "Buy column not set."
elif df_len != len(dataframe):
message = message_template.format("length")
2020-03-26 06:05:30 +00:00
elif df_close != dataframe["close"].iloc[-1]:
message = message_template.format("last close price")
2020-03-26 06:05:30 +00:00
elif df_date != dataframe["date"].iloc[-1]:
message = message_template.format("last date")
2020-03-26 06:05:30 +00:00
if message:
if self.disable_dataframe_checks:
logger.warning(message)
else:
raise StrategyError(message)
2021-07-21 19:23:34 +00:00
def get_signal(
self,
pair: str,
timeframe: str,
dataframe: DataFrame
) -> Tuple[bool, bool, Optional[str], Optional[str]]:
"""
2020-06-14 05:27:13 +00:00
Calculates current signal based based on the buy / sell columns of the dataframe.
Used by Bot to get the signal to buy or sell
:param pair: pair in format ANT/BTC
:param timeframe: timeframe to use
2020-06-18 06:01:09 +00:00
:param dataframe: Analyzed dataframe to get signal from.
:return: (Buy, Sell) A bool-tuple indicating buy/sell signal
"""
if not isinstance(dataframe, DataFrame) or dataframe.empty:
2020-06-18 05:03:30 +00:00
logger.warning(f'Empty candle (OHLCV) data for pair {pair}')
2021-10-12 22:22:53 +00:00
return False, False, None, None
2020-05-19 19:20:53 +00:00
latest_date = dataframe['date'].max()
latest = dataframe.loc[dataframe['date'] == latest_date].iloc[-1]
# Explicitly convert to arrow object to ensure the below comparison does not fail
2020-05-26 10:54:45 +00:00
latest_date = arrow.get(latest_date)
2020-03-11 16:28:03 +00:00
# Check if dataframe is out of date
timeframe_minutes = timeframe_to_minutes(timeframe)
2018-08-20 18:01:57 +00:00
offset = self.config.get('exchange', {}).get('outdated_offset', 5)
if latest_date < (arrow.utcnow().shift(minutes=-(timeframe_minutes * 2 + offset))):
logger.warning(
'Outdated history for pair %s. Last tick is %s minutes old',
pair, int((arrow.utcnow() - latest_date).total_seconds() // 60)
)
2021-10-12 22:22:53 +00:00
return False, False, None, None
buy = latest[SignalType.BUY.value] == 1
sell = False
if SignalType.SELL.value in latest:
sell = latest[SignalType.SELL.value] == 1
buy_tag = latest.get(SignalTagType.BUY_TAG.value, None)
exit_tag = latest.get(SignalTagType.EXIT_TAG.value, None)
2020-06-12 12:02:21 +00:00
logger.debug('trigger: %s (pair=%s) buy=%s sell=%s',
latest['date'], pair, str(buy), str(sell))
2021-01-05 13:49:35 +00:00
timeframe_seconds = timeframe_to_seconds(timeframe)
2021-01-05 14:30:29 +00:00
if self.ignore_expired_candle(latest_date=latest_date,
current_time=datetime.now(timezone.utc),
2021-01-05 14:30:29 +00:00
timeframe_seconds=timeframe_seconds,
buy=buy):
return False, sell, buy_tag, exit_tag
return buy, sell, buy_tag, exit_tag
def ignore_expired_candle(self, latest_date: datetime, current_time: datetime,
timeframe_seconds: int, buy: bool):
2021-01-12 07:24:11 +00:00
if self.ignore_buying_expired_candle_after and buy:
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, date: datetime, buy: bool,
sell: bool, low: float = None, high: float = None,
2018-11-07 17:15:04 +00:00
force_stoploss: float = 0) -> SellCheckTuple:
"""
2019-08-14 04:07:03 +00:00
This function evaluates if one of the conditions required to trigger a sell
2019-08-12 17:50:22 +00:00
has been reached, which can either be a stop-loss, ROI or sell-signal.
2019-03-17 15:02:13 +00:00
:param low: Only used during backtesting to simulate stoploss
:param high: Only used during backtesting, to simulate ROI
:param force_stoploss: Externally provided stoploss
:return: True if trade should be sold, False otherwise
"""
current_rate = rate
current_profit = trade.calc_profit_ratio(current_rate)
2018-11-19 19:02:26 +00:00
trade.adjust_min_max_rates(high or current_rate, low or current_rate)
2019-03-16 18:54:34 +00:00
stoplossflag = self.stop_loss_reached(current_rate=current_rate, trade=trade,
current_time=date, current_profit=current_profit,
force_stoploss=force_stoploss, low=low, high=high)
2019-03-17 15:02:13 +00:00
# Set current rate to high for backtesting sell
2018-10-30 19:23:31 +00:00
current_rate = high or rate
current_profit = trade.calc_profit_ratio(current_rate)
2020-11-27 08:18:03 +00:00
# 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)
2020-11-27 08:18:03 +00:00
and self.min_roi_reached(trade=trade, current_profit=current_profit,
current_time=date))
sell_signal = SellType.NONE
2021-04-26 07:42:24 +00:00
custom_reason = ''
# use provided rate in backtesting, not high/low.
current_rate = rate
current_profit = trade.calc_profit_ratio(current_rate)
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
else:
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)
if custom_reason:
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.')
custom_reason = custom_reason[:CUSTOM_SELL_MAX_LENGTH]
else:
custom_reason = None
2020-11-27 08:18:03 +00:00
# TODO: return here if sell-signal should be favored over ROI
# Start evaluations
# Sequence:
# ROI (if not stoploss)
# Sell-signal
# 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 sell_signal != SellType.NONE:
logger.debug(f"{trade.pair} - Sell signal received. "
f"sell_type=SellType.{sell_signal.name}" +
(f", custom_reason={custom_reason}" if custom_reason else ""))
return SellCheckTuple(sell_type=sell_signal, sell_reason=custom_reason)
2020-11-27 08:18:03 +00:00
if stoplossflag.sell_flag:
logger.debug(f"{trade.pair} - Stoploss hit. sell_type={stoplossflag.sell_type}")
2020-11-27 08:18:03 +00:00
return stoplossflag
2019-09-10 07:42:45 +00:00
# This one is noisy, commented out...
# logger.debug(f"{trade.pair} - No sell signal.")
return SellCheckTuple(sell_type=SellType.NONE)
def stop_loss_reached(self, current_rate: float, trade: Trade,
2019-03-23 15:51:36 +00:00
current_time: datetime, current_profit: float,
force_stoploss: float, low: float = None,
high: float = None) -> SellCheckTuple:
"""
Based on current profit of the trade and configured (trailing) stoploss,
decides to sell or not
2020-02-28 09:36:39 +00:00
: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
"""
2019-03-23 15:23:32 +00:00
stop_loss_value = force_stoploss if force_stoploss else self.stoploss
2019-03-23 15:48:17 +00:00
# Initiate stoploss with open_rate. Does nothing if stoploss is already set.
2019-03-23 15:23:32 +00:00
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):
2020-12-20 10:17:50 +00:00
stop_loss_value = strategy_safe_wrapper(self.custom_stoploss, default_retval=None
)(pair=trade.pair, trade=trade,
current_time=current_time,
current_rate=current_rate,
current_profit=current_profit)
# Sanity check - error cases will return None
if stop_loss_value:
# logger.info(f"{trade.pair} {stop_loss_value=} {current_profit=}")
trade.adjust_stop_loss(current_rate, stop_loss_value)
else:
logger.warning("CustomStoploss function did not return valid stoploss")
if self.trailing_stop and trade.stop_loss < (low or current_rate):
2019-03-23 15:51:36 +00:00
# trailing stoploss handling
2019-10-11 06:55:31 +00:00
sl_offset = self.trailing_stop_positive_offset
2018-07-16 19:23:35 +00:00
# Make sure current_profit is calculated using high for backtesting.
high_profit = current_profit if not high else trade.calc_profit_ratio(high)
2019-03-23 15:48:17 +00:00
# Don't update stoploss if trailing_only_offset_is_reached is true.
2019-10-11 06:55:31 +00:00
if not (self.trailing_only_offset_is_reached and high_profit < sl_offset):
2019-10-13 07:54:03 +00:00
# Specific handling for trailing_stop_positive
if self.trailing_stop_positive is not None and high_profit > sl_offset:
stop_loss_value = self.trailing_stop_positive
2019-09-11 20:32:08 +00:00
logger.debug(f"{trade.pair} - Using positive stoploss: {stop_loss_value} "
2019-03-23 15:48:17 +00:00
f"offset: {sl_offset:.4g} profit: {current_profit:.4f}%")
trade.adjust_stop_loss(high or current_rate, stop_loss_value)
2019-03-23 15:21:58 +00:00
# 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
(not self.order_types.get('stoploss_on_exchange') or self.config['dry_run'])):
2019-03-23 15:21:58 +00:00
2019-09-10 07:42:45 +00:00
sell_type = SellType.STOP_LOSS
2019-06-02 11:27:31 +00:00
# If initial stoploss is not the same as current one then it is trailing.
if trade.initial_stop_loss != trade.stop_loss:
2019-09-10 07:42:45 +00:00
sell_type = SellType.TRAILING_STOP_LOSS
2019-03-23 15:21:58 +00:00
logger.debug(
f"{trade.pair} - HIT STOP: current price at {(low or current_rate):.6f}, "
2019-09-10 07:42:45 +00:00
f"stoploss is {trade.stop_loss:.6f}, "
f"initial stoploss was at {trade.initial_stop_loss:.6f}, "
2019-03-23 15:21:58 +00:00
f"trade opened at {trade.open_rate:.6f}")
2019-09-11 20:32:08 +00:00
logger.debug(f"{trade.pair} - Trailing stop saved "
2019-09-10 07:42:45 +00:00
f"{trade.stop_loss - trade.initial_stop_loss:.6f}")
2019-03-23 15:21:58 +00:00
return SellCheckTuple(sell_type=sell_type)
2019-03-23 15:21:58 +00:00
return SellCheckTuple(sell_type=SellType.NONE)
2019-12-07 14:18:12 +00:00
def min_roi_reached_entry(self, trade_dur: int) -> Tuple[Optional[int], Optional[float]]:
"""
Based on trade duration defines the ROI entry that may have been reached.
:param trade_dur: trade duration in minutes
:return: minimal ROI entry value or None if none proper ROI entry was found.
"""
2019-06-20 00:26:25 +00:00
# Get highest entry in ROI dict where key <= trade-duration
roi_list = list(filter(lambda x: x <= trade_dur, self.minimal_roi.keys()))
if not roi_list:
2019-12-07 14:18:12 +00:00
return None, None
2019-06-20 00:26:25 +00:00
roi_entry = max(roi_list)
2019-12-07 14:18:12 +00:00
return roi_entry, self.minimal_roi[roi_entry]
def min_roi_reached(self, trade: Trade, current_profit: float, current_time: datetime) -> bool:
"""
2020-02-28 09:36:39 +00:00
Based on trade duration, current profit of the trade and ROI configuration,
decides whether bot should sell.
:param current_profit: current profit as ratio
2019-06-23 20:10:37 +00:00
:return: True if bot should sell at current rate
"""
# Check if time matches and current rate is above threshold
trade_dur = int((current_time.timestamp() - trade.open_date_utc.timestamp()) // 60)
2019-12-07 14:18:12 +00:00
_, roi = self.min_roi_reached_entry(trade_dur)
if roi is None:
return False
else:
return current_profit > roi
def advise_all_indicators(self, data: Dict[str, DataFrame]) -> Dict[str, DataFrame]:
"""
2020-06-13 05:09:44 +00:00
Populates indicators for given candle (OHLCV) data (for multiple pairs)
Does not run advise_buy or advise_sell!
2019-10-27 09:56:38 +00:00
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.
Has positive effects on memory usage for whatever reason - also when
using only one strategy.
"""
return {pair: self.advise_indicators(pair_data.copy(), {'pair': pair}).copy()
for pair, pair_data in data.items()}
def advise_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Populate indicators that will be used in the Buy and Sell strategy
This method should not be overridden.
: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
"""
logger.debug(f"Populating indicators for pair {metadata.get('pair')}.")
# call populate_indicators_Nm() which were tagged with @informative decorator.
for inf_data, populate_fn in self._ft_informative:
dataframe = _create_and_merge_informative_pair(
self, dataframe, metadata, inf_data, populate_fn)
if self._populate_fun_len == 2:
warnings.warn("deprecated - check out the Sample strategy to see "
"the current function headers!", DeprecationWarning)
return self.populate_indicators(dataframe) # type: ignore
else:
return self.populate_indicators(dataframe, metadata)
def advise_buy(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the buy signal for the given dataframe
This method should not be overridden.
:param dataframe: DataFrame
2021-06-25 17:13:31 +00:00
: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')}.")
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
else:
return self.populate_buy_trend(dataframe, metadata)
def advise_sell(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the sell signal for the given dataframe
This method should not be overridden.
:param dataframe: DataFrame
2021-06-25 17:13:31 +00:00
: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')}.")
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
else:
return self.populate_sell_trend(dataframe, metadata)