},
{
"endCharacter": 92,
- "endLine": 571,
+ "endLine": 581,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Argument of type \"RewardParamValue\" cannot be assigned to parameter \"x\" of type \"ConvertibleToInt\" in function \"__new__\"\n Type \"RewardParamValue\" is not assignable to type \"ConvertibleToInt\"\n Type \"None\" is not assignable to type \"ConvertibleToInt\"\n \"None\" is not assignable to \"str\"\n \"None\" is incompatible with protocol \"Buffer\"\n \"__buffer__\" is not present\n \"None\" is incompatible with protocol \"SupportsInt\"\n \"__int__\" is not present\n \"None\" is incompatible with protocol \"SupportsIndex\"",
"rule": "reportArgumentType",
"severity": "error",
"startCharacter": 22,
- "startLine": 571
+ "startLine": 581
},
{
"endCharacter": 92,
- "endLine": 571,
+ "endLine": 581,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Argument of type \"RewardParamValue\" cannot be assigned to parameter \"x\" of type \"ConvertibleToInt\" in function \"__new__\"\n Type \"RewardParamValue\" is not assignable to type \"ConvertibleToInt\"\n Type \"None\" is not assignable to type \"ConvertibleToInt\"\n \"None\" is not assignable to \"str\"\n \"None\" is incompatible with protocol \"Buffer\"\n \"__buffer__\" is not present\n \"None\" is incompatible with protocol \"SupportsInt\"\n \"__int__\" is not present\n \"None\" is incompatible with protocol \"SupportsIndex\"\n ...",
"rule": "reportArgumentType",
"severity": "error",
"startCharacter": 22,
- "startLine": 571
+ "startLine": 581
},
{
"endCharacter": 5,
- "endLine": 2140,
+ "endLine": 2156,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "No overloads for \"cut\" match the provided arguments",
"rule": "reportCallIssue",
"severity": "error",
"startCharacter": 17,
- "startLine": 2135
+ "startLine": 2151
},
{
"endCharacter": 21,
- "endLine": 2137,
+ "endLine": 2153,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Argument of type \"_Array1D[Any]\" cannot be assigned to parameter \"bins\" of type \"int | Sequence[float] | Index[int] | Index[float] | IntervalIndex[Interval[Any]] | Series[Any]\" in function \"cut\"\n Type \"_Array1D[Any]\" is not assignable to type \"int | Sequence[float] | Index[int] | Index[float] | IntervalIndex[Interval[Any]] | Series[Any]\"\n \"ndarray[tuple[int], dtype[Any]]\" is not assignable to \"int\"\n \"ndarray[tuple[int], dtype[Any]]\" is not assignable to \"Sequence[float]\"\n \"ndarray[tuple[int], dtype[Any]]\" is not assignable to \"Index[int]\"\n \"ndarray[tuple[int], dtype[Any]]\" is not assignable to \"Index[float]\"\n \"ndarray[tuple[int], dtype[Any]]\" is not assignable to \"IntervalIndex[Interval[Any]]\"\n \"ndarray[tuple[int], dtype[Any]]\" is not assignable to \"Series[Any]\"",
"rule": "reportArgumentType",
"severity": "error",
"startCharacter": 13,
- "startLine": 2137
+ "startLine": 2153
},
{
"endCharacter": 5,
- "endLine": 2162,
+ "endLine": 2178,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Type \"dict[bytes, bytes] | dict[str, RewardParamValue]\" is not assignable to declared type \"RewardParams\"\n Type \"dict[bytes, bytes] | dict[str, RewardParamValue]\" is not assignable to type \"RewardParams\"\n \"dict[bytes, bytes]\" is not assignable to \"dict[str, RewardParamValue]\"\n Type parameter \"_KT@dict\" is invariant, but \"bytes\" is not the same as \"str\"\n Type parameter \"_VT@dict\" is invariant, but \"bytes\" is not the same as \"RewardParamValue\"",
"rule": "reportAssignmentType",
"severity": "error",
"startCharacter": 34,
- "startLine": 2158
+ "startLine": 2174
},
{
"endCharacter": 43,
- "endLine": 2159,
+ "endLine": 2175,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "No overloads for \"__init__\" match the provided arguments",
"rule": "reportCallIssue",
"severity": "error",
"startCharacter": 8,
- "startLine": 2159
+ "startLine": 2175
},
{
"endCharacter": 42,
- "endLine": 2159,
+ "endLine": 2175,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Argument of type \"Any | None\" cannot be assigned to parameter \"iterable\" of type \"Iterable[list[bytes]]\" in function \"__init__\"\n Type \"Any | None\" is not assignable to type \"Iterable[list[bytes]]\"\n \"None\" is incompatible with protocol \"Iterable[list[bytes]]\"\n \"__iter__\" is not present",
"rule": "reportArgumentType",
"severity": "error",
"startCharacter": 13,
- "startLine": 2159
+ "startLine": 2175
},
{
"endCharacter": 60,
- "endLine": 2419,
+ "endLine": 2435,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Cannot access attribute \"importances_mean\" for class \"dict[Unknown, Bunch]\"\n Attribute \"importances_mean\" is unknown",
"rule": "reportAttributeAccessIssue",
"severity": "error",
"startCharacter": 44,
- "startLine": 2419
+ "startLine": 2435
},
{
"endCharacter": 58,
- "endLine": 2420,
+ "endLine": 2436,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Cannot access attribute \"importances_std\" for class \"dict[Unknown, Bunch]\"\n Attribute \"importances_std\" is unknown",
"rule": "reportAttributeAccessIssue",
"severity": "error",
"startCharacter": 43,
- "startLine": 2420
+ "startLine": 2436
},
{
"endCharacter": 88,
- "endLine": 2439,
+ "endLine": 2455,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Cannot access attribute \"columns\" for class \"NDArray[Unknown]\"\n Attribute \"columns\" is unknown",
"rule": "reportAttributeAccessIssue",
"severity": "error",
"startCharacter": 81,
- "startLine": 2439
+ "startLine": 2455
},
{
"endCharacter": 88,
- "endLine": 2439,
+ "endLine": 2455,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Cannot access attribute \"columns\" for class \"list[Unknown]\"\n Attribute \"columns\" is unknown",
"rule": "reportAttributeAccessIssue",
"severity": "error",
"startCharacter": 81,
- "startLine": 2439
+ "startLine": 2455
},
{
"endCharacter": 17,
- "endLine": 2448,
+ "endLine": 2464,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Object of type \"None\" cannot be called",
"rule": "reportOptionalCall",
"severity": "error",
"startCharacter": 28,
- "startLine": 2442
+ "startLine": 2458
},
{
"endCharacter": 29,
- "endLine": 3047,
+ "endLine": 3063,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "No overloads for \"ptp\" match the provided arguments",
"rule": "reportCallIssue",
"severity": "error",
"startCharacter": 11,
- "startLine": 3047
+ "startLine": 3063
},
{
"endCharacter": 28,
- "endLine": 3047,
+ "endLine": 3063,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Argument of type \"np_1darray[Any] | ExtensionArray | Categorical[object]\" cannot be assigned to parameter \"a\" of type \"_ArrayLikeNumeric_co\" in function \"ptp\"\n Type \"np_1darray[Any] | ExtensionArray | Categorical[object]\" is not assignable to type \"_ArrayLikeNumeric_co\"\n Type \"ExtensionArray\" is not assignable to type \"_ArrayLikeNumeric_co\"\n \"ExtensionArray\" is incompatible with protocol \"_SupportsArray[dtype[number[Any, Any] | numpy.bool[builtins.bool] | object_ | timedelta64[Any]]]\"\n \"__array__\" is not present\n \"ExtensionArray\" is incompatible with protocol \"_NestedSequence[_SupportsArray[dtype[number[Any, Any] | numpy.bool[builtins.bool] | object_ | timedelta64[Any]]]]\"\n \"__reversed__\" is not present\n \"count\" is not present\n \"index\" is not present\n ...",
"rule": "reportArgumentType",
"severity": "error",
"startCharacter": 18,
- "startLine": 3047
+ "startLine": 3063
},
{
"endCharacter": 65,
- "endLine": 3056,
+ "endLine": 3072,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Cannot access attribute \"mean\" for class \"Categorical[object]\"\n Attribute \"mean\" is unknown",
"rule": "reportAttributeAccessIssue",
"severity": "error",
"startCharacter": 61,
- "startLine": 3056
+ "startLine": 3072
},
{
"endCharacter": 65,
- "endLine": 3056,
+ "endLine": 3072,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Cannot access attribute \"mean\" for class \"ExtensionArray\"\n Attribute \"mean\" is unknown",
"rule": "reportAttributeAccessIssue",
"severity": "error",
"startCharacter": 61,
- "startLine": 3056
+ "startLine": 3072
},
{
"endCharacter": 5,
- "endLine": 3956,
+ "endLine": 3972,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Type \"dict[bytes, bytes] | dict[str, RewardParamValue]\" is not assignable to declared type \"RewardParams\"\n Type \"dict[bytes, bytes] | dict[str, RewardParamValue]\" is not assignable to type \"RewardParams\"\n \"dict[bytes, bytes]\" is not assignable to \"dict[str, RewardParamValue]\"\n Type parameter \"_KT@dict\" is invariant, but \"bytes\" is not the same as \"str\"\n Type parameter \"_VT@dict\" is invariant, but \"bytes\" is not the same as \"RewardParamValue\"",
"rule": "reportAssignmentType",
"severity": "error",
"startCharacter": 34,
- "startLine": 3952
+ "startLine": 3968
},
{
"endCharacter": 43,
- "endLine": 3953,
+ "endLine": 3969,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "No overloads for \"__init__\" match the provided arguments",
"rule": "reportCallIssue",
"severity": "error",
"startCharacter": 8,
- "startLine": 3953
+ "startLine": 3969
},
{
"endCharacter": 42,
- "endLine": 3953,
+ "endLine": 3969,
"file": "ReforceXY/reward_space_analysis/reward_space_analysis.py",
"message": "Argument of type \"Any | None\" cannot be assigned to parameter \"iterable\" of type \"Iterable[list[bytes]]\" in function \"__init__\"\n Type \"Any | None\" is not assignable to type \"Iterable[list[bytes]]\"\n \"None\" is incompatible with protocol \"Iterable[list[bytes]]\"\n \"__iter__\" is not present",
"rule": "reportArgumentType",
"severity": "error",
"startCharacter": 13,
- "startLine": 3953
+ "startLine": 3969
},
{
"endCharacter": 22,
show_missing = True
precision = 1
include_namespace_packages = True
-# measured 2026-10-02 at 75.6% — raise only, never lower
+# measured 2026-10-02 at 76.0% — raise only, never lower
fail_under = 70
\ No newline at end of file
- **`--rf_n_jobs`** / **`--perm_n_jobs`** (int, default: -1) – Parallel worker
counts for RandomForest and permutation importance (-1 = all cores).
+Programmatic reward parameters accept Python and NumPy real integer/floating
+scalars. Strict validation checks a numeric gamma in its native precision before
+normalizing it. Analysis-specific numeric string/bool coercion is retained; it
+does not replace the stricter runtime configuration contract. Direct gamma
+calculation falls back to 0.95 for invalid/non-finite input and clamps finite
+out-of-range values; relaxed parameter validation records its resets/clamps.
+For a trained or resumed learner, supply its effective discount as
+`potential_gamma`, not a stale reward/config gamma.
+
Inferential helpers are available through the programmatic API only and require
`independent_observations=True`. Programmatic callers may pass `stats_seed`
to isolate bootstrap resampling from the simulation seed; the descriptive CLI
**Note:** In ReforceXY, `risk_reward_ratio` maps to `rr`.
+The CLI does not import the runtime config template. To reproduce its reward
+settings, use `--profit_aim 0.025 --max_trade_duration_candles 96
+--idle_penalty_ratio 0` with the same effective gamma and entry/exit fees. These
+are explicit scenario overrides, not different canonical reward defaults.
+
**Formula:**
-Let `pnl_target = profit_aim · risk_reward_ratio` and
-`pnl_ratio = pnl / pnl_target` when `pnl_target > 0`. For losses, let
-`effective_rr = risk_reward_ratio` if positive, or `2.0` otherwise (the runtime
-fallback). Then `loss_threshold = pnl_target / effective_rr` and
+For complete reward calculations, let `pnl_target = profit_aim · risk_reward_ratio`;
+replace it by `0.01` when the product is nonpositive, as in the runtime. This
+effective target is also used for representativity statistics and the manifest.
+The supplied profit aim and risk/reward ratio remain unchanged in idle/hold
+penalties. Low-level helpers receiving an explicit nonpositive target retain
+their neutral coefficient/zero-signal guards.
+
+Let `pnl_ratio = pnl / pnl_target` when `pnl_target > 0`. The low-level loss
+coefficient uses `effective_rr = risk_reward_ratio` if positive, or `2.0` otherwise.
+That helper fallback does not make a zero ratio valid for the complete reward
+path. Then `loss_threshold = pnl_target / effective_rr` and
`loss_ratio = |pnl| / loss_threshold = |pnl_ratio| · effective_rr`.
- If `pnl_target ≤ 0`: `pnl_target_coefficient = 1.0`
`efficiency_coefficient = 1 + efficiency_weight · (efficiency_center - ratio)`
- Else: `efficiency_coefficient = 1`
+The calculated coefficient is clamped to a finite nonnegative value. Validated
+analysis parameters additionally require
+`efficiency_weight · max(efficiency_center, 1 − efficiency_center) ≤ 1`: strict
+validation rejects combinations that could need a clamp at either ratio endpoint;
+relaxed validation records an adjustment and disables weighting (`weight = 0`).
+This accepted-domain restriction is stronger than the runtime's per-state clamp;
+direct calculation preserves the runtime formula without applying that validator.
+
In synthetic `unrealized_pnl` mode, sampled market prices accumulate each
candle's return independently of the transformed, retained price. The
fee-aware sampled PnL is bounded, scaled by the duration-dependent tanh
| Parameter | Default | Description |
| ---------------------------- | ------- | ----------------------------------------------------------------------------------------- |
| `max_trade_duration_candles` | 128 | Trade duration cap |
-| `max_idle_duration_candles` | None | Idle hazard threshold (4× trade duration fallback); the idle clock keeps counting past it |
+| `max_idle_duration_candles` | None | Missing/None derives 4× trade cap; explicit zero uses denominator 1 and does not disable the penalty |
| `idle_penalty_ratio` | 1.0 | Idle penalty ratio |
| `idle_penalty_power` | 1.025 | Idle penalty exponent |
| `hold_penalty_ratio` | 1.0 | Hold penalty ratio |
"exit_additive_gain": {"min": 0.0},
}
-RewardParamValue = float | str | bool | None
+RewardParamValue = float | np.integer[Any] | np.floating[Any] | str | bool | None
RewardParams = dict[str, RewardParamValue]
if isinstance(value, bool):
return float(int(value))
# Numeric
- if isinstance(value, (int, float)):
+ if isinstance(value, (int, float, np.integer, np.floating)):
try:
fval = float(value)
except (ValueError, TypeError):
default: Fallback value. If None, looks up from DEFAULT_MODEL_REWARD_PARAMETERS.
Behavior:
- - Accept bool/int/float/str numeric representations.
+ - Accept Python/NumPy real scalars and bool/str numeric representations.
- Non-finite floats -> fallback to default coerced to int (or 0).
- Strings: strip then parse float/int; on failure fallback.
- None -> fallback.
- - Final value is clamped to a signed 64-bit range implicitly by int().
+ - Return a Python int without an implicit fixed-width clamp.
"""
if default is None:
default = DEFAULT_MODEL_REWARD_PARAMETERS.get(key)
value = params.get(key, default)
if value is None:
- return int(default) if isinstance(default, (int, float)) else 0
+ return int(default) if isinstance(default, (int, float, np.integer, np.floating)) else 0
if isinstance(value, bool):
return int(value)
- if isinstance(value, int):
- return value
- if isinstance(value, float):
+ if isinstance(value, (int, np.integer)):
+ return int(value)
+ if isinstance(value, (float, np.floating)):
if not np.isfinite(value):
- return int(default) if isinstance(default, (int, float)) else 0
+ return int(default) if isinstance(default, (int, float, np.integer, np.floating)) else 0
try:
return int(value)
except (OverflowError, ValueError):
- return int(default) if isinstance(default, (int, float)) else 0
+ return int(default) if isinstance(default, (int, float, np.integer, np.floating)) else 0
if isinstance(value, str):
stripped = value.strip()
if stripped == "":
- return int(default) if isinstance(default, (int, float)) else 0
+ return int(default) if isinstance(default, (int, float, np.integer, np.floating)) else 0
try:
if any(ch in stripped for ch in (".", "e", "E")):
fval = float(stripped)
if not np.isfinite(fval):
- return int(default) if isinstance(default, (int, float)) else 0
+ return (
+ int(default)
+ if isinstance(default, (int, float, np.integer, np.floating))
+ else 0
+ )
return int(fval)
return int(stripped)
except (ValueError, OverflowError):
- return int(default) if isinstance(default, (int, float)) else 0
+ return int(default) if isinstance(default, (int, float, np.integer, np.floating)) else 0
# Unsupported type
- return int(default) if isinstance(default, (int, float)) else 0
+ return int(default) if isinstance(default, (int, float, np.integer, np.floating)) else 0
def _get_str_param(params: RewardParams, key: str, default: str | None = None) -> str:
return trade_duration / max(1, max_trade_duration_candles)
+def _resolve_pnl_target(profit_aim: float, risk_reward_ratio: float) -> float:
+ """Resolve the effective target used by runtime rewards and analysis diagnostics."""
+ target = float(profit_aim * risk_reward_ratio)
+ return 0.01 if target <= 0.0 else target
+
+
def _is_short_allowed(trading_mode: str) -> bool:
mode = trading_mode.lower()
if mode in TRADING_MODES[1:]: # "margin", "futures"
) -> int:
mtd = (
int(max_trade_duration_candles)
- if isinstance(max_trade_duration_candles, (int, float))
+ if isinstance(max_trade_duration_candles, (int, float, np.integer, np.floating))
else None
)
if mtd is None or mtd <= 0:
default_mid = int(DEFAULT_IDLE_DURATION_MULTIPLIER * int(mtd))
mid = _get_int_param(params, "max_idle_duration_candles", default_mid)
- if mid <= 0:
- mid = default_mid
return int(mid)
continue
original_val = sanitized[key]
+ # Check the native discount before normalization can round it into [0, 1].
+ if (
+ key == "potential_gamma"
+ and strict
+ and isinstance(original_val, (int, float, np.integer, np.floating))
+ and not bounds["min"] <= original_val <= bounds["max"]
+ ):
+ raise ValueError(f"Param: '{key}'={original_val!r} outside [0, 1]")
# Robust coercion to float using helper (handles None/str/bool/non-finite)
coerced_val = _get_float_param({key: original_val}, key, np.nan)
Attributes
----------
current_pnl : float
- Unrealized PnL at the current tick (state s').
+ Unrealized PnL of the represented execution or marked next state.
"""
current_pnl: float
elif "rr" in params:
risk_reward_ratio = _get_float_param(params, "rr", float(risk_reward_ratio))
- pnl_target = float(profit_aim * risk_reward_ratio)
+ pnl_target = _resolve_pnl_target(profit_aim, risk_reward_ratio)
idle_factor = base_factor * (profit_aim / risk_reward_ratio)
hold_factor = idle_factor
risk_reward_ratio: float,
) -> dict[str, Any]:
"""Compute representativity statistics for the reward space."""
- pnl_target = float(profit_aim * risk_reward_ratio)
+ pnl_target = _resolve_pnl_target(profit_aim, risk_reward_ratio)
total = len(df)
# Map numeric position codes to readable labels to avoid casting Neutral (0.5) to 0
pos_label_map = {0.0: "Short", 0.5: "Neutral", 1.0: "Long"}
"generated_at": pd.Timestamp.now().isoformat(),
"num_samples": len(df),
"seed": int(args.seed),
- "pnl_target": float(profit_aim * risk_reward_ratio),
+ "pnl_target": _resolve_pnl_target(profit_aim, risk_reward_ratio),
"parameter_adjustments": adjustments,
"reward_params": resolved_reward_params,
"effective": effective_params,
| statistics-rank-direction-141 | statistics | Rank-biserial effect follows the named first-group advantage | statistics/test_statistics.py::test_pnl_rank_biserial_direction_matches_named_first_group | Checks both directions |
| statistics-bootstrap-count-142 | statistics | Zero and negative resample counts fail for variable and constant data | statistics/test_statistics.py::test_bootstrap_rejects_nonpositive_resample_count | Rejects missing bootstrap |
| api-unmasked-sample-probabilities-144 | api | Unmasked sample probabilities match marginal valid-action frequencies | api/test_api_helpers.py::test_unmasked_sampling_probabilities_match_action_frequencies | Spot/futures entries, long/short exits, and neutral probability |
+| cli-effective-reward-boundaries-145 | cli | Effective profit-target fallback and explicit zero idle denominator reach artifacts | cli/test_cli_params_and_csv.py::test_zero_target_and_idle_cap_reach_rewards_and_manifest | Metadata and near-target market coverage agree with runtime boundaries |
### Non-Owning Smoke / Reference Checks
while integration tests focus on report formatting. The mapping above defines
current ownership; multi-path and non-owning references are called out explicitly.
+Real runtime/reference parity is owned by
+`ReforceXY/tests/test_model_pbrs_transitions.py::PbrsTransitionsTest::test_analysis_matches_real_rewards_across_modes_and_parameter_boundaries`.
+Run it in the Freqtrade RL QA image through the canonical runtime suite; the
+standalone analysis environment deliberately does not require Freqtrade/SB3.
+The comparator exercises direct/strict/relaxed parameter paths, real fills and
+terminal liquidation, carries reference potential independently, and checks tiny
+shaping components more tightly than the report-verification tolerance.
+Raw random shaping sums are not discounted invariance oracles. Complete-episode
+discounted identities and report continuity checks retain their existing owners.
+
## When to Run Tests
Run after changes to: reward component logic, PBRS mechanics, CLI
from reward_space_analysis import (
DEFAULT_MODEL_REWARD_PARAMETERS,
Actions,
+ _compute_representativity_stats,
get_max_idle_duration_candles,
)
from test_reward_space_analysis_cli import _is_warning_header, run_scenario
-from ..constants import SCENARIOS, SEEDS, TOLERANCE
+from ..constants import PARAMS, SCENARIOS, SEEDS, TOLERANCE
from ..test_base import RewardSpaceTestBase
# Pytest marker for taxonomy classification
- PBRS invariance summary section when reward_shaping present
"""
+ def test_zero_target_and_idle_cap_reach_rewards_and_manifest(self):
+ """Keep configured idle cap zero and expose the effective runtime profit target.
+
+ **Invariant:** cli-effective-reward-boundaries-145
+ **Setup:** fixed seed, zero profit aim and explicit idle cap zero.
+ **Assertions:** target metadata agrees with near-target market coverage;
+ idle ratios use a one-candle denominator rather than the derived cap.
+ """
+ out_dir = self.output_path / "zero_reward_boundaries"
+ result = _run_cli(
+ out_dir=out_dir,
+ args=[
+ "--num_samples",
+ str(SCENARIOS.CLI_NUM_SAMPLES_STANDARD),
+ "--seed",
+ str(SEEDS.BASE),
+ "--profit_aim",
+ "0",
+ "--max_idle_duration_candles",
+ "0",
+ "--hold_potential_enabled",
+ "1",
+ "--skip_feature_analysis",
+ "--skip_partial_dependence",
+ ],
+ )
+ _assert_cli_success(self, result)
+ manifest = json.loads((out_dir / "manifest.json").read_text())
+ samples = pd.read_csv(out_dir / "reward_samples.csv")
+ self.assertEqual(manifest["pnl_target"], PARAMS.PNL_TARGET_FALLBACK)
+ self.assertEqual(manifest["reward_params"]["max_idle_duration_candles"], 0)
+ pd.testing.assert_series_equal(
+ samples["idle_ratio"], samples["idle_duration"].astype(float), check_names=False
+ )
+ expected_near_target = (
+ (samples["pnl"] >= 0.8 * PARAMS.PNL_TARGET_FALLBACK)
+ & (samples["pnl"] <= 1.2 * PARAMS.PNL_TARGET_FALLBACK)
+ ).mean()
+ stats = _compute_representativity_stats(samples, 0.0, PARAMS.RISK_REWARD_RATIO)
+ self.assertAlmostEqualFloat(
+ stats["pnl_near_target"], expected_near_target, tolerance=TOLERANCE.GENERIC_EQ
+ )
+
def test_skip_feature_analysis_summary_branch(self):
"""CLI run with --skip_feature_analysis should mark feature importance skipped in summary and omit feature_importance.csv."""
out_dir = self.output_path / "skip_feature_analysis"
Attributes:
BH_FP_RATE_THRESHOLD: Benjamini-Hochberg false positive rate threshold (0.15)
BOOTSTRAP_DEFAULT_ITERATIONS: Default bootstrap resampling count (100)
- EXIT_PROBABILITY_THRESHOLD: Probability threshold for exit events (0.15)
"""
BH_FP_RATE_THRESHOLD: float = 0.15
BOOTSTRAP_DEFAULT_ITERATIONS: int = 100
- EXIT_PROBABILITY_THRESHOLD: float = 0.15
@dataclass(frozen=True)
Attributes:
BASE_FACTOR: Default base factor for reward scaling (90.0)
PROFIT_AIM: Target profit threshold (0.06)
+ PNL_TARGET_FALLBACK: Effective target for nonpositive profit aim products (0.01)
RISK_REWARD_RATIO: Standard risk/reward ratio (2.0)
RISK_REWARD_RATIO_HIGH: High risk/reward ratio for stress tests (4.0)
PNL_STD: Standard deviation for PnL generation (0.02)
BASE_FACTOR: float = 90.0
PROFIT_AIM: float = 0.06
+ PNL_TARGET_FALLBACK: float = 0.01
RISK_REWARD_RATIO: float = 2.0
RISK_REWARD_RATIO_HIGH: float = 4.0
PNL_STD: float = 0.02
SAMPLE_SIZE_LARGE: Large sample size for statistical power (800)
SAMPLE_SIZE_CONST_DF: Sample size for constant dataframes (64)
SAMPLE_SIZE_SHIFT_SCALE: Sample size for shift/scale tests (256)
- PBRS_SIMULATION_STEPS: Number of steps for PBRS simulation tests (500)
MONTE_CARLO_ITERATIONS: Monte Carlo simulation iterations (160)
PBRS_SWEEP_ITERATIONS: Number of iterations for PBRS sweep tests (120)
BOOTSTRAP_MINIMAL_ITERATIONS: Minimal bootstrap iterations for quick tests (25)
SAMPLE_SIZE_SHIFT_SCALE: int = 256
# Specialized test scenario sizes
- PBRS_SIMULATION_STEPS: int = 500
MONTE_CARLO_ITERATIONS: int = 160
PBRS_SWEEP_ITERATIONS: int = 120
BOOTSTRAP_MINIMAL_ITERATIONS: int = 25
import reward_space_analysis
from reward_space_analysis import (
- DEFAULT_IDLE_DURATION_MULTIPLIER,
DEFAULT_MODEL_REWARD_PARAMETERS,
INTERNAL_GUARDS,
PBRS_INVARIANCE_TOL,
_compute_unrealized_pnl_estimate,
_get_potential_gamma,
apply_potential_shaping,
- get_max_idle_duration_candles,
simulate_samples,
validate_reward_parameters,
write_complete_statistical_analysis,
PBRS,
SCENARIOS,
SEEDS,
- STATISTICAL,
TOLERANCE,
)
from ..helpers import (
# ---------------- Potential transform mechanics ---------------- #
def test_pbrs_progressive_release_decay_clamped(self):
- """Verifies progressive_release mode decay clamps at terminal.
+ """Verify progressive_release decay clamps on a nonterminal exit.
- Tolerance rationale: IDENTITY_RELAXED used for PBRS terminal state checks
+ Tolerance rationale: IDENTITY_RELAXED used for discounted exit-potential checks
due to accumulated errors from gamma discounting and potential calculations.
"""
params = self.DEFAULT_PARAMS.copy()
reward_shaping, -prev_potential, tolerance=TOLERANCE.IDENTITY_RELAXED
)
- def test_pbrs_spike_cancel_invariance(self):
- """Verifies spike_cancel mode produces near-zero terminal shaping."""
+ def test_spike_cancel_cancels_voluntary_exit_shaping(self):
+ """Cancel shaping on a nonterminal exit, without claiming terminal invariance."""
params = self.DEFAULT_PARAMS.copy()
params.update(
{
)
self.assertTrue(evidence["verified"], evidence["reason"])
- def test_non_canonical_flag_false_and_sum_nonzero(self):
- """Non-canonical mode -> invariant flags False and Σ shaping non-zero."""
+ def test_progressive_release_is_not_classified_as_zero_exit_mode(self):
+ """Residual-potential modes do not receive the zero-exit configuration flag."""
params = self.base_params(
exit_potential_mode="progressive_release",
)
unique_flags = set(df["pbrs_invariant"].unique().tolist())
self.assertEqual(unique_flags, {False}, f"Unexpected invariant flags: {unique_flags}")
- abs_sum = float(df["reward_shaping"].abs().sum())
- self.assertGreater(
- abs_sum,
- PBRS_INVARIANCE_TOL * 2,
- f"Expected non-trivial shaping magnitude (got {abs_sum})",
- )
# ---------------- Additives and canonical path mechanics ---------------- #
"Canonical shaping magnitude should exceed spike_cancel",
)
- def test_pbrs_retain_previous_cumulative_drift(self):
- """retain_previous mode accumulates negative shaping drift (non-invariant)."""
- params = self.base_params(
- exit_potential_mode="retain_previous",
- hold_potential_enabled=True,
- entry_additive_enabled=False,
- exit_additive_enabled=False,
- potential_gamma=0.9,
- )
- gamma = _get_potential_gamma(params)
- rng = np.random.default_rng(SEEDS.ALTERNATE_1)
- potentials = rng.uniform(0.05, 0.85, size=220)
- deltas = [gamma * p - p for p in potentials]
- cumulative = float(np.sum(deltas))
- self.assertLess(cumulative, -TOLERANCE.NEGLIGIBLE)
- self.assertGreater(abs(cumulative), 10 * TOLERANCE.IDENTITY_RELAXED)
-
def test_exit_step_shaping_matches_exit_step_rules(self):
"""Exit step: shaping uses stored prev_potential.
# |Φ(s)| <= base_factor and |Δ| <= (1+γ)*base_factor
self.assertLessEqual(abs(float(shap)), (1.0 + gamma) * PARAMS.BASE_FACTOR)
- def test_report_cumulative_invariance_aggregation(self):
- """Canonical telescoping term: small per-step mean drift, bounded increments."""
-
- params = self.base_params(
- hold_potential_enabled=True,
- entry_additive_enabled=False,
- exit_additive_enabled=False,
- exit_potential_mode="canonical",
- )
- gamma = _get_potential_gamma(params)
- rng = np.random.default_rng(SEEDS.REPORT_FORMAT_2)
- prev_potential = 0.0
- telescoping_sum = 0.0
- max_abs_step = 0.0
- steps = 0
- for _ in range(SCENARIOS.PBRS_SIMULATION_STEPS):
- is_exit = rng.uniform() < 0.1
- current_pnl = float(rng.normal(0, 0.05))
- current_dur = float(rng.uniform(0, 1))
- next_pnl = 0.0 if is_exit else float(rng.normal(0, 0.05))
- next_dur = 0.0 if is_exit else float(rng.uniform(0, 1))
- _tot, _shap, next_potential, _pbrs_delta, _entry_additive, _exit_additive = (
- apply_potential_shaping(
- base_reward=0.0,
- current_pnl=current_pnl,
- pnl_target=PARAMS.PROFIT_AIM * PARAMS.RISK_REWARD_RATIO,
- current_duration_ratio=current_dur,
- next_pnl=next_pnl,
- entry_pnl=next_pnl,
- next_duration_ratio=next_dur,
- risk_reward_ratio=PARAMS.RISK_REWARD_RATIO,
- base_factor=PARAMS.BASE_FACTOR,
- is_exit=is_exit,
- prev_potential=prev_potential,
- params=params,
- )
- )
- inc = gamma * next_potential - prev_potential
- telescoping_sum += inc
- if abs(inc) > max_abs_step:
- max_abs_step = abs(inc)
- steps += 1
- prev_potential = 0.0 if is_exit else next_potential
- mean_drift = telescoping_sum / max(1, steps)
- self.assertLess(
- abs(mean_drift),
- 0.02,
- f"Per-step telescoping drift too large (mean={mean_drift}, steps={steps})",
- )
- self.assertLessEqual(
- max_abs_step,
- PBRS.MAX_ABS_SHAPING,
- f"Unexpected large telescoping increment (max={max_abs_step})",
- )
-
- def test_report_explicit_non_invariance_progressive_release(self):
- """progressive_release cumulative shaping non-zero (release leak)."""
-
- params = self.base_params(
- hold_potential_enabled=True,
- entry_additive_enabled=False,
- exit_additive_enabled=False,
- exit_potential_mode="progressive_release",
- exit_potential_decay=0.25,
- )
- rng = np.random.default_rng(SEEDS.REPORT_FORMAT_2)
- prev_potential = 0.0
- shaping_sum = 0.0
-
- for _ in range(SCENARIOS.MONTE_CARLO_ITERATIONS):
- is_exit = rng.uniform() < STATISTICAL.EXIT_PROBABILITY_THRESHOLD
- next_pnl = 0.0 if is_exit else float(rng.normal(0, 0.07))
- next_dur = 0.0 if is_exit else float(rng.uniform(0, 1))
- _tot, shap, next_pot, _pbrs_delta, _entry_additive, _exit_additive = (
- apply_potential_shaping(
- base_reward=0.0,
- current_pnl=float(rng.normal(0, 0.07)),
- pnl_target=PARAMS.PROFIT_AIM * PARAMS.RISK_REWARD_RATIO,
- current_duration_ratio=float(rng.uniform(0, 1)),
- next_pnl=next_pnl,
- entry_pnl=next_pnl,
- next_duration_ratio=next_dur,
- risk_reward_ratio=PARAMS.RISK_REWARD_RATIO,
- base_factor=PARAMS.BASE_FACTOR,
- is_exit=is_exit,
- prev_potential=prev_potential,
- params=params,
- )
- )
- shaping_sum += shap
- prev_potential = 0.0 if is_exit else next_pot
- self.assertGreater(
- abs(shaping_sum),
- PBRS_INVARIANCE_TOL * 50,
- f"Expected non-zero shaping (got {shaping_sum})",
- )
-
# Non-owning smoke; ownership: robustness/test_robustness.py:43 (robustness-decomposition-integrity-101)
# Owns invariant: pbrs-canonical-near-zero-report-116
@pytest.mark.smoke
self.assertIn("_PBRS components not present in this analysis._", content)
self.assertIn("_Not performed (no real episodes provided)._", content)
- def test_get_max_idle_duration_candles_negative_or_zero_fallback(self):
- """Explicit mid<=0 fallback path returns derived default multiplier."""
- base = DEFAULT_MODEL_REWARD_PARAMETERS.copy()
- base["max_trade_duration_candles"] = 64
- base["max_idle_duration_candles"] = 0
- result = get_max_idle_duration_candles(base)
- expected = DEFAULT_IDLE_DURATION_MULTIPLIER * 64
- self.assertEqual(
- result, expected, f"Expected fallback {expected} for mid<=0 (got {result})"
- )
-
if __name__ == "__main__":
unittest.main()
self.assertGreater(value, 0.0)
self.assertLessEqual(value, 100.0)
- def test_the_floor_carries_its_provenance(self):
- # The measured date is what tells a reader whether the number is current, and
- # it is the only thing distinguishing a real floor from a provisional one.
- lines = [line.strip() for line in self.text.splitlines()]
- index = next(i for i, line in enumerate(lines) if line.startswith(f"{FLOOR} ="))
- annotation = " ".join(lines[max(0, index - 2) : index])
- self.assertIn("measured", annotation, f"the line above {FLOOR} must record the measurement")
- self.assertNotIn("PROVISIONAL", annotation)
-
def test_branch_tracing_is_enabled(self):
# Without it, every guard — an early return, a raise — counts as covered the
# moment its `if` is evaluated, and the gate measures nothing useful.
def test_the_floor_is_bound_in_the_section_coverage_reads_it_from(self):
# coverage binds fail_under as `report:fail_under` ONLY. The same key under [run]
# is silently inert, which would leave the gate dead while the value, the
- # provenance and the placeholder checks all still passed.
+ # numeric floor and range checks still passed.
self.assertIn(FLOOR, self.parser["report"])
self.assertNotIn(FLOOR, self.parser["run"])
# policy allows.
self.assertGreaterEqual(float(self._raw_floor()), MINIMUM_FLOOR)
- def test_the_provenance_records_a_date_and_a_measurement_that_clears_the_floor(self):
- # Recency is deliberately NOT asserted: refusing a date older than N months would
- # make this suite fail on a repository nobody has touched. What is enforceable is
- # that the annotation records a real measurement and that the measurement clears
- # the floor it justifies.
- lines = [line.strip() for line in self.text.splitlines()]
- index = next(i for i, line in enumerate(lines) if line.startswith(f"{FLOOR} ="))
- annotation = " ".join(lines[max(0, index - 2) : index])
- with self.subTest(requirement="a well-formed measurement date"):
- self.assertRegex(annotation, r"\d{4}-\d{2}-\d{2}")
- with self.subTest(requirement="a recorded measurement"):
- found = re.search(r"measured[^%]*?([0-9]+(?:\.[0-9]+)?)%", annotation)
- self.assertIsNotNone(found, f"no percentage recorded above {FLOOR}: {annotation!r}")
- self.assertGreaterEqual(
- float(found.group(1)),
- float(self._raw_floor()),
- "the annotated measurement must clear the floor it justifies",
- )
-
def test_no_production_source_widens_coverage_s_default_exclusions(self):
# The SOURCE side of the same lever. coverage applies DEFAULT_EXCLUDE by
# regex-searching the raw file text, so a pragma inside a string literal removes a
from sb3_contrib import MaskablePPO
from sb3_contrib.common.maskable.evaluation import evaluate_policy
+from ReforceXY.reward_space_analysis import reward_space_analysis as analysis
from ReforceXY.user_data.freqaimodels.ReforceXY import Actions, MyRLEnv, Positions, ReforceXY
class PbrsTransitionsTest(QaTestCase):
- def env(self, *, short=False, fee=0.0, **parameters):
- prices = pd.DataFrame({"open": [100.0, 100.0] + [90.0 if short else 110.0] * 10})
+ def env(self, *, short=False, fee=0.0, profit_aim=0.03, open_prices=None, **parameters):
+ if open_prices is None:
+ open_prices = [100.0, 100.0] + [90.0 if short else 110.0] * 10
+ prices = pd.DataFrame({"open": open_prices})
env = MyRLEnv(
df=prices.copy(),
prices=prices,
df_raw=prices.copy(),
window_size=1,
- reward_kwargs={"rr": 2.0, "profit_aim": 0.03},
+ reward_kwargs={"rr": 2.0, "profit_aim": profit_aim},
fee=fee,
can_short=True,
config={
env.reset()
return env
+ def test_analysis_matches_real_rewards_across_modes_and_parameter_boundaries(self):
+ """Compare independent rewards on real fills, marks and terminal liquidation.
+
+ The reference carries its own stored potential. Only the physical next state
+ is observed before liquidation; neither reward implementation is replaced.
+ Tight component tolerances also detect the tiny-terminal correction regression.
+ """
+ # short, terminal loss, explicit exit, gamma, ratio, additives, masking, aim, idle cap
+ scenarios = (
+ (False, False, False, 0.8, 0.4, True, False, 0.03, 12),
+ (True, True, True, 0.8, 1e-10, False, True, 0.03, 12),
+ (False, False, True, 0.0, 0.4, True, False, 0.03, 12),
+ (True, False, False, 1.0, 1e-10, True, True, 0.03, 12),
+ (False, True, False, 1e-12, 1e-10, True, True, 0.03, 12),
+ (True, True, True, 1e-9, 0.4, False, False, 0.03, 12),
+ (False, False, True, np.float32(0.8), np.float32(0.4), True, False, 0.03, 0),
+ (True, False, False, np.int64(1), 0.4, False, True, 0.0, 12),
+ (False, False, True, np.longdouble(0.8), 0.4, True, False, -0.03, 12),
+ )
+ components = (
+ ("invalid_penalty", "_last_invalid_penalty"),
+ ("idle_penalty", "_last_idle_penalty"),
+ ("hold_penalty", "_last_hold_penalty"),
+ ("exit_component", "_last_exit_reward"),
+ ("entry_additive", "_last_entry_additive"),
+ ("exit_additive", "_last_exit_additive"),
+ ("reward_shaping", "_last_reward_shaping"),
+ ("prev_potential", "_last_prev_potential"),
+ ("next_potential", "_last_next_potential"),
+ )
+ for mode in ReforceXY._EXIT_POTENTIAL_MODES:
+ for scenario in scenarios:
+ short, terminal_loss, explicit, gamma, ratio, additives, masking, aim, idle_cap = (
+ scenario
+ )
+ with self.subTest(mode=mode, scenario=scenario):
+ opens = [
+ 100.0,
+ 100.0,
+ 100.0,
+ 104.0,
+ 110.0,
+ 90.0,
+ 108.0,
+ 105.0,
+ 100.0,
+ 98.0,
+ 112.0,
+ 107.0,
+ 96.0,
+ 115.0,
+ 103.0,
+ 108.0,
+ ]
+ if short != terminal_loss:
+ opens = [200.0 - price for price in opens]
+ parameters = {
+ "exit_potential_mode": mode,
+ "potential_gamma": gamma,
+ "hold_potential_enabled": True,
+ "hold_potential_ratio": ratio,
+ "hold_potential_gain": np.float32(1.0),
+ "efficiency_weight": np.float32(1.0),
+ "entry_additive_enabled": additives,
+ "exit_additive_enabled": additives,
+ "max_trade_duration_candles": np.int64(3),
+ "max_idle_duration_candles": np.int64(idle_cap),
+ }
+ env = self.env(open_prices=opens, profit_aim=aim, fee=0.0015, **parameters)
+ env.action_masking = masking
+ params = {
+ **analysis.DEFAULT_MODEL_REWARD_PARAMETERS,
+ **parameters,
+ "entry_fee_rate": env.fee,
+ "exit_fee_rate": env.fee,
+ }
+ references = {
+ "direct": params,
+ "strict": analysis.validate_reward_parameters(params, strict=True)[0],
+ "relaxed": analysis.validate_reward_parameters(params, strict=False)[0],
+ }
+ potentials = dict.fromkeys(references, 0.0)
+ for tick in range(len(opens)):
+ action = Actions.Neutral
+ if tick == 2:
+ action = Actions.Short_enter if short else Actions.Long_enter
+ elif explicit and tick == 4:
+ action = Actions.Short_exit if short else Actions.Long_exit
+ elif tick == 6:
+ action = Actions.Long_exit if short else Actions.Short_exit
+ context = analysis.RewardContext(
+ env.get_unrealized_profit(),
+ env.get_trade_duration(),
+ env.get_idle_duration(),
+ env.get_max_unrealized_profit(),
+ env.get_min_unrealized_profit(),
+ analysis.Positions(env._position.value),
+ analysis.Actions(action.value),
+ )
+ next_context = None
+ compute_components = env._compute_pbrs_components
+
+ def observe_next_state(
+ *, observed_env=env, compute=compute_components, **transition
+ ):
+ nonlocal next_context
+ pnl = transition["next_pnl"]
+ next_context = analysis.RewardContext(
+ pnl,
+ transition["next_trade_duration"],
+ observed_env.get_idle_duration(),
+ max(observed_env.get_max_unrealized_profit(), pnl),
+ min(observed_env.get_min_unrealized_profit(), pnl),
+ analysis.Positions(transition["next_position"].value),
+ analysis.Actions.Neutral,
+ )
+ return compute(**transition)
+
+ with mock.patch.object(
+ env, "_compute_pbrs_components", side_effect=observe_next_state
+ ):
+ _, reward, done, truncated, info = env.step(action.value)
+ self.assertIsNotNone(next_context)
+ for validation, reference_params in references.items():
+ with self.subTest(validation=validation, tick=tick):
+ expected = analysis.calculate_reward(
+ context,
+ reference_params,
+ 100.0,
+ aim,
+ 2.0,
+ short_allowed=True,
+ action_masking=masking,
+ prev_potential=potentials[validation],
+ next_context=next_context,
+ terminated=done or truncated,
+ )
+ self.assertTrue(
+ math.isclose(
+ reward, expected.total, rel_tol=1e-12, abs_tol=1e-12
+ ),
+ (reward, expected.total),
+ )
+ for field, attribute in components:
+ actual = getattr(env, attribute)
+ predicted = getattr(expected, field)
+ # Relative error for composed arithmetic; absolute floor
+ # below tiny shaping signals, unlike report tolerance.
+ self.assertTrue(
+ math.isclose(
+ actual, predicted, rel_tol=1e-12, abs_tol=1e-20
+ ),
+ (field, actual, predicted),
+ )
+ self.assertEqual(
+ info["terminal_liquidation"], expected.terminal_liquidation
+ )
+ if expected.exit_pnl is not None:
+ self.assertAlmostEqual(
+ info["exit_pnl"], expected.exit_pnl, places=12
+ )
+ potentials[validation] = expected.next_potential
+ if done or truncated:
+ break
+ self.assertTrue(done or truncated)
+ self.assertEqual(set(potentials.values()), {0.0})
+
+ def test_analysis_rejects_gamma_outside_bounds_before_float_rounding(self):
+ """Extended-precision out-of-range discounts cannot become valid by rounding."""
+ for gamma in (
+ np.nextafter(np.longdouble(1), np.longdouble(math.inf)),
+ np.nextafter(np.longdouble(0), np.longdouble(-1)),
+ ):
+ with self.subTest(gamma=gamma), self.assertRaises(ValueError):
+ analysis.validate_reward_parameters({"potential_gamma": gamma}, strict=True)
+
def test_retained_potential_shapes_every_nonterminal_neutral_step(self):
for mode in ("retain_previous", "progressive_release", "spike_cancel"):
with self.subTest(mode=mode):
repositories / freqai-strategies.git / commitdiff
