From f646ebe8ecfd0dfab400a71ff5d3c4eb1e75d034 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Tue, 7 Oct 2025 00:27:55 +0200 Subject: [PATCH] perf(reforcexy): untangle exit plateau logic in reward code MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- ReforceXY/reward_space_analysis/README.md | 53 ++- .../reward_space_analysis.py | 227 ++++++------- .../test_reward_space_analysis.py | 312 +++++++++++------- ReforceXY/user_data/freqaimodels/ReforceXY.py | 62 ++-- 4 files changed, 390 insertions(+), 264 deletions(-) diff --git a/ReforceXY/reward_space_analysis/README.md b/ReforceXY/reward_space_analysis/README.md index ce5a3be..fc6175a 100644 --- a/ReforceXY/reward_space_analysis/README.md +++ b/ReforceXY/reward_space_analysis/README.md @@ -19,7 +19,7 @@ This tool helps you understand and validate how the ReforceXY reinforcement lear --- -**New to this tool?** Start with [Common Use Cases](#-common-use-cases) then explore [CLI Parameters](#️-cli-parameters-reference). For runtime guardrails see [Validation Layers](#-validation-layers-runtime). The exit factor attenuation logic is now centralized through a single internal helper ensuring analytical parity with the live environment (parity date: 2025‑10‑06). +**New to this tool?** Start with [Common Use Cases](#-common-use-cases) then explore [CLI Parameters](#️-cli-parameters-reference). For runtime guardrails see [Validation Layers](#-validation-layers-runtime). The exit factor attenuation logic is now centralized through a single internal helper ensuring analytical parity with the live environment. --- @@ -226,15 +226,40 @@ _Holding penalty configuration:_ - `holding_penalty_scale` (default: 0.5) - Scale of holding penalty - `holding_penalty_power` (default: 1.0) - Power applied to holding penalty scaling -_Exit factor configuration:_ +_Exit attenuation configuration:_ -- `exit_factor_mode` (default: piecewise) - Time attenuation mode for exit factor (legacy|sqrt|linear|power|piecewise|half_life) -- `exit_linear_slope` (default: 1.0) - Slope for linear exit attenuation -- `exit_piecewise_grace` (default: 1.0) - Grace region boundary (duration ratio); values >1.0 extend no-attenuation period -- `exit_piecewise_slope` (default: 1.0) - Slope after grace for piecewise mode (0 ⇒ flat beyond grace) -- `exit_power_tau` (default: 0.5) - Tau in (0,1] mapped to alpha = -ln(tau)/ln(2) -- `exit_half_life` (default: 0.5) - Half-life for exponential decay exit mode (factor *= 2^(-r/half_life)) -- `exit_factor_threshold` (default: 10000.0) - Warning-only threshold; no capping occurs (emits RuntimeWarning if |factor| exceeds) +- `exit_attenuation_mode` (default: linear) - Selects attenuation kernel (see table below: legacy|sqrt|linear|power|half_life). +- `exit_plateau` (default: true) - Enables plateau (no attenuation until `exit_plateau_grace`). +- `exit_plateau_grace` (default: 1.0) - Duration ratio boundary of full‑strength region (may exceed 1.0). +- `exit_linear_slope` (default: 1.0) - Slope parameter used only when mode = linear. +- `exit_power_tau` (default: 0.5) - Tau ∈ (0,1]; internally mapped to alpha (see kernel table). +- `exit_half_life` (default: 0.5) - Half‑life parameter for the half_life kernel. +- `exit_factor_threshold` (default: 10000.0) - Warning-only soft threshold (emits RuntimeWarning; no capping). + +Attenuation kernels: + +Let r be the raw duration ratio and grace = `exit_plateau_grace`. + +``` +effective_r = 0 if exit_plateau and r <= grace +effective_r = r - grace if exit_plateau and r > grace +effective_r = r if not exit_plateau +``` + +| Mode | Multiplier (applied to base_factor * pnl * pnl_factor * efficiency) | Monotonic ↓ | Notes | +|------|---------------------------------------------------------------------|-------------|-------| +| legacy | step: ×1.5 if r* ≤ 1 else ×0.5 | No | Historical discontinuity retained (not smoothed) | +| sqrt | 1 / sqrt(1 + r*) | Yes | Sub-linear decay | +| linear | 1 / (1 + slope * r*) | Yes | slope = `exit_linear_slope` (≥0) | +| power | (1 + r*)^(-alpha) | Yes | alpha = -ln(tau)/ln(2), tau = `exit_power_tau` ∈ (0,1]; tau=1 ⇒ alpha=0 (flat) | +| half_life | 2^(- r* / hl) | Yes | hl = `exit_half_life`; r* = hl ⇒ factor × 0.5 | + +Where r* = `effective_r` above. + +Notes: +- Plateau guarantees continuity at the boundary r = grace for all monotonic kernels; only `legacy` may jump. +- A single implementation in code (`_get_exit_factor`) mirrors this table; this README is the canonical human-readable mapping. +- Continuity tests assert small‑epsilon bounded attenuation onset (excluding `legacy`). _Efficiency configuration:_ @@ -400,10 +425,10 @@ Implementation details: Test different reward parameter configurations to understand their impact: ```shell -# Test power-based exit factor with custom tau +# Test power-based exit attenuation with custom tau python reward_space_analysis.py \ --num_samples 25000 \ - --params exit_factor_mode=power exit_power_tau=0.5 efficiency_weight=0.8 \ + --params exit_attenuation_mode=power exit_power_tau=0.5 efficiency_weight=0.8 \ --output custom_test # Test aggressive holding penalties @@ -452,7 +477,7 @@ done python test_reward_space_analysis.py ``` -The suite currently contains 53 tests (current state; this number evolves as new invariants and attenuation modes are added). Always run the full suite after modifying reward logic or attenuation parameters. +The suite currently contains 54 tests (current state; this number evolves as new invariants and attenuation modes are added). Always run the full suite after modifying reward logic or attenuation parameters. ### Test Categories @@ -468,6 +493,7 @@ The suite currently contains 53 tests (current state; this number evolves as new | Private Functions (via public API) | TestPrivateFunctions | Idle / holding / invalid penalties, exit scenarios | | Robustness | TestRewardRobustness | Monotonic attenuation (where applicable), decomposition integrity, boundary regimes | | Parameter Validation | TestParameterValidation | Bounds clamping, warning threshold, penalty power scaling | +| Continuity | TestContinuityPlateau | Plateau boundary continuity & small‑epsilon attenuation scaling | ### Test Architecture @@ -673,8 +699,7 @@ Before simulation (early in `main()`), `validate_reward_parameters` enforces num | `holding_penalty_scale` | 0.0 | — | Scale ≥ 0 | | `holding_penalty_power` | 0.0 | — | Power exponent ≥ 0 | | `exit_linear_slope` | 0.0 | — | Slope ≥ 0 | -| `exit_piecewise_grace` | 0.0 | — | Grace boundary expressed in duration ratio units (can exceed 1.0 to extend full-strength region) | -| `exit_piecewise_slope` | 0.0 | — | Slope ≥ 0 | +| `exit_plateau_grace` | 0.0 | — | Plateau grace boundary (full strength until this duration ratio) | | `exit_power_tau` | 1e-6 | 1.0 | Mapped to alpha = -ln(tau) | | `exit_half_life` | 1e-6 | — | Half-life in duration ratio units | | `efficiency_weight` | 0.0 | 2.0 | Blend weight | diff --git a/ReforceXY/reward_space_analysis/reward_space_analysis.py b/ReforceXY/reward_space_analysis/reward_space_analysis.py index 261a397..17fdc55 100644 --- a/ReforceXY/reward_space_analysis/reward_space_analysis.py +++ b/ReforceXY/reward_space_analysis/reward_space_analysis.py @@ -77,36 +77,33 @@ def _to_bool(value: Any) -> bool: def _get_param_float(params: Dict[str, float | str], key: str, default: float) -> float: """Extract float parameter with type safety and default fallback.""" value = params.get(key, default) + # None -> default + if value is None: + return default + # Bool: treat explicitly (avoid surprising True->1.0 unless intentional) + if isinstance(value, bool): + return float(int(value)) + # Numeric if isinstance(value, (int, float)): - return float(value) - if isinstance(value, str): try: - return float(value) + fval = float(value) except (ValueError, TypeError): return default + return fval if math.isfinite(fval) else default + # String parsing + if isinstance(value, str): + stripped = value.strip() + if stripped == "": + return default + try: + fval = float(stripped) + except ValueError: + return default + return fval if math.isfinite(fval) else default + # Unsupported type return default -def _piecewise_duration_divisor( - duration_ratio: float, params: Dict[str, float | str] -) -> float: - """Compute divisor for piecewise attenuation (single source of truth). - - Ensures consistent fallback behaviour across the primary code path and - exception fallback in ``_get_exit_factor`` without duplicating logic. - """ - exit_piecewise_grace = _get_param_float(params, "exit_piecewise_grace", 1.0) - # Only enforce a lower bound; values >1.0 extend the grace region beyond max duration ratio. - if exit_piecewise_grace < 0.0: - exit_piecewise_grace = 0.0 - exit_piecewise_slope = _get_param_float(params, "exit_piecewise_slope", 1.0) - if exit_piecewise_slope < 0.0: # sanitize slope sign - exit_piecewise_slope = 1.0 - if duration_ratio <= exit_piecewise_grace: - return 1.0 - return 1.0 + exit_piecewise_slope * (duration_ratio - exit_piecewise_grace) - - def _compute_duration_ratio(trade_duration: int, max_trade_duration: int) -> float: """Compute duration ratio with safe division.""" return trade_duration / max(1, max_trade_duration) @@ -135,11 +132,11 @@ DEFAULT_MODEL_REWARD_PARAMETERS: Dict[str, float | str] = { # Holding keys (env defaults) "holding_penalty_scale": 0.5, "holding_penalty_power": 1.0, - # Exit factor configuration (env defaults) - "exit_factor_mode": "piecewise", + # Exit attenuation configuration (env default) + "exit_attenuation_mode": "linear", + "exit_plateau": True, + "exit_plateau_grace": 1.0, "exit_linear_slope": 1.0, - "exit_piecewise_grace": 1.0, - "exit_piecewise_slope": 1.0, "exit_power_tau": 0.5, "exit_half_life": 0.5, # Efficiency keys (env defaults) @@ -161,11 +158,10 @@ DEFAULT_MODEL_REWARD_PARAMETERS_HELP: Dict[str, str] = { "max_idle_duration_candles": "Maximum idle duration candles before full idle penalty scaling; 0 = use 2 * max_trade_duration_candles.", "holding_penalty_scale": "Scale of holding penalty.", "holding_penalty_power": "Power applied to holding penalty scaling.", - "exit_factor_mode": "Time attenuation mode for exit factor.", + "exit_attenuation_mode": "Attenuation kernel (legacy|sqrt|linear|power|half_life).", + "exit_plateau": "Enable plateau. If true, full strength until grace boundary then apply attenuation.", + "exit_plateau_grace": "Grace boundary duration ratio for plateau (full strength until this boundary).", "exit_linear_slope": "Slope for linear exit attenuation.", - # exit_piecewise_grace: duration ratio boundary; >1 extends full-strength region - "exit_piecewise_grace": "Grace boundary (duration ratio; >1 extends no-attenuation region).", - "exit_piecewise_slope": "Slope after grace for piecewise mode (0 = flat).", "exit_power_tau": "Tau in (0,1] to derive alpha for power mode.", "exit_half_life": "Half-life for exponential decay exit mode.", "efficiency_weight": "Weight for efficiency factor in exit reward.", @@ -191,8 +187,7 @@ _PARAMETER_BOUNDS: Dict[str, Dict[str, float]] = { "holding_penalty_scale": {"min": 0.0}, "holding_penalty_power": {"min": 0.0}, "exit_linear_slope": {"min": 0.0}, - "exit_piecewise_grace": {"min": 0.0}, - "exit_piecewise_slope": {"min": 0.0}, + "exit_plateau_grace": {"min": 0.0}, "exit_power_tau": {"min": 1e-6, "max": 1.0}, # open (0,1] approximated "exit_half_life": {"min": 1e-6}, "efficiency_weight": {"min": 0.0, "max": 2.0}, @@ -250,7 +245,7 @@ def add_tunable_cli_args(parser: argparse.ArgumentParser) -> None: Rules: - Use the same underscored names as option flags (e.g., --idle_penalty_scale). - Defaults are None so only user-provided values override params. - - For exit_factor_mode, enforce allowed choices and lowercase conversion. + - For exit_attenuation_mode, enforce allowed choices and lowercase conversion. - Skip keys already managed as top-level options (e.g., base_factor) to avoid duplicates. """ skip_keys = {"base_factor"} # already defined as top-level @@ -260,11 +255,19 @@ def add_tunable_cli_args(parser: argparse.ArgumentParser) -> None: help_text = DEFAULT_MODEL_REWARD_PARAMETERS_HELP.get( key, f"Override tunable '{key}'." ) - if key == "exit_factor_mode": + if key == "exit_attenuation_mode": parser.add_argument( f"--{key}", type=str.lower, - choices=["legacy", "sqrt", "linear", "power", "piecewise", "half_life"], + choices=["legacy", "sqrt", "linear", "power", "half_life"], + default=None, + help=help_text, + ) + elif key == "exit_plateau": + parser.add_argument( + f"--{key}", + type=int, + choices=[0, 1], default=None, help=help_text, ) @@ -307,41 +310,16 @@ def _get_exit_factor( duration_ratio: float, params: Dict[str, float | str], ) -> float: - """Compute the complete exit factor (time attenuation + PnL scaling). - - Synchronization - --------------- - Mirrors the environment implementation `ReforceXY._get_exit_factor` (parity date: 2025-10-06). - Any upstream change MUST be ported here to keep analytical results consistent with live behavior. - - Processing steps - ---------------- - 1. Clamp negative ``duration_ratio`` to 0. - 2. Apply time-based attenuation according to ``exit_factor_mode``. - 3. Multiply by ``pnl_factor`` (already includes profit amplification & efficiency component). - 4. Enforce invariants (finite; non-negative when pnl>=0) and optionally emit a warning if - ``|factor| > exit_factor_threshold`` (warning-only, no capping). - - Modes (attenuation formulas) - ---------------------------- - legacy : factor *= 1.5 if r <= 1 else *= 0.5 (step change) - sqrt : factor /= sqrt(1 + r) - linear : factor /= (1 + slope * r) - power : alpha = -log(tau)/log(2) with tau in (0,1]; factor /= (1 + r)^alpha (fallback alpha=1) - piecewise : grace g in [0,1]; if r <= g then divisor=1 else divisor = 1 + slope * (r - g) - half_life : factor *= 2^(-r / half_life) - - Fallback: Unknown modes default to piecewise. - - Invariants - ---------- - - Factor set to 0 if non-finite. - - Factor clamped to >=0 when pnl >= 0. - - Threshold exceedance triggers RuntimeWarning only. + """Compute exit factor = time attenuation kernel (with optional plateau) * pnl_factor. - Returns - ------- - float : attenuated & pnl-scaled factor (reward exit component = pnl * factor). + Parity: mirrors `ReforceXY._get_exit_factor`. + + Steps: + 1. Sanitize inputs (finite, non-negative duration_ratio). + 2. Derive effective duration ratio: if plateau enabled and r <= grace ⇒ 0 else r' = r - grace. + 3. Apply kernel (legacy|sqrt|linear|power|half_life). Unknown ⇒ linear. + 4. Multiply by externally supplied pnl_factor (includes profit amplification & efficiency). + 5. Enforce invariants (finite, non-negative when pnl ≥ 0, warn if |factor| exceeds threshold). """ # Basic finiteness checks if ( @@ -355,50 +333,75 @@ def _get_exit_factor( if duration_ratio < 0.0: duration_ratio = 0.0 - exit_factor_mode = str(params.get("exit_factor_mode", "piecewise")).lower() + exit_attenuation_mode = str(params.get("exit_attenuation_mode", "linear")).lower() + exit_plateau = _to_bool(params.get("exit_plateau", True)) + + exit_plateau_grace = _get_param_float(params, "exit_plateau_grace", 1.0) + if exit_plateau_grace < 0.0: + exit_plateau_grace = 1.0 + exit_linear_slope = _get_param_float(params, "exit_linear_slope", 1.0) + if exit_linear_slope < 0.0: + exit_linear_slope = 1.0 + + def _legacy_kernel(f: float, dr: float) -> float: + return f * (1.5 if dr <= 1.0 else 0.5) + + def _sqrt_kernel(f: float, dr: float) -> float: + return f / math.sqrt(1.0 + dr) + + def _linear_kernel(f: float, dr: float) -> float: + return f / (1.0 + exit_linear_slope * dr) + + def _power_kernel(f: float, dr: float) -> float: + alpha = params.get("exit_power_alpha") + if isinstance(alpha, (int, float)) and alpha < 0.0: + alpha = None + if alpha is None: + tau = params.get("exit_power_tau") + if isinstance(tau, (int, float)): + tau = float(tau) + if 0.0 < tau <= 1.0: + alpha = -math.log(tau) / _LOG_2 + if not isinstance(alpha, (int, float)): + alpha = 1.0 + else: + alpha = float(alpha) + return f / math.pow(1.0 + dr, alpha) + + def _half_life_kernel(f: float, dr: float) -> float: + hl = _get_param_float(params, "exit_half_life", 0.5) + if hl <= 0.0: + hl = 0.5 + return f * math.pow(2.0, -dr / hl) + + kernels = { + "legacy": _legacy_kernel, + "sqrt": _sqrt_kernel, + "linear": _linear_kernel, + "power": _power_kernel, + "half_life": _half_life_kernel, + } + if exit_plateau: + if duration_ratio <= exit_plateau_grace: + effective_dr = 0.0 + else: + effective_dr = duration_ratio - exit_plateau_grace + else: + effective_dr = duration_ratio + + kernel = kernels.get(exit_attenuation_mode, None) + if kernel is None: + kernel = _linear_kernel try: - if exit_factor_mode == "legacy": - factor *= 1.5 if duration_ratio <= 1.0 else 0.5 - elif exit_factor_mode == "sqrt": - factor /= math.sqrt(1.0 + duration_ratio) - elif exit_factor_mode == "linear": - slope = _get_param_float(params, "exit_linear_slope", 1.0) - if slope < 0.0: - slope = 1.0 - factor /= 1.0 + slope * duration_ratio - elif exit_factor_mode == "power": - exit_power_alpha = params.get("exit_power_alpha") - if isinstance(exit_power_alpha, (int, float)) and exit_power_alpha < 0.0: - exit_power_alpha = None - if exit_power_alpha is None: - exit_power_tau = params.get("exit_power_tau") - if isinstance(exit_power_tau, (int, float)): - exit_power_tau = float(exit_power_tau) - if 0.0 < exit_power_tau <= 1.0: - exit_power_alpha = -math.log(exit_power_tau) / _LOG_2 - if not isinstance(exit_power_alpha, (int, float)): - exit_power_alpha = 1.0 - else: - exit_power_alpha = float(exit_power_alpha) - factor /= math.pow(1.0 + duration_ratio, exit_power_alpha) - elif exit_factor_mode == "piecewise" or exit_factor_mode not in { - "legacy", - "sqrt", - "linear", - "power", - "half_life", - }: - # Default behaviour - factor /= _piecewise_duration_divisor(duration_ratio, params) - elif exit_factor_mode == "half_life": - exit_half_life = _get_param_float(params, "exit_half_life", 0.5) - if exit_half_life <= 0.0: - exit_half_life = 0.5 - factor *= math.pow(2.0, -duration_ratio / exit_half_life) - except Exception: - # Safe fallback to piecewise logic if any unexpected error arises (centralized) - factor /= _piecewise_duration_divisor(duration_ratio, params) + factor = kernel(factor, effective_dr) + except Exception as e: + warnings.warn( + f"exit_attenuation_mode '{exit_attenuation_mode}' failed ({e!r}); fallback linear (effective_dr={effective_dr:.5f})", + RuntimeWarning, + stacklevel=2, + ) + factor = _linear_kernel(factor, effective_dr) # Apply pnl_factor after time attenuation factor *= pnl_factor diff --git a/ReforceXY/reward_space_analysis/test_reward_space_analysis.py b/ReforceXY/reward_space_analysis/test_reward_space_analysis.py index 86a98d2..88a2117 100644 --- a/ReforceXY/reward_space_analysis/test_reward_space_analysis.py +++ b/ReforceXY/reward_space_analysis/test_reward_space_analysis.py @@ -669,7 +669,7 @@ class TestRewardAlignment(RewardSpaceTestBase): for mode in modes_to_test: test_params = self.DEFAULT_PARAMS.copy() - test_params["exit_factor_mode"] = mode + test_params["exit_attenuation_mode"] = mode factor = rsa.compute_exit_factor( base_factor=1.0, pnl=0.02, @@ -684,7 +684,7 @@ class TestRewardAlignment(RewardSpaceTestBase): self.assertGreater(factor, 0, f"Exit factor for {mode} should be positive") def test_negative_slope_sanitization(self): - """Negative slopes for linear/piecewise must be sanitized to positive default (1.0).""" + """Negative slopes for linear must be sanitized to positive default (1.0).""" from reward_space_analysis import compute_exit_factor base_factor = 100.0 @@ -695,9 +695,13 @@ class TestRewardAlignment(RewardSpaceTestBase): # Linear mode: slope -5.0 should behave like slope=1.0 (sanitized) params_lin_neg = self.DEFAULT_PARAMS.copy() - params_lin_neg.update({"exit_factor_mode": "linear", "exit_linear_slope": -5.0}) + params_lin_neg.update( + {"exit_attenuation_mode": "linear", "exit_linear_slope": -5.0} + ) params_lin_pos = self.DEFAULT_PARAMS.copy() - params_lin_pos.update({"exit_factor_mode": "linear", "exit_linear_slope": 1.0}) + params_lin_pos.update( + {"exit_attenuation_mode": "linear", "exit_linear_slope": 1.0} + ) val_lin_neg = compute_exit_factor( base_factor, pnl, pnl_factor, duration_ratio_linear, params_lin_neg ) @@ -711,34 +715,36 @@ class TestRewardAlignment(RewardSpaceTestBase): msg="Negative linear slope not sanitized to default behavior", ) - # Piecewise mode: negative slope sanitized to 1.0 - params_pw_neg = self.DEFAULT_PARAMS.copy() - params_pw_neg.update( + # Plateau+linear: negative slope sanitized similarly + params_pl_neg = self.DEFAULT_PARAMS.copy() + params_pl_neg.update( { - "exit_factor_mode": "piecewise", - "exit_piecewise_grace": 1.0, - "exit_piecewise_slope": -3.0, + "exit_attenuation_mode": "linear", + "exit_plateau": True, + "exit_plateau_grace": 1.0, + "exit_linear_slope": -3.0, } ) - params_pw_pos = self.DEFAULT_PARAMS.copy() - params_pw_pos.update( + params_pl_pos = self.DEFAULT_PARAMS.copy() + params_pl_pos.update( { - "exit_factor_mode": "piecewise", - "exit_piecewise_grace": 1.0, - "exit_piecewise_slope": 1.0, + "exit_attenuation_mode": "linear", + "exit_plateau": True, + "exit_plateau_grace": 1.0, + "exit_linear_slope": 1.0, } ) - val_pw_neg = compute_exit_factor( - base_factor, pnl, pnl_factor, duration_ratio_piecewise, params_pw_neg + val_pl_neg = compute_exit_factor( + base_factor, pnl, pnl_factor, duration_ratio_piecewise, params_pl_neg ) - val_pw_pos = compute_exit_factor( - base_factor, pnl, pnl_factor, duration_ratio_piecewise, params_pw_pos + val_pl_pos = compute_exit_factor( + base_factor, pnl, pnl_factor, duration_ratio_piecewise, params_pl_pos ) self.assertAlmostEqualFloat( - val_pw_neg, - val_pw_pos, + val_pl_neg, + val_pl_pos, tolerance=1e-9, - msg="Negative piecewise slope not sanitized to default behavior", + msg="Negative plateau+linear slope not sanitized to default behavior", ) def test_idle_penalty_zero_when_profit_target_zero(self): @@ -781,7 +787,13 @@ class TestRewardAlignment(RewardSpaceTestBase): pnl = 0.03 pnl_factor = 1.0 # isolate attenuation params = self.DEFAULT_PARAMS.copy() - params.update({"exit_factor_mode": "power", "exit_power_tau": tau}) + params.update( + { + "exit_attenuation_mode": "power", + "exit_power_tau": tau, + "exit_plateau": False, + } + ) observed = compute_exit_factor(base_factor, pnl, pnl_factor, r, params) expected = base_factor / (1.0 + r) ** alpha self.assertAlmostEqualFloat( @@ -1089,8 +1101,9 @@ class TestStatisticalValidation(RewardSpaceTestBase): duration_ratio = 50 / 100 # 0.5 # Test power mode with known tau - params["exit_factor_mode"] = "power" + params["exit_attenuation_mode"] = "power" params["exit_power_tau"] = 0.5 + params["exit_plateau"] = False reward_power = calculate_reward( context, params, 100.0, 0.03, 1.0, short_allowed=True, action_masking=True @@ -1105,7 +1118,7 @@ class TestStatisticalValidation(RewardSpaceTestBase): ) # Test half_life mode - params["exit_factor_mode"] = "half_life" + params["exit_attenuation_mode"] = "half_life" params["exit_half_life"] = 0.5 reward_half_life = calculate_reward( @@ -1117,7 +1130,7 @@ class TestStatisticalValidation(RewardSpaceTestBase): self.assertAlmostEqual(expected_half_life_factor, 0.5, places=6) # Test that different modes produce different results (mathematical diversity) - params["exit_factor_mode"] = "linear" + params["exit_attenuation_mode"] = "linear" params["exit_linear_slope"] = 1.0 reward_linear = calculate_reward( @@ -1388,12 +1401,12 @@ class TestBoundaryConditions(RewardSpaceTestBase): def test_different_exit_factor_modes(self): """Test different exit factor calculation modes.""" - modes = ["legacy", "sqrt", "linear", "power", "piecewise", "half_life"] + modes = ["legacy", "sqrt", "linear", "power", "half_life"] for mode in modes: with self.subTest(mode=mode): test_params = self.DEFAULT_PARAMS.copy() - test_params["exit_factor_mode"] = mode + test_params["exit_attenuation_mode"] = mode context = RewardContext( pnl=0.02, @@ -1426,55 +1439,6 @@ class TestBoundaryConditions(RewardSpaceTestBase): f"Total reward should be finite for mode {mode}", ) - def test_unknown_exit_factor_mode_fallback_piecewise(self): - """Unknown exit_factor_mode must fallback to piecewise attenuation.""" - base_factor = 150.0 - context = RewardContext( - pnl=0.05, - trade_duration=160, - idle_duration=0, - max_trade_duration=100, - max_unrealized_profit=0.07, - min_unrealized_profit=0.0, - position=Positions.Long, - action=Actions.Long_exit, - force_action=None, - ) - params_piecewise = self.DEFAULT_PARAMS.copy() - params_piecewise["exit_factor_mode"] = "piecewise" - params_unknown = self.DEFAULT_PARAMS.copy() - params_unknown["exit_factor_mode"] = "unrecognized_mode_xyz" - - reward_piecewise = calculate_reward( - context, - params_piecewise, - base_factor=base_factor, - profit_target=0.03, - risk_reward_ratio=1.0, - short_allowed=True, - action_masking=True, - ) - reward_unknown = calculate_reward( - context, - params_unknown, - base_factor=base_factor, - profit_target=0.03, - risk_reward_ratio=1.0, - short_allowed=True, - action_masking=True, - ) - self.assertGreater( - reward_piecewise.exit_component, - 0.0, - "Piecewise exit reward should be positive with positive pnl", - ) - self.assertAlmostEqualFloat( - reward_piecewise.exit_component, - reward_unknown.exit_component, - tolerance=1e-9, - msg="Fallback for unknown mode should produce identical result to piecewise", - ) - class TestHelperFunctions(RewardSpaceTestBase): """Test utility and helper functions.""" @@ -2154,23 +2118,26 @@ class TestRewardRobustness(RewardSpaceTestBase): def test_exit_factor_monotonic_attenuation(self): """For attenuation modes: factor should be non-increasing w.r.t duration_ratio. - Modes covered: sqrt, linear, power, half_life, piecewise (after grace). - Legacy is excluded (non-monotonic by design: step change). Piecewise includes flat grace then monotonic. + Modes covered: sqrt, linear, power, half_life, plateau+linear (after grace). + Legacy is excluded (non-monotonic by design). Plateau+linear includes flat grace then monotonic. """ from reward_space_analysis import compute_exit_factor - modes = ["sqrt", "linear", "power", "half_life", "piecewise"] + modes = ["sqrt", "linear", "power", "half_life", "plateau_linear"] base_factor = 100.0 pnl = 0.05 pnl_factor = 1.0 for mode in modes: params = self.DEFAULT_PARAMS.copy() - params["exit_factor_mode"] = mode + if mode in ("sqrt", "linear", "power", "half_life"): + params["exit_attenuation_mode"] = mode if mode == "linear": params["exit_linear_slope"] = 1.2 - if mode == "piecewise": - params["exit_piecewise_grace"] = 0.2 - params["exit_piecewise_slope"] = 1.0 + if mode == "plateau_linear": + params["exit_attenuation_mode"] = "linear" + params["exit_plateau"] = True + params["exit_plateau_grace"] = 0.2 + params["exit_linear_slope"] = 1.0 if mode == "power": params["exit_power_tau"] = 0.5 if mode == "half_life": @@ -2181,9 +2148,9 @@ class TestRewardRobustness(RewardSpaceTestBase): compute_exit_factor(base_factor, pnl, pnl_factor, r, params) for r in ratios ] - # Piecewise: ignore initial flat region when checking monotonic decrease - if mode == "piecewise": - grace = float(params["exit_piecewise_grace"]) # type: ignore[index] + # Plateau+linear: ignore initial flat region when checking monotonic decrease + if mode == "plateau_linear": + grace = float(params["exit_plateau_grace"]) # type: ignore[index] filtered = [(r, v) for r, v in zip(ratios, values) if r >= grace - 1e-9] values_to_check = [v for _, v in filtered] else: @@ -2196,7 +2163,7 @@ class TestRewardRobustness(RewardSpaceTestBase): ) def test_exit_factor_boundary_parameters(self): - """Test parameter edge cases: tau extremes, grace edges, slope zero.""" + """Test parameter edge cases: tau extremes, plateau grace edges, slope zero.""" from reward_space_analysis import compute_exit_factor base_factor = 50.0 @@ -2204,9 +2171,9 @@ class TestRewardRobustness(RewardSpaceTestBase): pnl_factor = 1.0 # Tau near 1 (minimal attenuation) vs tau near 0 (strong attenuation) params_hi = self.DEFAULT_PARAMS.copy() - params_hi.update({"exit_factor_mode": "power", "exit_power_tau": 0.999999}) + params_hi.update({"exit_attenuation_mode": "power", "exit_power_tau": 0.999999}) params_lo = self.DEFAULT_PARAMS.copy() - params_lo.update({"exit_factor_mode": "power", "exit_power_tau": 1e-6}) + params_lo.update({"exit_attenuation_mode": "power", "exit_power_tau": 1e-6}) r = 1.5 hi_val = compute_exit_factor(base_factor, pnl, pnl_factor, r, params_hi) lo_val = compute_exit_factor(base_factor, pnl, pnl_factor, r, params_lo) @@ -2215,21 +2182,23 @@ class TestRewardRobustness(RewardSpaceTestBase): lo_val, "Power mode: higher tau (≈1) should attenuate less than tiny tau", ) - # Piecewise grace 0 vs 1 + # Plateau grace 0 vs 1 params_g0 = self.DEFAULT_PARAMS.copy() params_g0.update( { - "exit_factor_mode": "piecewise", - "exit_piecewise_grace": 0.0, - "exit_piecewise_slope": 1.0, + "exit_attenuation_mode": "linear", + "exit_plateau": True, + "exit_plateau_grace": 0.0, + "exit_linear_slope": 1.0, } ) params_g1 = self.DEFAULT_PARAMS.copy() params_g1.update( { - "exit_factor_mode": "piecewise", - "exit_piecewise_grace": 1.0, - "exit_piecewise_slope": 1.0, + "exit_attenuation_mode": "linear", + "exit_plateau": True, + "exit_plateau_grace": 1.0, + "exit_linear_slope": 1.0, } ) val_g0 = compute_exit_factor(base_factor, pnl, pnl_factor, 0.5, params_g0) @@ -2242,9 +2211,21 @@ class TestRewardRobustness(RewardSpaceTestBase): ) # Linear slope zero vs positive params_lin0 = self.DEFAULT_PARAMS.copy() - params_lin0.update({"exit_factor_mode": "linear", "exit_linear_slope": 0.0}) + params_lin0.update( + { + "exit_attenuation_mode": "linear", + "exit_linear_slope": 0.0, + "exit_plateau": False, + } + ) params_lin1 = self.DEFAULT_PARAMS.copy() - params_lin1.update({"exit_factor_mode": "linear", "exit_linear_slope": 2.0}) + params_lin1.update( + { + "exit_attenuation_mode": "linear", + "exit_linear_slope": 2.0, + "exit_plateau": False, + } + ) val_lin0 = compute_exit_factor(base_factor, pnl, pnl_factor, 1.0, params_lin0) val_lin1 = compute_exit_factor(base_factor, pnl, pnl_factor, 1.0, params_lin1) self.assertGreater( @@ -2253,16 +2234,17 @@ class TestRewardRobustness(RewardSpaceTestBase): "Linear slope=0 should yield no attenuation vs slope>0", ) - def test_piecewise_slope_zero_constant_after_grace(self): - """Piecewise slope=0 should yield flat factor after grace boundary.""" + def test_plateau_linear_slope_zero_constant_after_grace(self): + """Plateau+linear slope=0 should yield flat factor after grace boundary (no attenuation).""" from reward_space_analysis import compute_exit_factor params = self.DEFAULT_PARAMS.copy() params.update( { - "exit_factor_mode": "piecewise", - "exit_piecewise_grace": 0.3, - "exit_piecewise_slope": 0.0, + "exit_attenuation_mode": "linear", + "exit_plateau": True, + "exit_plateau_grace": 0.3, + "exit_linear_slope": 0.0, } ) base_factor = 100.0 @@ -2279,19 +2261,20 @@ class TestRewardRobustness(RewardSpaceTestBase): v, first, tolerance=1e-9, - msg=f"Piecewise slope=0 factor drift at ratio set {ratios} => {values}", + msg=f"Plateau+linear slope=0 factor drift at ratio set {ratios} => {values}", ) - def test_piecewise_grace_extends_beyond_one(self): - """Grace >1.0 should keep divisor=1 (no attenuation) past duration_ratio=1.""" + def test_plateau_grace_extends_beyond_one(self): + """Plateau grace >1.0 should keep full strength (no attenuation) past duration_ratio=1.""" from reward_space_analysis import compute_exit_factor params = self.DEFAULT_PARAMS.copy() params.update( { - "exit_factor_mode": "piecewise", - "exit_piecewise_grace": 1.5, # extend grace beyond max duration ratio 1.0 - "exit_piecewise_slope": 2.0, + "exit_attenuation_mode": "linear", + "exit_plateau": True, + "exit_plateau_grace": 1.5, # extend grace beyond max duration ratio 1.0 + "exit_linear_slope": 2.0, } ) base_factor = 80.0 @@ -2319,7 +2302,8 @@ class TestRewardRobustness(RewardSpaceTestBase): from reward_space_analysis import compute_exit_factor params = self.DEFAULT_PARAMS.copy() - params["exit_factor_mode"] = "legacy" + params["exit_attenuation_mode"] = "legacy" + params["exit_plateau"] = False base_factor = 100.0 pnl = 0.02 pnl_factor = 1.0 @@ -2530,7 +2514,8 @@ class TestParameterValidation(RewardSpaceTestBase): from reward_space_analysis import compute_exit_factor params = self.DEFAULT_PARAMS.copy() - params["exit_factor_mode"] = "sqrt" + params["exit_attenuation_mode"] = "sqrt" + params["exit_plateau"] = False f1 = compute_exit_factor(100.0, 0.02, 1.0, 0.0, params) f2 = compute_exit_factor(100.0, 0.02, 1.0, 1.0, params) self.assertGreater( @@ -2538,6 +2523,109 @@ class TestParameterValidation(RewardSpaceTestBase): ) +class TestContinuityPlateau(RewardSpaceTestBase): + """Continuity tests for plateau-enabled exit attenuation (excluding legacy).""" + + def test_plateau_continuity_at_grace_boundary(self): + import math + + from reward_space_analysis import compute_exit_factor + + modes = ["sqrt", "linear", "power", "half_life"] + grace = 0.8 + eps = 1e-4 + base_factor = 100.0 + pnl = 0.01 + pnl_factor = 1.0 + tau = 0.5 # for power + half_life = 0.5 + slope = 1.3 + + for mode in modes: + with self.subTest(mode=mode): + params = self.DEFAULT_PARAMS.copy() + params.update( + { + "exit_attenuation_mode": mode, + "exit_plateau": True, + "exit_plateau_grace": grace, + "exit_linear_slope": slope, + "exit_power_tau": tau, + "exit_half_life": half_life, + } + ) + + left = compute_exit_factor( + base_factor, pnl, pnl_factor, grace - eps, params + ) + boundary = compute_exit_factor( + base_factor, pnl, pnl_factor, grace, params + ) + right = compute_exit_factor( + base_factor, pnl, pnl_factor, grace + eps, params + ) + + self.assertAlmostEqualFloat( + left, + boundary, + tolerance=1e-9, + msg=f"Left/boundary mismatch for mode {mode}", + ) + self.assertLess( + right, + boundary, + f"No attenuation detected just after grace for mode {mode}", + ) + + diff = boundary - right + if mode == "linear": + bound = base_factor * slope * eps * 2.0 + elif mode == "sqrt": + bound = base_factor * 0.5 * eps * 2.0 + elif mode == "power": + alpha = -math.log(tau) / math.log(2.0) + bound = base_factor * alpha * eps * 2.0 + elif mode == "half_life": + bound = base_factor * (math.log(2.0) / half_life) * eps * 2.5 + else: + bound = base_factor * eps * 5.0 + + self.assertLessEqual( + diff, + bound, + f"Attenuation jump too large at boundary for mode {mode} (diff={diff:.6e} > bound={bound:.6e})", + ) + + def test_plateau_continuity_multiple_eps_scaling(self): + """Verify attenuation difference scales approximately linearly with epsilon (first-order continuity heuristic).""" + from reward_space_analysis import compute_exit_factor + + mode = "linear" + grace = 0.6 + eps1 = 1e-3 + eps2 = 1e-4 + base_factor = 80.0 + pnl = 0.02 + params = self.DEFAULT_PARAMS.copy() + params.update( + { + "exit_attenuation_mode": mode, + "exit_plateau": True, + "exit_plateau_grace": grace, + "exit_linear_slope": 1.1, + } + ) + f_boundary = compute_exit_factor(base_factor, pnl, 1.0, grace, params) + f1 = compute_exit_factor(base_factor, pnl, 1.0, grace + eps1, params) + f2 = compute_exit_factor(base_factor, pnl, 1.0, grace + eps2, params) + + diff1 = f_boundary - f1 + diff2 = f_boundary - f2 + ratio = diff1 / max(diff2, 1e-12) + self.assertGreater(ratio, 5.0, f"Scaling ratio too small (ratio={ratio:.2f})") + self.assertLess(ratio, 15.0, f"Scaling ratio too large (ratio={ratio:.2f})") + + if __name__ == "__main__": # Configure test discovery and execution loader = unittest.TestLoader() diff --git a/ReforceXY/user_data/freqaimodels/ReforceXY.py b/ReforceXY/user_data/freqaimodels/ReforceXY.py index 3f3e055..21d3bd4 100644 --- a/ReforceXY/user_data/freqaimodels/ReforceXY.py +++ b/ReforceXY/user_data/freqaimodels/ReforceXY.py @@ -121,6 +121,7 @@ class ReforceXY(BaseReinforcementLearningModel): - pip install optuna-dashboard """ + _LOG_2 = math.log(2.0) _action_masks_cache: Dict[Tuple[str, int, Optional[int]], NDArray[np.bool_]] = {} def __init__(self, *args, **kwargs): @@ -1392,7 +1393,18 @@ class MyRLEnv(Base5ActionRLEnv): duration_ratio = 0.0 model_reward_parameters = self.rl_config.get("model_reward_parameters", {}) - exit_factor_mode = model_reward_parameters.get("exit_factor_mode", "piecewise") + exit_attenuation_mode = str( + model_reward_parameters.get("exit_attenuation_mode", "linear") + ) + exit_plateau = bool(model_reward_parameters.get("exit_plateau", True)) + exit_plateau_grace = float( + model_reward_parameters.get("exit_plateau_grace", 1.0) + ) + if exit_plateau_grace < 0.0: + exit_plateau_grace = 1.0 + exit_linear_slope = float(model_reward_parameters.get("exit_linear_slope", 1.0)) + if exit_linear_slope < 0.0: + exit_linear_slope = 1.0 def _legacy(f: float, dr: float, p: Mapping) -> float: return f * (1.5 if dr <= 1.0 else 0.5) @@ -1415,26 +1427,13 @@ class MyRLEnv(Base5ActionRLEnv): if isinstance(tau, (int, float)): tau = float(tau) if 0.0 < tau <= 1.0: - alpha = -math.log(tau) / math.log(2.0) + alpha = -math.log(tau) / ReforceXY._LOG_2 if not isinstance(alpha, (int, float)): alpha = 1.0 else: alpha = float(alpha) return f / math.pow(1.0 + dr, alpha) - def _piecewise(f: float, dr: float, p: Mapping) -> float: - grace = float(p.get("exit_piecewise_grace", 1.0)) - if grace < 0.0: - grace = 1.0 - slope = float(p.get("exit_piecewise_slope", 1.0)) - if slope < 0.0: - slope = 1.0 - if dr <= grace: - divisor = 1.0 - else: - divisor = 1.0 + slope * (dr - grace) - return f / divisor - def _half_life(f: float, dr: float, p: Mapping) -> float: hl = float(p.get("exit_half_life", 0.5)) if hl <= 0.0: @@ -1446,30 +1445,41 @@ class MyRLEnv(Base5ActionRLEnv): "sqrt": _sqrt, "linear": _linear, "power": _power, - "piecewise": _piecewise, "half_life": _half_life, } - strategy_fn = strategies.get(exit_factor_mode, _piecewise) + if exit_plateau: + if duration_ratio <= exit_plateau_grace: + effective_dr = 0.0 + else: + effective_dr = duration_ratio - exit_plateau_grace + else: + effective_dr = duration_ratio + + strategy_fn = strategies.get(exit_attenuation_mode, None) + if strategy_fn is None: + logger.debug( + "Unknown exit_attenuation_mode '%s'; defaulting to linear", + exit_attenuation_mode, + ) + strategy_fn = _linear + try: - factor = strategy_fn(factor, duration_ratio, model_reward_parameters) + factor = strategy_fn(factor, effective_dr, model_reward_parameters) except Exception as e: logger.warning( - "exit_factor_mode '%s' failed (%r), falling back to piecewise", - exit_factor_mode, + "exit_attenuation_mode '%s' failed (%r); fallback linear (effective_dr=%.5f)", + exit_attenuation_mode, e, + effective_dr, ) - factor = _piecewise(factor, duration_ratio, model_reward_parameters) + factor = _linear(factor, effective_dr, model_reward_parameters) factor *= self._get_pnl_factor(pnl, self.profit_aim * self.rr) check_invariants = model_reward_parameters.get("check_invariants", True) check_invariants = ( - check_invariants - if isinstance(check_invariants, bool) - else bool(int(check_invariants)) - if isinstance(check_invariants, (int, float)) - else True + check_invariants if isinstance(check_invariants, bool) else True ) if check_invariants: if not np.isfinite(factor): -- 2.53.0