-# Reward Space Analysis (Specification)
-Concise operational guide. No marketing language. Single source of truth for tunables and validation guarantees. Exit factor parity date: 2025‑10‑06.
-## 1. Prérequis
-Python ≥3.8. Recommended: 8GB RAM. GPU non requis.
-Setup minimal:
-```shell
-cd ReforceXY/reward_space_analysis
-python -m venv .venv
-source .venv/bin/activate
-pip install pandas numpy scipy scikit-learn
-Run:
-```shell
-python reward_space_analysis.py --num_samples 20000 --output run1
-python test_reward_space_analysis.py
-## 2. Commandes Rapides
-Basique:
-```shell
-python reward_space_analysis.py --num_samples 10000
-```
-Sensibilité `win_reward_factor`:
-```shell
-python reward_space_analysis.py --num_samples 30000 --params win_reward_factor=2.0 --output wf2
-python reward_space_analysis.py --num_samples 30000 --params win_reward_factor=4.0 --output wf4
-```
-Comparaison réel vs synthétique:
-```shell
-python reward_space_analysis.py --num_samples 80000 --real_episodes ../user_data/models/ReforceXY-PPO/*/episode_rewards.pkl --output real_vs_syn
-```
-Batch simple:
-```shell
-for f in 1.5 2 3; do python reward_space_analysis.py --num_samples 20000 --params win_reward_factor=$f --output wf_$f; done
-```
-## 3. Paramètres (Tous optionnels)
-Paramètres CLI explicites + overrides `--params key=value`. Precedence: individual flag < `--params`.
-| Name | Default | Min | Max | Notes |
-|------|---------|-----|-----|-------|
-| num_samples | 20000 | 1 | — | Nombre d'échantillons synthétiques |
-| seed | 42 | 0 | — | Graine globale (simulation + RF) |
-| stats_seed | (seed) | 0 | — | Graine analytique (tests / bootstrap) |
-| max_trade_duration | 128 | 1 | — | Durée trade référence |
-| holding_max_ratio | 2.5 | >0 | — | Étendue d'échantillonnage durées |
-| pnl_base_std | 0.02 | 0 | — | Volatilité de base PnL |
-| pnl_duration_vol_scale | 0.5 | 0 | — | Amplification hétéroscédasticité |
-| trading_mode | spot | — | — | spot|margin|futures |
-| action_masking | true | — | — | Booléen |
-| base_factor | 100.0 | 0 | — | Facteur commun |
-| profit_target | 0.03 | 0 | — | Objectif profit |
-| risk_reward_ratio | 1.0 | 0 | — | Multiplicateur objectif |
-| invalid_action | -2.0 | — | 0 | Pénalité action invalide |
-| idle_penalty_scale | 1.0 | 0 | — | Échelle idle |
-| idle_penalty_power | 1.0 | 0 | — | Puissance idle |
-| max_idle_duration_candles | 0 | 0 | — | 0 ⇒ fallback max_trade_duration |
-| holding_penalty_scale | 0.5 | 0 | — | Échelle holding |
-| holding_penalty_power | 1.0 | 0 | — | Puissance holding |
-| exit_factor_mode | piecewise | — | — | legacy|sqrt|linear|power|piecewise|half_life |
-| exit_linear_slope | 1.0 | 0 | — | Pente linéaire |
-| exit_piecewise_grace | 1.0 | 0 | — | Frontière sans atténuation (>1 accepté) |
-| exit_piecewise_slope | 1.0 | 0 | — | Pente après grâce (0=plat) |
-| exit_power_tau | 0.5 | >0 | 1 | Tau ⇒ alpha = -ln(tau)/ln 2 |
-| exit_half_life | 0.5 | >0 | — | Demi‑vie exponentielle |
-| exit_factor_threshold | 10000 | >0 | — | Seuil warning-only |
-| efficiency_weight | 0.75 | 0 | 2 | Pondération efficacité |
-| efficiency_center | 0.75 | 0 | 1 | Centre sigmoïde |
-| win_reward_factor | 2.0 | 0 | — | Amplification asymptotique (1+val) |
-| pnl_factor_beta | 0.5 | >0 | — | Sensibilité tanh |
-| check_invariants | true | — | — | Active validations runtime |
-Notes:
-- `win_reward_factor` non plafonné mais borne effective via tanh.
-- `exit_piecewise_grace` >1 étend la zone plein facteur.
-- `exit_factor_threshold` génère un RuntimeWarning uniquement.
-## 4. Reproductibilité
-## 5. Overrides
-## 6. Exemples
-## 7. Résultats (Artifacts)
-## 8. Avancé
-## 9. Tests
-## 10. Dépannage (Condensé)
-## 11. Référence Rapide
-### Couches de Validation
-### Méthodes Statistiques
-### Validation Paramètres
-#### Bornes (rappel)
# 📊 Reward Space Analysis - User Guide
**Analyze and validate ReforceXY reward logic with synthetic data**
- `win_reward_factor` (default: 2.0) - Amplification for PnL above target (no upper bound; effective profit_target_factor ∈ [1, 1 + win_reward_factor] because tanh ≤ 1)
- `pnl_factor_beta` (default: 0.5) - Sensitivity of amplification around target
+_Invariant / safety controls:_
+
+- `check_invariants` (default: true) - Enable/disable runtime invariant & safety validations (simulation invariants, mathematical bounds, distribution checks). Set to `false` only for performance experiments; not recommended for production validation.
+
**`--real_episodes`** (path, optional)
- Path to real episode rewards pickle file for distribution comparison
-- Enables distribution shift analysis (KL divergence, JS distance, Wasserstein distance)
+- Enables distribution shift analysis (KL(synthetic‖real), JS distance, Wasserstein distance, KS test)
- Example: `../user_data/models/ReforceXY-PPO/sub_train_SYMBOL_DATE/episode_rewards.pkl`
**`--pvalue_adjust`** (choice: none|benjamini_hochberg, default: none)
Use `params_hash` to verify reproducibility across runs; identical seeds + identical overrides ⇒ identical hash.
+#### Distribution Shift Metric Conventions
+
+| Metric | Definition | Notes |
+|--------|------------|-------|
+| `*_kl_divergence` | KL(synthetic‖real) = Σ p_synth log(p_synth / p_real) | Asymmetric; 0 iff identical histograms (after binning). |
+| `*_js_distance` | √(JS(p_synth, p_real)) | Symmetric, bounded [0,1]; distance form (sqrt of JS divergence). |
+| `*_wasserstein` | 1D Earth Mover's Distance | Non-negative; same units as feature. |
+| `*_ks_statistic` | KS two-sample statistic | ∈ [0,1]; higher = greater divergence. |
+| `*_ks_pvalue` | KS test p-value | ∈ [0,1]; small ⇒ reject equality (at α). |
+
+Implementation details:
+- Histograms: 50 uniform bins spanning min/max across both samples.
+- Probabilities: counts + ε (1e‑10) then normalized ⇒ avoids log(0) and division by zero.
+- Degenerate (constant) distributions short‑circuit to zeros (divergences) / p-value 1.0.
+- JS distance is reported (not raw divergence) for bounded interpretability.
+
---
## 🔬 Advanced Usage
python test_reward_space_analysis.py
```
-The suite currently contains 49 focused tests (coverage ~84% — dynamic; see manifest + future reports). The number evolves as new invariants and edge cases are added. Always prefer running the full suite after modifying reward logic or attenuation parameters.
+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.
### Test Categories
| Statistical Coherence | TestStatisticalCoherence | Distribution shift, diagnostics, hypothesis basics |
| Reward Alignment | TestRewardAlignment | Component correctness & exit factors |
| Public API | TestPublicAPI | Core API functions and interfaces |
-| Statistical Validation | TestStatisticalValidation | Mathematical bounds and validation |
-| Boundary Conditions | TestBoundaryConditions | Extreme params & edge cases |
-| Helper Functions | TestHelperFunctions | I/O writers, model analysis, utility conversions |
-| Private Functions | TestPrivateFunctions | Penalty logic & internal reward calculations |
+| Statistical Validation | TestStatisticalValidation | Mathematical bounds, heteroscedasticity, invariants |
+| Boundary Conditions | TestBoundaryConditions | Extreme params & unknown mode fallback |
+| Helper Functions | TestHelperFunctions | Report writers, model analysis, utility conversions |
+| 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 |
### Test Architecture
### Code Coverage Analysis
-**Current Coverage: ~84% (approximate; re-run coverage locally for exact figures)**
-
-To analyze code coverage in detail:
-
```shell
-# Install coverage tool (if not already installed)
pip install coverage
-
-# Run tests with coverage
coverage run --source=. test_reward_space_analysis.py
-
-# Generate coverage report
coverage report -m
-
-# Generate HTML report for detailed analysis
-coverage html
-# View htmlcov/index.html in browser
+coverage html # open htmlcov/index.html
```
**Coverage Focus Areas:**
2. Reset to min if non-finite.
3. Recorded in `manifest.json` under `parameter_adjustments` with fields: `original`, `adjusted`, `reason` (a comma‑separated list of clamp reasons like `min=0.0`, `max=1.0`, `non_finite_reset`).
-Design intent: maintain a single canonical defaults map + explicit bounds; no silent acceptance of pathological inputs. (The earlier `_reason_text` placeholder has been removed; use `reason`.)
#### Parameter Bounds Summary
def validate_reward_parameters(
params: Dict[str, float | str],
-) -> Tuple[Dict[str, float | str], Dict[str, Dict[str, float]]]:
+) -> Tuple[Dict[str, float | str], Dict[str, Dict[str, Any]]]:
"""Validate and clamp reward parameter values.
Returns
Mapping param -> {original, adjusted, reason} for every modification.
"""
sanitized = dict(params)
- adjustments: Dict[str, Dict[str, float]] = {}
+ adjustments: Dict[str, Dict[str, Any]] = {}
for key, bounds in _PARAMETER_BOUNDS.items():
if key not in sanitized:
continue
"""Mirror the environment's idle penalty behaviour."""
idle_penalty_scale = _get_param_float(params, "idle_penalty_scale", 1.0)
idle_penalty_power = _get_param_float(params, "idle_penalty_power", 1.0)
- max_idle_duration = int(
- params.get(
- "max_idle_duration_candles", params.get("max_trade_duration_candles", 128)
+ max_trade_duration = int(params.get("max_trade_duration_candles", 128))
+ max_idle_duration_candles = params.get("max_idle_duration_candles")
+ try:
+ max_idle_duration = (
+ int(max_idle_duration_candles)
+ if max_idle_duration_candles is not None
+ else max_trade_duration
)
- )
+ except (TypeError, ValueError):
+ max_idle_duration = max_trade_duration
+ if max_idle_duration <= 0:
+ max_idle_duration = max_trade_duration
idle_duration_ratio = context.idle_duration / max(1, max_idle_duration)
return -idle_factor * idle_penalty_scale * idle_duration_ratio**idle_penalty_power
for mode in modes_to_test:
test_params = self.DEFAULT_PARAMS.copy()
test_params["exit_factor_mode"] = mode
-
- factor = rsa._get_exit_factor(
- factor=1.0,
+ factor = rsa.compute_exit_factor(
+ base_factor=1.0,
pnl=0.02,
pnl_factor=1.5,
duration_ratio=0.3,
)
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)."""
+ from reward_space_analysis import compute_exit_factor
+
+ base_factor = 100.0
+ pnl = 0.04
+ pnl_factor = 1.0
+ duration_ratio_linear = 1.2 # any positive ratio
+ duration_ratio_piecewise = 1.3 # > grace so slope matters
+
+ # 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_pos = self.DEFAULT_PARAMS.copy()
+ params_lin_pos.update({"exit_factor_mode": "linear", "exit_linear_slope": 1.0})
+ val_lin_neg = compute_exit_factor(
+ base_factor, pnl, pnl_factor, duration_ratio_linear, params_lin_neg
+ )
+ val_lin_pos = compute_exit_factor(
+ base_factor, pnl, pnl_factor, duration_ratio_linear, params_lin_pos
+ )
+ self.assertAlmostEqualFloat(
+ val_lin_neg,
+ val_lin_pos,
+ tolerance=1e-9,
+ 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(
+ {
+ "exit_factor_mode": "piecewise",
+ "exit_piecewise_grace": 1.0,
+ "exit_piecewise_slope": -3.0,
+ }
+ )
+ params_pw_pos = self.DEFAULT_PARAMS.copy()
+ params_pw_pos.update(
+ {
+ "exit_factor_mode": "piecewise",
+ "exit_piecewise_grace": 1.0,
+ "exit_piecewise_slope": 1.0,
+ }
+ )
+ val_pw_neg = compute_exit_factor(
+ base_factor, pnl, pnl_factor, duration_ratio_piecewise, params_pw_neg
+ )
+ val_pw_pos = compute_exit_factor(
+ base_factor, pnl, pnl_factor, duration_ratio_piecewise, params_pw_pos
+ )
+ self.assertAlmostEqualFloat(
+ val_pw_neg,
+ val_pw_pos,
+ tolerance=1e-9,
+ msg="Negative piecewise slope not sanitized to default behavior",
+ )
+
+ def test_idle_penalty_zero_when_profit_target_zero(self):
+ """If profit_target=0 → idle_factor=0 → idle penalty must be exactly 0 for neutral idle state."""
+ context = RewardContext(
+ pnl=0.0,
+ trade_duration=0,
+ idle_duration=30,
+ max_trade_duration=100,
+ max_unrealized_profit=0.0,
+ min_unrealized_profit=0.0,
+ position=Positions.Neutral,
+ action=Actions.Neutral,
+ force_action=None,
+ )
+ br = calculate_reward(
+ context,
+ self.DEFAULT_PARAMS,
+ base_factor=100.0,
+ profit_target=0.0, # critical case
+ risk_reward_ratio=1.0,
+ short_allowed=True,
+ action_masking=True,
+ )
+ self.assertEqual(
+ br.idle_penalty, 0.0, "Idle penalty should be zero when profit_target=0"
+ )
+ self.assertEqual(
+ br.total, 0.0, "Total reward should be zero in this configuration"
+ )
+
+ def test_power_mode_alpha_formula(self):
+ """Validate power mode: factor ≈ base_factor / (1+r)^alpha where alpha=-log(tau)/log(2)."""
+ from reward_space_analysis import compute_exit_factor
+
+ tau = 0.5
+ r = 1.2
+ alpha = -math.log(tau) / math.log(2.0)
+ base_factor = 100.0
+ pnl = 0.03
+ pnl_factor = 1.0 # isolate attenuation
+ params = self.DEFAULT_PARAMS.copy()
+ params.update({"exit_factor_mode": "power", "exit_power_tau": tau})
+ observed = compute_exit_factor(base_factor, pnl, pnl_factor, r, params)
+ expected = base_factor / (1.0 + r) ** alpha
+ self.assertAlmostEqualFloat(
+ observed,
+ expected,
+ tolerance=1e-9,
+ msg=f"Power mode attenuation mismatch (obs={observed}, exp={expected}, alpha={alpha})",
+ )
+
class TestPublicAPI(RewardSpaceTestBase):
"""Test public API functions and interfaces."""
pnl_factor = 1.1
# Ratios straddling 1.0 but below grace=1.5 plus one beyond grace
ratios = [0.8, 1.0, 1.2, 1.4, 1.6]
- vals = [compute_exit_factor(base_factor, pnl, pnl_factor, r, params) for r in ratios]
+ vals = [
+ compute_exit_factor(base_factor, pnl, pnl_factor, r, params) for r in ratios
+ ]
# All ratios <=1.5 should yield identical factor
ref = vals[0]
for i, r in enumerate(ratios[:-1]): # exclude last (1.6)
self.assertAlmostEqualFloat(
- vals[i], ref, 1e-9, msg=f"Unexpected attenuation before grace end at ratio {r}"
+ vals[i],
+ ref,
+ 1e-9,
+ msg=f"Unexpected attenuation before grace end at ratio {r}",
)
# Last ratio (1.6) should be attenuated (strictly less than ref)
self.assertLess(vals[-1], ref, "Attenuation should begin after grace boundary")
params["exit_factor_threshold"] = 10.0 # low threshold to trigger easily
# Remove base_factor to allow argument override
params.pop("base_factor", None)
- from reward_space_analysis import RewardContext, Actions, Positions
+ from reward_space_analysis import Actions, Positions, RewardContext
+
context = RewardContext(
pnl=0.06,
trade_duration=10,
"""
Compute the reward factor at trade exit
"""
- if (
- not np.isfinite(factor)
- or not np.isfinite(pnl)
- or not np.isfinite(duration_ratio)
+ if not (
+ np.isfinite(factor) and np.isfinite(pnl) and np.isfinite(duration_ratio)
):
return 0.0
+ if duration_ratio < 0.0:
+ 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")
-
- if exit_factor_mode == "legacy":
- if duration_ratio <= 1.0:
- factor *= 1.5
- else:
- factor *= 0.5
- elif exit_factor_mode == "sqrt":
- factor /= math.sqrt(1.0 + duration_ratio)
- elif exit_factor_mode == "linear":
- exit_linear_slope = float(
- model_reward_parameters.get("exit_linear_slope", 1.0)
- )
- if exit_linear_slope < 0.0:
- exit_linear_slope = 1.0
- factor /= 1.0 + exit_linear_slope * duration_ratio
- elif exit_factor_mode == "power":
- exit_power_alpha = model_reward_parameters.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 = model_reward_parameters.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) / math.log(2.0)
- if not isinstance(exit_power_alpha, (int, float)):
- exit_power_alpha = 1.0
+ exit_factor_mode = str(
+ model_reward_parameters.get("exit_factor_mode", "piecewise")
+ ).lower()
+
+ def _legacy(f: float, dr: float, p: Mapping) -> float:
+ return f * (1.5 if dr <= 1.0 else 0.5)
+
+ def _sqrt(f: float, dr: float, p: Mapping) -> float:
+ return f / math.sqrt(1.0 + dr)
+
+ def _linear(f: float, dr: float, p: Mapping) -> float:
+ slope = float(p.get("exit_linear_slope", 1.0))
+ if slope < 0.0:
+ slope = 1.0
+ return f / (1.0 + slope * dr)
+
+ def _power(f: float, dr: float, p: Mapping) -> float:
+ alpha = p.get("exit_power_alpha")
+ if isinstance(alpha, (int, float)) and alpha < 0.0:
+ alpha = None
+ if alpha is None:
+ tau = p.get("exit_power_tau")
+ if isinstance(tau, (int, float)):
+ tau = float(tau)
+ if 0.0 < tau <= 1.0:
+ alpha = -math.log(tau) / math.log(2.0)
+ if not isinstance(alpha, (int, float)):
+ 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":
- exit_piecewise_grace = float(
- model_reward_parameters.get("exit_piecewise_grace", 1.0)
- )
- if not (0.0 <= exit_piecewise_grace <= 1.0):
- exit_piecewise_grace = 1.0
- exit_piecewise_slope = float(
- model_reward_parameters.get("exit_piecewise_slope", 1.0)
- )
- if exit_piecewise_slope < 0.0:
- exit_piecewise_slope = 1.0
- if duration_ratio <= exit_piecewise_grace:
- duration_divisor = 1.0
+ 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:
- duration_divisor = 1.0 + exit_piecewise_slope * (
- duration_ratio - exit_piecewise_grace
- )
- factor /= duration_divisor
- elif exit_factor_mode == "half_life":
- exit_half_life = float(model_reward_parameters.get("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)
+ 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:
+ hl = 0.5
+ return f * math.pow(2.0, -dr / hl)
+
+ strategies: Dict[str, Callable[[float, float, Mapping], float]] = {
+ "legacy": _legacy,
+ "sqrt": _sqrt,
+ "linear": _linear,
+ "power": _power,
+ "piecewise": _piecewise,
+ "half_life": _half_life,
+ }
+
+ strategy_fn = strategies.get(exit_factor_mode, _piecewise)
+ try:
+ factor = strategy_fn(factor, duration_ratio, model_reward_parameters)
+ except Exception as e:
+ logger.warning(
+ "exit_factor_mode '%s' failed (%r), falling back to piecewise",
+ exit_factor_mode,
+ e,
+ )
+ factor = _piecewise(factor, duration_ratio, 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
+ )
+ if check_invariants:
+ if not np.isfinite(factor):
+ logger.debug(
+ "_get_exit_factor produced non-finite factor; resetting to 0.0"
+ )
+ return 0.0
+ if factor < 0.0 and pnl >= 0.0:
+ logger.debug(
+ "_get_exit_factor negative with positive pnl (factor=%.5f, pnl=%.5f); clamping to 0.0",
+ factor,
+ pnl,
+ )
+ factor = 0.0
+ exit_factor_threshold = float(
+ model_reward_parameters.get("exit_factor_threshold", 10_000.0)
+ )
+ if exit_factor_threshold > 0 and abs(factor) > exit_factor_threshold:
+ logger.warning(
+ "_get_exit_factor |factor|=%.2f exceeds threshold %.2f",
+ factor,
+ exit_factor_threshold,
+ )
+
return factor
def _get_pnl_factor(self, pnl: float, pnl_target: float) -> float:
"max_idle_duration_candles", max_trade_duration
)
)
+ if max_idle_duration <= 0:
+ max_idle_duration = max_trade_duration
idle_penalty_scale = float(
model_reward_parameters.get("idle_penalty_scale", 1.0)
)
order: Literal["entry", "exit"],
rate: float,
min_natr_ratio_percent: float = 0.009,
- max_natr_ratio_percent: float = 0.03,
+ max_natr_ratio_percent: float = 0.035,
lookback_period: int = 1,
decay_ratio: float = 0.5,
) -> bool:
repositories / freqai-strategies.git / commitdiff
