# EXPERIMENTAL CODE # $ freqtrade hyperopt --userdir user_data \ # --config user_data/config.json --config user_data/HRPStrategy-pairs.json \ # --strategy HRPStrategy --timerange 20230101- --spaces buy sell \ # --epochs 1024 --print-all --disable-param-export \ # --hyperopt-loss CalmarHyperOptLoss --timeframe-detail 1h import logging from datetime import datetime, timedelta from typing import Callable, Dict, Iterable, List, Optional, Tuple, Union import numpy as np import riskfolio as rp import talib.abstract as ta from freqtrade.constants import Config from freqtrade.exchange import timeframe_to_minutes, timeframe_to_prev_date from freqtrade.optimize.space import Categorical, Dimension, SKDecimal from freqtrade.persistence import Trade from freqtrade.strategy import CategoricalParameter, IntParameter, IStrategy from pandas import DataFrame, Series logger = logging.getLogger(__name__) # fmt: on RISK_MEASURES = ["vol", "MV", "MAD", "GMD", "MSV", "FLPM", "SLPM", "VaR", "CVaR", "TG", "EVaR", "WR", "RG", "CVRG", "TGRG", "MDD", "ADD", "DaR", "CDaR", "EDaR", "UCI", "MDD_Rel", "ADD_Rel", "DaR_Rel", "CDaR_Rel", "EDaR_Rel", "UCI_Rel", ] # fmt: off class HRPStrategy(IStrategy): INTERFACE_VERSION: int = 3 timeframe: str = "4h" can_short: bool = False process_only_new_candles: bool = True use_exit_signal: bool = True ignore_buying_expired_candle_after: int = 600 position_adjustment_enable: bool = True startup_candle_count: int = 500 minimal_roi: Dict[str, float] = {} stoploss: float = -1.0 entry_dayofweek = IntParameter(0, 6, default=0, space="buy") rebalance_candle = CategoricalParameter(range(50, 351, 50), default=100, space="buy") history_length = CategoricalParameter(range(100, 501, 100), default=200, space="buy") risk_measure = CategoricalParameter(RISK_MEASURES, default="MV", space="buy", optimize=False) custom_stop_loss = CategoricalParameter( np.arange(-0.5, 0, 0.025).round(3), default=-0.5, space="sell" ) custom_trailing_stop = CategoricalParameter( np.arange(0.025, 0.175, 0.025).round(3), default=0.05, space="sell" ) custom_trailing_offset = CategoricalParameter( np.arange(0.125, 0.225, 0.025).round(3), default=0.2, space="sell" ) def __init__(self, config: Config) -> None: super().__init__(config) self.config = config for idx in range(self.config["max_open_trades"]): setattr( self, f"pair_{idx:02d}", CategoricalParameter( self.config["exchange"]["pair_whitelist"], default="ETH/BTC", space="buy" ) ) @property def selected_pairs(self): pairs = [getattr(self, f"pair_{i:02d}").value for i in range(self.max_open_trades)] if len(set(pairs)) < self.max_open_trades: return [] return pairs def populate_indicators(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: dataframe["dayofweek"] = dataframe["date"].dt.dayofweek dataframe["hour"] = dataframe["date"].dt.hour return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: dataframe.loc[ ((dataframe["dayofweek"] == self.entry_dayofweek.value) & (dataframe["hour"] == 0)), "enter_long", ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: Dict) -> DataFrame: return dataframe def is_fresh_candle(self, current_time: datetime, delta_minutes: int = 5) -> bool: prev_candle_time = timeframe_to_prev_date(self.timeframe, current_time) if (current_time - prev_candle_time) >= timedelta(minutes=delta_minutes): return False return True def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, current_time: datetime, **kwargs) -> bool: if not self.is_fresh_candle(current_time): return False return True def custom_exit(self, pair: str, trade: "Trade", current_time: "datetime", current_rate: float, current_profit: float, **kwargs): if not self.is_fresh_candle(current_time): return None dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) current_candle = dataframe.iloc[-1].squeeze() current_rate = current_candle["close"] current_profit = trade.calc_profit_ratio(current_rate) if current_profit < self.custom_stop_loss.value: logging.info( f"{current_time} - custom_exit - {pair} - " f"stop_loss with current_profit: {current_profit:.08f}" ) return "custom_stop_loss" max_rate = trade.min_rate if trade.is_short else trade.max_rate max_profit = trade.calc_profit_ratio(max_rate) if not max_profit > self.custom_trailing_offset.value: return None if (max_profit - current_profit) < self.custom_trailing_stop.value: logging.info( f"{current_time} - custom_exit - {pair} - " f"trailing stop with current_profit: {current_profit:.08f}" ) return "custom_trailing_stop" return None def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: weights = self.calculate_weights() pair_weight = weights.get(pair, 0.0) if pair_weight == 0: return 0.0 capital = self.wallets.get_total_stake_amount() stake_amount = pair_weight * capital logging.info( f"{current_time} - custom_stake_amount - {pair} - " f"pair_weight: {pair_weight:.03f}, " f"stake_amount: {stake_amount:.08f}") return stake_amount def adjust_trade_position(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: Optional[float], max_stake: float, current_entry_rate: float, current_exit_rate: float, current_entry_profit: float, current_exit_profit: float, **kwargs ) -> Union[Optional[float], Tuple[Optional[float], Optional[str]]]: tf_minutes = timeframe_to_minutes(self.timeframe) trade_date = max(t.date_last_filled_utc for t in Trade.get_trades_proxy(is_open=True)) trade_dur = int((current_time - trade_date).total_seconds() // 60) if trade_dur < self.rebalance_candle.value * tf_minutes: return None weights = self.calculate_weights() pair_weight = weights.get(trade.pair, 0.0) if pair_weight <= 0: return -trade.stake_amount capital = self.wallets.get_total_stake_amount() pair_allocation = pair_weight * capital adjustment_amount = pair_allocation - trade.stake_amount logging.info( f"{current_time} - adjust_trade_position - {trade.pair} - " f"pair_weight: {pair_weight:.03f}, " f"trade_stake_amount: {trade.stake_amount:.08f}, " f"adjustment_amount: {adjustment_amount:.08f}, " f"capital: {capital:.08f}") return adjustment_amount def calculate_weights(self) -> Dict: data = {} if not self.selected_pairs: return {} for pair in self.selected_pairs: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) data[pair] = dataframe.iloc[-self.startup_candle_count:]["close"] returns = DataFrame(data).pct_change(fill_method=None).dropna() portfolio = rp.HCPortfolio(returns=returns) weights = portfolio.optimization( model="HRP", codependence="pearson", rm=self.risk_measure.value, rf=0, linkage="single", max_k=10, leaf_order=True, ) weights = weights["weights"].to_dict() return weights