From 22d27429eca52d2444c07bbca19c5185cc9e99b3 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Tue, 28 Jul 2026 22:35:17 +0200 Subject: [PATCH] fix(quickadapter): use directional PnL momentum gate (#149) Replace the velocity/acceleration t-statistic exit gate with a deterministic mean per-candle PnL velocity direction rule: once the take-profit target is reached and the recent window is complete, allow the exit iff recent[-1] < recent[0] (equivalent to strictly negative mean velocity via telescoping). Remove the now-dead machinery: decline_quantile, t-critical/effective-df, acceleration and zero-variance fallbacks, get_pnl_momentum, get_trade_unrealized_pnl_history, the thresholds_calibration config specs, and the unrealized_pnl_timeframe_minutes field. A legacy exit_pricing.thresholds_calibration setting now warns as obsolete and ignored. Fail-open only where the horizon is unmeasurable (missing candle date, incomplete window, non-finite value in the selected window); complete finite windows are deterministic (falling exits; flat or rising blocks). Fixes #128 --- README.md | 1 - .../user_data/strategies/QuickAdapterV3.py | 266 +++--------------- quickadapter/user_data/strategies/Utils.py | 48 +--- 3 files changed, 52 insertions(+), 263 deletions(-) diff --git a/README.md b/README.md index 255fb1b..07d369e 100644 --- a/README.md +++ b/README.md @@ -51,7 +51,6 @@ docker compose up -d --build | leverage | `proposed_leverage` | float [1.0, max_leverage] | Leverage. Fallback to `proposed_leverage` for the pair. | | _Exit pricing_ | | | | | exit_pricing.trade_price_target_method | `moving_average` | enum {`moving_average`,`quantile_interpolation`,`weighted_average`} | Trade NATR computation method. | -| exit_pricing.thresholds_calibration.decline_quantile | 0.5 | float (0,1) | PnL decline quantile threshold. | | _Reversal confirmation_ | | | | | reversal_confirmation.lookback_period_candles | 0 | int >= 0 | Prior confirming candles; 0 = none. | | reversal_confirmation.decay_fraction | 0.5 | float (0,1] | Geometric per-candle volatility adjusted reversal threshold relaxation factor. | diff --git a/quickadapter/user_data/strategies/QuickAdapterV3.py b/quickadapter/user_data/strategies/QuickAdapterV3.py index db177c1..2caff75 100644 --- a/quickadapter/user_data/strategies/QuickAdapterV3.py +++ b/quickadapter/user_data/strategies/QuickAdapterV3.py @@ -39,10 +39,8 @@ from LabelTransformer import ( get_label_column_config, ) from pandas import DataFrame, Series, isna, to_numeric -from scipy.stats import pearsonr, t from technical.pivots_points import pivots_points from Utils import ( - DEFAULTS_EXIT_THRESHOLDS_CALIBRATION, _CACHE_MAXSIZE_LARGE, _OPTUNA_NAMESPACES, EXTREMA_COLUMN, @@ -69,7 +67,6 @@ from Utils import ( get_distance, get_custom_protections_config, get_exit_pricing_config, - get_exit_thresholds_calibration_config, get_fit_live_predictions_candles, get_label_defaults, get_label_horizon_candles, @@ -110,13 +107,12 @@ _TakeProfitHistoryEntry = float | tuple[int, float] | list[int | float] class _TradeHistory(TypedDict): # Key names must mirror the _UNREALIZED_PNL_CANDLE_DATE_KEY / - # _UNREALIZED_PNL_TIMEFRAME_KEY / _LEGACY_UNREALIZED_PNL_TIMEFRAME_MINUTES_KEY - # constants (a TypedDict field cannot reference a constant). + # _UNREALIZED_PNL_TIMEFRAME_KEY constants (a TypedDict field cannot + # reference a constant). unrealized_pnl: list[float] take_profit_price: list[_TakeProfitHistoryEntry] unrealized_pnl_candle_date: NotRequired[str] unrealized_pnl_timeframe: NotRequired[str] - unrealized_pnl_timeframe_minutes: NotRequired[int] logger = logging.getLogger(__name__) @@ -181,15 +177,6 @@ class QuickAdapterV3(IStrategy): stoploss = -0.025 use_custom_stoploss = True - default_exit_thresholds: ClassVar[dict[str, float]] = { - "t_decl_v": 0.675, - "t_decl_a": 0.675, - } - - default_exit_thresholds_calibration: ClassVar[dict[str, float]] = ( - DEFAULTS_EXIT_THRESHOLDS_CALIBRATION.copy() - ) - position_adjustment_enable = True # {stage: (natr_multiplier_fraction, stake_percent, color)} @@ -214,15 +201,6 @@ class QuickAdapterV3(IStrategy): _TAKE_PROFIT_ORDER_TAG_PREFIX: Final[str] = "take_profit_" _UNREALIZED_PNL_CANDLE_DATE_KEY: Final[str] = "unrealized_pnl_candle_date" _UNREALIZED_PNL_TIMEFRAME_KEY: Final[str] = "unrealized_pnl_timeframe" - _LEGACY_UNREALIZED_PNL_TIMEFRAME_MINUTES_KEY: Final[str] = ( - "unrealized_pnl_timeframe_minutes" - ) - - # get_pnl_momentum differences the window twice: velocity needs >=2 first - # diffs (window>=3), acceleration >=2 second diffs (window>=4). 4 is the - # binding floor so both t-statistics are computable (n>=2); with fewer - # samples the acceleration t-statistic is structurally NaN. - _MIN_PNL_MOMENTUM_WINDOW_SIZE: Final[int] = 4 # Rounding margin so the sized partial-exit remainder clears freqtrade's # strict `remaining < min_exit_stake` guard. @@ -472,36 +450,11 @@ class QuickAdapterV3(IStrategy): # 30-minute velocity span needs ceil(30/tf)+1 samples (ceil so a # timeframe not dividing 30 still spans >=30 min). nominal_pnl_momentum_window_size = math.ceil(30 / self.timeframe_minutes) + 1 - self._pnl_momentum_window_size = max( - QuickAdapterV3._MIN_PNL_MOMENTUM_WINDOW_SIZE, - nominal_pnl_momentum_window_size, - ) + self._pnl_momentum_window_size = nominal_pnl_momentum_window_size self._max_history_size = max( self._pnl_momentum_window_size, int(12 * 60 / self.timeframe_minutes), ) - if ( - nominal_pnl_momentum_window_size - < QuickAdapterV3._MIN_PNL_MOMENTUM_WINDOW_SIZE - ): - velocity_span_minutes = ( - self._pnl_momentum_window_size - 1 - ) * self.timeframe_minutes - logger.warning( - f"Timeframe {self.timeframe}: the nominal 30-minute PnL momentum " - f"window resolves to only {nominal_pnl_momentum_window_size} samples " - f"(< {QuickAdapterV3._MIN_PNL_MOMENTUM_WINDOW_SIZE} needed " - f"to compute an acceleration t-statistic); flooring to " - f"{self._pnl_momentum_window_size} candles " - f"(~{velocity_span_minutes} min velocity span)." - ) - self._exit_thresholds_calibration: dict[str, float] = ( - get_exit_thresholds_calibration_config( - self.config.get("exit_pricing"), - logger, - self.default_exit_thresholds_calibration, - ) - ) self._candle_deviation_cache: dict[CandleDeviationCacheKey, float] = {} self._candle_threshold_cache: dict[CandleThresholdCacheKey, float] = {} self._cached_df_signature: dict[str, DfSignature] = {} @@ -618,9 +571,6 @@ class QuickAdapterV3(IStrategy): logger.info("Exit Pricing:") logger.info(f" trade_price_target_method: {self.trade_price_target_method}") - logger.info( - f" thresholds_calibration: {format_dict(self._exit_thresholds_calibration, style='dict')}" - ) logger.info("Custom Stoploss:") logger.info( @@ -1539,11 +1489,6 @@ class QuickAdapterV3(IStrategy): "history", {"unrealized_pnl": [], "take_profit_price": []} ) - @staticmethod - def get_trade_unrealized_pnl_history(trade: Trade) -> list[float]: - history = QuickAdapterV3._get_trade_history(trade) - return history.get("unrealized_pnl", []) - @staticmethod def get_trade_take_profit_price_history( trade: Trade, @@ -1564,7 +1509,6 @@ class QuickAdapterV3(IStrategy): candle_date.isoformat() ) history[QuickAdapterV3._UNREALIZED_PNL_TIMEFRAME_KEY] = self.timeframe - history.pop(QuickAdapterV3._LEGACY_UNREALIZED_PNL_TIMEFRAME_MINUTES_KEY, None) trade.set_custom_data("history", history) return pnl_history @@ -2069,61 +2013,32 @@ class QuickAdapterV3(IStrategy): return True @staticmethod - def get_pnl_momentum( + def is_pnl_declining( unrealized_pnl_history: Sequence[float], window_size: int - ) -> tuple[ - tuple[float, ...], - float, - float, - tuple[float, ...], - float, - float, - ]: - """Compute velocity (first derivative) and acceleration (second) from PnL history. + ) -> Optional[bool]: + """Return whether mean per-candle PnL velocity is strictly negative. ``window_size > 0`` truncates to the most recent window before - differencing. Returns - ``(velocity_values, velocity_mean, velocity_std, acceleration_values, - acceleration_mean, acceleration_std)``. + evaluating the direction. Because the mean of consecutive first + differences telescopes, this is equivalent to comparing the last and + first samples. A short window or one containing non-numeric or + non-finite samples is unmeasurable and returns ``None``. """ - unrealized_pnl_history_array = np.asarray(unrealized_pnl_history, dtype=float) - - if window_size > 0 and unrealized_pnl_history_array.size > window_size: - unrealized_pnl_history_array = unrealized_pnl_history_array[-window_size:] - - velocity = np.diff(unrealized_pnl_history_array) - velocity_mean = np.nanmean(velocity) if velocity.size > 0 else 0.0 - velocity_std = np.nanstd(velocity, ddof=1) if velocity.size > 1 else 0.0 - - acceleration = np.diff(velocity) - acceleration_mean = np.nanmean(acceleration) if acceleration.size > 0 else 0.0 - acceleration_std = ( - np.nanstd(acceleration, ddof=1) if acceleration.size > 1 else 0.0 - ) - - return ( - tuple(velocity.tolist()), - velocity_mean, - velocity_std, - tuple(acceleration.tolist()), - acceleration_mean, - acceleration_std, - ) - - @staticmethod - def _t_statistic(mean: float, std: float, n: int) -> float: - """Compute t-statistic for H0: mu = 0 as ``mean * sqrt(n) / std``. - - Returns NaN when ``n < 2``, ``std`` is approximately zero, or any - input is non-finite. - """ - if n < 2: - return np.nan - if not np.isfinite(mean) or not np.isfinite(std): - return np.nan - if np.isclose(std, 0.0): - return np.nan - return mean * math.sqrt(n) / std + try: + recent_unrealized_pnl_history = ( + unrealized_pnl_history[-window_size:] + if window_size > 0 + else unrealized_pnl_history + ) + if len(recent_unrealized_pnl_history) < 2 or not all( + is_finite_number(pnl) for pnl in recent_unrealized_pnl_history + ): + return None + return bool( + recent_unrealized_pnl_history[-1] < recent_unrealized_pnl_history[0] + ) + except (TypeError, ValueError, OverflowError, IndexError): + return None @staticmethod @lru_cache(maxsize=_CACHE_MAXSIZE_LARGE) @@ -2136,61 +2051,6 @@ class QuickAdapterV3(IStrategy): return False return True - @staticmethod - def _effective_df(x: tuple[float, ...]) -> float: - """Effective degrees of freedom with Bartlett's autocorrelation correction. - - Computes ``df_eff = (n - 1) * (1 - rho1) / (1 + rho1)`` where ``rho1`` - is the lag-1 autocorrelation clamped to ``[-0.99, 0.99]``. Falls back - to ``n - 1`` when ``n < 4`` or pearsonr fails. Result is bounded - below by 1. - """ - n = len(x) - if n < 4: - return max(1.0, n - 1) - - x_arr = np.asarray(x, dtype=float) - x_centered = x_arr - np.nanmean(x_arr) - - try: - rho1, _ = pearsonr(x_centered[:-1], x_centered[1:]) - except (ValueError, TypeError) as exc: - logger.debug( - "[%s] pearsonr failed, using standard df: %r", "effective_df", exc - ) - return n - 1 - - if not np.isfinite(rho1): - return n - 1 - - # Clamp to avoid division by zero or negative n_eff - rho1 = np.clip(rho1, -0.99, 0.99) - correction_factor = (1 - rho1) / (1 + rho1) - - n_eff = n * correction_factor - df_eff = max(1.0, n_eff - 1) - - return df_eff - - @staticmethod - def _t_critical(q: float, df: float, default_t: float) -> float: - """Critical t-value from Student's t-distribution at quantile ``q``. - - Returns ``default_t`` on invalid inputs or scipy failure. - """ - if not (0.0 < q < 1.0): - return default_t - if df < 1: - return default_t - try: - t_crit = float(t.ppf(q, df)) - if not np.isfinite(t_crit): - return default_t - return t_crit - except (ValueError, TypeError, OverflowError) as exc: - logger.debug("[%s] t.ppf failed, using default_t: %r", "t_critical", exc) - return default_t - def custom_exit( self, pair: str, @@ -2314,7 +2174,7 @@ class QuickAdapterV3(IStrategy): if len(trade_unrealized_pnl_history) < self._pnl_momentum_window_size: # Warm-up: without a full momentum window a 30-minute decline is not # measurable yet; fail open (never block a profitable take-profit - # exit) rather than gate on a partial, low-power series. + # exit) rather than gate on a partial-horizon series. self.throttle_callback( pair=pair, current_time=current_time, @@ -2329,63 +2189,27 @@ class QuickAdapterV3(IStrategy): return QuickAdapterV3._take_profit_order_tag( trade.trade_direction, trade_exit_stage ) - ( - trade_recent_velocity_values, - trade_recent_velocity_mean, - trade_recent_velocity_std, - trade_recent_acceleration_values, - trade_recent_acceleration_mean, - trade_recent_acceleration_std, - ) = QuickAdapterV3.get_pnl_momentum( + trade_recent_pnl_declining = QuickAdapterV3.is_pnl_declining( trade_unrealized_pnl_history, self._pnl_momentum_window_size ) - q_decl = self._exit_thresholds_calibration.get("decline_quantile") - - n_trade_recent_velocity = len(trade_recent_velocity_values) - n_trade_recent_acceleration = len(trade_recent_acceleration_values) - - t_trade_recent_velocity = QuickAdapterV3._t_statistic( - trade_recent_velocity_mean, - trade_recent_velocity_std, - n_trade_recent_velocity, - ) - t_trade_recent_acceleration = QuickAdapterV3._t_statistic( - trade_recent_acceleration_mean, - trade_recent_acceleration_std, - n_trade_recent_acceleration, - ) - - df_eff_trade_recent_velocity = QuickAdapterV3._effective_df( - trade_recent_velocity_values - ) - df_eff_trade_recent_acceleration = QuickAdapterV3._effective_df( - trade_recent_acceleration_values - ) - - t_crit_trade_recent_velocity = QuickAdapterV3._t_critical( - q_decl, - df_eff_trade_recent_velocity, - QuickAdapterV3.default_exit_thresholds["t_decl_v"], - ) - t_crit_trade_recent_acceleration = QuickAdapterV3._t_critical( - q_decl, - df_eff_trade_recent_acceleration, - QuickAdapterV3.default_exit_thresholds["t_decl_a"], - ) - - # Declining if t_stat ≤ -t_crit (one-sided test for μ < 0) - decl_checks: list[bool] = [] - if np.isfinite(t_trade_recent_velocity): - decl_checks.append(t_trade_recent_velocity <= -t_crit_trade_recent_velocity) - if np.isfinite(t_trade_recent_acceleration): - decl_checks.append( - t_trade_recent_acceleration <= -t_crit_trade_recent_acceleration + if trade_recent_pnl_declining is None: + # A full but invalid history is still unmeasurable. Preserve the + # profitable-exit fail-open policy used for missing and warm-up + # history instead of trapping the trade on corrupted observations. + self.throttle_callback( + pair=pair, + current_time=current_time, + callback=lambda: logger.info( + f"[{pair}] Trade {trade.trade_direction} stage " + f"{trade_exit_stage} | PnL momentum gate unmeasurable " + "(invalid history); take-profit exit not gated " + "(fail-open)" + ), + ) + return QuickAdapterV3._take_profit_order_tag( + trade.trade_direction, trade_exit_stage ) - if len(decl_checks) == 0: - trade_recent_pnl_declining = True - else: - trade_recent_pnl_declining = all(decl_checks) trade_exit = trade_take_profit_exit and trade_recent_pnl_declining @@ -2397,7 +2221,9 @@ class QuickAdapterV3(IStrategy): f"[{pair}] Trade {trade.trade_direction} stage {trade_exit_stage} | " f"Take Profit: {format_number(trade_take_profit_price)}, Rate: {format_number(current_rate)} | " f"Declining: {trade_recent_pnl_declining} " - f"(tV:{format_number(t_trade_recent_velocity)}<=-t:{format_number(-t_crit_trade_recent_velocity)}, tA:{format_number(t_trade_recent_acceleration)}<=-t:{format_number(-t_crit_trade_recent_acceleration)})" + f"(window end: " + f"{format_number(trade_unrealized_pnl_history[-1])} < start: " + f"{format_number(trade_unrealized_pnl_history[-self._pnl_momentum_window_size])})" ), ) diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py index 0a5e6fb..2e88da4 100644 --- a/quickadapter/user_data/strategies/Utils.py +++ b/quickadapter/user_data/strategies/Utils.py @@ -1373,6 +1373,12 @@ _EXIT_PRICING_SPECS: Final[dict[str, _ParamSpec]] = { def get_exit_pricing_config(config: Any, logger: Logger) -> dict[str, str]: + if isinstance(config, dict) and "thresholds_calibration" in config: + logger.warning( + "exit_pricing.thresholds_calibration is obsolete and ignored: " + "the PnL momentum gate now uses the direction of mean per-candle " + "PnL velocity" + ) return _validate_params( as_config_section(config, "exit_pricing", logger), logger, @@ -1382,48 +1388,6 @@ def get_exit_pricing_config(config: Any, logger: Logger) -> dict[str, str]: ) -DEFAULTS_EXIT_THRESHOLDS_CALIBRATION: Final[dict[str, Any]] = { - "decline_quantile": 0.5, -} - -_EXIT_THRESHOLDS_CALIBRATION_SPECS: Final[dict[str, _ParamSpec]] = { - "decline_quantile": _ParamSpec( - _NumericValidator( - min_value=0, max_value=1, min_exclusive=True, max_exclusive=True - ), - output_type=float, - ), -} - - -def get_exit_thresholds_calibration_config( - config: Any, - logger: Logger, - overrides: dict[str, Any] | None = None, -) -> dict[str, float]: - # exit_pricing mapping warning owned by get_exit_pricing_config (avoid double-warn) - config = as_dict(config) - # validate the override so an invalid subclass default falls back to the canonical default - defaults = _validate_params( - overrides or {}, - logger, - "exit_pricing.thresholds_calibration", - _EXIT_THRESHOLDS_CALIBRATION_SPECS, - DEFAULTS_EXIT_THRESHOLDS_CALIBRATION, - ) - return _validate_params( - as_config_section( - config.get("thresholds_calibration"), - "exit_pricing.thresholds_calibration", - logger, - ), - logger, - "exit_pricing.thresholds_calibration", - _EXIT_THRESHOLDS_CALIBRATION_SPECS, - defaults, - ) - - DEFAULTS_CUSTOM_PROTECTIONS: Final[dict[str, Any]] = { "trade_duration_candles": 72, "lookback_period_fraction": 0.5, -- 2.53.0