stable/freqtrade/strategy/interface.py

422 lines
17 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
from abc import ABC, abstractmethod
from datetime import datetime
from enum import Enum
2018-12-11 18:47:48 +00:00
from typing import Dict, List, NamedTuple, Tuple
2018-06-15 16:59:34 +00:00
import warnings
2018-03-17 21:44:47 +00:00
import arrow
2018-01-15 08:35:11 +00:00
from pandas import DataFrame
2018-12-26 13:32:17 +00:00
from freqtrade.data.dataprovider import DataProvider
from freqtrade.exchange import timeframe_to_minutes
from freqtrade.persistence import Trade
2018-12-26 13:32:17 +00:00
from freqtrade.wallets import Wallets
logger = logging.getLogger(__name__)
class SignalType(Enum):
"""
Enum to distinguish between buy and sell signals
"""
BUY = "buy"
SELL = "sell"
2018-01-15 08:35:11 +00:00
2018-07-11 17:22:34 +00:00
class SellType(Enum):
"""
Enum to distinguish between sell reasons
"""
ROI = "roi"
STOP_LOSS = "stop_loss"
2018-11-26 17:28:13 +00:00
STOPLOSS_ON_EXCHANGE = "stoploss_on_exchange"
2018-07-11 17:22:34 +00:00
TRAILING_STOP_LOSS = "trailing_stop_loss"
SELL_SIGNAL = "sell_signal"
2018-07-11 17:57:01 +00:00
FORCE_SELL = "force_sell"
2018-07-11 17:59:30 +00:00
NONE = ""
2018-07-11 17:22:34 +00:00
2018-07-12 20:21:52 +00:00
class SellCheckTuple(NamedTuple):
"""
NamedTuple for Sell type + reason
"""
sell_flag: bool
sell_type: SellType
2018-01-15 08:35:11 +00:00
class IStrategy(ABC):
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
2018-05-31 19:59:22 +00:00
ticker_interval -> str: value of the ticker interval to use for the strategy
2018-01-28 05:26:57 +00:00
"""
2018-01-15 08:35:11 +00:00
_populate_fun_len: int = 0
_buy_fun_len: int = 0
_sell_fun_len: int = 0
# 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: float
trailing_stop_positive_offset: float
2019-03-12 14:43:53 +00:00
trailing_only_offset_is_reached = False
2019-01-05 06:10:25 +00:00
# associated ticker interval
2018-05-31 19:59:22 +00:00
ticker_interval: str
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
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.
dp: DataProvider
wallets: Wallets
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] = {}
@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: Raw data from the exchange and parsed by parse_ticker_dataframe()
: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
"""
@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
"""
@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
"""
2018-07-12 18:38:14 +00:00
def informative_pairs(self) -> List[Tuple[str, str]]:
2019-01-21 19:22:27 +00:00
"""
Define additional, informative pair/interval combinations to be cached from the exchange.
2019-01-21 19:22:27 +00:00
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 []
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__
2018-12-11 18:47:48 +00:00
def analyze_ticker(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Parses the given ticker history and returns a populated DataFrame
add several TA indicators and buy signal to it
2019-06-23 20:10:37 +00:00
:return: DataFrame with ticker data and indicator data
"""
2018-09-01 17:53:49 +00:00
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.
2019-02-10 18:02:53 +00:00
logger.debug("TA Analysis Launched")
dataframe = self.advise_indicators(dataframe, metadata)
dataframe = self.advise_buy(dataframe, metadata)
dataframe = self.advise_sell(dataframe, metadata)
2018-09-01 17:50:45 +00:00
self._last_candle_seen_per_pair[pair] = dataframe.iloc[-1]['date']
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
# 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
2018-08-02 18:11:27 +00:00
def get_signal(self, pair: str, interval: str,
2018-12-11 18:47:48 +00:00
dataframe: DataFrame) -> Tuple[bool, bool]:
"""
Calculates current signal based several technical analysis indicators
:param pair: pair in format ANT/BTC
:param interval: Interval to use (in min)
2018-12-11 18:47:48 +00:00
:param dataframe: Dataframe to analyze
:return: (Buy, Sell) A bool-tuple indicating buy/sell signal
"""
if not isinstance(dataframe, DataFrame) or dataframe.empty:
logger.warning('Empty ticker history for pair %s', pair)
return False, False
try:
2018-12-11 18:47:48 +00:00
dataframe = self.analyze_ticker(dataframe, {'pair': pair})
except ValueError as error:
logger.warning(
'Unable to analyze ticker for pair %s: %s',
pair,
str(error)
)
return False, False
except Exception as error:
logger.exception(
'Unexpected error when analyzing ticker for pair %s: %s',
pair,
str(error)
)
return False, False
if dataframe.empty:
logger.warning('Empty dataframe for pair %s', pair)
return False, False
latest = dataframe.iloc[-1]
# Check if dataframe is out of date
signal_date = arrow.get(latest['date'])
interval_minutes = timeframe_to_minutes(interval)
2018-08-20 18:01:57 +00:00
offset = self.config.get('exchange', {}).get('outdated_offset', 5)
if signal_date < (arrow.utcnow().shift(minutes=-(interval_minutes * 2 + offset))):
logger.warning(
'Outdated history for pair %s. Last tick is %s minutes old',
pair,
(arrow.utcnow() - signal_date).seconds // 60
)
return False, False
(buy, sell) = latest[SignalType.BUY.value] == 1, latest[SignalType.SELL.value] == 1
logger.debug(
'trigger: %s (pair=%s) buy=%s sell=%s',
latest['date'],
pair,
str(buy),
str(sell)
)
return buy, sell
2018-07-11 17:57:01 +00:00
def should_sell(self, trade: Trade, rate: float, date: datetime, buy: bool,
2018-11-07 17:15:04 +00:00
sell: bool, low: float = None, high: float = None,
force_stoploss: float = 0) -> SellCheckTuple:
"""
This function evaluate if on the condition required to trigger a sell has been reached
if the threshold is reached and updates the trade record.
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
"""
# Set current rate to low for backtesting sell
2018-10-30 19:23:31 +00:00
current_rate = low or rate
current_profit = trade.calc_profit_percent(current_rate)
2018-11-19 19:02:26 +00:00
trade.adjust_min_max_rates(high 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, high=high)
2018-11-22 15:24:40 +00:00
2018-07-12 20:21:52 +00:00
if stoplossflag.sell_flag:
return stoplossflag
2018-11-22 15:24:40 +00:00
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_percent(current_rate)
experimental = self.config.get('experimental', {})
if buy and experimental.get('ignore_roi_if_buy_signal', False):
logger.debug('Buy signal still active - not selling.')
2018-07-12 20:21:52 +00:00
return SellCheckTuple(sell_flag=False, sell_type=SellType.NONE)
# Check if minimal roi has been reached and no longer in buy conditions (avoiding a fee)
if self.min_roi_reached(trade=trade, current_profit=current_profit, current_time=date):
logger.debug('Required profit reached. Selling..')
2018-07-12 20:21:52 +00:00
return SellCheckTuple(sell_flag=True, sell_type=SellType.ROI)
if experimental.get('sell_profit_only', False):
logger.debug('Checking if trade is profitable..')
if trade.calc_profit(rate=rate) <= 0:
2018-07-12 20:21:52 +00:00
return SellCheckTuple(sell_flag=False, sell_type=SellType.NONE)
if sell and not buy and experimental.get('use_sell_signal', False):
logger.debug('Sell signal received. Selling..')
2018-07-12 20:21:52 +00:00
return SellCheckTuple(sell_flag=True, sell_type=SellType.SELL_SIGNAL)
2018-07-12 20:21:52 +00:00
return SellCheckTuple(sell_flag=False, sell_type=SellType.NONE)
2019-03-23 15:51:36 +00:00
def stop_loss_reached(self, current_rate: float, trade: Trade,
current_time: datetime, current_profit: float,
force_stoploss: float, high: float = None) -> SellCheckTuple:
"""
Based on current profit of the trade and configured (trailing) stoploss,
decides to sell or not
2018-07-16 19:23:35 +00:00
:param current_profit: current profit in percent
"""
trailing_stop = self.config.get('trailing_stop', False)
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)
2019-03-23 15:23:32 +00:00
if trailing_stop:
2019-03-23 15:51:36 +00:00
# trailing stoploss handling
sl_offset = self.config.get('trailing_stop_positive_offset') or 0.0
2019-03-23 15:23:32 +00:00
tsl_only_offset = self.config.get('trailing_only_offset_is_reached', False)
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_percent(high)
2019-03-23 15:48:17 +00:00
# Don't update stoploss if trailing_only_offset_is_reached is true.
if not (tsl_only_offset and high_profit < sl_offset):
2019-03-23 15:51:36 +00:00
# Specific handling for trailing_stop_positive
if 'trailing_stop_positive' in self.config and high_profit > sl_offset:
2019-03-23 15:48:17 +00:00
# Ignore mypy error check in configuration that this is a float
stop_loss_value = self.config.get('trailing_stop_positive') # type: ignore
logger.debug(f"using positive stop loss: {stop_loss_value} "
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
if ((self.stoploss is not None) and
(trade.stop_loss >= current_rate) and
(not self.order_types.get('stoploss_on_exchange'))):
selltype = 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-03-23 15:21:58 +00:00
selltype = SellType.TRAILING_STOP_LOSS
logger.debug(
f"HIT STOP: current price at {current_rate:.6f}, "
f"stop loss is {trade.stop_loss:.6f}, "
f"initial stop loss was at {trade.initial_stop_loss:.6f}, "
f"trade opened at {trade.open_rate:.6f}")
logger.debug(f"trailing stop saved {trade.stop_loss - trade.initial_stop_loss:.6f}")
logger.debug('Stop loss hit.')
return SellCheckTuple(sell_flag=True, sell_type=selltype)
2018-07-12 20:21:52 +00:00
return SellCheckTuple(sell_flag=False, sell_type=SellType.NONE)
def min_roi_reached(self, trade: Trade, current_profit: float, current_time: datetime) -> bool:
"""
Based an earlier trade and current price and ROI configuration, decides whether bot should
sell. Requires current_profit to be in percent!!
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 = (current_time.timestamp() - trade.open_date.timestamp()) / 60
# Get highest entry in ROI dict where key >= trade-duration
roi_entry = max(list(filter(lambda x: trade_dur >= x, self.minimal_roi.keys())))
threshold = self.minimal_roi[roi_entry]
if current_profit > threshold:
return True
return False
def tickerdata_to_dataframe(self, tickerdata: Dict[str, List]) -> Dict[str, DataFrame]:
"""
Creates a dataframe and populates indicators for given ticker data
"""
2018-12-15 13:28:37 +00:00
return {pair: self.advise_indicators(pair_data, {'pair': pair})
for pair, pair_data in tickerdata.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: Raw data from the exchange and parsed by parse_ticker_dataframe()
: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')}.")
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
:param pair: Additional information, 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
:param pair: Additional information, 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)