---
-**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.
---
- `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:_
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
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
| 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
| `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 |
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)
# 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)
"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.",
"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},
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
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,
)
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 (
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
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,
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
# 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
)
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):
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(
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
)
# 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(
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(
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,
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."""
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":
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:
)
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
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)
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)
)
# 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(
"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
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
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
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(
)
+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()
- 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):
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)
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:
"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):
repositories / freqai-strategies.git / commitdiff
