from __future__ import annotations import numpy as np from pandas import DataFrame from freqtrade.strategy import DecimalParameter, IStrategy class ClawQuantMeanReversionV2FreqAI(IStrategy): """ ClawTrade FreqAI Mean-Reversion Strategy V2. Model : LightGBMRegressor (freqai.prediction_models.LightGBMRegressor) Timeframe : 5m Target : &-mean_reversion_edge = normalized forward return (label_period_candles=6 → 30m) Entry/Exit: Model-driven (do_predict + prediction threshold). NO raw indicator triggers. Safety : spot-only, can_short=False, dry_run=True (enforced by config) Lane : quant_meanrev_ai Identifier: claw_mr_ai_v1 ← rotate when feature_engineering_*() or set_freqai_targets() changes startup_candle_count = 600: - 5m lookbacks: max 96 candles (zscore_96) - 1h via include_timeframes: 40 periods * 12 = 480 5m equivalents - Safety buffer: +120 → total 600 """ INTERFACE_VERSION = 3 freqai_info: dict = {} can_short = False timeframe = "5m" process_only_new_candles = True startup_candle_count: int = 600 use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # ROI: realistis vs Binance spot fee 0.1%+0.1%=0.2% round-trip # Target 0.5%-1.0% per trade → net 0.3%-0.8% setelah fee minimal_roi = { "0": 0.008, # 0.8% take-profit kapan saja "30": 0.005, # 0.5% setelah 30 menit "60": 0.003, # 0.3% setelah 60 menit "120": 0.0, # breakeven setelah 2 jam (cap position time) } # Stoploss ketat: mean-reversion yang salah arah tidak dibiarkan stoploss = -0.04 trailing_stop = False position_adjustment_enable = False max_entry_position_adjustment = 0 # ── Hyperopt parameters ────────────────────────────────────────────────── # HANYA threshold dan risk params yang boleh di-hyperopt. # JANGAN letakkan feature/label params di sini. # Entry: minimum prediksi normalized forward return dari model entry_threshold = DecimalParameter( 0.002, 0.020, default=0.008, decimals=3, space="buy", optimize=True ) # Exit: model prediksi edge sudah habis (bisa negatif = reversal risk) exit_threshold = DecimalParameter( -0.005, 0.010, default=0.002, decimals=3, space="sell", optimize=True ) # DI gate: outlier filter — higher = lebih konservatif di_threshold = DecimalParameter( 0.5, 2.0, default=0.9, decimals=1, space="buy", optimize=True ) # ── FreqAI lifecycle ───────────────────────────────────────────────────── def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ self.freqai.start() dipanggil HANYA di sini. FreqAI mengisi kolom: &-mean_reversion_edge__mean : prediksi expected forward return (normalized) &-mean_reversion_edge__std : uncertainty prediksi do_predict : 1 = confident, 0 = outlier/uncertain DI_values : distance index (outlier score) """ self.freqai_info = self.config.get("freqai", {}) dataframe = self.freqai.start(dataframe, metadata, self) return dataframe def feature_engineering_expand_all( self, dataframe: DataFrame, period: int, metadata: dict, **kwargs ) -> DataFrame: """ Dipanggil untuk setiap period di indicator_periods_candles=[10, 20, 40]. Semua output wajib prefix %%. Jangan bocor future data di sini. """ close = dataframe["close"] volume = dataframe["volume"] # ── Z-Score family ────────────────────────────────────────────────── roll_mean = close.rolling(period).mean() roll_std = close.rolling(period).std(ddof=0).replace(0.0, np.nan) dataframe[f"%%zscore_{period}"] = ( (close - roll_mean) / roll_std ).replace([np.inf, -np.inf], np.nan) # ── Band distance family ───────────────────────────────────────────── # % distance dari rolling mean (signed: negatif = below mean) dataframe[f"%%pct_from_mean_{period}"] = ( (close - roll_mean) / roll_mean.replace(0.0, np.nan) ).replace([np.inf, -np.inf], np.nan) # % below BB lower band (clipped to 0 if above band) bb_lower = roll_mean - 2.0 * roll_std dataframe[f"%%pct_below_bb2_{period}"] = ( (bb_lower - close) / close.replace(0.0, np.nan) ).clip(lower=0).replace([np.inf, -np.inf], np.nan) # ── Realized volatility family ─────────────────────────────────────── log_ret = np.log(close / close.shift(1)) dataframe[f"%%realized_vol_{period}"] = ( log_ret.rolling(period).std() * np.sqrt(period) ) # ── Volume divergence family ───────────────────────────────────────── vol_mean = volume.rolling(period).mean().replace(0.0, np.nan) dataframe[f"%%vol_ratio_{period}"] = ( volume / vol_mean ).replace([np.inf, -np.inf], np.nan) # Volume-price correlation: apakah volume naik seiring harga naik/turun? price_chg = close.pct_change() dataframe[f"%%vol_price_corr_{period}"] = ( price_chg.rolling(period).corr(volume) ) return dataframe def feature_engineering_expand_basic( self, dataframe: DataFrame, metadata: dict, **kwargs ) -> DataFrame: """ Feature yang tidak di-expand per-period. Dipanggil sekali per pair+timeframe. Prefix %%. """ close = dataframe["close"] high = dataframe["high"] low = dataframe["low"] volume = dataframe["volume"] # ── Shifted candles: lagged returns ───────────────────────────────── for lag in [1, 2, 3, 5, 10]: dataframe[f"%%close_lag_{lag}"] = close.pct_change(lag) # ── Intrabar shape ─────────────────────────────────────────────────── # High-Low range as fraction of close (short-term volatility proxy) dataframe["%%hl_range"] = ( (high - low) / close.replace(0.0, np.nan) ).replace([np.inf, -np.inf], np.nan) # Open-to-close direction (bullish/bearish candle body) dataframe["%%oc_direction"] = ( (close - dataframe["open"]) / dataframe["open"].replace(0.0, np.nan) ).replace([np.inf, -np.inf], np.nan) # Candle body vs wick ratio (upper wick dominance = rejection) body = (close - dataframe["open"]).abs() wick_upper = high - close.where(close > dataframe["open"], dataframe["open"]) dataframe["%%upper_wick_ratio"] = ( wick_upper / (body + 1e-10) ).replace([np.inf, -np.inf], np.nan).clip(0, 10) # ── Intrabar VWAP deviation ────────────────────────────────────────── typical_price = (high + low + close) / 3.0 dataframe["%%vwap_deviation"] = ( (close - typical_price) / close.replace(0.0, np.nan) ).replace([np.inf, -np.inf], np.nan) return dataframe def feature_engineering_standard( self, dataframe: DataFrame, metadata: dict, **kwargs ) -> DataFrame: """ Feature global: regime + cross-asset context. Prefix %%. """ close = dataframe["close"] # ── Regime filter ───────────────────────────────────────────────────── sma48 = close.rolling(48).mean() dataframe["%%regime_above_sma48"] = (close > sma48).astype(int) # Trend slope: RoC of SMA10 over last 10 bars sma10 = close.rolling(10).mean() dataframe["%%trend_slope_10"] = ( (sma10 - sma10.shift(10)) / sma10.shift(10).replace(0.0, np.nan) ).replace([np.inf, -np.inf], np.nan) # ── Price context ───────────────────────────────────────────────────── # How far is current price from its 96-bar mean? mean96 = close.rolling(96).mean() dataframe["%%price_vs_96ma"] = ( (close - mean96) / mean96.replace(0.0, np.nan) ).replace([np.inf, -np.inf], np.nan) # ── Time of day feature ─────────────────────────────────────────────── # Proxy for market session (Asia/Europe/US) dataframe["%%hour"] = dataframe["date"].dt.hour / 23.0 # ── Day of week ─────────────────────────────────────────────────────── dataframe["%%dayofweek"] = dataframe["date"].dt.dayofweek / 6.0 return dataframe def set_freqai_targets(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame: """ Target label dengan prefix &. Target: &-mean_reversion_edge = normalized forward return dalam label_period_candles ke depan (default 6 = 30m) Justifikasi pilihan REGRESSOR vs CLASSIFIER: - Forward return kontinu → regressor lebih natural - Threshold entry/exit bisa di-hyperopt TANPA retrain model - Gradasi sinyal: magnitude prediksi = proxy confidence - Untuk thin-margin (0.5-1.0%), batas label diskrit terlalu sensitif vs fee Normalisasi oleh rolling std: - Menghilangkan heteroskedastisitas antar pair dan antar regime - Edge terukur dalam unit standar deviasi, bukan absolut - FreqAI menangani shift() dengan benar selama training (tidak bocor) """ label_period = self.freqai_info.get("feature_parameters", {}).get( "label_period_candles", 6 ) future_close = dataframe["close"].shift(-label_period) current_close = dataframe["close"] raw_return = (future_close - current_close) / current_close.replace(0.0, np.nan) # Normalisasi oleh rolling std return (40 candle window) ret_std = ( dataframe["close"] .pct_change() .rolling(40) .std() .replace(0.0, np.nan) ) dataframe["&-mean_reversion_edge"] = ( raw_return / ret_std ).replace([np.inf, -np.inf], np.nan) return dataframe # ── Entry / Exit ───────────────────────────────────────────────────────── def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Entry WAJIB model-driven. Tiga kondisi AND: 1. do_predict == 1 (FreqAI confident, bukan outlier) 2. &-mean_reversion_edge__mean >= entry_threshold (predicted edge) 3. DI_values <= di_threshold (tidak outlier berdasarkan distance index) TIDAK ada kondisi raw indicator (zscore, BB, SMA, dll) sebagai trigger utama. """ dataframe["enter_long"] = 0 dataframe["enter_tag"] = "" entry_thresh = float(self.entry_threshold.value) di_thresh = float(self.di_threshold.value) # FreqAI predictions — diisi oleh self.freqai.start() di populate_indicators() pred_mean = dataframe.get( "&-mean_reversion_edge__mean", dataframe["close"] * 0.0, ) pred_di = dataframe.get( "DI_values", dataframe["close"] * 0.0 + 999.0, # default tinggi = semua dianggap outlier ) do_predict = dataframe.get( "do_predict", dataframe["close"] * 0.0, # default 0 = tidak confident ) cond_confident = do_predict == 1 cond_edge = pred_mean >= entry_thresh cond_not_outlier = pred_di <= di_thresh entry_mask = cond_confident & cond_edge & cond_not_outlier dataframe.loc[entry_mask, "enter_long"] = 1 dataframe.loc[entry_mask, "enter_tag"] = "freqai_mr_edge" return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Exit model-driven. Dua kondisi OR: 1. pred_mean <= exit_threshold (edge sudah habis / berbalik) 2. do_predict == 0 (model tidak confident → exit defensif) """ dataframe["exit_long"] = 0 dataframe["exit_tag"] = "" exit_thresh = float(self.exit_threshold.value) pred_mean = dataframe.get( "&-mean_reversion_edge__mean", dataframe["close"] * 0.0, ) do_predict = dataframe.get( "do_predict", dataframe["close"] * 0.0 + 1.0, # default 1 = confident (konservatif) ) cond_edge_exhausted = pred_mean <= exit_thresh cond_uncertain = do_predict == 0 exit_mask = cond_edge_exhausted | cond_uncertain dataframe.loc[exit_mask, "exit_long"] = 1 dataframe.loc[cond_edge_exhausted, "exit_tag"] = "freqai_edge_exhausted" dataframe.loc[ cond_uncertain & ~cond_edge_exhausted, "exit_tag" ] = "freqai_uncertain" return dataframe