from __future__ import annotations import sys import unittest from pathlib import Path import numpy as np import pandas as pd TRAINING_ROOT = Path(__file__).resolve().parents[1] if str(TRAINING_ROOT) not in sys.path: sys.path.insert(0, str(TRAINING_ROOT)) from trader_training.labels import DEFAULT_LABEL_CONFIG, _path_stats_for_group from trader_training.pm import _probability_implied_edge, _simulate_open_trades, _threshold_candidates, default_pm_config class RiskPmFixTest(unittest.TestCase): def test_path_stats_never_writes_negative_adverse_or_favorable_move(self) -> None: frame = pd.DataFrame( { "event_time": pd.date_range("2026-01-01", periods=4, freq="min", tz="UTC"), "open_time_ms": np.arange(4, dtype=np.int64) * 60_000, "symbol": "BTC-USDT-PERP", "close": [100.0, 101.0, 102.0, 103.0], "high": [100.0, 101.0, 102.0, 103.0], "low": [100.0, 101.0, 102.0, 103.0], "spread_bps": [1.0, 1.0, 1.0, 1.0], } ) long_stats = _path_stats_for_group(frame, "LONG", horizon=2, target_bps=500.0, stop_bps=500.0) short_stats = _path_stats_for_group(frame, "SHORT", horizon=2, target_bps=500.0, stop_bps=500.0) self.assertGreaterEqual(float(long_stats["mae_bps"].min()), 0.0) self.assertGreaterEqual(float(long_stats["mfe_bps"].min()), 0.0) self.assertGreaterEqual(float(short_stats["mae_bps"].min()), 0.0) self.assertGreaterEqual(float(short_stats["mfe_bps"].min()), 0.0) def test_default_risk_labels_match_design_thresholds(self) -> None: self.assertEqual(45, DEFAULT_LABEL_CONFIG["continue"]["horizon_minutes"]) self.assertEqual(60.0, DEFAULT_LABEL_CONFIG["risk"]["market_drawdown_bps"]) self.assertEqual(35.0, DEFAULT_LABEL_CONFIG["risk"]["position_path_risk_bps"]) self.assertEqual(80.0, DEFAULT_LABEL_CONFIG["risk"]["spike_bps"]) self.assertEqual(1.8, DEFAULT_LABEL_CONFIG["risk"]["vol_expansion_ratio"]) def test_pm_search_covers_low_entry_probability_without_allowing_negative_edge(self) -> None: candidates = _threshold_candidates() self.assertTrue(candidates) self.assertLessEqual(max(item["max_market_risk_prob"] for item in candidates), 0.98) self.assertLessEqual(min(item["min_entry_prob"] for item in candidates), 0.03) self.assertGreaterEqual(min(item["min_expected_edge_bps"] for item in candidates), 0.0) def test_probability_implied_edge_uses_price_plan_payoff(self) -> None: edge = _probability_implied_edge( pd.Series([0.10, 0.50]), {"targetDistanceBps": 120.0, "stopDistanceBps": 2.0, "costBps": 6.5}, ) self.assertAlmostEqual(3.7, float(edge.iloc[0]), places=6) self.assertAlmostEqual(52.5, float(edge.iloc[1]), places=6) def test_pm_backtest_sizing_uses_position_manager_formula_not_fixed_floor(self) -> None: frame = pd.DataFrame( { "sample_id": ["s0"], "symbol": ["BTC-USDT-PERP"], "event_time": pd.to_datetime(["2026-01-01T00:00:00Z"]), "split_id": ["tune_inner"], "long_prob": [0.70], "short_prob": [0.10], "neutral_prob": [0.20], "long_entry_prob": [0.80], "short_entry_prob": [0.20], "market_risk_prob": [0.20], "long_position_risk_prob": [0.10], "short_position_risk_prob": [0.10], "pred_long_expected_net_edge_bps": [40.0], "pred_short_expected_net_edge_bps": [1.0], "actual_long_expected_net_edge_bps": [30.0], "actual_short_expected_net_edge_bps": [-10.0], "long_trade_net_edge_bps": [11.0], "short_trade_net_edge_bps": [-14.5], "long_target_hit": [1], "short_target_hit": [0], "long_stop_hit": [0], "short_stop_hit": [1], "long_time_to_target_ms": [300_000], "short_time_to_target_ms": [-1], "long_time_to_stop_ms": [-1], "short_time_to_stop_ms": [180_000], "long_entry_target": [1], "short_entry_target": [0], } ) thresholds = { "long_open_prob": 0.55, "short_open_prob": 0.55, "min_entry_prob": 0.55, "max_market_risk_prob": 0.55, "min_expected_edge_bps": 3.0, "min_direction_margin": 0.02, } trades = _simulate_open_trades( frame, thresholds, default_pm_config(), {"stopDistanceBps": 8.0, "costBps": 6.5}, ) self.assertEqual(1, len(trades)) self.assertAlmostEqual(11.0, float(trades.iloc[0]["actual_edge_bps"])) self.assertAlmostEqual(30.0, float(trades.iloc[0]["label_max_edge_bps"])) self.assertGreater(float(trades.iloc[0]["planned_ratio"]), 0.05) self.assertLessEqual(float(trades.iloc[0]["planned_ratio"]), 0.20) def test_pm_backtest_blocks_overlapping_open_trades_until_exit_and_cooldown(self) -> None: frame = pd.DataFrame( { "sample_id": ["s0", "s1"], "symbol": ["BTC-USDT-PERP", "BTC-USDT-PERP"], "event_time": pd.to_datetime(["2026-01-01T00:00:00Z", "2026-01-01T00:01:00Z"]), "split_id": ["tune_inner", "tune_inner"], "long_prob": [0.70, 0.72], "short_prob": [0.10, 0.10], "neutral_prob": [0.20, 0.18], "long_entry_prob": [0.80, 0.82], "short_entry_prob": [0.20, 0.20], "market_risk_prob": [0.20, 0.20], "long_position_risk_prob": [0.10, 0.10], "short_position_risk_prob": [0.10, 0.10], "pred_long_expected_net_edge_bps": [40.0, 42.0], "pred_short_expected_net_edge_bps": [1.0, 1.0], "actual_long_expected_net_edge_bps": [30.0, 31.0], "actual_short_expected_net_edge_bps": [-10.0, -10.0], "long_trade_net_edge_bps": [11.0, 12.0], "short_trade_net_edge_bps": [-14.5, -14.5], "long_target_hit": [1, 1], "short_target_hit": [0, 0], "long_stop_hit": [0, 0], "short_stop_hit": [1, 1], "long_time_to_target_ms": [300_000, 300_000], "short_time_to_target_ms": [-1, -1], "long_time_to_stop_ms": [-1, -1], "short_time_to_stop_ms": [180_000, 180_000], "long_entry_target": [1, 1], "short_entry_target": [0, 0], } ) thresholds = { "long_open_prob": 0.55, "short_open_prob": 0.55, "min_entry_prob": 0.55, "max_market_risk_prob": 0.55, "min_expected_edge_bps": 3.0, "min_direction_margin": 0.02, } trades = _simulate_open_trades( frame, thresholds, default_pm_config(), {"stopDistanceBps": 8.0, "costBps": 6.5, "maxHoldMinutes": 45}, ) self.assertEqual(1, len(trades)) self.assertEqual("s0", trades.iloc[0]["sample_id"]) if __name__ == "__main__": unittest.main()