From fa41285c975b8e3181ac56f32a9514632a99eb41 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Sun, 21 Dec 2025 20:54:21 +0100 Subject: [PATCH] refactor(ReforceXY): cleanup reward space analysis additive enablement handling MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- .../reward_space_analysis.py | 82 ++++++++++++++--- .../reward_space_analysis/tests/README.md | 2 +- .../components/test_reward_components.py | 4 +- .../tests/helpers/assertions.py | 6 +- .../tests/helpers/configs.py | 26 +++--- .../tests/helpers/test_internal_branches.py | 12 +-- .../tests/helpers/warnings.py | 15 ++-- .../tests/pbrs/test_pbrs.py | 88 +++++++++++++++---- .../tests/robustness/test_branch_coverage.py | 40 ++++++--- .../tests/robustness/test_robustness.py | 2 +- .../test_feature_analysis_failures.py | 35 ++++---- .../tests/statistics/test_statistics.py | 22 +++-- .../reward_space_analysis/tests/test_base.py | 34 +++++-- 13 files changed, 260 insertions(+), 108 deletions(-) diff --git a/ReforceXY/reward_space_analysis/reward_space_analysis.py b/ReforceXY/reward_space_analysis/reward_space_analysis.py index 822ab66..e4e8d03 100644 --- a/ReforceXY/reward_space_analysis/reward_space_analysis.py +++ b/ReforceXY/reward_space_analysis/reward_space_analysis.py @@ -301,6 +301,32 @@ def _get_bool_param(params: RewardParams, key: str, default: bool) -> bool: return bool(default) +def _resolve_additive_enablement( + exit_potential_mode: str, + entry_additive_enabled_raw: bool, + exit_additive_enabled_raw: bool, +) -> tuple[bool, bool, bool]: + """Resolve effective additive enablement. + + Canonical exit mode suppresses additive terms at runtime, even if they were + requested/enabled in configuration. + + Returns: + (entry_additive_effective, exit_additive_effective, additives_suppressed) + """ + + entry_additive_effective = ( + bool(entry_additive_enabled_raw) if exit_potential_mode != "canonical" else False + ) + exit_additive_effective = ( + bool(exit_additive_enabled_raw) if exit_potential_mode != "canonical" else False + ) + additives_suppressed = exit_potential_mode == "canonical" and bool( + entry_additive_enabled_raw or exit_additive_enabled_raw + ) + return entry_additive_effective, exit_additive_effective, additives_suppressed + + def _is_strict_validation(params: RewardParams) -> bool: """Return strict validation flag from params (default True).""" return _get_bool_param(params, "strict_validation", True) @@ -1333,8 +1359,11 @@ def simulate_samples( bool(DEFAULT_MODEL_REWARD_PARAMETERS.get("exit_additive_enabled", False)), ) - entry_enabled = bool(entry_enabled_raw) if exit_mode != "canonical" else False - exit_enabled = bool(exit_enabled_raw) if exit_mode != "canonical" else False + entry_enabled, exit_enabled, _additives_suppressed = _resolve_additive_enablement( + exit_mode, + entry_enabled_raw, + exit_enabled_raw, + ) pbrs_invariant = bool(exit_mode == "canonical" and not (entry_enabled or exit_enabled)) max_idle_duration_candles = get_max_idle_duration_candles( @@ -3608,41 +3637,64 @@ def write_complete_statistical_analysis( # Get configuration for proper invariance assessment reward_params = df.attrs.get("reward_params", {}) if hasattr(df, "attrs") else {} exit_potential_mode = _get_str_param(reward_params, "exit_potential_mode", "canonical") - entry_additive_enabled = _get_bool_param(reward_params, "entry_additive_enabled", False) - exit_additive_enabled = _get_bool_param(reward_params, "exit_additive_enabled", False) + entry_additive_enabled_raw = _get_bool_param( + reward_params, "entry_additive_enabled", False + ) + exit_additive_enabled_raw = _get_bool_param( + reward_params, "exit_additive_enabled", False + ) - # True invariance requires canonical mode AND no additives + ( + entry_additive_effective, + exit_additive_effective, + additives_suppressed, + ) = _resolve_additive_enablement( + exit_potential_mode, + entry_additive_enabled_raw, + exit_additive_enabled_raw, + ) + + # True invariance requires canonical mode AND no effective additives. is_theoretically_invariant = exit_potential_mode == "canonical" and not ( - entry_additive_enabled or exit_additive_enabled + entry_additive_effective or exit_additive_effective ) shaping_near_zero = abs(total_shaping) < PBRS_INVARIANCE_TOL + suppression_note = "" + if additives_suppressed: + suppression_note = ( + " Additives are suppressed in canonical mode" + f" (requested entry_additive_enabled={bool(entry_additive_enabled_raw)}," + f" exit_additive_enabled={bool(exit_additive_enabled_raw)})." + ) + # Prepare invariance summary markdown block if is_theoretically_invariant: if shaping_near_zero: invariance_status = "✅ Canonical" invariance_note = ( - "Theoretical invariance preserved (canonical mode, no additives, Σ≈0)" + "Theoretical invariance preserved (canonical mode, no additives, Σ≈0)." + + suppression_note ) else: invariance_status = "⚠️ Canonical (with warning)" invariance_note = ( - f"Canonical mode but unexpected shaping sum = {total_shaping:.6f}" + f"Canonical mode but unexpected shaping sum = {total_shaping:.6f}." + + suppression_note ) else: invariance_status = "❌ Non-canonical" reasons = [] if exit_potential_mode != "canonical": reasons.append(f"exit_potential_mode='{exit_potential_mode}'") - if entry_additive_enabled or exit_additive_enabled: + if entry_additive_effective or exit_additive_effective: additive_types = [] - if entry_additive_enabled: + if entry_additive_effective: additive_types.append("entry") - if exit_additive_enabled: + if exit_additive_effective: additive_types.append("exit") reasons.append(f"additives={additive_types}") invariance_note = f"Modified for flexibility: {', '.join(reasons)}" - # Summarize PBRS invariance f.write("**PBRS Invariance Summary:**\n\n") f.write("| Field | Value |\n") @@ -3650,8 +3702,10 @@ def write_complete_statistical_analysis( f.write(f"| Invariance Status | {invariance_status} |\n") f.write(f"| Analysis Note | {invariance_note} |\n") f.write(f"| Exit Potential Mode | {exit_potential_mode} |\n") - f.write(f"| Entry Additive Enabled | {entry_additive_enabled} |\n") - f.write(f"| Exit Additive Enabled | {exit_additive_enabled} |\n") + f.write(f"| Entry Additive Enabled | {bool(entry_additive_enabled_raw)} |\n") + f.write(f"| Exit Additive Enabled | {bool(exit_additive_enabled_raw)} |\n") + f.write(f"| Entry Additive Effective | {bool(entry_additive_effective)} |\n") + f.write(f"| Exit Additive Effective | {bool(exit_additive_effective)} |\n") f.write(f"| Σ Shaping Reward | {total_shaping:.6f} |\n") f.write(f"| Abs Σ Shaping Reward | {abs(total_shaping):.6e} |\n") f.write(f"| Σ Entry Additive | {entry_add_total:.6f} |\n") diff --git a/ReforceXY/reward_space_analysis/tests/README.md b/ReforceXY/reward_space_analysis/tests/README.md index e20f7c1..c3d1597 100644 --- a/ReforceXY/reward_space_analysis/tests/README.md +++ b/ReforceXY/reward_space_analysis/tests/README.md @@ -183,7 +183,7 @@ Columns: | robustness-negative-grace-clamp-103 | robustness | Negative exit_plateau_grace clamps to 0.0 w/ warning | robustness/test_robustness.py:555 | | | robustness-invalid-power-tau-104 | robustness | Invalid power tau falls back alpha=1.0 w/ warning | robustness/test_robustness.py:592 | | | robustness-near-zero-half-life-105 | robustness | Near-zero half life yields no attenuation (factor≈base) | robustness/test_robustness.py:621 | | -| pbrs-canonical-exit-semantic-106 | pbrs | Canonical exit uses shaping=-prev_potential and next_potential=0.0 | pbrs/test_pbrs.py:449 | Uses stored potential across steps; no drift correction applied | +| pbrs-canonical-exit-semantic-106 | pbrs | Canonical exit uses shaping=-prev_potential and next_potential=0.0 | pbrs/test_pbrs.py:449 | Uses stored potential across steps; no drift correction applied | | pbrs-canonical-near-zero-report-116 | pbrs | Canonical near-zero cumulative shaping classification | pbrs/test_pbrs.py:748 | Full report classification | | statistics-partial-deps-skip-107 | statistics | skip_partial_dependence => empty PD structures | statistics/test_statistics.py:28 | Docstring line | | helpers-duplicate-rows-drop-108 | helpers | Duplicate rows dropped w/ warning counting removals | helpers/test_utilities.py:26 | Docstring line | diff --git a/ReforceXY/reward_space_analysis/tests/components/test_reward_components.py b/ReforceXY/reward_space_analysis/tests/components/test_reward_components.py index a2833fc..a7b9b4b 100644 --- a/ReforceXY/reward_space_analysis/tests/components/test_reward_components.py +++ b/ReforceXY/reward_space_analysis/tests/components/test_reward_components.py @@ -9,7 +9,6 @@ import pytest from reward_space_analysis import ( Actions, Positions, - RewardContext, _compute_efficiency_coefficient, _compute_hold_potential, _compute_pnl_target_coefficient, @@ -273,7 +272,7 @@ class TestRewardComponents(RewardSpaceTestBase): modes_to_test = ["linear", "power"] pnl = 0.02 pnl_target = 0.045 # 0.03 * 1.5 coefficient - context = RewardContext( + context = self.make_ctx( pnl=pnl, trade_duration=50, idle_duration=0, @@ -282,6 +281,7 @@ class TestRewardComponents(RewardSpaceTestBase): position=Positions.Neutral, action=Actions.Neutral, ) + for mode in modes_to_test: test_params = self.base_params(exit_attenuation_mode=mode) factor = _get_exit_factor( diff --git a/ReforceXY/reward_space_analysis/tests/helpers/assertions.py b/ReforceXY/reward_space_analysis/tests/helpers/assertions.py index c70e8b3..7eff926 100644 --- a/ReforceXY/reward_space_analysis/tests/helpers/assertions.py +++ b/ReforceXY/reward_space_analysis/tests/helpers/assertions.py @@ -1029,13 +1029,13 @@ def assert_exit_factor_invariant_suite( cases = [ { "base_factor": 90.0, "pnl": 0.08, "pnl_target": 0.03, - "context": RewardContext(...), + "context": make_ctx(...), "duration_ratio": 0.5, "params": {...}, "expectation": "non_negative", "tolerance": 1e-09 }, { "base_factor": 90.0, "pnl": 0.0, "pnl_target": 0.03, - "context": RewardContext(...), + "context": make_ctx(...), "duration_ratio": 0.5, "params": {...}, "expectation": "safe_zero" }, @@ -1103,7 +1103,7 @@ def assert_exit_factor_kernel_fallback( Example: # After monkeypatching kernel to fail: - test_context = RewardContext(pnl=0.08, ...) + test_context = make_ctx(pnl=0.08, ...) assert_exit_factor_kernel_fallback( self, _get_exit_factor, 90.0, 0.08, 0.03, 0.5, test_context, bad_params={"exit_attenuation_mode": "power", "exit_power_tau": -1.0}, diff --git a/ReforceXY/reward_space_analysis/tests/helpers/configs.py b/ReforceXY/reward_space_analysis/tests/helpers/configs.py index 6be3402..1b336ab 100644 --- a/ReforceXY/reward_space_analysis/tests/helpers/configs.py +++ b/ReforceXY/reward_space_analysis/tests/helpers/configs.py @@ -6,19 +6,19 @@ function signatures in test helpers, following the DRY principle and reducing parameter proliferation. Usage: - from tests.helpers.configs import RewardScenarioConfig - from tests.constants import PARAMS, TOLERANCE - - config = RewardScenarioConfig( - base_factor=PARAMS.BASE_FACTOR, - profit_aim=PARAMS.PROFIT_AIM, - risk_reward_ratio=PARAMS.RISK_REWARD_RATIO, - tolerance_relaxed=TOLERANCE.IDENTITY_RELAXED - ) - - assert_reward_calculation_scenarios( - test_case, scenarios, config, validation_fn - ) + >>> from tests.helpers.configs import RewardScenarioConfig + >>> from tests.constants import PARAMS, TOLERANCE + + >>> config = RewardScenarioConfig( + ... base_factor=PARAMS.BASE_FACTOR, + ... profit_aim=PARAMS.PROFIT_AIM, + ... risk_reward_ratio=PARAMS.RISK_REWARD_RATIO, + ... tolerance_relaxed=TOLERANCE.IDENTITY_RELAXED + ... ) + + >>> assert_reward_calculation_scenarios( + ... test_case, scenarios, config, validation_fn + ... ) """ from dataclasses import dataclass diff --git a/ReforceXY/reward_space_analysis/tests/helpers/test_internal_branches.py b/ReforceXY/reward_space_analysis/tests/helpers/test_internal_branches.py index 6cb1339..ce00aa4 100644 --- a/ReforceXY/reward_space_analysis/tests/helpers/test_internal_branches.py +++ b/ReforceXY/reward_space_analysis/tests/helpers/test_internal_branches.py @@ -5,12 +5,14 @@ import numpy as np from reward_space_analysis import ( Actions, Positions, - RewardContext, + RewardParams, _get_bool_param, _get_float_param, calculate_reward, ) +from ..test_base import make_ctx + def test_get_bool_param_none_and_invalid_literal(): """Verify _get_bool_param handles None and invalid literals correctly. @@ -27,11 +29,11 @@ def test_get_bool_param_none_and_invalid_literal(): - None coerces to False (covers _to_bool None path) - Invalid literal returns default (ValueError fallback path) """ - params_none = {"check_invariants": None} + params_none: RewardParams = {"check_invariants": None} # None should coerce to False (coverage for _to_bool None path) assert _get_bool_param(params_none, "check_invariants", True) is False - params_invalid = {"check_invariants": "not_a_bool"} + params_invalid: RewardParams = {"check_invariants": "not_a_bool"} # Invalid literal triggers ValueError in _to_bool; fallback returns default (True) assert _get_bool_param(params_invalid, "check_invariants", True) is True @@ -50,7 +52,7 @@ def test_get_float_param_invalid_string_returns_nan(): **Assertions:** - Result is NaN (covers float conversion ValueError path) """ - params = {"idle_penalty_scale": "abc"} + params: RewardParams = {"idle_penalty_scale": "abc"} val = _get_float_param(params, "idle_penalty_scale", 0.5) assert math.isnan(val) @@ -73,7 +75,7 @@ def test_calculate_reward_unrealized_pnl_hold_path(): - At least one potential is non-zero (shaping should activate) """ # Exercise unrealized_pnl branch during hold to cover next_pnl tanh path - context = RewardContext( + context = make_ctx( pnl=0.01, trade_duration=5, idle_duration=0, diff --git a/ReforceXY/reward_space_analysis/tests/helpers/warnings.py b/ReforceXY/reward_space_analysis/tests/helpers/warnings.py index 908fa0a..9de1319 100644 --- a/ReforceXY/reward_space_analysis/tests/helpers/warnings.py +++ b/ReforceXY/reward_space_analysis/tests/helpers/warnings.py @@ -6,10 +6,10 @@ capturing and validating warnings in tests, reducing boilerplate code and ensuring consistent warning handling patterns. Usage: - from tests.helpers.warnings import assert_diagnostic_warning + >>> from tests.helpers.warnings import assert_diagnostic_warning - with assert_diagnostic_warning(["exit_factor", "threshold"]) as caught: - result = calculate_something_that_warns() + >>> with assert_diagnostic_warning(["exit_factor", "threshold"]) as caught: + ... result = calculate_something_that_warns() # Assertions are automatic; caught warnings available for inspection """ @@ -18,10 +18,11 @@ import warnings from contextlib import contextmanager from typing import Any, Optional -try: - from reward_space_analysis import RewardDiagnosticsWarning -except ImportError: - RewardDiagnosticsWarning = RuntimeWarning +import reward_space_analysis + +RewardDiagnosticsWarning = getattr( + reward_space_analysis, "RewardDiagnosticsWarning", RuntimeWarning +) @contextmanager diff --git a/ReforceXY/reward_space_analysis/tests/pbrs/test_pbrs.py b/ReforceXY/reward_space_analysis/tests/pbrs/test_pbrs.py index 2c69ec8..ef7b356 100644 --- a/ReforceXY/reward_space_analysis/tests/pbrs/test_pbrs.py +++ b/ReforceXY/reward_space_analysis/tests/pbrs/test_pbrs.py @@ -8,6 +8,7 @@ import numpy as np import pandas as pd import pytest +import reward_space_analysis from reward_space_analysis import ( DEFAULT_IDLE_DURATION_MULTIPLIER, DEFAULT_MODEL_REWARD_PARAMETERS, @@ -927,6 +928,65 @@ class TestPBRS(RewardSpaceTestBase): # Non-owning smoke; ownership: robustness/test_robustness.py:35 (robustness-decomposition-integrity-101) @pytest.mark.smoke + def test_pbrs_canonical_suppresses_additives_in_report(self): + """Canonical exit mode suppresses additives for classification. + + The reward engine suppresses additive terms when exit_potential_mode is "canonical". + The report should align: classification stays canonical and should not claim + non-canonical additives involvement. + """ + + small_vals = [1.0e-7, -2.0e-7, 3.0e-7] # sum = 2.0e-7 < tolerance + total_shaping = float(sum(small_vals)) + self.assertLess(abs(total_shaping), PBRS_INVARIANCE_TOL) + n = len(small_vals) + df = pd.DataFrame( + { + "reward": np.random.normal(0, 1, n), + "reward_idle": np.zeros(n), + "reward_hold": np.random.normal(-0.2, 0.05, n), + "reward_exit": np.random.normal(0.4, 0.15, n), + "pnl": np.random.normal(0.01, 0.02, n), + "trade_duration": np.random.uniform(5, 30, n), + "idle_duration": np.zeros(n), + "position": np.random.choice([0.0, 0.5, 1.0], n), + "action": np.random.randint(0, 3, n), + "reward_shaping": small_vals, + "reward_entry_additive": [0.0] * n, + "reward_exit_additive": [0.0] * n, + "reward_invalid": np.zeros(n), + "duration_ratio": np.random.uniform(0.2, 1.0, n), + "idle_ratio": np.zeros(n), + } + ) + df.attrs["reward_params"] = { + "exit_potential_mode": "canonical", + "entry_additive_enabled": True, + "exit_additive_enabled": True, + } + out_dir = self.output_path / "canonical_additives_suppressed" + write_complete_statistical_analysis( + df, + output_dir=out_dir, + profit_aim=PARAMS.PROFIT_AIM, + risk_reward_ratio=PARAMS.RISK_REWARD_RATIO, + seed=SEEDS.BASE, + skip_feature_analysis=True, + skip_partial_dependence=True, + bootstrap_resamples=SCENARIOS.BOOTSTRAP_MINIMAL_ITERATIONS, + ) + report_path = out_dir / "statistical_analysis.md" + self.assertTrue(report_path.exists(), "Report file missing for canonical additives test") + content = report_path.read_text(encoding="utf-8") + assert_pbrs_invariance_report_classification( + self, content, "Canonical", expect_additives=False + ) + self.assertIn("Additives are suppressed in canonical mode", content) + self.assertIn("| Entry Additive Enabled | True |", content) + self.assertIn("| Exit Additive Enabled | True |", content) + self.assertIn("| Entry Additive Effective | False |", content) + self.assertIn("| Exit Additive Effective | False |", content) + def test_pbrs_canonical_warning_report(self): """Canonical mode + no additives but |Σ shaping| > tolerance -> warning classification.""" @@ -1114,25 +1174,19 @@ class TestPBRS(RewardSpaceTestBase): } ) out_dir = self.output_path / "pbrs_absence_and_shift_placeholder" - # Import here to mock _compute_summary_stats function - import reward_space_analysis as rsa - - original_compute_summary_stats = rsa._compute_summary_stats + original_compute_summary_stats = reward_space_analysis._compute_summary_stats def _minimal_summary_stats(_df): - # Use _pd alias to avoid conflicts with global pd - import pandas as _pd - - comp_share = _pd.Series([], dtype=float) - action_summary = _pd.DataFrame( - columns=_pd.Index(["count", "mean", "std", "min", "max"]), - index=_pd.Index([], name="action"), + comp_share = pd.Series([], dtype=float) + action_summary = pd.DataFrame( + columns=pd.Index(["count", "mean", "std", "min", "max"]), + index=pd.Index([], name="action"), ) - component_bounds = _pd.DataFrame( - columns=_pd.Index(["component_min", "component_mean", "component_max"]), - index=_pd.Index([], name="component"), + component_bounds = pd.DataFrame( + columns=pd.Index(["component_min", "component_mean", "component_max"]), + index=pd.Index([], name="component"), ) - global_stats = _pd.Series([], dtype=float) + global_stats = pd.Series([], dtype=float) return { "global_stats": global_stats, "action_summary": action_summary, @@ -1140,7 +1194,7 @@ class TestPBRS(RewardSpaceTestBase): "component_bounds": component_bounds, } - rsa._compute_summary_stats = _minimal_summary_stats + reward_space_analysis._compute_summary_stats = _minimal_summary_stats try: write_complete_statistical_analysis( df, @@ -1153,7 +1207,7 @@ class TestPBRS(RewardSpaceTestBase): bootstrap_resamples=SCENARIOS.BOOTSTRAP_MINIMAL_ITERATIONS // 2, ) finally: - rsa._compute_summary_stats = original_compute_summary_stats + reward_space_analysis._compute_summary_stats = original_compute_summary_stats report_path = out_dir / "statistical_analysis.md" self.assertTrue(report_path.exists(), "Report file missing for PBRS absence test") content = report_path.read_text(encoding="utf-8") diff --git a/ReforceXY/reward_space_analysis/tests/robustness/test_branch_coverage.py b/ReforceXY/reward_space_analysis/tests/robustness/test_branch_coverage.py index 1cf1ea5..96d4034 100644 --- a/ReforceXY/reward_space_analysis/tests/robustness/test_branch_coverage.py +++ b/ReforceXY/reward_space_analysis/tests/robustness/test_branch_coverage.py @@ -1,3 +1,4 @@ +import math import unittest import pytest @@ -7,6 +8,7 @@ from reward_space_analysis import ( Positions, RewardContext, RewardDiagnosticsWarning, + RewardParams, _get_exit_factor, _hold_penalty, validate_reward_parameters, @@ -18,6 +20,7 @@ from ..helpers import ( run_relaxed_validation_adjustment_cases, run_strict_validation_failure_cases, ) +from ..test_base import make_ctx class _PyTestAdapter(unittest.TestCase): @@ -72,10 +75,14 @@ def test_get_exit_factor_negative_plateau_grace_warning(): - Warning emitted (RewardDiagnosticsWarning) - Factor is non-negative despite invalid parameter """ - params = {"exit_attenuation_mode": "linear", "exit_plateau": True, "exit_plateau_grace": -1.0} + params: RewardParams = { + "exit_attenuation_mode": "linear", + "exit_plateau": True, + "exit_plateau_grace": -1.0, + } pnl = 0.01 pnl_target = 0.03 - context = RewardContext( + context = make_ctx( pnl=pnl, trade_duration=50, idle_duration=0, @@ -110,10 +117,10 @@ def test_get_exit_factor_negative_linear_slope_warning(): - Warning emitted (RewardDiagnosticsWarning) - Factor is non-negative despite invalid parameter """ - params = {"exit_attenuation_mode": "linear", "exit_linear_slope": -5.0} + params: RewardParams = {"exit_attenuation_mode": "linear", "exit_linear_slope": -5.0} pnl = 0.01 pnl_target = 0.03 - context = RewardContext( + context = make_ctx( pnl=pnl, trade_duration=50, idle_duration=0, @@ -149,10 +156,14 @@ def test_get_exit_factor_invalid_power_tau_relaxed(): - Warning emitted (RewardDiagnosticsWarning) - Factor is positive (fallback to default tau) """ - params = {"exit_attenuation_mode": "power", "exit_power_tau": 0.0, "strict_validation": False} + params: RewardParams = { + "exit_attenuation_mode": "power", + "exit_power_tau": 0.0, + "strict_validation": False, + } pnl = 0.02 pnl_target = 0.03 - context = RewardContext( + context = make_ctx( pnl=pnl, trade_duration=50, idle_duration=0, @@ -188,14 +199,14 @@ def test_get_exit_factor_half_life_near_zero_relaxed(): - Warning emitted (RewardDiagnosticsWarning) - Factor is non-zero (fallback to sensible value) """ - params = { + params: RewardParams = { "exit_attenuation_mode": "half_life", "exit_half_life": 1e-12, "strict_validation": False, } pnl = 0.02 pnl_target = 0.03 - context = RewardContext( + context = make_ctx( pnl=pnl, trade_duration=50, idle_duration=0, @@ -229,7 +240,7 @@ def test_hold_penalty_short_duration_returns_zero(): **Assertions:** - Penalty equals 0.0 (no penalty for short duration holds) """ - context = RewardContext( + context = make_ctx( pnl=0.0, trade_duration=1, # shorter than default max trade duration (128) idle_duration=0, @@ -238,7 +249,7 @@ def test_hold_penalty_short_duration_returns_zero(): position=Positions.Long, action=Actions.Neutral, ) - params = {"max_trade_duration_candles": 128} + params: RewardParams = {"max_trade_duration_candles": 128.0} penalty = _hold_penalty(context, hold_factor=1.0, params=params) assert penalty == 0.0 @@ -249,13 +260,14 @@ def test_exit_factor_invariant_suite_grouped(): def make_context(pnl: float) -> RewardContext: """Helper to create context for test cases.""" - return RewardContext( + max_profit = 0.03 + if isinstance(pnl, float) and math.isfinite(pnl): + max_profit = max(pnl * 1.2, 0.03) + return make_ctx( pnl=pnl, trade_duration=50, idle_duration=0, - max_unrealized_profit=max(pnl * 1.2, 0.03) - if not (isinstance(pnl, float) and (pnl != pnl or pnl == float("inf"))) - else 0.03, + max_unrealized_profit=max_profit, min_unrealized_profit=0.0, position=Positions.Neutral, action=Actions.Neutral, diff --git a/ReforceXY/reward_space_analysis/tests/robustness/test_robustness.py b/ReforceXY/reward_space_analysis/tests/robustness/test_robustness.py index ed2ad5b..34217ca 100644 --- a/ReforceXY/reward_space_analysis/tests/robustness/test_robustness.py +++ b/ReforceXY/reward_space_analysis/tests/robustness/test_robustness.py @@ -347,7 +347,7 @@ class TestRewardRobustnessAndBoundaries(RewardSpaceTestBase): - Uses assertFinite which checks for non-NaN, non-Inf values only """ extreme_params = self.base_params(win_reward_factor=1000.0, base_factor=10000.0) - context = RewardContext( + context = self.make_ctx( pnl=0.05, trade_duration=50, idle_duration=0, diff --git a/ReforceXY/reward_space_analysis/tests/statistics/test_feature_analysis_failures.py b/ReforceXY/reward_space_analysis/tests/statistics/test_feature_analysis_failures.py index 8e7dde0..a27f233 100644 --- a/ReforceXY/reward_space_analysis/tests/statistics/test_feature_analysis_failures.py +++ b/ReforceXY/reward_space_analysis/tests/statistics/test_feature_analysis_failures.py @@ -12,11 +12,15 @@ Covers early stub returns and guarded exception branches to raise coverage: - scikit-learn import fallback (RandomForestRegressor/train_test_split/permutation_importance/r2_score unavailable) """ +import builtins +import importlib +import sys + import numpy as np import pandas as pd import pytest -from reward_space_analysis import _perform_feature_analysis # type: ignore +from reward_space_analysis import RandomForestRegressor, _perform_feature_analysis # type: ignore from tests.constants import SEEDS pytestmark = pytest.mark.statistics @@ -102,7 +106,7 @@ def test_feature_analysis_single_feature_path(): ) assert stats["n_features"] == 1 # Importance stub path returns NaNs - assert importance_df["importance_mean"].isna().all() + assert bool(importance_df["importance_mean"].isna().all()) assert model is None @@ -132,7 +136,7 @@ def test_feature_analysis_nans_present_path(): ) # Should hit NaN stub path (model_fitted False) assert stats["model_fitted"] is False - assert importance_df["importance_mean"].isna().all() + assert bool(importance_df["importance_mean"].isna().all()) assert model is None @@ -153,7 +157,8 @@ def test_feature_analysis_model_fitting_failure(monkeypatch): - importance_mean is all NaN """ # Monkeypatch model fit to raise - from reward_space_analysis import RandomForestRegressor # type: ignore + if RandomForestRegressor is None: # type: ignore[comparison-overlap] + pytest.skip("sklearn components unavailable; skipping model fitting failure test") _ = RandomForestRegressor.fit # preserve reference for clarity (unused) @@ -167,7 +172,7 @@ def test_feature_analysis_model_fitting_failure(monkeypatch): ) assert stats["model_fitted"] is False assert model is None - assert importance_df["importance_mean"].isna().all() + assert bool(importance_df["importance_mean"].isna().all()) # Restore (pytest monkeypatch will revert automatically at teardown) @@ -200,7 +205,7 @@ def test_feature_analysis_permutation_failure_partial_dependence(monkeypatch): ) assert stats["model_fitted"] is True # Importance should be NaNs due to failure - assert importance_df["importance_mean"].isna().all() + assert bool(importance_df["importance_mean"].isna().all()) # Partial dependencies should still attempt and produce entries for available features listed in function assert len(partial_deps) >= 1 # at least one PD computed assert model is not None @@ -226,7 +231,7 @@ def test_feature_analysis_success_partial_dependence(): df, seed=SEEDS.FEATURE_PRIME_47, skip_partial_dependence=False ) # Expect at least one non-NaN importance (model fitted path) - assert importance_df["importance_mean"].notna().any() + assert bool(importance_df["importance_mean"].notna().any()) assert stats["model_fitted"] is True assert len(partial_deps) >= 1 assert model is not None @@ -254,10 +259,6 @@ def test_module_level_sklearn_import_failure_reload(): - Call its _perform_feature_analysis to confirm ImportError path surfaces. - Restore original importer and original module to avoid side-effects on other tests. """ - import builtins - import importlib - import sys - orig_mod = sys.modules.get("reward_space_analysis") orig_import = builtins.__import__ @@ -271,17 +272,17 @@ def test_module_level_sklearn_import_failure_reload(): # Drop existing module to force fresh execution of top-level imports if "reward_space_analysis" in sys.modules: del sys.modules["reward_space_analysis"] - import reward_space_analysis as rsa_fallback # noqa: F401 + reloaded_module = importlib.import_module("reward_space_analysis") # Fallback assigns sklearn symbols to None - assert getattr(rsa_fallback, "RandomForestRegressor") is None - assert getattr(rsa_fallback, "train_test_split") is None - assert getattr(rsa_fallback, "permutation_importance") is None - assert getattr(rsa_fallback, "r2_score") is None + assert getattr(reloaded_module, "RandomForestRegressor") is None + assert getattr(reloaded_module, "train_test_split") is None + assert getattr(reloaded_module, "permutation_importance") is None + assert getattr(reloaded_module, "r2_score") is None # Perform feature analysis should raise ImportError under missing components df = _minimal_df(15) with pytest.raises(ImportError): - rsa_fallback._perform_feature_analysis( + reloaded_module._perform_feature_analysis( df, seed=SEEDS.FEATURE_SMALL_3, skip_partial_dependence=True ) # type: ignore[attr-defined] finally: diff --git a/ReforceXY/reward_space_analysis/tests/statistics/test_statistics.py b/ReforceXY/reward_space_analysis/tests/statistics/test_statistics.py index 3931f15..2e1d830 100644 --- a/ReforceXY/reward_space_analysis/tests/statistics/test_statistics.py +++ b/ReforceXY/reward_space_analysis/tests/statistics/test_statistics.py @@ -7,6 +7,7 @@ import numpy as np import pandas as pd import pytest +import reward_space_analysis from reward_space_analysis import ( RewardDiagnosticsWarning, _binned_stats, @@ -29,6 +30,8 @@ from ..constants import ( from ..helpers import assert_diagnostic_warning from ..test_base import RewardSpaceTestBase +_perform_feature_analysis = getattr(reward_space_analysis, "_perform_feature_analysis", None) + pytestmark = pytest.mark.statistics @@ -37,9 +40,7 @@ class TestStatistics(RewardSpaceTestBase): def test_statistics_feature_analysis_skip_partial_dependence(self): """Invariant 107: skip_partial_dependence=True yields empty partial_deps.""" - try: - from reward_space_analysis import _perform_feature_analysis # type: ignore - except ImportError: + if _perform_feature_analysis is None: self.skipTest("sklearn not available; skipping feature analysis invariance test") # Use existing helper to get synthetic stats df (small for speed) df = self.make_stats_df(n=120, seed=SEEDS.BASE, idle_pattern="mixed") @@ -250,12 +251,16 @@ class TestStatistics(RewardSpaceTestBase): js_key = next((k for k in metrics if k.endswith("pnl_js_distance")), None) if js_key is None: self.skipTest("JS distance key not present in metrics output") + assert js_key is not None + metrics_swapped = compute_distribution_shift_metrics(df2, df1) js_key_swapped = next((k for k in metrics_swapped if k.endswith("pnl_js_distance")), None) self.assertIsNotNone(js_key_swapped) + assert js_key_swapped is not None + self.assertAlmostEqualFloat( - metrics[js_key], - metrics_swapped[js_key_swapped], + float(metrics[js_key]), + float(metrics_swapped[js_key_swapped]), tolerance=TOLERANCE.IDENTITY_STRICT, rtol=TOLERANCE.RELATIVE, ) @@ -294,9 +299,10 @@ class TestStatistics(RewardSpaceTestBase): """Batch mean additivity.""" df_a = self._shift_scale_df(120) df_b = self._shift_scale_df(180, shift=0.2) - m_concat = pd.concat([df_a["pnl"], df_b["pnl"]]).mean() - m_weighted = (df_a["pnl"].mean() * len(df_a) + df_b["pnl"].mean() * len(df_b)) / ( - len(df_a) + len(df_b) + m_concat = float(pd.concat([df_a["pnl"], df_b["pnl"]]).mean()) + m_weighted = float( + (df_a["pnl"].mean() * len(df_a) + df_b["pnl"].mean() * len(df_b)) + / (len(df_a) + len(df_b)) ) self.assertAlmostEqualFloat( m_concat, m_weighted, tolerance=TOLERANCE.IDENTITY_STRICT, rtol=TOLERANCE.RELATIVE diff --git a/ReforceXY/reward_space_analysis/tests/test_base.py b/ReforceXY/reward_space_analysis/tests/test_base.py index 79fc075..e02bc17 100644 --- a/ReforceXY/reward_space_analysis/tests/test_base.py +++ b/ReforceXY/reward_space_analysis/tests/test_base.py @@ -28,6 +28,31 @@ from .constants import ( TOLERANCE, ) +# Helper functions + + +def make_ctx( + *, + pnl: float = 0.0, + trade_duration: int = 0, + idle_duration: int = 0, + max_unrealized_profit: float = 0.0, + min_unrealized_profit: float = 0.0, + position: Positions = Positions.Neutral, + action: Actions = Actions.Neutral, +) -> RewardContext: + """Create a RewardContext with neutral defaults.""" + return RewardContext( + pnl=pnl, + trade_duration=trade_duration, + idle_duration=idle_duration, + max_unrealized_profit=max_unrealized_profit, + min_unrealized_profit=min_unrealized_profit, + position=position, + action=action, + ) + + # Global constants PBRS_INTEGRATION_PARAMS = [ "potential_gamma", @@ -46,9 +71,6 @@ class RewardSpaceTestBase(unittest.TestCase): def setUpClass(cls): """Set up class-level constants.""" cls.DEFAULT_PARAMS = DEFAULT_MODEL_REWARD_PARAMETERS.copy() - # Constants used in helper methods - cls.PBRS_TERMINAL_PROB = PBRS.TERMINAL_PROBABILITY - cls.PBRS_SWEEP_ITER = SCENARIOS.PBRS_SWEEP_ITERATIONS cls.JS_DISTANCE_UPPER_BOUND = math.sqrt(math.log(2.0)) def setUp(self): @@ -73,7 +95,7 @@ class RewardSpaceTestBase(unittest.TestCase): action: Actions = Actions.Neutral, ) -> RewardContext: """Create a RewardContext with neutral defaults.""" - return RewardContext( + return make_ctx( pnl=pnl, trade_duration=trade_duration, idle_duration=idle_duration, @@ -101,8 +123,8 @@ class RewardSpaceTestBase(unittest.TestCase): Returns (terminal_next_potentials, shaping_values). """ - iters = iterations or self.PBRS_SWEEP_ITER - term_p = terminal_prob or self.PBRS_TERMINAL_PROB + iters = iterations or SCENARIOS.PBRS_SWEEP_ITERATIONS + term_p = terminal_prob or PBRS.TERMINAL_PROBABILITY rng = np.random.default_rng(seed) prev_potential = 0.0 terminal_next: list[float] = [] -- 2.43.0