# source: https://raw.githubusercontent.com/GoodyNick/Freqai-Strategy/7fd33daeeee90ffa1f11d1418314d9f4681e74cc/strategies/ExampleLSTMStrategy_v1.py import logging from functools import reduce from typing import Dict import joblib import os from datetime import datetime import numpy as np import pandas as pd import talib.abstract as ta from technical import qtpylib from pandas import DataFrame from technical import qtpylib from sklearn.preprocessing import StandardScaler from sklearn.linear_model import Ridge from freqtrade import data from freqtrade.exchange.exchange_utils import * from freqtrade.strategy import IStrategy, RealParameter from freqtrade.persistence import Trade logger = logging.getLogger(__name__) class Github_GoodyNick_Freqai_Strategy__ExampleLSTMStrategy_v1__20250226_175652(IStrategy): """ This is an example strategy that uses the LSTMRegressor model to predict the target score. Use at your own risk. This is a simple example strategy and should be used for educational purposes only. """ plot_config = { "main_plot": {}, "subplots": { "predictions": { "T": {"color": "blue", "plot_type": "line"}, # Model's Label "&-s_target": {"color": "red", "plot_type": "line"}, # Model's prediction "&-s_target_mean": {"color": "green", "plot_type": "line"}, # Model's prediction, average # "do_predict": {"color": "black", "plot_type": "scatter"}, # Show predictions as dots }, }, } # ROI table: minimal_roi = { "0": 1 # we let the model decide when to exit } # Stoploss: stoploss = -1 # Were letting the model decide when to sell # Trailing stop: trailing_stop = False trailing_stop_positive = 0.001 trailing_stop_positive_offset = 0.0139 trailing_only_offset_is_reached = True threshold_buy = RealParameter(-1, 1, default=0, space='buy') threshold_sell = RealParameter(-1, 1, default=0, space='sell') timeframe = "1h" can_short = True use_exit_signal = True process_only_new_candles = True use_custom_stoploss = True startup_candle_count = 20 # Set to True to remove highly correlated features(enabling this causes # mismatch between features in trained models and backtest features). # also, these should be enabled without using PCA do_remove_highly_correlated_features = False do_filter_important_features = True prediction_metrics_storage = [] # Class-level storage for all pairs def feature_engineering_expand_all(self, dataframe: DataFrame, period: int, metadata: Dict, **kwargs): """ Expands all features for FreqAI and applies filtering before training. """ dataframe["%-cci-period"] = ta.CCI(dataframe, timeperiod=20) dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=10) dataframe["%-momentum-period"] = ta.MOM(dataframe, timeperiod=4) dataframe['%-ma-period'] = ta.SMA(dataframe, timeperiod=10) dataframe['%-macd-period'], dataframe['%-macdsignal-period'], dataframe['%-macdhist-period'] = ta.MACD( dataframe['close'], slowperiod=12, fastperiod=26 ) dataframe['%-roc-period'] = ta.ROC(dataframe, timeperiod=2) bollinger = qtpylib.bollinger_bands( qtpylib.typical_price(dataframe), window=period, stds=2.2 ) dataframe["bb_lowerband-period"] = bollinger["lower"] dataframe["bb_middleband-period"] = bollinger["mid"] dataframe["bb_upperband-period"] = bollinger["upper"] dataframe["%-bb_width-period"] = ( dataframe["bb_upperband-period"] - dataframe["bb_lowerband-period"] ) / dataframe["bb_middleband-period"] dataframe["%-close-bb_lower-period"] = dataframe["close"] / dataframe["bb_lowerband-period"] # ✅ Remove highly correlated features (if enabled) if self.do_remove_highly_correlated_features: logger.info(f"🔍 Removing highly correlated features.") dataframe = self.remove_highly_correlated_features(dataframe) # ✅ Apply feature filtering HERE if self.do_filter_important_features: dataframe = self.filter_important_features(dataframe) logger.info(f"🔍 Remaining Features After Filtering: {list(dataframe.columns)}") return dataframe def feature_engineering_expand_basic(self, dataframe: DataFrame, metadata: Dict, **kwargs): dataframe["%-pct-change"] = dataframe["close"].pct_change() dataframe["%-raw_volume"] = dataframe["volume"] dataframe["%-raw_price"] = dataframe["close"] return dataframe def feature_engineering_standard(self, dataframe: DataFrame, metadata: Dict, **kwargs): dataframe['date'] = pd.to_datetime(dataframe['date']) dataframe["%-day_of_week"] = dataframe["date"].dt.dayofweek dataframe["%-hour_of_day"] = dataframe["date"].dt.hour return dataframe def set_freqai_targets(self, dataframe: DataFrame, metadata: Dict, **kwargs) -> DataFrame: # ✅ Assign `&-s_target` for FreqAI dataframe['&-s_target'] = self.create_target_T(dataframe) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: self.freqai_info = self.config["freqai"] dataframe['T'] = self.create_target_T(dataframe) dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) dataframe['dynamic_stoploss'] = np.clip(-dataframe['atr'] * 1.75, -0.03, -0.10) dataframe = self.freqai.start(dataframe, metadata, self) logger.info(f"🔍 do_predict distribution: {dataframe['do_predict'].value_counts()}") self.compute_prediction_metrics(dataframe, metadata) self.save_prediction_metrics() return dataframe def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame: confidence_threshold = 0.8 # Minimum confidence required to enter trades # ✅ Compute dynamic thresholds using rolling statistics of the prediction (`&-s_target`) prediction_mean = df["&-s_target"].rolling(100).mean() prediction_std = df["&-s_target"].rolling(100).std() long_threshold = prediction_mean + 0.5 * prediction_std # ✅ Dynamic long threshold short_threshold = prediction_mean - 0.5 * prediction_std # ✅ Dynamic short threshold # ✅ Use crossover conditions for trade triggers enter_long_conditions = [ df["do_predict"] == 1, qtpylib.crossed_above(df["&-s_target"], long_threshold), # ✅ Long trigger using crossover df["prediction_confidence"] > confidence_threshold, df["volume"] > 0 ] enter_short_conditions = [ df["do_predict"] == 1, qtpylib.crossed_below(df["&-s_target"], short_threshold), # ✅ Short trigger using crossover df["prediction_confidence"] > confidence_threshold, df["volume"] > 0 ] # ✅ Apply entry conditions df.loc[reduce(lambda x, y: x & y, enter_long_conditions), ["enter_long", "enter_tag"]] = (1, "long") df.loc[reduce(lambda x, y: x & y, enter_short_conditions), ["enter_short", "enter_tag"]] = (1, "short") return df def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame: confidence_threshold = 0.85 # Slightly higher confidence for exits # ✅ Compute adaptive exit threshold based on rolling median of predictions (`&-s_target`) exit_threshold = df["&-s_target"].rolling(100).median() # ✅ Use crossovers instead of static comparisons strong_exit_long_conditions = [ df["do_predict"] == 1, qtpylib.crossed_below(df["&-s_target"], exit_threshold), # ✅ Exit long only on crossover df["prediction_confidence"] > confidence_threshold ] strong_exit_short_conditions = [ df["do_predict"] == 1, qtpylib.crossed_above(df["&-s_target"], exit_threshold), # ✅ Exit short only on crossover df["prediction_confidence"] > confidence_threshold ] # ✅ Apply exit conditions df.loc[reduce(lambda x, y: x & y, strong_exit_long_conditions), ["exit_long", "exit_tag"]] = (1, "strong_exit_long") df.loc[reduce(lambda x, y: x & y, strong_exit_short_conditions), ["exit_short", "exit_tag"]] = (1, "strong_exit_short") return df def create_target_T(self, dataframe: DataFrame): """ Creates a target label (T) using scaled log future returns. """ lookahead = 10 # Predicting 10 periods ahead # ✅ Compute log return target dataframe['future_return'] = np.log(dataframe['close'].shift(-lookahead) / dataframe['close']) # ✅ Scale `T` so that values are in a reasonable range for training dataframe['T'] = dataframe['future_return'].rolling(window=5).mean() * 100 # Multiply by 100 # ✅ Optional: Normalize `T` (test with and without) # dataframe['T'] = np.tanh(dataframe['T']) # ✅ Fill NaNs dataframe['T'] = dataframe['T'].fillna(0) return dataframe['T'] def custom_stoploss(self, pair, trade, current_time, current_rate, current_profit, **kwargs) -> float: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if 'atr' not in df.columns: logger.warning(f"⚠️ ATR indicator missing for {pair}. Returning default stoploss.") return -0.05 # Default static stoploss atr_value = df.iloc[-1]['atr'] trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 # Convert to hours if current_profit > 0.02: atr_multiplier = max(1.8, current_profit * 5) # ✅ Increase stop buffer with profit elif current_profit > 0: atr_multiplier = max(1.3, current_profit * 3) # ✅ Protect small profits elif current_profit < -0.01: atr_multiplier = 0.8 # ❌ Tighten stop for losing trades else: atr_multiplier = 1.0 # ✅ If trade is open for more than 24 hours, tighten stoploss if trade_duration > 24: atr_multiplier *= 0.7 # ✅ Reduce stoploss buffer by 30% if trade.is_short: stoploss_value = current_rate + (atr_value * atr_multiplier) else: stoploss_value = current_rate - (atr_value * atr_multiplier) return max(stoploss_value, trade.stop_loss) # ✅ Ensures stoploss only tightens def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time, entry_tag, side: str, **kwargs) -> bool: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = df.iloc[-1] confidence = last_candle["prediction_confidence"] if "prediction_confidence" in last_candle else 0.5 # Adjust position size dynamically (scale trade size with confidence) adjusted_size = amount * confidence return super().confirm_trade_entry(pair, order_type, adjusted_size, rate, time_in_force, current_time, entry_tag, side, **kwargs) def compute_prediction_metrics(self, dataframe: pd.DataFrame, metadata: dict, label_col: str= "T", prediction_col: str = "&-s_target") -> pd.DataFrame: """ Computes and stores prediction accuracy metrics for all trading pairs. Saves the results to a CSV file after backtesting. """ prediction_mean = prediction_col + "_mean" prediction_std = prediction_col + "_std" logger.info(f"🔍 {label_col} mean: {dataframe[label_col].mean()}, min: {dataframe[label_col].min()}, max: {dataframe[label_col].max()}") logger.info(f"🔍 {prediction_col} mean: {dataframe[prediction_col].mean()}, min: {dataframe[prediction_col].min()}, max: {dataframe[prediction_col].max()}") logger.info(f"🔍 {prediction_mean} mean: {dataframe[prediction_mean].mean()}, min: {dataframe[prediction_mean].min()}, max: {dataframe[prediction_mean].max()}") logger.info(f"🔍 {prediction_std} mean: {dataframe[prediction_std].mean()}, min: {dataframe[prediction_std].min()}, max: {dataframe[prediction_std].max()}") # Ensure required columns exist if prediction_col not in dataframe.columns: logger.warning(f"❌ Column '{prediction_col}' not found in dataframe. Skipping prediction metrics.") return dataframe # ✅ Step 1: Directional Accuracy (Sign Match) dataframe["prediction_correct"] = (np.sign(dataframe[label_col]) == np.sign(dataframe[prediction_col])).astype(int) # ✅ Step 2: Rolling Accuracy (Last 50 candles) dataframe["rolling_accuracy"] = dataframe["prediction_correct"].rolling(50, min_periods=1).mean() # ✅ Step 3: Mean Absolute Error (MAE) dataframe["mae"] = np.abs(dataframe[label_col] - dataframe[prediction_col]).rolling(100, min_periods=1).mean() # ✅ Step 4: Prediction Confidence (Normalized by Standard Deviation) std_col = prediction_std if std_col in dataframe.columns: dataframe["prediction_confidence"] = (np.abs(dataframe[prediction_col]) / (dataframe[std_col] + 1e-6)).clip(0, 1) # Confidence score is only counted for correct predictions dataframe["confidence_correct"] = np.where( dataframe["prediction_correct"] == 1, dataframe["prediction_confidence"], 0 ) # Normalize avg confidence over correct predictions correct_preds = dataframe["prediction_correct"].rolling(100, min_periods=1).sum() dataframe["avg_confidence_correct"] = dataframe["confidence_correct"].rolling(100, min_periods=1).sum() / (correct_preds + 1e-6) else: logger.warning(f"⚠️ Column '{std_col}' not found. Skipping confidence tracking.") dataframe["avg_confidence_correct"] = np.nan # ✅ Step 5: Calculate Fraction of Predicted Targets total_predictions = (dataframe["do_predict"] == 1).sum() logger.info(f"🔍 `do_predict=1` Count: {total_predictions}, `do_predict=-1` Count: {(dataframe['do_predict'] == -1).sum()}") total_targets_available = dataframe[label_col].notna().sum() fraction_predicted = total_predictions / total_targets_available if total_targets_available > 0 else 0 # ✅ Step 6: Store Metrics in Class-Level List pair = metadata["pair"] metrics = { "pair": pair, "total_predictions": total_predictions, "fraction_predicted": fraction_predicted, "rolling_accuracy": dataframe["rolling_accuracy"].iloc[-1], "mae": dataframe["mae"].iloc[-1], "avg_confidence_correct": dataframe["avg_confidence_correct"].iloc[-1] if "avg_confidence_correct" in dataframe.columns else np.nan, "correlation": dataframe[prediction_col].corr(dataframe[label_col]) # ✅ Step 8: Correlation between Target and Predictions } self.prediction_metrics_storage.append(metrics) # ✅ Step 7: Log Key Statistics logger.info( "🔍 Prediction Metrics | Pair: %s | Total Predictions: %s | Fraction Predicted: %.4f | Rolling Accuracy: %.4f | MAE: %.6f | Avg Confidence: %.4f | Correlation: %.4f", pair, total_predictions, fraction_predicted, metrics["rolling_accuracy"], metrics["mae"], metrics["avg_confidence_correct"], metrics["correlation"] ) return dataframe def save_prediction_metrics(self, filename="prediction_metrics.csv"): """ Saves the accumulated prediction metrics to a CSV file after backtesting. """ if not self.prediction_metrics_storage: logger.warning("⚠️ No prediction metrics found to save.") return df = pd.DataFrame(self.prediction_metrics_storage) output_path = os.path.join(self.config["user_data_dir"], filename) df.to_csv(output_path, index=False) logger.info(f"✅ Prediction metrics saved to {output_path}") def remove_highly_correlated_features(self, dataframe, threshold=0.85): """ Drops one feature from each highly correlated pair, except essential features. """ essential_features = {"high", "low", "close", "open", "volume"} # Features FreqAI depends on corr_matrix = dataframe.corr().abs() upper = corr_matrix.where(np.triu(np.ones(corr_matrix.shape), k=1).astype(bool)) to_drop = [column for column in upper.columns if any(upper[column] > threshold) and column not in essential_features] if to_drop: logger.info(f"Dropping {len(to_drop)} highly correlated features: {to_drop}") dataframe = dataframe.drop(columns=to_drop) return dataframe def filter_important_features(self, dataframe): """ Removes all columns that start with '%' unless they are in the important features list. """ important_features = { "%-hour_of_day", "%-day_of_week", "%-cci-period_50_BTC/USDTUSDT_4h", "%-pct-change_gen_BTC/USDTUSDT_1h", "%-roc-period_20_BTC/USDTUSDT_2h", "%-rsi-period_10_BTC/USDTUSDT_4h", "%-rsi-period_50_ETH/USDTUSDT_4h", "%-bb_width-period_50_BTC/USDTUSDT_4h" } # Drop all columns starting with '%' unless they are in the important_features set columns_to_keep = [col for col in dataframe.columns if not col.startswith("%") or col in important_features] return dataframe[columns_to_keep]