# source: https://raw.githubusercontent.com/raulcarreno/trader-bot/1e811ea1c84acdabce4bafe74c14ce1920da7549/user_data/strategies/FreqaiExampleStrategyV3.py from user_data.strategies.v3.core.thread_env import configure_safe_native_threads configure_safe_native_threads() import logging from datetime import datetime, timedelta from functools import reduce from pathlib import Path import talib.abstract as ta from pandas import DataFrame from technical import qtpylib from freqtrade.strategy import IStrategy, Trade, timeframe_to_minutes from user_data.strategies.v3.core.account_tracker import AccountTracker from user_data.strategies.v3.core.config import build_v3_config from user_data.strategies.v3.core.orchestrator import OrchestratorV3 from user_data.strategies.v3.experts.news.features import build_news_features from user_data.strategies.v3.core.debug_columns import V3_BLOCK_REASON, V3_WOULD_ENTER_LONG from user_data.strategies.v3.core.debug_log import log_v3_decision, log_v3_experts from user_data.strategies.v3.core.expert_trace import build_expert_trace_from_row from user_data.strategies.v3.execution.entry_gates import entry_gate_masks, validate_live_entry from user_data.strategies.v3.execution.exit_policy import ExitPolicy from user_data.strategies.v3.experts.quant.features import build_quant_features from user_data.strategies.v3.mlops.registry import build_registry_snapshot, write_registry_snapshot from user_data.strategies.v3.news_pipeline.live_scheduler import NewsIngestScheduler logger = logging.getLogger(__name__) class Github_raulcarreno_trader_bot__FreqaiExampleStrategyV3__20260610_164544(IStrategy): """ V3 adapter: offline quant ensemble (LGBM + XGB + MLP) + BS + News via OrchestratorV3. Does not modify V1/V2 strategies. """ timeframe = "5m" TARGET_LEVERAGE = 10.0 MAX_STAKE_PRED_MULT = 2.0 DI_MAX_DEFAULT = 0.95 MIN_ADX = 22 MAX_ATR_PCT = 0.012 INDICATOR_PERIODS = [10, 20] stoploss = -0.15 use_custom_stoploss = True trailing_stop = False _tf_min = timeframe_to_minutes(timeframe) minimal_roi = { "0": 0.10, str(_tf_min * 12): 0.06, str(_tf_min * 18): 0.04, str(_tf_min * 24): 0.02, str(_tf_min * 30): 0, } plot_config = { "main_plot": {}, "subplots": { "ensemble_score": {"ensemble_score": {"color": "green"}}, "opportunity_score": {"opportunity_score": {"color": "cyan"}}, "quant1_blend": {"quant1_blend": {"color": "blue"}}, "quant2_hit_prob": {"quant2_hit_prob": {"color": "orange"}}, "do_predict": {"do_predict": {"color": "brown"}}, "v3_debug": { V3_WOULD_ENTER_LONG: {"color": "lime"}, }, }, } process_only_new_candles = True use_exit_signal = True startup_candle_count: int = 310 can_short = True def __init__(self, config: dict) -> None: super().__init__(config) self.v3_config = build_v3_config(config) self.orchestrator = OrchestratorV3(self.v3_config) self.account_tracker = self.orchestrator.account_tracker self.registry_path = ( Path("user_data/models") / self.v3_config.identifier / "registry_snapshot.json" ) self._pair_trades_today: dict[str, int] = {} self._last_trade_day = None self._pair_cooldown_until: dict[str, datetime] = {} self._runmode: str = "unknown" self._last_entry_reject_log: dict[str, str] = {} self._last_expert_log_key: dict[str, str] = {} self.exit_policy = ExitPolicy(self.v3_config.strategy_execution) self._news_ingest_scheduler = NewsIngestScheduler(self.v3_config) def _sync_news_ingest_runmode(self) -> None: if getattr(self, "dp", None) is not None and getattr(self.dp, "runmode", None) is not None: self._runmode = str(self.dp.runmode.value) self._news_ingest_scheduler.set_runmode(self._runmode) def _maybe_run_news_ingest(self, *, on_start: bool = False) -> None: scheduler = self._news_ingest_scheduler whitelist = self.config.get("exchange", {}).get("pair_whitelist", []) if whitelist: scheduler.set_exchange_pairs(whitelist) scheduler.set_runmode(self._runmode) scheduler.maybe_run_async(on_start=on_start) def ft_bot_start(self, **kwargs) -> None: super().ft_bot_start(**kwargs) self._sync_news_ingest_runmode() self._maybe_run_news_ingest(on_start=True) def _di_threshold(self) -> float: return float(self.v3_config.feature_filter.di_threshold) def _entry_gates(self, df: DataFrame, side: str) -> list: masks = entry_gate_masks( df, side, exec_cfg=self.v3_config.strategy_execution, hybrid=self.v3_config.hybrid_quant, di_threshold=self._di_threshold(), max_atr_pct=self.MAX_ATR_PCT, min_adx=self.MIN_ADX, ) return list(masks.values()) @property def protections(self): return [ {"method": "CooldownPeriod", "stop_duration_candles": 2}, { "method": "StoplossGuard", "lookback_period_candles": 36, "trade_limit": 4, "stop_duration_candles": 12, "only_per_pair": False, }, { "method": "MaxDrawdown", "calculation_mode": "equity", "lookback_period_candles": 288, "trade_limit": 3, "stop_duration_candles": 72, "max_allowed_drawdown": 0.08, }, ] def _build_base_features(self, dataframe: DataFrame, pair: str = "") -> DataFrame: for period in self.INDICATOR_PERIODS: dataframe = build_quant_features(dataframe, period) dataframe["%-pct-change"] = dataframe["close"].pct_change() dataframe["%-raw_volume"] = dataframe["volume"] dataframe["%-raw_price"] = dataframe["close"] news = self.v3_config.news_feed dataframe = build_news_features( dataframe, pair=pair or None, news_store_root=news.store_root, news_enabled=news.enabled, ) if "date" in dataframe.columns: dataframe["%-day_of_week"] = dataframe["date"].dt.dayofweek dataframe["%-hour_of_day"] = dataframe["date"].dt.hour dataframe["atr_pct"] = ta.ATR(dataframe, timeperiod=14) / dataframe["close"] dataframe["adx"] = ta.ADX(dataframe, timeperiod=14) return dataframe def feature_engineering_expand_all( self, dataframe: DataFrame, period: int, metadata: dict, **kwargs ) -> DataFrame: return build_quant_features(dataframe, period) def feature_engineering_expand_basic( self, dataframe: DataFrame, metadata: dict, **kwargs ) -> DataFrame: dataframe["%-pct-change"] = dataframe["close"].pct_change() dataframe["%-raw_volume"] = dataframe["volume"] dataframe["%-raw_price"] = dataframe["close"] news = self.v3_config.news_feed return build_news_features( dataframe, pair=metadata.get("pair"), news_store_root=news.store_root, news_enabled=news.enabled, ) def feature_engineering_standard( self, dataframe: DataFrame, metadata: dict, **kwargs ) -> DataFrame: 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: dataframe["&-s_close"] = dataframe["close"].pct_change() return dataframe def _refresh_account_tracker(self, current_time: datetime | None = None) -> None: if getattr(self, "wallets", None) is not None: total = float(self.wallets.get_total(self.config["stake_currency"])) open_stake = 0.0 if getattr(self, "trade_pool", None) is not None: open_stake = sum(float(t.stake_amount) for t in self.trade_pool.open_trades) self.account_tracker.update_from_wallet(total, open_stake) if current_time is not None: day = current_time.date() if hasattr(current_time, "date") else current_time self.account_tracker.roll_day_if_needed(day) def _is_backtest_like_runmode(self) -> bool: runmode = getattr(self, "_runmode", "unknown") return runmode in ("backtest", "hyperopt", "plot") def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata.get("pair", "") if getattr(self, "dp", None) is not None and getattr(self.dp, "runmode", None) is not None: self._runmode = str(self.dp.runmode.value) self._sync_news_ingest_runmode() self._refresh_account_tracker() if self.v3_config.freqai_enabled: dataframe = self._build_base_features(dataframe, pair=pair) dataframe = self.freqai.start(dataframe, metadata, self) infer_latest_only = ( self.v3_config.performance.infer_latest_only_live and not self._is_backtest_like_runmode() ) dataframe = self.orchestrator.infer_dataframe( dataframe, pair, infer_latest_only=infer_latest_only, ) self._log_expert_snapshot(pair, dataframe) return dataframe def _log_expert_snapshot(self, pair: str, dataframe: DataFrame) -> None: debug_cfg = self.v3_config.debug if not debug_cfg.log_experts or dataframe.empty: return last = dataframe.iloc[-1] do_predict = int(last.get("do_predict", 0) or 0) if not debug_cfg.log_every_candle and do_predict != 1: return candle_key = str(last.get("date", "")) dedupe_key = f"{pair}:{candle_key}" if self._last_expert_log_key.get(pair) == dedupe_key: return self._last_expert_log_key[pair] = dedupe_key snapshot = build_expert_trace_from_row(pair, last, self.v3_config) log_v3_experts(logger, snapshot, runmode=self._runmode) def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame: long_conditions = self._entry_gates(df, "long") short_conditions = self._entry_gates(df, "short") if long_conditions: df.loc[reduce(lambda x, y: x & y, long_conditions), ["enter_long", "enter_tag"]] = ( 1, "ensemble_long", ) if short_conditions: df.loc[reduce(lambda x, y: x & y, short_conditions), ["enter_short", "enter_tag"]] = ( 1, "ensemble_short", ) return df def _confirmed_exit_mask(self, df: DataFrame, raw_exit: DataFrame) -> DataFrame: exec_cfg = self.v3_config.strategy_execution n = max(1, int(exec_cfg.exit_confirm_candles)) if n <= 1: return raw_exit confirmed = raw_exit.astype(int).rolling(window=n, min_periods=n).sum() >= n return confirmed.fillna(False) def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame: exec_cfg = self.v3_config.strategy_execution exit_thr = df["exit_thr_eff"] if "exit_thr_eff" in df.columns else exec_cfg.exit_score_threshold weak_thr = df["exit_weak_eff"] if "exit_weak_eff" in df.columns else exec_cfg.weak_score_threshold score_col = df["score_adj"] if "score_adj" in df.columns else df["ensemble_score"] flip_long = (df["ensemble_side"] == "short") & (score_col < -exit_thr) weak_long = score_col < -weak_thr raw_long = flip_long | weak_long long_exit = self._confirmed_exit_mask(df, raw_long) flip_short = (df["ensemble_side"] == "long") & (score_col > exit_thr) weak_short = score_col > weak_thr raw_short = flip_short | weak_short short_exit = self._confirmed_exit_mask(df, raw_short) df.loc[long_exit, "exit_long"] = 1 df.loc[short_exit, "exit_short"] = 1 return df def _max_leverage_cap(self) -> float: return min( self.TARGET_LEVERAGE, self.v3_config.strategy_execution.validation_max_leverage, self.v3_config.risk.max_leverage, ) def leverage( self, pair: str, current_time, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str | None, side: str, **kwargs, ) -> float: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df.empty: return min(self._max_leverage_cap(), max_leverage) score = abs(float(df.iloc[-1].get("score_adj", 0.0) or 0.0)) return min(self._max_leverage_cap(), max_leverage, 1.0 + score * 5.0) def custom_stake_amount( self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: float | None, max_stake: float, leverage: float, entry_tag: str | None, side: str, **kwargs, ) -> float: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df.empty: return proposed_stake last = df.iloc[-1] if int(last.get("do_predict", 0) or 0) != 1: return 0.0 score = abs(float(last.get("score_adj", 0.0) or 0.0)) multiplier = min(self.MAX_STAKE_PRED_MULT, max(0.25, score * 15.0)) stake = proposed_stake * multiplier floor = min_stake if min_stake is not None else 0.0 return max(floor, min(stake, max_stake)) def custom_stoploss( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs, ) -> float | None: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df.empty: return None from user_data.strategies.v3.execution.stoploss_policy import resolve_custom_stoploss_value exec_cfg = self.v3_config.strategy_execution atr_pct = float(df.iloc[-1].get("atr_pct", 0.006) or 0.006) dynamic_stop = -min( exec_cfg.max_stoploss_pct, max(exec_cfg.min_stoploss_pct, atr_pct * 2.5), ) return resolve_custom_stoploss_value( exec_cfg, dynamic_stop=dynamic_stop, strategy_stoploss=self.stoploss, current_profit=current_profit, ) def custom_exit( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> str | bool | None: return None def _roll_pair_trade_day(self, current_time) -> None: day = current_time.date() if hasattr(current_time, "date") else current_time if self._last_trade_day != day: self._pair_trades_today = {} self._last_trade_day = day def _validate_entry( self, pair: str, side: str, last, current_time, ) -> tuple[bool, str]: return validate_live_entry( pair, side, last, current_time, exec_cfg=self.v3_config.strategy_execution, hybrid=self.v3_config.hybrid_quant, di_threshold=self._di_threshold(), max_atr_pct=self.MAX_ATR_PCT, min_adx=self.MIN_ADX, pair_cooldown_until=self._pair_cooldown_until.get(pair), pair_trades_today=self._pair_trades_today.get(pair, 0), pair_trade_limits=self.v3_config.risk.pair_trade_limits, ) def _log_entry_rejection(self, pair: str, current_time, reason: str, last) -> None: debug_cfg = self.v3_config.debug if not debug_cfg.enabled or not debug_cfg.log_rejected_entries: return candle_key = str(current_time) dedupe_key = f"{pair}:{candle_key}:{reason}" if self._last_entry_reject_log.get(pair) == dedupe_key: return self._last_entry_reject_log[pair] = dedupe_key log_v3_decision( logger, pair, "entry_rejected", candle_time=candle_key, reason=reason, runmode=self._runmode, score_adj=float(last.get("score_adj", 0.0) or 0.0) if last is not None else 0.0, block_reason=str(last.get(V3_BLOCK_REASON, "") or "") if last is not None else "", ) 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: self._refresh_account_tracker(current_time) self._roll_pair_trade_day(current_time) df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df.empty: if self.v3_config.debug.log_rejected_entries: self._log_entry_rejection(pair, current_time, "empty_dataframe", None) return False last = df.iloc[-1].squeeze() allowed, reason = self._validate_entry(pair, side, last, current_time) if not allowed: self._log_entry_rejection(pair, current_time, reason, last) return False self._pair_trades_today[pair] = self._pair_trades_today.get(pair, 0) + 1 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: exec_cfg = self.v3_config.strategy_execution if exit_reason == "exit_signal" and exec_cfg.min_hold_candles > 0: open_dt = trade.open_date_utc if trade.open_date_utc.tzinfo else trade.open_date hold_minutes = (current_time - open_dt).total_seconds() / 60.0 min_minutes = exec_cfg.min_hold_candles * timeframe_to_minutes(self.timeframe) if hold_minutes < min_minutes: debug_cfg = self.v3_config.debug if debug_cfg.enabled and debug_cfg.log_rejected_entries: log_v3_decision( logger, pair, "exit_rejected", candle_time=str(current_time), reason="min_hold", hold_minutes=round(hold_minutes, 2), min_minutes=min_minutes, runmode=self._runmode, ) return False if exec_cfg.post_exit_cooldown_candles > 0: cooldown_min = exec_cfg.post_exit_cooldown_candles * timeframe_to_minutes(self.timeframe) self._pair_cooldown_until[pair] = current_time + timedelta(minutes=cooldown_min) if trade is not None: try: pnl = float(trade.calc_profit_ratio(rate)) except Exception: pnl = 0.0 self.orchestrator.record_trade_result(pnl, ()) return True def bot_loop_start(self, current_time: datetime, **kwargs) -> None: super().bot_loop_start(current_time, **kwargs) self._refresh_account_tracker(current_time) self._maybe_run_news_ingest(on_start=False) metrics = { "max_drawdown": float(self.config.get("go_no_go", {}).get("max_drawdown", 0.0)), "profit_factor": float(self.config.get("go_no_go", {}).get("profit_factor", 0.0)), "monthly_stability": float( self.config.get("go_no_go", {}).get("monthly_stability", 0.0) ), "constraint_breaches": float( self.config.get("go_no_go", {}).get("constraint_breaches", 0.0) ), } snapshot = build_registry_snapshot(self.v3_config.identifier, metrics, {}, self.v3_config.go_no_go) write_registry_snapshot(self.registry_path, snapshot)