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)
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(
# 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")
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")
| 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 |
from reward_space_analysis import (
Actions,
Positions,
- RewardContext,
_compute_efficiency_coefficient,
_compute_hold_potential,
_compute_pnl_target_coefficient,
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,
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(
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"
},
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},
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
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.
- 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
**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)
- 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,
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
"""
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
import pandas as pd
import pytest
+import reward_space_analysis
from reward_space_analysis import (
DEFAULT_IDLE_DURATION_MULTIPLIER,
DEFAULT_MODEL_REWARD_PARAMETERS,
# 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."""
}
)
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,
"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,
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")
+import math
import unittest
import pytest
Positions,
RewardContext,
RewardDiagnosticsWarning,
+ RewardParams,
_get_exit_factor,
_hold_penalty,
validate_reward_parameters,
run_relaxed_validation_adjustment_cases,
run_strict_validation_failure_cases,
)
+from ..test_base import make_ctx
class _PyTestAdapter(unittest.TestCase):
- 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,
- 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,
- 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,
- 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,
**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,
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
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,
- 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,
- 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
)
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
)
# 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
- 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)
)
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)
)
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
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
- 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__
# 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:
import pandas as pd
import pytest
+import reward_space_analysis
from reward_space_analysis import (
RewardDiagnosticsWarning,
_binned_stats,
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
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")
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,
)
"""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
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",
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):
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,
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] = []
repositories / freqai-strategies.git / commitdiff
