# source: https://raw.githubusercontent.com/phamminhducxt/bot-ai/e98f45b2950dd9633292a17d7a825dc21eaf10fd/user_data/strategies/FreqAI_Futures_Strategy.py import logging from functools import reduce from typing import Dict, Optional from pathlib import Path from pandas import DataFrame import pandas as pd import numpy as np import talib.abstract as ta from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter from freqtrade.persistence import Trade from collections import deque from datetime import datetime, timedelta, timezone logger = logging.getLogger(__name__) class Github_phamminhducxt_bot_ai__FreqAI_Futures_Strategy__20251226_093758(IStrategy): """ ============================================================= FREQAI FUTURES STRATEGY – SNOWBALL GROWTH VERSION ============================================================= Mục tiêu: Bắt đầu vốn nhỏ → Tích lũy dần → Profit cao Chiến lược "Quả cầu tuyết": - Vốn < $500: Aggressive (risk 3-5%, leverage 3-5x) - Vốn $500-2000: Balanced (risk 2-3%, leverage 2-3x) - Vốn > $2000: Conservative (risk 1-2%, leverage 1-2x) ============================================================= """ # ================= BASIC CONFIG ================= timeframe = "5m" startup_candle_count = 300 process_only_new_candles = True can_short = True # ROI tiers - AGGRESSIVE: Chốt lời sớm để quay vòng nhanh minimal_roi = { "0": 0.05, # 5% - chốt ngay nếu đạt "15": 0.035, # 3.5% sau 15 phút "30": 0.025, # 2.5% sau 30 phút "60": 0.018, # 1.8% sau 1 giờ "120": 0.012, # 1.2% sau 2 giờ "180": 0.008, # 0.8% sau 3 giờ } stoploss = -0.022 # -2.2% stoploss (tight hơn để bảo vệ vốn) trailing_stop = True trailing_stop_positive = 0.008 # Bắt đầu trailing sớm hơn khi +0.8% trailing_stop_positive_offset = 0.015 # Kích hoạt khi +1.5% trailing_only_offset_is_reached = True max_open_trades = 3 # Cho phép 3 lệnh cùng lúc để tăng cơ hội use_custom_stake_amount = True use_custom_stoploss = True # ================= WALLET GROWTH MILESTONES ================= # Các mốc vốn để điều chỉnh risk - SNOWBALL STRATEGY WALLET_TIERS = { "nano": 10, # < $10: YOLO mode - All-in để thoát vùng nguy hiểm "micro": 50, # $10-50: Super aggressive "mini": 200, # $50-200: Aggressive "small": 500, # $200-500: Moderate aggressive "medium": 2000, # $500-2000: Balanced "large": 10000, # $2000-10000: Conservative "whale": 50000 # > $10000: Very conservative } # ================= HYPEROPT PARAMETERS ================= # Confidence thresholds - Nới lỏng hơn để có nhiều lệnh confidence_high = DecimalParameter(0.65, 0.85, default=0.72, space="buy", optimize=True) confidence_low = DecimalParameter(0.55, 0.75, default=0.65, space="buy", optimize=True) # ADX thresholds - Nới lỏng hơn adx_threshold = IntParameter(15, 28, default=20, space="buy", optimize=True) # ATRP threshold - Nới lỏng hơn atrp_threshold = DecimalParameter(0.004, 0.010, default=0.005, space="buy", optimize=True) # ================= REGIME ADAPTATION (FUTURES) ================= # Mục tiêu: kết hợp 2 thứ: # 1) Regime gate: tránh chop / volatility spike / fakeout # 2) EV shrink: nếu model xuống phong độ thì tự siết (threshold↑, stake/leverage↓) # Nếu regime xấu (chop/spike) thì tăng threshold và giảm stake/leverage. REGIME_MULTIPLIERS = { "TREND": {"conf": 0.95, "stake": 1.05, "lev": 1.05}, "NORMAL": {"conf": 1.00, "stake": 1.00, "lev": 1.00}, "CHOP": {"conf": 1.10, "stake": 0.75, "lev": 0.75}, "SPIKE": {"conf": 1.20, "stake": 0.55, "lev": 0.60}, } # region Init & Protections def __init__(self, config: Dict) -> None: super().__init__(config) # ===== Runtime state (must exist before any strategy callbacks) ===== self.ai_pause_until: Optional[datetime] = None self.loss_streak: int = 0 self.win_streak: int = 0 self.ai_results: deque = deque(maxlen=30) # Wallet / growth tracking self.initial_wallet: float = float(config.get("dry_run_wallet", 1000)) self.peak_wallet: float = float(self.initial_wallet) self.total_profit: float = 0.0 # EV / regime tracking self._ewma_ev: float = 0.0 self._last_regime: Dict[str, str] = {} # Dynamic trades tracking (updated in bot_loop_start) self._dynamic_max_trades: int = int(getattr(self, "max_open_trades", 1) or 1) # One-time compatibility checks for FreqAI state on disk self._historic_predictions_checked: bool = False try: self._ensure_freqai_historic_predictions_compatible() except Exception: pass try: self._quarantine_incompatible_models() except Exception: pass # ================= PROTECTIONS ================= @property def protections(self): """Bảo vệ tài khoản khỏi các chuỗi thua liên tiếp""" return [ { "method": "StoplossGuard", "lookback_period_candles": 24, # 2 giờ với timeframe 5m "trade_limit": 3, # Dừng sau 3 lệnh chạm stoploss "stop_duration_candles": 12, # Nghỉ 1 giờ "only_per_pair": False }, { "method": "CooldownPeriod", "stop_duration_candles": 2 # Nghỉ 10 phút giữa các lệnh }, { "method": "MaxDrawdown", "lookback_period_candles": 48, # 4 giờ "trade_limit": 20, "stop_duration_candles": 24, # Nghỉ 2 giờ "max_allowed_drawdown": 0.10 # Dừng nếu drawdown > 10% }, { "method": "LowProfitPairs", "lookback_period_candles": 288, # 24 giờ "trade_limit": 4, "stop_duration_candles": 144, # Nghỉ 12 giờ "required_profit": -0.02 # Dừng pair nếu lỗ > 2% } ] # endregion Init & Protections # region FreqAI Disk Helpers def _get_freqai_identifier(self) -> Optional[str]: """Return FreqAI identifier used to locate model storage directory.""" try: freqai_cfg = self.config.get("freqai", {}) if hasattr(self, "config") and self.config else {} identifier = freqai_cfg.get("identifier") return str(identifier) if identifier else None except Exception: return None def _get_user_data_dir(self) -> Optional[Path]: """Best-effort locate user_data directory from this strategy file path.""" try: # /freqtrade/user_data/strategies/.py -> parent.parent == /freqtrade/user_data return Path(__file__).resolve().parent.parent except Exception: return None def _quarantine_file(self, file_path: Path, reason: str) -> None: """Rename a problematic file so FreqAI can recreate it fresh.""" try: ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") except Exception: ts = "unknown" new_name = f"{file_path.name}.bad-{reason}-{ts}" quarantined = file_path.with_name(new_name) try: file_path.rename(quarantined) logger.warning( "Quarantined incompatible FreqAI file: %s -> %s (reason=%s)", str(file_path), str(quarantined), reason, ) except Exception: # If rename fails (e.g. permissions), do not hard-fail. logger.warning("Failed to quarantine FreqAI file: %s (reason=%s)", str(file_path), reason) def _ensure_freqai_historic_predictions_compatible(self) -> None: """Fix common FreqAI startup crashes caused by stale historic_predictions pickles. Symptoms addressed: - KeyError: 'date_pred' (old pickle schema missing date_pred) - AttributeError: 'DataFrame' object has no attribute 'dtype' (duplicate columns -> df[label] returns DataFrame) Approach: - Load pickle(s) if present - If schema is incompatible, rename the file(s) so FreqAI rebuilds them cleanly """ if getattr(self, "_historic_predictions_checked", False): return self._historic_predictions_checked = True identifier = self._get_freqai_identifier() if not identifier: return user_data_dir = self._get_user_data_dir() if not user_data_dir: return models_dir = user_data_dir / "models" / identifier if not models_dir.exists(): return # Validate both primary and backup historic predictions. for fname in ("historic_predictions.pkl", "historic_predictions.backup.pkl"): fp = models_dir / fname if not fp.exists(): continue try: hist = pd.read_pickle(fp) except Exception: self._quarantine_file(fp, reason="read_error") continue if not isinstance(hist, pd.DataFrame): self._quarantine_file(fp, reason="not_dataframe") continue missing_date_pred = "date_pred" not in hist.columns has_dupes = bool(getattr(hist.columns, "has_duplicates", False)) if missing_date_pred: self._quarantine_file(fp, reason="missing_date_pred") continue if has_dupes: self._quarantine_file(fp, reason="dup_cols") continue def _quarantine_incompatible_models(self) -> None: """ Kiểm tra và xóa các model files có LabelEncoder không tương thích. Gọi trong __init__ để đảm bảo clean state. """ identifier = self._get_freqai_identifier() if not identifier: return user_data_dir = self._get_user_data_dir() if not user_data_dir: return models_dir = user_data_dir / "models" / identifier if not models_dir.exists(): return import pickle try: import joblib # type: ignore except Exception: joblib = None def _iter_candidate_files(folder: Path): # Search recursively because FreqAI may store encoders in nested paths. for ext in ("*.pkl", "*.pickle", "*.joblib"): try: yield from folder.rglob(ext) except Exception: continue def _has_single_class_encoder(obj) -> Optional[list]: """Return classes list if obj looks like a LabelEncoder with <2 classes.""" try: if hasattr(obj, "classes_"): classes = list(getattr(obj, "classes_")) if len(classes) < 2: return classes except Exception: return None return None # Tìm tất cả các sub-directories chứa model (mỗi pair/timestamp có 1 folder) for model_folder in models_dir.iterdir(): if not model_folder.is_dir(): continue incompatible = False found_classes = None for fpath in _iter_candidate_files(model_folder): try: data = None if fpath.suffix == ".joblib" and joblib is not None: data = joblib.load(fpath) else: with open(fpath, "rb") as f: data = pickle.load(f) # Common: dict with encoders/metadata if isinstance(data, dict): for _k, v in data.items(): classes = _has_single_class_encoder(v) if classes is not None: incompatible = True found_classes = classes break else: classes = _has_single_class_encoder(data) if classes is not None: incompatible = True found_classes = classes except Exception as e: logger.debug(f"Could not check model artifact {fpath}: {e}") if incompatible: break if incompatible: logger.warning( f"🗑️ Found incompatible model (single class {found_classes}) -> removing folder: {model_folder.name}" ) try: import shutil shutil.rmtree(model_folder, ignore_errors=True) except Exception as e: logger.debug(f"Failed to remove incompatible model folder {model_folder}: {e}") # endregion FreqAI Disk Helpers # ================= 1. FEATURE ENGINEERING (PHẦN CÒN THIẾU) ================= # Đây là phần quan trọng để AI biết cần học cái gì. # Các cột bắt đầu bằng %- sẽ được AI sử dụng làm features. def feature_engineering_expand_all(self, dataframe: DataFrame, period: int, metadata: dict, **kwargs) -> DataFrame: """ Tạo ra các chỉ báo kỹ thuật trên nhiều khung thời gian (5m, 15m, 1h) để AI học. Sử dụng period để tạo features đa dạng theo indicator_periods_candles trong config. """ # Momentum dataframe[f"%-rsi-{period}"] = ta.RSI(dataframe, timeperiod=period) dataframe[f"%-mfi-{period}"] = ta.MFI(dataframe, timeperiod=period) dataframe[f"%-roc-{period}"] = ta.ROC(dataframe, timeperiod=period) dataframe[f"%-willr-{period}"] = ta.WILLR(dataframe, timeperiod=period) # MACD macd = ta.MACD(dataframe, fastperiod=period, slowperiod=period*2, signalperiod=int(period*0.9)) dataframe[f"%-macd-{period}"] = macd["macd"] dataframe[f"%-macdsignal-{period}"] = macd["macdsignal"] dataframe[f"%-macdhist-{period}"] = macd["macdhist"] # Stochastic stoch = ta.STOCH(dataframe, fastk_period=period, slowk_period=3, slowd_period=3) dataframe[f"%-slowk-{period}"] = stoch["slowk"] dataframe[f"%-slowd-{period}"] = stoch["slowd"] # Trend dataframe[f"%-adx-{period}"] = ta.ADX(dataframe, timeperiod=period) dataframe[f"%-cci-{period}"] = ta.CCI(dataframe, timeperiod=period) dataframe[f"%-aroon-up-{period}"] = ta.AROON(dataframe, timeperiod=period)["aroonup"] dataframe[f"%-aroon-down-{period}"] = ta.AROON(dataframe, timeperiod=period)["aroondown"] dataframe[f"%-dx-{period}"] = ta.DX(dataframe, timeperiod=period) # Plus/Minus DI dataframe[f"%-plus-di-{period}"] = ta.PLUS_DI(dataframe, timeperiod=period) dataframe[f"%-minus-di-{period}"] = ta.MINUS_DI(dataframe, timeperiod=period) # Volatility dataframe[f"%-atr-{period}"] = ta.ATR(dataframe, timeperiod=period) bollinger = ta.BBANDS(dataframe, timeperiod=period, nbdevup=2.0, nbdevdn=2.0) dataframe[f"%-bb-upper-{period}"] = bollinger["upperband"] dataframe[f"%-bb-middle-{period}"] = bollinger["middleband"] dataframe[f"%-bb-lower-{period}"] = bollinger["lowerband"] dataframe[f"%-bb-width-{period}"] = (bollinger["upperband"] - bollinger["lowerband"]) / bollinger["middleband"] dataframe[f"%-bb-percent-{period}"] = (dataframe["close"] - bollinger["lowerband"]) / (bollinger["upperband"] - bollinger["lowerband"]) # Keltner Channel keltner_mid = ta.EMA(dataframe, timeperiod=period) keltner_atr = ta.ATR(dataframe, timeperiod=period) dataframe[f"%-kc-upper-{period}"] = keltner_mid + (keltner_atr * 2) dataframe[f"%-kc-lower-{period}"] = keltner_mid - (keltner_atr * 2) # Volume indicators dataframe[f"%-obv-{period}"] = ta.OBV(dataframe) dataframe[f"%-ad-{period}"] = ta.AD(dataframe) dataframe[f"%-adosc-{period}"] = ta.ADOSC(dataframe, fastperiod=3, slowperiod=period) # Price patterns dataframe[f"%-sar-{period}"] = ta.SAR(dataframe) # EMA crossover features dataframe[f"%-ema-{period}"] = ta.EMA(dataframe, timeperiod=period) dataframe[f"%-sma-{period}"] = ta.SMA(dataframe, timeperiod=period) dataframe[f"%-close-ema-dist-{period}"] = (dataframe["close"] - dataframe[f"%-ema-{period}"]) / dataframe["close"] return dataframe def feature_engineering_expand_basic(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame: """ Tạo các features cơ bản về giá và volume. """ # Price features dataframe["%-pct-change"] = dataframe["close"].pct_change() dataframe["%-pct-change-2"] = dataframe["close"].pct_change(2) dataframe["%-pct-change-5"] = dataframe["close"].pct_change(5) # Raw values (normalized by rolling stats) dataframe["%-raw_volume"] = dataframe["volume"] / dataframe["volume"].rolling(20).mean() dataframe["%-raw_price"] = dataframe["close"] / dataframe["close"].rolling(20).mean() # High/Low features dataframe["%-high-low-pct"] = (dataframe["high"] - dataframe["low"]) / dataframe["close"] dataframe["%-close-open-pct"] = (dataframe["close"] - dataframe["open"]) / dataframe["open"] # Candle body and wick dataframe["%-body-pct"] = abs(dataframe["close"] - dataframe["open"]) / dataframe["close"] dataframe["%-upper-wick"] = (dataframe["high"] - dataframe[["close", "open"]].max(axis=1)) / dataframe["close"] dataframe["%-lower-wick"] = (dataframe[["close", "open"]].min(axis=1) - dataframe["low"]) / dataframe["close"] # Volume momentum dataframe["%-volume-pct-change"] = dataframe["volume"].pct_change() dataframe["%-volume-ratio-5"] = dataframe["volume"] / dataframe["volume"].rolling(5).mean() # Price position in recent range dataframe["%-price-position"] = (dataframe["close"] - dataframe["low"].rolling(20).min()) / \ (dataframe["high"].rolling(20).max() - dataframe["low"].rolling(20).min() + 1e-10) # Trend strength dataframe["%-higher-highs"] = (dataframe["high"] > dataframe["high"].shift(1)).astype(int).rolling(5).sum() dataframe["%-lower-lows"] = (dataframe["low"] < dataframe["low"].shift(1)).astype(int).rolling(5).sum() return dataframe def feature_engineering_standard(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame: """ Tạo features về thời gian và pair-specific (AI học thói quen thị trường theo giờ/ngày). """ # Time features dataframe["%-day_of_week"] = dataframe["date"].dt.dayofweek dataframe["%-hour_of_day"] = dataframe["date"].dt.hour dataframe["%-minute_of_hour"] = dataframe["date"].dt.minute dataframe["%-is_weekend"] = (dataframe["date"].dt.dayofweek >= 5).astype(int) # Session features (UTC) hour = dataframe["date"].dt.hour dataframe["%-asian_session"] = ((hour >= 0) & (hour < 8)).astype(int) dataframe["%-european_session"] = ((hour >= 8) & (hour < 16)).astype(int) dataframe["%-us_session"] = ((hour >= 13) & (hour < 22)).astype(int) # Pair-specific features # Lưu ý: metadata["pair"].startswith(...) trả về bool scalar -> gán trực tiếp (broadcast) cho toàn bộ cột. dataframe["%-is_btc"] = int(str(metadata.get("pair", "")).startswith("BTC")) dataframe["%-is_eth"] = int(str(metadata.get("pair", "")).startswith("ETH")) return dataframe # ================= 2. SET FREQAI TARGETS (MỤC TIÊU) ================= def set_freqai_targets(self, df: DataFrame, metadata: dict, **kwargs): """ Định nghĩa thế nào là Kèo Ngon (Target 1) để AI học. Có nhiều target khác nhau cho các mục đích khác nhau. QUAN TRỌNG: Phải đảm bảo cả 2 class (0 và 1) đều có trong training data để tránh lỗi "y contains previously unseen labels". """ # Bảo vệ: Tính lại chỉ báo nếu thiếu required_cols = ["adx", "atr", "atrp"] if not all(col in df.columns for col in required_cols): df["adx"] = ta.ADX(df, 14) df["atr"] = ta.ATR(df, 14) df["atrp"] = df["atr"] / df["close"] logger.debug(f"Calculated missing indicators for {metadata['pair']}") N = int(self.freqai_info["feature_parameters"].get("label_period_candles", 20)) # --- Future windows (forward-looking) --- close_fwd = df["close"].shift(-N) low_fwd = df["low"].shift(-1).rolling(N, min_periods=N).min().shift(-(N - 1)) high_fwd = df["high"].shift(-1).rolling(N, min_periods=N).max().shift(-(N - 1)) # Future return / forward excursions future_ret = (close_fwd / df["close"] - 1).replace([np.inf, -np.inf], np.nan) worst_dd = (low_fwd / df["close"] - 1).replace([np.inf, -np.inf], np.nan) best_up = (high_fwd / df["close"] - 1).replace([np.inf, -np.inf], np.nan) atrp = (ta.ATR(df, 14) / df["close"]).replace([np.inf, -np.inf], np.nan) # Valid rows for labeling (avoid NaNs at the tail or from divisions) valid_label = ( future_ret.notna() & worst_dd.notna() & best_up.notna() & atrp.notna() & df["close"].notna() ) # ========== MAIN TARGET: Trade quality ========== # ĐIỀU KIỆN RẤT NỚI LỎNG để đảm bảo có đủ samples cho cả 2 class # # Chiến lược: Label = 1 nếu future return > 0 VÀ drawdown chấp nhận được # Đây là định nghĩa đơn giản nhất: "trade có lãi" # Điều kiện cơ bản: return dương và drawdown không quá lớn trade_ok = ( (future_ret > 0.001) # Return > 0.1% (rất thấp) & (worst_dd > -0.03) # Drawdown < 3% (rất lỏng) ) # Khởi tạo target column df["&s-trade_ok"] = 0 df.loc[valid_label & trade_ok, "&s-trade_ok"] = 1 # ========== CRITICAL: Đảm bảo cả 2 class đều có trong data ========== # Nếu chỉ có 1 class, XGBoostClassifier sẽ crash khi inverse_transform class_counts = df.loc[valid_label, "&s-trade_ok"].value_counts() n_valid = valid_label.sum() n_class_0 = class_counts.get(0, 0) n_class_1 = class_counts.get(1, 0) logger.info(f"📊 TARGET DISTRIBUTION for {metadata.get('pair', 'unknown')}: " f"Valid={n_valid}, Class0={n_class_0} ({100*n_class_0/max(1,n_valid):.1f}%), " f"Class1={n_class_1} ({100*n_class_1/max(1,n_valid):.1f}%)") # Nếu một class bị thiếu hoàn toàn hoặc quá ít (<5%), tạo synthetic samples min_samples_per_class = max(50, int(n_valid * 0.05)) # Ít nhất 5% hoặc 50 samples if n_class_1 < min_samples_per_class: # Không đủ class 1 -> Chọn các samples có return cao nhất làm class 1 logger.warning(f"⚠️ Class imbalance detected: Only {n_class_1} positive samples. " f"Need at least {min_samples_per_class}. Relaxing conditions...") # Tìm top N samples có future_ret cao nhất (trong valid rows) need_more = min_samples_per_class - n_class_1 # Chỉ xét các rows valid mà hiện tại là class 0 candidates = valid_label & (df["&s-trade_ok"] == 0) & future_ret.notna() candidate_returns = future_ret.where(candidates, np.nan) available = int(candidate_returns.notna().sum()) if available > 0 and need_more > 0: k = min(need_more, available) top_idx = candidate_returns.nlargest(k).index df.loc[top_idx, "&s-trade_ok"] = 1 try: threshold = float(candidate_returns.loc[top_idx].min()) logger.info(f"✅ Promoted {k} samples to class 1 (return >= {threshold:.4f})") except Exception: logger.info(f"✅ Promoted {k} samples to class 1") if n_class_0 < min_samples_per_class: # Không đủ class 0 -> Chọn các samples có return thấp nhất làm class 0 logger.warning(f"⚠️ Class imbalance detected: Only {n_class_0} negative samples. " f"Need at least {min_samples_per_class}. Adjusting...") need_more = min_samples_per_class - n_class_0 candidates = valid_label & (df["&s-trade_ok"] == 1) & future_ret.notna() candidate_returns = future_ret.where(candidates, np.nan) available = int(candidate_returns.notna().sum()) if available > 0 and need_more > 0: k = min(need_more, available) bottom_idx = candidate_returns.nsmallest(k).index df.loc[bottom_idx, "&s-trade_ok"] = 0 try: threshold = float(candidate_returns.loc[bottom_idx].max()) logger.info(f"✅ Demoted {k} samples to class 0 (return <= {threshold:.4f})") except Exception: logger.info(f"✅ Demoted {k} samples to class 0") # Final check và log final_counts = df.loc[valid_label, "&s-trade_ok"].value_counts() final_0 = final_counts.get(0, 0) final_1 = final_counts.get(1, 0) if final_0 == 0 or final_1 == 0: # Last resort: Force some samples to ensure both classes exist logger.error(f"❌ CRITICAL: Still missing a class after adjustment! " f"Class0={final_0}, Class1={final_1}") # Force tạo ít nhất 1 sample cho mỗi class từ valid rows valid_indices = df.index[valid_label].tolist() if len(valid_indices) >= 2: if final_0 == 0: df.loc[valid_indices[0], "&s-trade_ok"] = 0 if final_1 == 0: df.loc[valid_indices[-1], "&s-trade_ok"] = 1 logger.warning("🔧 Force-created samples for missing class(es)") # Ensure correct dtype df["&s-trade_ok"] = df["&s-trade_ok"].fillna(0).astype(int) # Final distribution log final_counts = df["&s-trade_ok"].value_counts() logger.info(f"📊 FINAL TARGET: Class0={final_counts.get(0, 0)}, Class1={final_counts.get(1, 0)}") return df # ================= 3. INDICATORS (CHO STRATEGY LOGIC) ================= def populate_indicators(self, df: DataFrame, metadata: dict) -> DataFrame: # ---- Date hygiene (FreqAI compatibility) ---- # Một số version/pipeline của FreqAI kỳ vọng có cột `date` kiểu datetime64[ns, UTC] # và đôi khi sẽ reference `date_pred` để căn chỉnh prediction. # Nếu thiếu/khác dtype, có thể gây lỗi kiểu KeyError: 'date_pred'. if df is not None and not df.empty: # Ensure `date` exists and is timezone-aware if "date" in df.columns: try: # Prefer UTC-aware timestamps if not pd.api.types.is_datetime64_any_dtype(df["date"]): df["date"] = pd.to_datetime(df["date"], utc=True, errors="coerce") else: # If datetime but tz-naive, localize to UTC if getattr(df["date"].dt, "tz", None) is None: df["date"] = df["date"].dt.tz_localize("UTC") except Exception: # Best-effort: do not fail indicator population because of date parsing pass # Provide `date_pred` if missing (FreqAI may look for it in some pipelines) if "date_pred" not in df.columns and "date" in df.columns: df["date_pred"] = df["date"] # ---- Column hygiene (FreqAI compatibility) ---- # FreqAI internals sometimes do: `hist_preds_df[label].dtype` # If `label` exists multiple times (duplicate column names), then `df[label]` is a DataFrame # and `.dtype` will crash with: AttributeError: 'DataFrame' object has no attribute 'dtype'. # De-duplicate columns deterministically (keep first occurrence) to ensure `df[label]` is a Series. if df is not None and not df.empty: try: if df.columns.has_duplicates: dupes = df.columns[df.columns.duplicated()].unique().tolist() logger.warning( "Detected duplicate columns for %s; de-duplicating to avoid FreqAI dtype crash. duplicates=%s", metadata.get("pair"), dupes, ) df = df.loc[:, ~df.columns.duplicated(keep="first")].copy() except Exception: # Best effort only pass # --- TREND --- df["ema50"] = ta.EMA(df, 50) df["ema200"] = ta.EMA(df, 200) df["adx"] = ta.ADX(df, 14) # --- VOLATILITY --- df["atr"] = ta.ATR(df, 14) df["atrp"] = df["atr"] / df["close"] df["range_pct"] = (df["high"] - df["low"]) / df["close"] # --- MOMENTUM --- df["rsi"] = ta.RSI(df, 14) # --- VOLUME --- df["vol_mean"] = df["volume"].rolling(20).mean() df["vol_ratio"] = df["volume"] / df["vol_mean"] # --- CHOP SCORE --- df["chop_score"] = ( (df["adx"] < 20).astype(int) + (df["atrp"] < 0.006).astype(int) + (df["range_pct"] < 0.004).astype(int) ) # --- REGIME FEATURES (dùng cho filters, không feed trực tiếp như %- features) --- # Wick ratio: wick dài thường = stop hunt / noisy microstructure wick_up = (df["high"] - df[["close", "open"]].max(axis=1)).clip(lower=0) wick_dn = (df[["close", "open"]].min(axis=1) - df["low"]).clip(lower=0) body = (df["close"] - df["open"]).abs().clip(lower=1e-12) df["wick_ratio"] = ((wick_up + wick_dn) / body).replace([np.inf, -np.inf], np.nan).fillna(0.0) # Vol spike: ATRP so với median ngắn hạn atrp_med = df["atrp"].rolling(96, min_periods=20).median() df["vol_spike"] = (df["atrp"] / (atrp_med + 1e-12)).replace([np.inf, -np.inf], np.nan).fillna(1.0) # GỌI FREQAI START (Sau khi đã định nghĩa feature_engineering ở trên) # Instrumentation & compatibility: ensure date columns have correct dtype # NOTE: Do NOT set df.index from date column - having date as both index and column causes ambiguity. try: # Coerce `date` and `date_pred` to UTC datetimes (column only, not index) try: if "date" in df.columns: df["date"] = pd.to_datetime(df["date"], utc=True, errors="coerce") # Defensive: create date_pred as close as possible to freqai.start() # (some merge/pipeline steps may drop it earlier) if "date_pred" not in df.columns and "date" in df.columns: df["date_pred"] = df["date"].copy() if "date_pred" in df.columns: df["date_pred"] = pd.to_datetime(df["date_pred"], utc=True, errors="coerce") except Exception: logger.debug("Failed to coerce date/date_pred columns to datetime (best-effort).") # Add a small debug snapshot to logs to assist diagnosing KeyError: 'date_pred' / dtype errors try: cols = list(df.columns) dtypes = df.dtypes.apply(lambda x: x.name).to_dict() idx_type = type(df.index).__name__ sample_dates = None if "date" in df.columns or "date_pred" in df.columns: sample_dates = { "date_tail": None, "date_pred_tail": None, } if "date" in df.columns: sample_dates["date_tail"] = df["date"].tail(3).astype(str).tolist() if "date_pred" in df.columns: sample_dates["date_pred_tail"] = df["date_pred"].tail(3).astype(str).tolist() logger.debug( "FreqAI start: pair=%s, cols=%s, dtypes=%s, index=%s, sample_dates=%s", metadata.get("pair"), cols, dtypes, idx_type, sample_dates, ) except Exception: # never fail indicators population due to logging pass df = self.freqai.start(df, metadata, self) except Exception as e: # Extended error logging to capture dataframe shape/columns at failure time logger.error(f"❌ FreqAI error for {metadata['pair']}: {e}") try: import traceback logger.error(traceback.format_exc()) # Log a compact view of problematic dataframe columns/dtypes to help debugging try: cols = list(df.columns) dtypes = df.dtypes.apply(lambda x: x.name).to_dict() idx_type = type(df.index).__name__ logger.debug( "FreqAI failure snapshot: pair=%s, rows=%s, cols=%s, dtypes=%s, index=%s", metadata.get("pair"), len(df), cols, dtypes, idx_type, ) # Log last few date/date_pred values if present if "date" in df.columns: logger.debug("date tail: %s", df["date"].tail(5).astype(str).tolist()) if "date_pred" in df.columns: logger.debug("date_pred tail: %s", df["date_pred"].tail(5).astype(str).tolist()) except Exception: logger.debug("Failed to capture DataFrame diagnostic snapshot.") except Exception: # final fallback logger.exception("Unhandled exception while logging FreqAI error") return df # ================= REGIME / EV HELPERS ================= def _get_regime(self, last: dict) -> str: """Phân loại regime đơn giản (nhanh, ổn định) dựa trên indicators đã có.""" adx = float(last.get("adx", 0.0)) atrp = float(last.get("atrp", 0.0)) chop = int(last.get("chop_score", 0)) vol_spike = float(last.get("vol_spike", 1.0)) wick_ratio = float(last.get("wick_ratio", 0.0)) # Spike regime: volatility tăng đột ngột + wick lớn → dễ fakeout/liquidation wicks if vol_spike >= 1.8 or (atrp >= 0.015 and wick_ratio >= 2.0): return "SPIKE" # Chop regime: trend yếu + chop_score cao hoặc wick noise if chop >= 2 or (adx < 18 and wick_ratio >= 2.5): return "CHOP" # Trend clean regime if adx >= 28 and chop <= 1: return "TREND" return "NORMAL" def _get_ev(self, n: int = 20) -> float: """EV rolling (profit ratio trung bình) từ lịch sử đóng lệnh.""" if len(self.ai_results) < 3: return 0.0 xs = list(self.ai_results)[-min(n, len(self.ai_results)) :] return float(sum(xs) / max(1, len(xs))) def _get_ev_multipliers(self) -> Dict[str, float]: """Chuyển EV rolling thành multipliers cho conf/stake/leverage.""" ev = self._get_ev(20) # EWMA EV để phản ứng nhanh hơn khi regime đổi (futures rất hay "flip") # alpha lớn hơn => phản ứng nhanh hơn. alpha = 0.25 self._ewma_ev = (alpha * ev) + ((1 - alpha) * float(getattr(self, "_ewma_ev", 0.0))) # Nếu EWMA tụt nhanh, tăng mức shrink ngay cả khi rolling EV chưa kịp xấu. ew = float(self._ewma_ev) # Futures: khi EV âm, phải co nhanh; khi EV tốt, nới vừa phải. if ew <= -0.0045: return {"conf": 1.15, "stake": 0.55, "lev": 0.60} if ev <= -0.006: return {"conf": 1.18, "stake": 0.45, "lev": 0.55} if ev <= -0.003: return {"conf": 1.10, "stake": 0.65, "lev": 0.70} if ev >= 0.008: return {"conf": 0.95, "stake": 1.12, "lev": 1.08} if ev >= 0.004: return {"conf": 0.98, "stake": 1.06, "lev": 1.03} return {"conf": 1.00, "stake": 1.00, "lev": 1.00} # ================= FREQAI PREDICTION HELPERS ================= def _get_trade_ok_confidence(self, df: DataFrame) -> float: """Lấy confidence/probability dựa trên dự đoán của FreqAI. Lưu ý quan trọng: - Cột "&s-trade_ok" là *nhãn* (ground-truth) do strategy tạo ra để train. - Khi live/backtest, FreqAI sẽ thêm các cột dự đoán (tên cột phụ thuộc model/pipeline). Hàm này cố gắng đọc các biến thể thường gặp của cột probability/prediction. Nếu không tìm thấy thì fallback về NaN/0.0 để tránh vô tình dùng label. """ if df is None or df.empty: return 0.0 last = df.iloc[-1].to_dict() # Các tên cột dự đoán hay gặp trong FreqAI. # (tuỳ phiên bản/model, có thể là prob cho class=1, hoặc score liên tục 0..1) candidate_cols = [ "&s-trade_ok_prob", "&s-trade_ok_probability", "&s-trade_ok_proba", "&s-trade_ok_pred_prob", "&s-trade_ok_pred_probability", "&s-trade_ok_pred_proba", "&s-trade_ok_predict_prob", "&s-trade_ok_confidence", "&s-trade_ok_pred", "&s-trade_ok_prediction", ] for c in candidate_cols: if c in df.columns: v = last.get(c, np.nan) if v is None or (isinstance(v, float) and np.isnan(v)): continue try: return float(v) except Exception: continue # Một số pipeline đặt tên chung hơn (không theo target). Thử thêm vài fallback. generic_cols = [ "prediction", "predicted", "pred", "probability", "proba", "confidence", ] for c in generic_cols: if c in df.columns: try: v = float(last.get(c, 0.0)) # Nếu giá trị liên tục (không phải 0/1), sử dụng ngay. if 0.0 < v < 1.0: return v except Exception: continue # ==== FALLBACK CUỐI: dùng chính cột &s-trade_ok ==== # Sau khi freqai.start() chạy, cột này chứa DỰ ĐOÁN (0/1) của model, # không còn là ground-truth label nữa. Với XGBoostClassifier, giá trị là 0 hoặc 1. if "&s-trade_ok" in df.columns: v = last.get("&s-trade_ok", np.nan) if v is not None and not (isinstance(v, float) and np.isnan(v)): return float(v) return 0.0 # ================= 4. ENTRY LOGIC ================= def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame: # Defensive init for hot-reload / partial instantiation edge-cases if not hasattr(self, "ai_pause_until"): self.ai_pause_until = None if not hasattr(self, "loss_streak"): self.loss_streak = 0 if not hasattr(self, "win_streak"): self.win_streak = 0 if not hasattr(self, "ai_results"): self.ai_results = deque(maxlen=30) # Chỉ trade khi FreqAI đã dự đoán. Không yêu cầu label tồn tại. if "do_predict" not in df.columns: return df try: now = pd.to_datetime(df.iloc[-1]['date']) except Exception: now = datetime.now(timezone.utc) if self.ai_pause_until and now < self.ai_pause_until: return df if self.loss_streak >= 4: self.ai_pause_until = now + timedelta(hours=1) logger.warning("⏸️ AI PAUSED 1h – Loss streak limit") return df if len(self.ai_results) >= 10: ev = sum(self.ai_results) / len(self.ai_results) if ev < -0.002: self.ai_pause_until = now + timedelta(minutes=30) logger.warning(f"⏸️ AI PAUSED 30m – Negative EV ({ev:.2%})") return df is_eth = metadata["pair"].startswith("ETH") conf_th = self.get_confidence_threshold() # ===== REGIME GATE (hộp số) ===== last = df.iloc[-1].to_dict() regime = self._get_regime(last) reg_mul = self.REGIME_MULTIPLIERS.get(regime, self.REGIME_MULTIPLIERS["NORMAL"]) ev_mul = self._get_ev_multipliers() # Nếu regime xấu, siết điều kiện vào lệnh. conf_th_adj = min(0.92, conf_th * reg_mul["conf"] * ev_mul["conf"]) # Gate cứng: chop/spike nặng thì bỏ qua hoàn toàn (giảm "chết nhanh" khi đổi regime) if regime == "SPIKE" and float(last.get("atrp", 0.0)) > 0.018: return df if regime == "CHOP" and int(last.get("chop_score", 0)) >= 3: return df # Lấy confidence/prediction cho hàng cuối để log, nhưng entry filter dùng cột prediction gốc. # Với XGBoostClassifier, &s-trade_ok sau freqai.start() chứa 0 hoặc 1 (prediction). # => Dùng == 1 thay vì > threshold (vì không phải probability). conf_last = self._get_trade_ok_confidence(df) # dùng cho log/confirm base = [ df["do_predict"] == 1, df["&s-trade_ok"] == 1, # XGBoostClassifier: prediction = 1 nghĩa là model dự đoán "kèo ngon" df["adx"] > (22 if is_eth else 26), df["atrp"] > 0.007, df["chop_score"] < 2, df["vol_ratio"] > 1.1, # Avoid noisy wicks in futures df["wick_ratio"] < 3.5, ] df.loc[ reduce(lambda a, b: a & b, base + [ df["close"] > df["ema200"], df["ema50"] > df["ema200"], ]), "enter_long" ] = 1 df.loc[ reduce(lambda a, b: a & b, base + [ df["close"] < df["ema200"], df["ema50"] < df["ema200"], ]), "enter_short" ] = 1 return df # ================= 5. EXIT LOGIC ================= def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[ (df["close"] < df["ema50"]) & (df["rsi"] < 40), "exit_long" ] = 1 df.loc[ (df["close"] > df["ema50"]) & (df["rsi"] > 60), "exit_short" ] = 1 return df # ================= TRADE CALLBACKS (Đúng API của FreqTrade) ================= def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs) -> bool: """ Xác nhận trước khi vào lệnh. Return False để từ chối lệnh. """ try: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last = df.iloc[-1].to_dict() # Kiểm tra confidence conf = self._get_trade_ok_confidence(df) if conf < float(self.confidence_low.value): logger.info(f"❌ Entry rejected for {pair}: Low confidence ({conf:.2f})") return False # Kiểm tra volatility tối thiểu atrp = last.get("atrp", 0) if atrp < float(self.atrp_threshold.value): logger.info(f"❌ Entry rejected for {pair}: Low volatility ({atrp:.4f})") return False # Kiểm tra trend strength adx = last.get("adx", 0) if adx < int(self.adx_threshold.value): logger.info(f"❌ Entry rejected for {pair}: Weak trend (ADX={adx:.1f})") return False logger.info(f"✅ Entry confirmed for {pair} | Side: {side} | Conf: {conf:.2f} | ADX: {adx:.1f}") return True except Exception as e: logger.error(f"Error in confirm_trade_entry: {e}") return True # Allow entry on error to not block trading 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: """ Xác nhận trước khi thoát lệnh. """ # Log thông tin exit profit = trade.calc_profit_ratio(rate) logger.info(f"🔔 Exit signal for {pair} | Reason: {exit_reason} | Profit: {profit:+.2%}") # Không block stoploss if exit_reason in ["stop_loss", "trailing_stop_loss"]: return True # Có thể thêm logic để giữ lệnh nếu profit đang tốt return True def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[str]: """ Custom exit conditions. """ try: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last = df.iloc[-1].to_dict() # Take profit động dựa trên ATR atrp = last.get("atrp", 0.01) # Nếu profit > 3x ATR, chốt lời if current_profit > 3 * atrp: logger.info(f"💰 Take profit triggered for {pair}: {current_profit:+.2%}") return "take_profit_atr" # Nếu đã có lãi > 1.5% và RSI quá cao/thấp, chốt lời rsi = last.get("rsi", 50) if trade.is_short: if current_profit > 0.015 and rsi < 25: return "take_profit_rsi_oversold" else: if current_profit > 0.015 and rsi > 75: return "take_profit_rsi_overbought" # Timeout: Nếu lệnh mở quá lâu (12 giờ) và không có lãi đáng kể trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 if trade_duration > 12 and current_profit < 0.005: logger.info(f"⏰ Timeout exit for {pair}: Duration {trade_duration:.1f}h") return "timeout_exit" except Exception as e: logger.debug(f"Error in custom_exit: {e}") return None 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) -> Optional[float]: """ Điều chỉnh position size (DCA hoặc partial close). Return positive để add, negative để reduce. """ try: # Nếu lỗ > 1.5% và confidence vẫn cao, có thể DCA if current_profit < -0.015: df, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) conf = self._get_trade_ok_confidence(df) # Chỉ DCA nếu confidence > 0.85 và chưa DCA quá nhiều if conf > 0.85 and len(trade.orders) < 3: # Add 50% của stake hiện tại add_stake = trade.stake_amount * 0.5 if add_stake >= min_stake: logger.info(f"📈 DCA for {trade.pair}: Adding {add_stake:.2f}") return add_stake except Exception as e: logger.debug(f"Error in adjust_trade_position: {e}") return None # ================= HELPERS (Quản lý vốn & Kết quả) ================= def bot_loop_start(self, current_time: datetime, **kwargs) -> None: """ Được gọi ở đầu mỗi bot loop. Dùng để update state. """ # Defensive init for hot-reload / partial instantiation edge-cases if not hasattr(self, "ai_pause_until"): self.ai_pause_until = None if not hasattr(self, "loss_streak"): self.loss_streak = 0 if not hasattr(self, "win_streak"): self.win_streak = 0 if not hasattr(self, "ai_results"): self.ai_results = deque(maxlen=30) if not hasattr(self, "_dynamic_max_trades"): self._dynamic_max_trades = int(getattr(self, "max_open_trades", 1) or 1) if not hasattr(self, "initial_wallet"): try: self.initial_wallet = float(self.config.get("dry_run_wallet", 1000)) if getattr(self, "config", None) else 1000.0 except Exception: self.initial_wallet = 1000.0 if not hasattr(self, "peak_wallet"): self.peak_wallet = float(self.initial_wallet) if not hasattr(self, "total_profit"): self.total_profit = 0.0 # Reset pause nếu đã hết thời gian if self.ai_pause_until and current_time >= self.ai_pause_until: logger.info("▶️ AI RESUMED - Pause period ended") self.ai_pause_until = None # Khởi tạo tracking nếu chưa có (tương thích backward) if not hasattr(self, "_last_ev"): self._last_ev = 0.0 if not hasattr(self, "_prev_win_streak"): self._prev_win_streak = 0 if not hasattr(self, "_cooldown_reason"): self._cooldown_reason = None # Khởi tạo regime tracking nếu thiếu if not hasattr(self, "_last_regime"): self._last_regime = {} # EV shock detector: nếu EV tụt nhanh, pause để tránh trả profit if len(self.ai_results) >= 10: ev = sum(self.ai_results) / len(self.ai_results) ev_drop = ev - float(self._last_ev) if ev_drop < -0.006: # tụt hơn -0.6% so với EV trước self.ai_pause_until = current_time + timedelta(minutes=45) self._cooldown_reason = f"EV shock: {self._last_ev:+.2%} → {ev:+.2%}" logger.warning(f"⏸️ COOLDOWN 45m – {self._cooldown_reason}") self._last_ev = ev # ===== REGIME FLIP COOLDOWN (nhanh hơn) ===== # Nếu regime đổi từ TREND/NORMAL sang CHOP/SPIKE thì nghỉ ngắn để tránh bị fakeout. try: if hasattr(self, "dp") and self.dp: for pair in getattr(self, "dp", {}).current_whitelist() if hasattr(self.dp, "current_whitelist") else []: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df is None or df.empty: continue last = df.iloc[-1].to_dict() new_reg = self._get_regime(last) prev_reg = self._last_regime.get(pair, "NORMAL") flip_to_bad = (prev_reg in ["TREND", "NORMAL"] and new_reg in ["CHOP", "SPIKE"]) if flip_to_bad: # SPIKE nguy hiểm hơn CHOP => pause dài hơn dur = 45 if new_reg == "SPIKE" else 25 self.ai_pause_until = max(self.ai_pause_until or current_time, current_time) + timedelta(minutes=dur) self._cooldown_reason = f"Regime flip {prev_reg} → {new_reg} ({pair})" logger.warning(f"⏸️ COOLDOWN {dur}m – {self._cooldown_reason}") self._last_regime[pair] = new_reg except Exception as e: logger.debug(f"Regime flip cooldown check failed: {e}") # Hot streak ended: vừa kết thúc chuỗi thắng lớn → nghỉ chút if self._prev_win_streak >= 6 and self.win_streak == 0: self.ai_pause_until = current_time + timedelta(minutes=30) self._cooldown_reason = f"Hot streak ended ({self._prev_win_streak} wins)" logger.warning(f"⏸️ COOLDOWN 30m – {self._cooldown_reason}") self._prev_win_streak = self.win_streak # ===== DYNAMIC MAX_OPEN_TRADES theo wallet tier ===== # Cập nhật max_open_trades dựa trên vốn hiện tại try: new_max_trades = self.max_open_trades_for_current_wallet() if new_max_trades != self._dynamic_max_trades: old_max = self._dynamic_max_trades self._dynamic_max_trades = new_max_trades # Cập nhật class attribute để Freqtrade đọc được self.max_open_trades = min(new_max_trades, self.config.get("max_open_trades", 10)) wallet = float(self.wallets.get_total_stake_amount()) tier = self._get_wallet_tier(wallet) logger.info(f"📈 MAX_TRADES UPDATED: {old_max} → {new_max_trades} | Wallet: ${wallet:.0f} ({tier})") except Exception as e: logger.debug(f"Error updating max_open_trades: {e}") def order_filled(self, pair: str, trade: Trade, order, current_time: datetime, **kwargs) -> None: """ Được gọi khi order được fill. Dùng để tracking và compound logic. """ # Xác định order này là entry hay exit (long/short có side ngược nhau) is_entry = (order.ft_order_side == "buy" and not trade.is_short) or (order.ft_order_side == "sell" and trade.is_short) if is_entry: logger.info(f"📥 Entry filled: {pair} | Rate: {order.average:.6f} | Amount: {order.amount:.4f}") return else: # Exit filled - update tracking if trade.close_profit is not None: profit = trade.close_profit # Update streaks if profit < 0: self.loss_streak += 1 self.win_streak = 0 else: self.win_streak += 1 self.loss_streak = 0 self.total_profit += profit # Track results self.ai_results.append(profit) # Update peak wallet current_wallet = float(self.wallets.get_total_stake_amount()) if current_wallet > self.peak_wallet: self.peak_wallet = current_wallet logger.info(f"🏆 NEW PEAK WALLET: ${current_wallet:.2f}") # Log stats với growth info if self.ai_results: ev = sum(self.ai_results) / len(self.ai_results) winrate = sum(1 for x in self.ai_results if x > 0) / len(self.ai_results) growth = ((current_wallet / self.initial_wallet) - 1) * 100 tier = self._get_wallet_tier(current_wallet) logger.info(f"📊 CLOSED | PnL: {profit:+.2%} | Win: {self.win_streak} | Loss: {self.loss_streak}") logger.info(f"💰 Wallet: ${current_wallet:.2f} ({tier}) | Growth: {growth:+.1f}% | WR: {winrate:.1%} | EV: {ev:+.3%}") def _get_wallet_tier(self, wallet: float) -> str: """Xác định tier của wallet để điều chỉnh risk""" if wallet < self.WALLET_TIERS["nano"]: return "NANO" # < $10: YOLO mode elif wallet < self.WALLET_TIERS["micro"]: return "MICRO" # $10-50 elif wallet < self.WALLET_TIERS["mini"]: return "MINI" # $50-200 elif wallet < self.WALLET_TIERS["small"]: return "SMALL" # $200-500 elif wallet < self.WALLET_TIERS["medium"]: return "MEDIUM" # $500-2000 elif wallet < self.WALLET_TIERS["large"]: return "LARGE" # $2000-10000 else: return "WHALE" # > $10000 def _get_risk_for_tier(self, wallet: float) -> float: """ Tính risk % dựa trên wallet tier. Snowball "tích dần": ưu tiên sống sót, không all-in. """ tier = self._get_wallet_tier(wallet) base_risk = { # Vốn siêu nhỏ: vẫn aggressive nhưng không YOLO để tránh bay tài khoản "NANO": 0.30, # 30% risk "MICRO": 0.15, # 15% risk "MINI": 0.08, # 8% risk # Các tier lớn hơn "SMALL": 0.05, # 5% risk "MEDIUM": 0.025, # 2.5% risk "LARGE": 0.015, # 1.5% risk "WHALE": 0.01, # 1% risk } return base_risk.get(tier, 0.02) def _calc_max_trades_for_wallet(self, wallet: float) -> int: """ Tính max_open_trades động theo wallet size. Vốn nhỏ = ít lệnh (tập trung), vốn lớn = nhiều lệnh hơn. """ tier = self._get_wallet_tier(wallet) tier_max_trades = { "NANO": 1, # < $10: Chỉ 1 lệnh, tập trung tối đa "MICRO": 1, # $10-50: Vẫn 1 lệnh "MINI": 2, # $50-200: Có thể 2 lệnh "SMALL": 2, # $200-500: 2 lệnh "MEDIUM": 3, # $500-2000: 3 lệnh "LARGE": 4, # $2000-10000: 4 lệnh "WHALE": 5, # > $10000: 5 lệnh } return tier_max_trades.get(tier, 2) def max_open_trades_for_current_wallet(self) -> int: """ Trả về max_open_trades động dựa trên wallet hiện tại. Gọi trong bot_loop_start để update. """ try: wallet = float(self.wallets.get_total_stake_amount()) return self._calc_max_trades_for_wallet(wallet) except Exception: return self._dynamic_max_trades def get_confidence_threshold(self) -> float: """Động threshold dựa trên performance gần đây""" wallet = float(self.wallets.get_total_stake_amount()) if hasattr(self, 'wallets') and self.wallets else 1000 tier = self._get_wallet_tier(wallet) # Vốn siêu nhỏ = threshold cực thấp để vào lệnh nhanh tier_threshold = { "NANO": 0.55, # Vốn < $10: nới lỏng để có nhiều cơ hội "MICRO": 0.58, "MINI": 0.62, "SMALL": 0.65, "MEDIUM": 0.68, "LARGE": 0.72, "WHALE": 0.78, } base = tier_threshold.get(tier, 0.65) # ===== WIN STREAK BOOST - Nới threshold khi đang hot ===== if self.win_streak >= 5: base -= 0.10 # Thắng 5+ → threshold giảm 10% logger.info(f"🎯 WIN STREAK: Threshold reduced by 10%") elif self.win_streak >= 3: base -= 0.06 # Thắng 3+ → threshold giảm 6% # ===== EV BOOST - Nới threshold khi EV tốt ===== if len(self.ai_results) >= 5: ev = sum(self.ai_results) / len(self.ai_results) if ev > 0.008: # EV > 0.8% = rất tốt base -= 0.08 logger.info(f"💎 HIGH EV ({ev:.2%}): Threshold reduced by 8%") elif ev > 0.004: # EV > 0.4% base -= 0.05 elif ev < -0.003: # EV âm = siết chặt base += 0.12 # ===== LOSS STREAK PROTECTION ===== if self.loss_streak >= 3: base += 0.15 # Thua 3+ → threshold tăng 15% elif self.loss_streak >= 2: base += 0.08 # Clamp threshold return max(0.45, min(base, 0.90)) 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: """ SNOWBALL STAKE CALCULATION Vốn nhỏ → Risk cao, compound khi thắng Vốn lớn → Risk thấp, bảo vệ """ wallet = float(self.wallets.get_total_stake_amount()) tier = self._get_wallet_tier(wallet) # Base risk theo tier risk = self._get_risk_for_tier(wallet) # ===== REGIME/EV MULTIPLIER (toàn cục) ===== try: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last = df.iloc[-1].to_dict() regime = self._get_regime(last) except Exception: regime = "NORMAL" reg_mul = self.REGIME_MULTIPLIERS.get(regime, self.REGIME_MULTIPLIERS["NORMAL"]) ev_mul = self._get_ev_multipliers() # Co lại ngay khi regime xấu hoặc EV xấu risk *= reg_mul["stake"] * ev_mul["stake"] # ===== AGGRESSIVE COMPOUND - Chỉ boost mạnh khi regime & EV ổn ===== if self.win_streak >= 7: if regime in ["TREND", "NORMAL"] and ev_mul["stake"] >= 1.0: risk *= 2.0 # x2 stake khi hot hand và market clean else: risk *= 1.25 logger.info(f"🔥🔥 SUPER COMPOUND: Win streak {self.win_streak} → Risk x2.0") elif self.win_streak >= 5: risk *= 1.7 if regime in ["TREND", "NORMAL"] else 1.15 logger.info(f"🔥 COMPOUND: Win streak {self.win_streak} → Risk x1.7") elif self.win_streak >= 3: risk *= 1.4 if regime in ["TREND", "NORMAL"] else 1.10 logger.info(f"📈 COMPOUND: Win streak {self.win_streak} → Risk x1.4") elif self.win_streak >= 2: risk *= 1.15 # +15% khi thắng 2 lệnh # EV boost đã được đưa vào ev_mul ở trên để thống nhất (stake/leverage/threshold) # ===== RISK CAP - Không cho compound phóng quá mức (tránh trả lại profit) ===== # Cap theo tier: nhỏ thì cho cao hơn, lớn thì cap thấp hơn. risk_cap = { "NANO": 0.45, "MICRO": 0.30, "MINI": 0.18, "SMALL": 0.10, "MEDIUM": 0.06, "LARGE": 0.04, "WHALE": 0.025, }.get(tier, 0.08) if risk > risk_cap: logger.info(f"🧯 RISK CAP: {risk:.1%} → {risk_cap:.1%} (tier={tier})") risk = risk_cap # ===== PROTECTION LOGIC - Giảm stake khi thua ===== if self.loss_streak >= 4: risk *= 0.25 # Giảm 75% khi thua 4 lệnh logger.warning(f"⚠️ PROTECTION: Loss streak {self.loss_streak} → Risk x0.25") elif self.loss_streak == 3: risk *= 0.4 elif self.loss_streak == 2: risk *= 0.6 # ===== DRAWDOWN PROTECTION ===== if wallet < self.peak_wallet * 0.9: # Drawdown > 10% risk *= 0.5 logger.warning(f"⚠️ DRAWDOWN PROTECTION: Wallet ${wallet:.0f} < Peak ${self.peak_wallet:.0f} * 0.9") # ===== CONFIDENCE ADJUSTMENT ===== try: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) conf = self._get_trade_ok_confidence(df) if conf > 0.85 and self.loss_streak < 2 and regime in ["TREND", "NORMAL"]: risk *= 1.25 elif conf < 0.65: risk *= 0.6 except Exception as e: logger.debug(f"Error getting confidence: {e}") # Tính stop distance stop_dist = abs(self.stoploss) try: last_row = df.iloc[-1].to_dict() atrp = last_row.get("atrp", None) if atrp and atrp > 0: stop_dist = max(stop_dist, atrp * 1.5) # 1.5x ATR buffer except: pass # Tính final stake if stop_dist > 0: stake = (wallet * risk) / stop_dist else: stake = wallet * risk # Clamp to limits stake = max(min_stake or 0, min(stake, max_stake)) logger.info(f"💎 STAKE [{tier}]: ${stake:.2f} | Risk: {risk:.1%} | Wallet: ${wallet:.0f} | Win: {self.win_streak} | Loss: {self.loss_streak}") return stake def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs) -> Optional[float]: """ Dynamic trailing stoploss based on ATR và profit. """ try: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_row = df.iloc[-1].to_dict() atr = last_row.get("atr", 0) atrp = atr / current_rate if current_rate > 0 else 0.01 # Trailing stop khi có lãi if current_profit > 0.020: # > 2% profit # Lock 50% profit sl = -max(current_profit * 0.5, atrp) return max(sl, -0.004) elif current_profit > 0.012: # > 1.2% profit # Tighter trailing sl = -(atrp * 1.5) return max(sl, -0.006) elif current_profit > 0.006: # > 0.6% profit # Start trailing sl = -(atrp * 2) return max(sl, -0.010) # Dưới 0.6% profit, giữ stoploss mặc định return self.stoploss except Exception as e: logger.debug(f"Error in custom_stoploss: {e}") return self.stoploss def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: """ SNOWBALL LEVERAGE - Vốn nhỏ = Leverage cao hơn để tăng tốc growth Vốn lớn = Leverage thấp hơn để bảo vệ """ try: wallet = float(self.wallets.get_total_stake_amount()) tier = self._get_wallet_tier(wallet) df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_row = df.iloc[-1].to_dict() atrp = last_row.get("atrp", 0.01) adx = last_row.get("adx", 20) conf = self._get_trade_ok_confidence(df) regime = self._get_regime(last_row) reg_mul = self.REGIME_MULTIPLIERS.get(regime, self.REGIME_MULTIPLIERS["NORMAL"]) ev_mul = self._get_ev_multipliers() # ===== BASE LEVERAGE THEO TIER ===== tier_leverage = { "NANO": 6.0, # Vốn < $10: leverage vừa phải để sống sót "MICRO": 5.0, # $10-50 "MINI": 4.0, # $50-200 "SMALL": 3.0, # $200-500 "MEDIUM": 2.5, # $500-2000 "LARGE": 2.0, # $2000-10000 "WHALE": 1.5, # > $10000 } base_lev = tier_leverage.get(tier, 2.5) # Regime/EV multipliers - co lại trong chop/spike hoặc EV xấu base_lev *= reg_mul["lev"] * ev_mul["lev"] # Không dùng "NANO max leverage" nữa - vẫn cho phép điều chỉnh theo volatility/trend/conf # ===== ĐIỀU CHỈNH THEO VOLATILITY ===== if atrp > 0.015: # Rất volatile - giảm leverage base_lev *= 0.6 elif atrp > 0.010: # Volatile base_lev *= 0.8 elif atrp < 0.005: # Rất ít volatile - có thể tăng base_lev *= 1.2 # ===== ĐIỀU CHỈNH THEO TREND ===== if adx > 30: # Trend rất mạnh - có thể tăng leverage base_lev *= 1.15 elif adx < 18: # Trend yếu - giảm leverage base_lev *= 0.7 # ===== ĐIỀU CHỈNH THEO CONFIDENCE ===== if conf > 0.85: base_lev *= 1.1 elif conf < 0.65: base_lev *= 0.7 # ===== PROTECTION: Giảm leverage khi thua (không áp dụng cho MICRO) ===== if tier not in ["NANO", "MICRO"]: if self.loss_streak >= 3: base_lev *= 0.4 logger.warning(f"⚠️ Leverage reduced due to {self.loss_streak} loss streak") elif self.loss_streak >= 2: base_lev *= 0.6 # ===== COMPOUND: Tăng leverage khi thắng ===== if self.win_streak >= 6 and tier in ["NANO", "MICRO", "MINI", "SMALL"]: if regime in ["TREND", "NORMAL"] and ev_mul["lev"] >= 1.0: base_lev *= 1.5 else: base_lev *= 1.10 logger.info(f"🔥🔥 LEVERAGE SUPER BOOST: Win streak {self.win_streak} → x1.5") elif self.win_streak >= 4 and tier in ["NANO", "MICRO", "MINI", "SMALL"]: base_lev *= 1.35 if regime in ["TREND", "NORMAL"] else 1.08 logger.info(f"🔥 LEVERAGE BOOST: Win streak {self.win_streak} → x1.35") elif self.win_streak >= 2 and tier in ["NANO", "MICRO", "MINI"]: base_lev *= 1.15 # +15% leverage khi thắng 2+ lệnh # EV leverage handled through ev_mul for consistency # Clamp to max allowed final_lev = min(max(1.0, base_lev), max_leverage, 10.0) # Max 10x logger.info(f"⚡ LEVERAGE [{tier}]: {final_lev:.1f}x | ADX: {adx:.0f} | ATRP: {atrp:.3f}") return final_lev except Exception as e: logger.debug(f"Error in leverage calculation: {e}") # Default leverage theo wallet tier wallet = float(self.wallets.get_total_stake_amount()) if hasattr(self, 'wallets') and self.wallets else 1000 if wallet < 500: return min(4.0, max_leverage) elif wallet < 2000: return min(3.0, max_leverage) else: return min(2.0, max_leverage)