Improve Trader entry quality training diagnostics

training/scripts/23_train_ofi_feature_experiment.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+import _bootstrap  # noqa: F401
+from trader_training.io_utils import add_common_args, setup_logging
+from trader_training.ofi_feature_experiment import run_ofi_feature_experiment
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    add_common_args(parser)
+    parser.add_argument("--baseline-run-id", required=True)
+    parser.add_argument("--raw-root", type=Path)
+    parser.add_argument("--max-rows-per-split", type=int, default=0)
+    args = parser.parse_args()
+    setup_logging()
+    run_ofi_feature_experiment(args)
+
+
+if __name__ == "__main__":
+    main()

training/tests/test_risk_pm_fix.py

@@ -0,0 +1,173 @@
+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()

training/tests/test_training_contract.py

@@ -14,8 +14,10 @@ if str(TRAINING_ROOT) not in sys.path:
     sys.path.insert(0, str(TRAINING_ROOT))
 
 from trader_training.onnx_export import LinearHead, export_heads
+from trader_training.entry_feature_screen import _screen_edge_column
 from trader_training.io_utils import read_json, write_json
 from trader_training.labels import ENTRY_LABEL_METHOD, _path_stats_for_group, build_entry_labels
+from trader_training.ofi_feature_experiment import l1_snapshot_diff_ofi_quote
 from trader_training.promote import promote_artifact_bundle
 from trader_training.replay import build_splits
 from trader_training.schemas import FEATURE_ORDER, LATEST_STRESS_SPLIT, MODEL_OUTPUTS, OUTPUT_MAPPING, TRAINING_SPLITS, VALIDATION_LOCKED_SPLIT
@@ -33,6 +35,19 @@ class TrainingContractTest(unittest.TestCase):
             self.assertEqual(set(fields), set(OUTPUT_MAPPING[model_name]))
             self.assertEqual([f"prediction[{idx}]" for idx in range(len(fields))], [OUTPUT_MAPPING[model_name][field] for field in fields])
 
+    def test_entry_feature_screen_prefers_actual_plan_edge(self) -> None:
+        dataset = pd.DataFrame(
+            {
+                "long_expected_net_edge_bps": [20.0],
+                "short_expected_net_edge_bps": [15.0],
+                "long_actual_plan_net_edge_bps": [-3.0],
+                "short_actual_plan_net_edge_bps": [4.0],
+            }
+        )
+
+        self.assertEqual("long_actual_plan_net_edge_bps", _screen_edge_column(dataset, "LONG"))
+        self.assertEqual("short_actual_plan_net_edge_bps", _screen_edge_column(dataset, "SHORT"))
+
     def test_split_builder_uses_locked_validation_contract(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             data_root = Path(tmp)
@@ -90,7 +105,7 @@ class TrainingContractTest(unittest.TestCase):
         self.assertEqual(120_000, first["time_to_stop_ms"])
         self.assertAlmostEqual(-8.0, first["gross_edge_bps"])
 
-    def test_entry_label_uses_max_future_edge_not_fixed_target_hit(self) -> None:
+    def test_entry_label_uses_price_plan_outcome_not_max_future_edge(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             data_root = Path(tmp)
             run_root = data_root / "trader-v4" / "runs" / "unit-entry"
@@ -167,11 +182,200 @@ class TrainingContractTest(unittest.TestCase):
             labels = pd.read_parquet(run_root / "label" / "entry_labels.parquet")
             row = labels[labels["sample_id"].eq("s0") & labels["side"].eq("LONG")].iloc[0]
             self.assertEqual(0, row["target_hit"])
-            self.assertEqual(1, row["entry_target"])
+            self.assertEqual(0, row["entry_target"])
             self.assertEqual(ENTRY_LABEL_METHOD, row["label_method"])
-            self.assertAlmostEqual(13.5, row["expected_net_edge_bps"], places=6)
+            self.assertAlmostEqual(-6.5, row["expected_net_edge_bps"], places=6)
+            self.assertAlmostEqual(row["gross_edge_bps"] - row["cost_bps"], row["expected_net_edge_bps"], places=6)
             self.assertAlmostEqual(row["mfe_bps"] - row["cost_bps"], row["max_achievable_net_edge_bps"], places=6)
 
+    def test_entry_opportunity_label_keeps_plan_outcome_for_pm(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            data_root = Path(tmp)
+            run_root = data_root / "trader-v4" / "runs" / "unit-entry-opportunity"
+            feature_path = run_root / "feature" / "feature_frame.parquet"
+            replay_path = run_root / "replay" / "replay_1m.parquet"
+            plan_path = run_root / "label" / "price_plan_context.json"
+            config_path = data_root / "label_config.json"
+            feature_path.parent.mkdir(parents=True)
+            replay_path.parent.mkdir(parents=True)
+
+            times = pd.date_range("2026-01-01", periods=5, freq="min", tz="UTC")
+            pd.DataFrame(
+                {
+                    "sample_id": ["s0"],
+                    "symbol": "BTC-USDT-PERP",
+                    "event_time": [times[0]],
+                    "open_time_ms": [0],
+                    "split_id": "fit_inner",
+                    "walk_forward_fold": 0,
+                    "data_quality_flag": "OK",
+                    "spread_bps": 1.0,
+                    "spread_rank_24h_pct": 0.1,
+                    "realized_vol_15m_bps": 2.0,
+                }
+            ).to_parquet(feature_path, index=False)
+            pd.DataFrame(
+                {
+                    "event_time": times,
+                    "open_time_ms": np.arange(5, dtype=np.int64) * 60_000,
+                    "symbol": "BTC-USDT-PERP",
+                    "open": [100.0] * 5,
+                    "high": [100.0, 100.05, 100.19, 100.20, 100.0],
+                    "low": [100.0, 99.99, 99.98, 99.97, 100.0],
+                    "close": [100.0] * 5,
+                    "spread_bps": 1.0,
+                }
+            ).to_parquet(replay_path, index=False)
+            write_json(
+                config_path,
+                {
+                    "entry": {
+                        "max_hold_minutes": 3,
+                        "target_bps": 50.0,
+                        "stop_bps": 50.0,
+                        "min_expected_net_edge_bps": 3.0,
+                        "target_method": "OPPORTUNITY_MFE_V1",
+                    }
+                },
+            )
+            write_json(
+                plan_path,
+                {
+                    "pricePlanId": "unit-plan",
+                    "pricePlanConfigHash": "unit-hash",
+                    "targetDistanceBps": 50.0,
+                    "stopDistanceBps": 50.0,
+                    "maxHoldMinutes": 3,
+                    "costBps": 6.5,
+                    "entryLabelMethod": ENTRY_LABEL_METHOD,
+                    "entryTargetMethod": "OPPORTUNITY_MFE_V1",
+                },
+            )
+
+            build_entry_labels(
+                Namespace(
+                    data_root=data_root,
+                    run_id="unit-entry-opportunity",
+                    feature_path=feature_path,
+                    replay_path=replay_path,
+                    label_config_path=config_path,
+                    cost_config_path=None,
+                    price_plan_context_path=plan_path,
+                )
+            )
+
+            labels = pd.read_parquet(run_root / "label" / "entry_labels.parquet")
+            row = labels[labels["sample_id"].eq("s0") & labels["side"].eq("LONG")].iloc[0]
+            self.assertEqual(0, row["target_hit"])
+            self.assertEqual(1, row["entry_target"])
+            self.assertEqual("OPPORTUNITY_MFE_V1", row["label_method"])
+            self.assertAlmostEqual(row["mfe_bps"] - row["cost_bps"], row["expected_net_edge_bps"], places=6)
+            self.assertAlmostEqual(-6.5, row["gross_edge_bps"] - row["cost_bps"], places=6)
+
+    def test_entry_quality_label_rejects_untradable_opportunity(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            data_root = Path(tmp)
+            run_root = data_root / "trader-v4" / "runs" / "unit-entry-quality"
+            feature_path = run_root / "feature" / "feature_frame.parquet"
+            replay_path = run_root / "replay" / "replay_1m.parquet"
+            plan_path = run_root / "label" / "price_plan_context.json"
+            config_path = data_root / "label_config.json"
+            feature_path.parent.mkdir(parents=True)
+            replay_path.parent.mkdir(parents=True)
+
+            times = pd.date_range("2026-01-01", periods=5, freq="min", tz="UTC")
+            pd.DataFrame(
+                {
+                    "sample_id": ["s0"],
+                    "symbol": "BTC-USDT-PERP",
+                    "event_time": [times[0]],
+                    "open_time_ms": [0],
+                    "split_id": "fit_inner",
+                    "walk_forward_fold": 0,
+                    "data_quality_flag": "OK",
+                    "spread_bps": 1.0,
+                    "spread_rank_24h_pct": 0.1,
+                    "realized_vol_15m_bps": 2.0,
+                }
+            ).to_parquet(feature_path, index=False)
+            pd.DataFrame(
+                {
+                    "event_time": times,
+                    "open_time_ms": np.arange(5, dtype=np.int64) * 60_000,
+                    "symbol": "BTC-USDT-PERP",
+                    "open": [100.0] * 5,
+                    "high": [100.0, 100.05, 100.19, 100.20, 100.0],
+                    "low": [100.0, 99.99, 99.98, 99.97, 100.0],
+                    "close": [100.0] * 5,
+                    "spread_bps": 1.0,
+                }
+            ).to_parquet(replay_path, index=False)
+            write_json(
+                config_path,
+                {
+                    "entry": {
+                        "max_hold_minutes": 3,
+                        "target_bps": 50.0,
+                        "stop_bps": 50.0,
+                        "min_expected_net_edge_bps": 3.0,
+                        "min_plan_net_edge_bps": 0.0,
+                        "max_entry_mae_bps": 12.0,
+                        "target_method": "OPPORTUNITY_QUALITY_V1",
+                    }
+                },
+            )
+            write_json(
+                plan_path,
+                {
+                    "pricePlanId": "unit-plan",
+                    "pricePlanConfigHash": "unit-hash",
+                    "targetDistanceBps": 50.0,
+                    "stopDistanceBps": 50.0,
+                    "maxHoldMinutes": 3,
+                    "costBps": 6.5,
+                    "entryLabelMethod": ENTRY_LABEL_METHOD,
+                    "entryTargetMethod": "OPPORTUNITY_QUALITY_V1",
+                },
+            )
+
+            build_entry_labels(
+                Namespace(
+                    data_root=data_root,
+                    run_id="unit-entry-quality",
+                    feature_path=feature_path,
+                    replay_path=replay_path,
+                    label_config_path=config_path,
+                    cost_config_path=None,
+                    price_plan_context_path=plan_path,
+                )
+            )
+
+            labels = pd.read_parquet(run_root / "label" / "entry_labels.parquet")
+            row = labels[labels["sample_id"].eq("s0") & labels["side"].eq("LONG")].iloc[0]
+            self.assertEqual("OPPORTUNITY_QUALITY_V1", row["label_method"])
+            self.assertGreater(row["expected_net_edge_bps"], 3.0)
+            self.assertLess(row["actual_plan_net_edge_bps"], 0.0)
+            self.assertEqual(0, row["entry_target"])
+
+    def test_l1_snapshot_diff_ofi_uses_quote_notional_and_signed_ask_side(self) -> None:
+        bid_part, ask_part = l1_snapshot_diff_ofi_quote(
+            pd.Series([101.0, 101.0, 100.5]),
+            pd.Series([2.0, 3.0, 4.0]),
+            pd.Series([102.0, 101.5, 102.5]),
+            pd.Series([5.0, 6.0, 7.0]),
+            pd.Series([100.0, 101.0, 101.0]),
+            pd.Series([1.5, 2.0, 3.0]),
+            pd.Series([102.0, 102.0, 101.5]),
+            pd.Series([4.0, 5.0, 6.0]),
+        )
+
+        self.assertAlmostEqual(202.0, bid_part.iloc[0])
+        self.assertAlmostEqual(-102.0, ask_part.iloc[0])
+        self.assertAlmostEqual(101.0, bid_part.iloc[1])
+        self.assertAlmostEqual(-609.0, ask_part.iloc[1])
+        self.assertAlmostEqual(-303.0, bid_part.iloc[2])
+        self.assertAlmostEqual(609.0, ask_part.iloc[2])
+
     def test_exported_onnx_accepts_java_feature_shape(self) -> None:
         import onnxruntime as ort
 

training/trader_training/datasets.py

@@ -83,11 +83,25 @@ def build_train_datasets(args: Any) -> None:
 
 def _entry_pivot(entry: pd.DataFrame) -> pd.DataFrame:
     require_columns(entry, ("sample_id", "side", "entry_target", "expected_net_edge_bps"), "entry_labels")
-    long = entry[entry["side"] == "LONG"][["sample_id", "entry_target", "expected_net_edge_bps"]].rename(
-        columns={"entry_target": "long_entry_target", "expected_net_edge_bps": "long_expected_net_edge_bps"}
+    optional_columns = [column for column in ("actual_plan_net_edge_bps", "max_achievable_net_edge_bps", "mae_bps") if column in entry.columns]
+    selected_columns = ["sample_id", "entry_target", "expected_net_edge_bps", *optional_columns]
+    long = entry[entry["side"] == "LONG"][selected_columns].rename(
+        columns={
+            "entry_target": "long_entry_target",
+            "expected_net_edge_bps": "long_expected_net_edge_bps",
+            "actual_plan_net_edge_bps": "long_actual_plan_net_edge_bps",
+            "max_achievable_net_edge_bps": "long_max_achievable_net_edge_bps",
+            "mae_bps": "long_mae_bps",
+        }
     )
-    short = entry[entry["side"] == "SHORT"][["sample_id", "entry_target", "expected_net_edge_bps"]].rename(
-        columns={"entry_target": "short_entry_target", "expected_net_edge_bps": "short_expected_net_edge_bps"}
+    short = entry[entry["side"] == "SHORT"][selected_columns].rename(
+        columns={
+            "entry_target": "short_entry_target",
+            "expected_net_edge_bps": "short_expected_net_edge_bps",
+            "actual_plan_net_edge_bps": "short_actual_plan_net_edge_bps",
+            "max_achievable_net_edge_bps": "short_max_achievable_net_edge_bps",
+            "mae_bps": "short_mae_bps",
+        }
     )
     return long.merge(short, on="sample_id", how="inner")
 

training/trader_training/diagnostics.py

@@ -6,8 +6,8 @@ from typing import Any
 import numpy as np
 import pandas as pd
 
-from trader_training.io_utils import read_parquet, run_root, write_json, write_text
-from trader_training.pm import _pm_frame, _simulate_open_trades, _threshold_candidates, _trade_metrics
+from trader_training.io_utils import read_json, read_parquet, run_root, write_json, write_text
+from trader_training.pm import _pm_frame, _price_plan_context, _simulate_open_trades, _threshold_candidates, _thresholds_from_config, _trade_metrics, default_pm_config
 from trader_training.schemas import FIT_SPLIT, LATEST_STRESS_SPLIT, TUNE_SPLIT, VALIDATION_LOCKED_SPLIT
 
 
@@ -66,12 +66,19 @@ def _label_summary(root) -> dict[str, Any]:
 
 def _pm_summary(root) -> dict[str, Any]:
     summary: dict[str, Any] = {}
+    config_path = root / "pm-search" / "position_manager_config.json"
+    config = read_json(config_path)["config"] if config_path.is_file() else default_pm_config()
+    thresholds = _thresholds_from_config(config)
+    price_plan = _price_plan_context(root)
     for split_id in PM_EVAL_SPLITS:
         frame = _pm_frame(root, split_id)
+        selected_trades = _simulate_open_trades(frame, thresholds, config, price_plan)
         item = {
             "rows": len(frame),
             "score_distribution": _score_distribution(frame),
-            "gate_funnel": _gate_funnel(frame),
+            "active_thresholds": thresholds,
+            "gate_funnel": _gate_funnel(frame, thresholds),
+            "selected_trade_metrics": _trade_metrics(selected_trades),
             "relaxed_variants": _relaxed_variants(frame),
             "top_bucket_edge": _top_bucket_edge(frame),
             "grid_search_any_trade": _grid_trade_summary(frame),
@@ -89,34 +96,29 @@ def _score_distribution(frame: pd.DataFrame) -> dict[str, dict[str, float]]:
         "market_risk_prob",
         "pred_long_expected_net_edge_bps",
         "pred_short_expected_net_edge_bps",
+        "model_pred_long_expected_net_edge_bps",
+        "model_pred_short_expected_net_edge_bps",
         "actual_long_expected_net_edge_bps",
         "actual_short_expected_net_edge_bps",
     ]
-    return {column: _quantiles(frame[column], (0.0, 0.05, 0.5, 0.95, 1.0)) for column in columns}
+    return {column: _quantiles(frame[column], (0.0, 0.05, 0.5, 0.95, 1.0)) for column in columns if column in frame.columns}
 
 
-def _gate_funnel(frame: pd.DataFrame) -> dict[str, Any]:
-    thresholds = {
-        "long_open_prob": 0.54,
-        "short_open_prob": 0.54,
-        "min_entry_prob": 0.50,
-        "max_market_risk_prob": 0.55,
-        "min_expected_edge_bps": 1.0,
-        "min_direction_margin": 0.02,
-    }
+def _gate_funnel(frame: pd.DataFrame, thresholds: dict[str, float]) -> dict[str, Any]:
+    direction_margin = (frame["long_prob"] - frame["short_prob"]).abs()
     long_steps = {
-        "long_prob >= 0.54": frame["long_prob"] >= thresholds["long_open_prob"],
-        "long_prob - short_prob >= 0.02": (frame["long_prob"] - frame["short_prob"]) >= thresholds["min_direction_margin"],
-        "long_entry_prob >= 0.50": frame["long_entry_prob"] >= thresholds["min_entry_prob"],
-        "market_risk_prob <= 0.55": frame["market_risk_prob"] <= thresholds["max_market_risk_prob"],
-        "pred_long_expected_net_edge_bps >= 1.0": frame["pred_long_expected_net_edge_bps"] >= thresholds["min_expected_edge_bps"],
+        f"long_prob > {thresholds['long_open_prob']}": frame["long_prob"] > thresholds["long_open_prob"],
+        f"abs(long_prob - short_prob) > {thresholds['min_direction_margin']}": direction_margin > thresholds["min_direction_margin"],
+        f"long_entry_prob > {thresholds['min_entry_prob']}": frame["long_entry_prob"] > thresholds["min_entry_prob"],
+        f"market_risk_prob < {thresholds['max_market_risk_prob']}": frame["market_risk_prob"] < thresholds["max_market_risk_prob"],
+        f"pred_long_expected_net_edge_bps > {thresholds['min_expected_edge_bps']}": frame["pred_long_expected_net_edge_bps"] > thresholds["min_expected_edge_bps"],
     }
     short_steps = {
-        "short_prob >= 0.54": frame["short_prob"] >= thresholds["short_open_prob"],
-        "short_prob - long_prob >= 0.02": (frame["short_prob"] - frame["long_prob"]) >= thresholds["min_direction_margin"],
-        "short_entry_prob >= 0.50": frame["short_entry_prob"] >= thresholds["min_entry_prob"],
-        "market_risk_prob <= 0.55": frame["market_risk_prob"] <= thresholds["max_market_risk_prob"],
-        "pred_short_expected_net_edge_bps >= 1.0": frame["pred_short_expected_net_edge_bps"] >= thresholds["min_expected_edge_bps"],
+        f"short_prob > {thresholds['short_open_prob']}": frame["short_prob"] > thresholds["short_open_prob"],
+        f"abs(long_prob - short_prob) > {thresholds['min_direction_margin']}": direction_margin > thresholds["min_direction_margin"],
+        f"short_entry_prob > {thresholds['min_entry_prob']}": frame["short_entry_prob"] > thresholds["min_entry_prob"],
+        f"market_risk_prob < {thresholds['max_market_risk_prob']}": frame["market_risk_prob"] < thresholds["max_market_risk_prob"],
+        f"pred_short_expected_net_edge_bps > {thresholds['min_expected_edge_bps']}": frame["pred_short_expected_net_edge_bps"] > thresholds["min_expected_edge_bps"],
     }
     return {
         "thresholds": thresholds,
@@ -140,6 +142,7 @@ def _cumulative_gate_counts(steps: dict[str, pd.Series], total_rows: int) -> dic
 def _relaxed_variants(frame: pd.DataFrame) -> dict[str, Any]:
     variants = {
         "no_risk_no_edge": {"prob": 0.54, "entry": 0.50, "margin": 0.02, "risk": 1.0, "edge": -99.0},
+        "rare_entry_low_prob": {"prob": 0.50, "entry": 0.03, "margin": 0.02, "risk": 0.98, "edge": 0.0},
         "entry_only_55": {"prob": 0.0, "entry": 0.55, "margin": -99.0, "risk": 1.0, "edge": -99.0},
         "direction_only_54": {"prob": 0.54, "entry": 0.0, "margin": 0.02, "risk": 1.0, "edge": -99.0},
         "very_loose": {"prob": 0.50, "entry": 0.45, "margin": 0.0, "risk": 1.0, "edge": -99.0},
@@ -168,10 +171,10 @@ def _variant_trades(frame: pd.DataFrame, thresholds: dict[str, float]) -> pd.Dat
     )
     long = frame.loc[long_mask].copy()
     long["side"] = "LONG"
-    long["actual_edge_bps"] = long["actual_long_expected_net_edge_bps"]
+    long["actual_edge_bps"] = long["long_trade_net_edge_bps"]
     short = frame.loc[short_mask].copy()
     short["side"] = "SHORT"
-    short["actual_edge_bps"] = short["actual_short_expected_net_edge_bps"]
+    short["actual_edge_bps"] = short["short_trade_net_edge_bps"]
     return pd.concat([long, short], ignore_index=True)
 
 
@@ -189,7 +192,7 @@ def _plain_trade_metrics(trades: pd.DataFrame) -> dict[str, Any]:
 def _top_bucket_edge(frame: pd.DataFrame) -> dict[str, Any]:
     side = np.where(frame["long_prob"] >= frame["short_prob"], "LONG", "SHORT")
     side_prob = np.where(side == "LONG", frame["long_prob"], frame["short_prob"])
-    side_edge = np.where(side == "LONG", frame["actual_long_expected_net_edge_bps"], frame["actual_short_expected_net_edge_bps"])
+    side_edge = np.where(side == "LONG", frame["long_trade_net_edge_bps"], frame["short_trade_net_edge_bps"])
     direction_frame = pd.DataFrame({"score": side_prob, "actual_edge_bps": side_edge, "side": side})
     direction_top = {}
     for fraction in (0.01, 0.02, 0.05, 0.10):
@@ -243,27 +246,19 @@ def _grid_trade_summary(frame: pd.DataFrame) -> dict[str, Any]:
 
 
 def _diagnostic_conclusion(pm_summary: dict[str, Any]) -> dict[str, Any]:
-    tune = pm_summary.get(TUNE_SPLIT, {})
-    gate = tune.get("gate_funnel", {})
-    long_single = gate.get("long", {}).get("single_gate_pass", {})
-    short_single = gate.get("short", {}).get("single_gate_pass", {})
-    pred_edge_blocked = (
-        long_single.get("pred_long_expected_net_edge_bps >= 1.0", 0) == 0
-        and short_single.get("pred_short_expected_net_edge_bps >= 1.0", 0) == 0
-    )
-    relaxed = tune.get("relaxed_variants", {})
-    any_relaxed_positive = any(item.get("avg_actual_edge_bps", 0.0) > 0 for item in relaxed.values())
-    if pred_edge_blocked and not any_relaxed_positive:
+    validation = pm_summary.get(VALIDATION_LOCKED_SPLIT, {}).get("selected_trade_metrics", {})
+    stress = pm_summary.get(LATEST_STRESS_SPLIT, {}).get("selected_trade_metrics", {})
+    if validation.get("trade_count", 0) == 0:
         return {
-            "status": "MODEL_SIGNAL_NOT_TRADABLE",
-            "plain_reason": "Entry 预测的净收益基本都是负数；即使放松风险和收益门槛，选出来的样本平均仍亏。",
-            "next_action": "优先重查 Entry 标签和价格计划，再考虑更强模型；不要直接放松 PM 阈值上线。",
+            "status": "NO_VALIDATION_TRADE",
+            "plain_reason": "当前 PM 阈值在验证集没有选出交易，主要要看挡单漏斗。",
+            "next_action": "先看 Direction、Risk、Entry 哪个门槛挡住，再做阈值实验。",
         }
-    if pred_edge_blocked:
+    if validation.get("avg_weighted_edge_bps", 0.0) <= 0 and stress.get("avg_weighted_edge_bps", 0.0) <= 0:
         return {
-            "status": "ENTRY_EDGE_GATE_BLOCKED",
-            "plain_reason": "PM 没有交易主要是 Entry 预测净收益过低。",
-            "next_action": "重训 Entry 或调整价格计划后再搜索 PM 阈值。",
+            "status": "PRICE_PLAN_OR_ENTRY_NOT_TRADABLE",
+            "plain_reason": "按固定止盈止损真实收益算，验证集和压力集选出来的交易平均都不赚钱。",
+            "next_action": "优先重新搜索价格计划，再重建 Entry 标签和模型；不要只放松 PM 阈值。",
         }
     return {
         "status": "NEEDS_MANUAL_REVIEW",
@@ -312,6 +307,8 @@ def _markdown_report(payload: dict[str, Any]) -> str:
         lines.append(f"### {split_id}")
         lines.append("")
         lines.append(f"- 样本数: {item['rows']}")
+        lines.append(f"- 当前阈值: `{item['active_thresholds']}`")
+        lines.append(f"- 当前阈值选中交易: `{item['selected_trade_metrics']}`")
         lines.append(f"- 网格里有交易的候选数: {item['grid_search_any_trade']['candidates_with_trade']} / {item['grid_search_any_trade']['candidate_count']}")
         lines.append("")
         for side in ("long", "short"):

training/trader_training/entry_feature_screen.py

@@ -24,9 +24,11 @@ def screen_entry_features(args: Any) -> None:
 
     min_bucket_rows = int(args.min_bucket_rows or 300)
     rows: list[dict[str, Any]] = []
+    edge_source_by_side: dict[str, str] = {}
     for side in ("LONG", "SHORT"):
         target_col = "long_entry_target" if side == "LONG" else "short_entry_target"
-        edge_col = "long_expected_net_edge_bps" if side == "LONG" else "short_expected_net_edge_bps"
+        edge_col = _screen_edge_column(dataset, side)
+        edge_source_by_side[side] = edge_col
         baselines = _split_baselines(dataset, target_col, edge_col)
         for feature in FEATURE_ORDER:
             rows.extend(_feature_rows(dataset, feature, side, target_col, edge_col, baselines))
@@ -43,6 +45,7 @@ def screen_entry_features(args: Any) -> None:
         "bucket_metric_count": int(len(bucket_metrics)),
         "candidate_count": int(len(candidates)),
         "min_bucket_rows": min_bucket_rows,
+        "edge_source_by_side": edge_source_by_side,
         "selection_rule": "bucket boundaries are learned on fit_inner; candidate is picked by tune_inner and checked on validation_locked/latest_stress",
     }
     write_json(root / "diagnostics" / "entry_feature_screen_result.json", result)
@@ -59,6 +62,14 @@ def screen_entry_features(args: Any) -> None:
     )
 
 
+def _screen_edge_column(dataset: pd.DataFrame, side: str) -> str:
+    prefix = "long" if side == "LONG" else "short"
+    actual_col = f"{prefix}_actual_plan_net_edge_bps"
+    if actual_col in dataset.columns:
+        return actual_col
+    return f"{prefix}_expected_net_edge_bps"
+
+
 def _split_baselines(dataset: pd.DataFrame, target_col: str, edge_col: str) -> dict[str, dict[str, float]]:
     baselines: dict[str, dict[str, float]] = {}
     for split_id in ALL_SPLITS:
@@ -225,6 +236,7 @@ def _markdown_report(result: dict[str, Any], candidates: pd.DataFrame) -> str:
         "",
         "这份报告只回答一个问题：历史数据里，单个特征的某些区间有没有稳定变好。",
         "",
+        "- 如果数据里有真实出场净收益，本报告用真实出场净收益；没有时才退回训练收益标签。",
         "- `tune_inner` 用来挑候选区间。",
         "- `validation_locked` 和 `latest_stress` 用来检查这个区间是不是出了训练样本也还能站住。",
         "- `stable_positive_edge=true` 代表这个区间在三个检查集里的平均净收益都大于 0。",
@@ -237,6 +249,8 @@ def _markdown_report(result: dict[str, Any], candidates: pd.DataFrame) -> str:
         f"- 分桶明细数: `{result['bucket_metric_count']}`",
         f"- 候选数: `{result['candidate_count']}`",
         f"- 最小分桶行数: `{result['min_bucket_rows']}`",
+        f"- 做多收益来源: `{result['edge_source_by_side'].get('LONG')}`",
+        f"- 做空收益来源: `{result['edge_source_by_side'].get('SHORT')}`",
         "",
     ]
     if candidates.empty:

training/trader_training/labels.py

@@ -24,7 +24,21 @@ from trader_training.schemas import LABEL_VERSION
 
 DEFAULT_LABEL_CONFIG = {
     "direction": {"horizon_minutes": 45, "long_threshold_bps": 5.0, "short_threshold_bps": -5.0},
-    "entry": {"max_hold_minutes": 45, "target_bps": 12.0, "stop_bps": 8.0, "min_expected_net_edge_bps": 3.0},
+    "entry": {
+        "max_hold_minutes": 45,
+        "target_bps": 12.0,
+        "stop_bps": 8.0,
+        "min_expected_net_edge_bps": 3.0,
+        "plan_method": "FIXED_TARGET_STOP_V1",
+        "target_method": "PRICE_PLAN_OUTCOME_V1",
+        "min_plan_net_edge_bps": 0.0,
+        "max_entry_mae_bps": 12.0,
+        "partial_take_1_ratio": 0.50,
+        "partial_take_2_ratio": 0.25,
+        "second_target_bps": 24.0,
+        "trailing_stop_bps": 10.0,
+        "breakeven_after_first_target": True,
+    },
     "continue": {"horizon_minutes": 45, "min_expected_continue_edge_bps": 5.0},
     "exit": {"horizon_minutes": 45, "adverse_move_bps": 20.0, "stagnation_abs_return_bps": 5.0},
     "risk": {
@@ -92,6 +106,7 @@ PATH_STAT_COLUMNS = [
     "ambiguous_hit",
     "time_to_target_ms",
     "time_to_stop_ms",
+    "time_to_exit_ms",
     "gross_edge_bps",
     "future_return_bps",
     "mfe_bps",
@@ -112,7 +127,14 @@ def _first_hit_index(hit_window: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
     return hit_any, first_idx
 
 
-def _path_stats_for_group(group: pd.DataFrame, side: str, horizon: int, target_bps: float, stop_bps: float) -> pd.DataFrame:
+def _path_stats_for_group(
+    group: pd.DataFrame,
+    side: str,
+    horizon: int,
+    target_bps: float,
+    stop_bps: float,
+    plan_config: dict[str, Any] | None = None,
+) -> pd.DataFrame:
     if len(group) <= horizon:
         return _empty_path_stats_frame()
 
@@ -151,6 +173,29 @@ def _path_stats_for_group(group: pd.DataFrame, side: str, horizon: int, target_b
     else:
         realized_vol_bps = np.full(len(entry), np.nan)
 
+    method = str((plan_config or {}).get("plan_method", "FIXED_TARGET_STOP_V1"))
+    if method == "DYNAMIC_TRAILING_V1":
+        return _dynamic_path_stats_for_group(
+            grouped,
+            side,
+            horizon,
+            target_bps,
+            stop_bps,
+            close,
+            high,
+            low,
+            spread,
+            open_ms,
+            valid,
+            future_high,
+            future_low,
+            future_spread,
+            future_realized_vol_bps=realized_vol_bps,
+            plan_config=plan_config or {},
+        )
+    if method != "FIXED_TARGET_STOP_V1":
+        raise ValueError(f"unsupported entry plan_method: {method}")
+
     if side == "LONG":
         target_price = entry * (1.0 + target_bps / 10000.0)
         stop_price = entry * (1.0 - stop_bps / 10000.0)
@@ -175,6 +220,11 @@ def _path_stats_for_group(group: pd.DataFrame, side: str, horizon: int, target_b
     stop_hit = stop_any & (first_stop_idx <= first_target_idx)
     timeout_hit = ~(target_hit | stop_hit)
     gross_edge_bps = np.where(target_hit, target_bps, np.where(stop_hit, -stop_bps, future_return_bps))
+    time_to_exit_ms = np.where(
+        target_hit,
+        (first_target_idx + 1) * 60_000,
+        np.where(stop_hit, (first_stop_idx + 1) * 60_000, horizon * 60_000),
+    )
 
     out = pd.DataFrame(
         {
@@ -187,6 +237,7 @@ def _path_stats_for_group(group: pd.DataFrame, side: str, horizon: int, target_b
             "ambiguous_hit": ambiguous_hit.astype("int8"),
             "time_to_target_ms": np.where(target_hit, (first_target_idx + 1) * 60_000, -1).astype("int64"),
             "time_to_stop_ms": np.where(stop_hit, (first_stop_idx + 1) * 60_000, -1).astype("int64"),
+            "time_to_exit_ms": time_to_exit_ms.astype("int64"),
             "gross_edge_bps": gross_edge_bps.astype("float64"),
             "future_return_bps": future_return_bps.astype("float64"),
             "mfe_bps": mfe_bps.astype("float64"),
@@ -198,27 +249,184 @@ def _path_stats_for_group(group: pd.DataFrame, side: str, horizon: int, target_b
     return out.loc[valid, PATH_STAT_COLUMNS].reset_index(drop=True)
 
 
-def _build_path_stats(replay: pd.DataFrame, horizon: int, target_bps: float, stop_bps: float) -> pd.DataFrame:
+def _dynamic_path_stats_for_group(
+    grouped: pd.DataFrame,
+    side: str,
+    horizon: int,
+    target_bps: float,
+    stop_bps: float,
+    close: np.ndarray,
+    high: np.ndarray,
+    low: np.ndarray,
+    spread: np.ndarray,
+    open_ms: np.ndarray,
+    valid: np.ndarray,
+    future_high: np.ndarray,
+    future_low: np.ndarray,
+    future_spread: np.ndarray,
+    future_realized_vol_bps: np.ndarray,
+    plan_config: dict[str, Any],
+) -> pd.DataFrame:
+    entry = close[:-horizon]
+    exit_price = close[horizon:]
+    current_open_ms = open_ms[:-horizon]
+    future_close = sliding_window_view(close[1:], horizon)
+    with np.errstate(all="ignore"):
+        high_max = np.nanmax(future_high, axis=1)
+        low_min = np.nanmin(future_low, axis=1)
+        spread_p80 = np.nanquantile(future_spread, 0.8, axis=1)
+
+    take1_ratio = float(plan_config.get("partial_take_1_ratio", 0.50))
+    take2_ratio = float(plan_config.get("partial_take_2_ratio", 0.25))
+    take1_ratio = float(np.clip(take1_ratio, 0.0, 1.0))
+    take2_ratio = float(np.clip(take2_ratio, 0.0, max(0.0, 1.0 - take1_ratio)))
+    target2_bps = float(plan_config.get("second_target_bps", target_bps * 2.0))
+    trailing_stop_bps = float(plan_config.get("trailing_stop_bps", stop_bps))
+    breakeven_after_first = bool(plan_config.get("breakeven_after_first_target", True))
+
+    n = len(entry)
+    active = np.ones(n, dtype=bool)
+    first_target_done = np.zeros(n, dtype=bool)
+    second_target_done = np.zeros(n, dtype=bool)
+    bad_stop_done = np.zeros(n, dtype=bool)
+    trailing_exit_done = np.zeros(n, dtype=bool)
+    remaining = np.ones(n, dtype="float64")
+    gross = np.zeros(n, dtype="float64")
+    first_target_idx = np.full(n, horizon + 1, dtype="int64")
+    stop_idx = np.full(n, horizon + 1, dtype="int64")
+    exit_idx = np.full(n, horizon, dtype="int64")
+
+    if side == "LONG":
+        high_water = entry.copy()
+        for step in range(horizon):
+            h = future_high[:, step]
+            l = future_low[:, step]
+            prior_high_water = high_water.copy()
+            trailing_stop_price = prior_high_water * (1.0 - trailing_stop_bps / 10000.0)
+            if breakeven_after_first:
+                trailing_stop_price = np.maximum(trailing_stop_price, entry)
+            stop_price = np.where(first_target_done, trailing_stop_price, entry * (1.0 - stop_bps / 10000.0))
+            stop_now = active & (l <= stop_price)
+            stop_gross = (stop_price / entry - 1.0) * 10000.0
+            gross = np.where(stop_now, gross + remaining * stop_gross, gross)
+            trailing_exit_done |= stop_now & first_target_done
+            bad_stop_done |= stop_now & (~first_target_done)
+            stop_idx = np.where(stop_now, step, stop_idx)
+            exit_idx = np.where(stop_now, step, exit_idx)
+            remaining = np.where(stop_now, 0.0, remaining)
+            active &= ~stop_now
+
+            first_now = active & (~first_target_done) & (h >= entry * (1.0 + target_bps / 10000.0))
+            gross = np.where(first_now, gross + take1_ratio * target_bps, gross)
+            remaining = np.where(first_now, remaining - take1_ratio, remaining)
+            first_target_done |= first_now
+            first_target_idx = np.where(first_now, step, first_target_idx)
+
+            second_now = active & first_target_done & (~second_target_done) & (h >= entry * (1.0 + target2_bps / 10000.0))
+            gross = np.where(second_now, gross + take2_ratio * target2_bps, gross)
+            remaining = np.where(second_now, remaining - take2_ratio, remaining)
+            second_target_done |= second_now
+
+            high_water = np.maximum(high_water, h)
+        timeout_return = (exit_price / entry - 1.0) * 10000.0
+        future_return_bps = timeout_return
+        mfe_bps = np.maximum((high_max / entry - 1.0) * 10000.0, 0.0)
+        mae_bps = np.maximum((entry / low_min - 1.0) * 10000.0, 0.0)
+    else:
+        low_water = entry.copy()
+        for step in range(horizon):
+            h = future_high[:, step]
+            l = future_low[:, step]
+            prior_low_water = low_water.copy()
+            trailing_stop_price = prior_low_water * (1.0 + trailing_stop_bps / 10000.0)
+            if breakeven_after_first:
+                trailing_stop_price = np.minimum(trailing_stop_price, entry)
+            stop_price = np.where(first_target_done, trailing_stop_price, entry * (1.0 + stop_bps / 10000.0))
+            stop_now = active & (h >= stop_price)
+            stop_gross = (entry / stop_price - 1.0) * 10000.0
+            gross = np.where(stop_now, gross + remaining * stop_gross, gross)
+            trailing_exit_done |= stop_now & first_target_done
+            bad_stop_done |= stop_now & (~first_target_done)
+            stop_idx = np.where(stop_now, step, stop_idx)
+            exit_idx = np.where(stop_now, step, exit_idx)
+            remaining = np.where(stop_now, 0.0, remaining)
+            active &= ~stop_now
+
+            first_now = active & (~first_target_done) & (l <= entry * (1.0 - target_bps / 10000.0))
+            gross = np.where(first_now, gross + take1_ratio * target_bps, gross)
+            remaining = np.where(first_now, remaining - take1_ratio, remaining)
+            first_target_done |= first_now
+            first_target_idx = np.where(first_now, step, first_target_idx)
+
+            second_now = active & first_target_done & (~second_target_done) & (l <= entry * (1.0 - target2_bps / 10000.0))
+            gross = np.where(second_now, gross + take2_ratio * target2_bps, gross)
+            remaining = np.where(second_now, remaining - take2_ratio, remaining)
+            second_target_done |= second_now
+
+            low_water = np.minimum(low_water, l)
+        timeout_return = (entry / exit_price - 1.0) * 10000.0
+        future_return_bps = timeout_return
+        mfe_bps = np.maximum((entry / low_min - 1.0) * 10000.0, 0.0)
+        mae_bps = np.maximum((high_max / entry - 1.0) * 10000.0, 0.0)
+
+    timeout_now = active
+    gross = np.where(timeout_now, gross + remaining * timeout_return, gross)
+    exit_idx = np.where(timeout_now, horizon - 1, exit_idx)
+    target_hit = first_target_done
+    stop_hit = bad_stop_done
+    timeout_hit = timeout_now
+    ambiguous_hit = np.zeros(n, dtype=bool)
+    out = pd.DataFrame(
+        {
+            "symbol": grouped["symbol"].iloc[0],
+            "open_time_ms": current_open_ms,
+            "side": side,
+            "target_hit": target_hit.astype("int8"),
+            "stop_hit": stop_hit.astype("int8"),
+            "timeout_hit": timeout_hit.astype("int8"),
+            "ambiguous_hit": ambiguous_hit.astype("int8"),
+            "time_to_target_ms": np.where(target_hit, (first_target_idx + 1) * 60_000, -1).astype("int64"),
+            "time_to_stop_ms": np.where(stop_hit | trailing_exit_done, (stop_idx + 1) * 60_000, -1).astype("int64"),
+            "time_to_exit_ms": ((exit_idx + 1) * 60_000).astype("int64"),
+            "gross_edge_bps": gross.astype("float64"),
+            "future_return_bps": future_return_bps.astype("float64"),
+            "mfe_bps": mfe_bps.astype("float64"),
+            "mae_bps": mae_bps.astype("float64"),
+            "future_spread_p80": spread_p80.astype("float64"),
+            "future_realized_vol_bps": future_realized_vol_bps.astype("float64"),
+        }
+    )
+    return out.loc[valid, PATH_STAT_COLUMNS].reset_index(drop=True)
+
+
+def _build_path_stats(replay: pd.DataFrame, horizon: int, target_bps: float, stop_bps: float, plan_config: dict[str, Any] | None = None) -> pd.DataFrame:
     frames: list[pd.DataFrame] = []
     for symbol, group in replay.groupby("symbol", sort=False, observed=False):
         logging.info(
-            "trader.training.path_stats_group_start symbol=%s horizonMinutes=%s rowCount=%s",
+            "trader.training.path_stats_group_start symbol=%s horizonMinutes=%s planMethod=%s rowCount=%s",
             symbol,
             horizon,
+            (plan_config or {}).get("plan_method", "FIXED_TARGET_STOP_V1"),
             len(group),
         )
         for side in ("LONG", "SHORT"):
-            stats = _path_stats_for_group(group, side, horizon, target_bps, stop_bps)
+            stats = _path_stats_for_group(group, side, horizon, target_bps, stop_bps, plan_config=plan_config)
             frames.append(stats)
             logging.info(
-                "trader.training.path_stats_side_done symbol=%s side=%s horizonMinutes=%s rowCount=%s",
+                "trader.training.path_stats_side_done symbol=%s side=%s horizonMinutes=%s planMethod=%s rowCount=%s",
                 symbol,
                 side,
                 horizon,
+                (plan_config or {}).get("plan_method", "FIXED_TARGET_STOP_V1"),
                 len(stats),
             )
     out = pd.concat(frames, ignore_index=True) if frames else _empty_path_stats_frame()
-    logging.info("trader.training.path_stats_built horizonMinutes=%s rowCount=%s", horizon, len(out))
+    logging.info(
+        "trader.training.path_stats_built horizonMinutes=%s planMethod=%s rowCount=%s",
+        horizon,
+        (plan_config or {}).get("plan_method", "FIXED_TARGET_STOP_V1"),
+        len(out),
+    )
     return out
 
 
@@ -235,8 +443,17 @@ def write_price_plan_context(args: Any) -> None:
         "targetDistanceBps": float(entry["target_bps"]),
         "maxHoldMinutes": int(entry["max_hold_minutes"]),
         "minExpectedNetEdgeBps": float(entry["min_expected_net_edge_bps"]),
+        "minPlanNetEdgeBps": float(entry.get("min_plan_net_edge_bps", 0.0)),
+        "maxEntryMaeBps": float(entry.get("max_entry_mae_bps", entry["stop_bps"])),
         "costBps": cost_bps,
         "entryLabelMethod": ENTRY_LABEL_METHOD,
+        "entryTargetMethod": str(entry.get("target_method", ENTRY_LABEL_METHOD)),
+        "entryPlanMethod": str(entry.get("plan_method", "FIXED_TARGET_STOP_V1")),
+        "partialTake1Ratio": float(entry.get("partial_take_1_ratio", 0.50)),
+        "partialTake2Ratio": float(entry.get("partial_take_2_ratio", 0.25)),
+        "secondTargetBps": float(entry.get("second_target_bps", float(entry["target_bps"]) * 2.0)),
+        "trailingStopBps": float(entry.get("trailing_stop_bps", float(entry["stop_bps"]))),
+        "breakevenAfterFirstTarget": bool(entry.get("breakeven_after_first_target", True)),
     }
     path = root / "label" / "price_plan_context.json"
     write_json(path, context)
@@ -247,8 +464,17 @@ def write_price_plan_context(args: Any) -> None:
         "stop_bps": context["stopDistanceBps"],
         "max_hold_minutes": context["maxHoldMinutes"],
         "min_expected_net_edge_bps": context["minExpectedNetEdgeBps"],
+        "min_plan_net_edge_bps": context["minPlanNetEdgeBps"],
+        "max_entry_mae_bps": context["maxEntryMaeBps"],
         "cost_bps": context["costBps"],
         "entry_label_method": context["entryLabelMethod"],
+        "entry_target_method": context["entryTargetMethod"],
+        "entry_plan_method": context["entryPlanMethod"],
+        "partial_take_1_ratio": context["partialTake1Ratio"],
+        "partial_take_2_ratio": context["partialTake2Ratio"],
+        "second_target_bps": context["secondTargetBps"],
+        "trailing_stop_bps": context["trailingStopBps"],
+        "breakeven_after_first_target": context["breakevenAfterFirstTarget"],
     }])
     write_parquet(root / "label" / "price_plan_context.parquet", frame)
     logging.info("trader.training.price_plan_written runId=%s path=%s", args.run_id, path)
@@ -308,6 +534,7 @@ def build_entry_labels(args: Any) -> None:
         int(entry_conf["max_hold_minutes"]),
         float(entry_conf["target_bps"]),
         float(entry_conf["stop_bps"]),
+        plan_config=entry_conf,
     )
     feature_columns = [
         "sample_id",
@@ -321,14 +548,28 @@ def build_entry_labels(args: Any) -> None:
         "realized_vol_15m_bps",
     ]
     merged = features[feature_columns].merge(stats, on=["symbol", "open_time_ms"], how="inner")
+    merged["actual_plan_net_edge_bps"] = merged["gross_edge_bps"] - cost_bps
     merged["max_achievable_gross_edge_bps"] = merged["mfe_bps"]
     merged["max_achievable_net_edge_bps"] = merged["max_achievable_gross_edge_bps"] - cost_bps
-    merged["expected_net_edge_bps"] = merged["gross_edge_bps"] - cost_bps
-    merged["entry_target"] = (merged["expected_net_edge_bps"] >= float(entry_conf["min_expected_net_edge_bps"])).astype("int8")
+    target_method = str(entry_conf.get("target_method", ENTRY_LABEL_METHOD))
+    if target_method == "PRICE_PLAN_OUTCOME_V1":
+        merged["expected_net_edge_bps"] = merged["actual_plan_net_edge_bps"]
+    elif target_method in {"OPPORTUNITY_MFE_V1", "OPPORTUNITY_QUALITY_V1"}:
+        merged["expected_net_edge_bps"] = merged["max_achievable_net_edge_bps"]
+    else:
+        raise ValueError(f"unsupported entry target_method: {target_method}")
+    opportunity = merged["expected_net_edge_bps"] >= float(entry_conf["min_expected_net_edge_bps"])
+    if target_method == "OPPORTUNITY_QUALITY_V1":
+        # MFE 只说明价格曾经给过机会；真实开仓还要确认这笔机会按计划能拿走，
+        # 并且过程中没有先承受过大的反向波动。
+        min_plan_net_edge_bps = float(entry_conf.get("min_plan_net_edge_bps", 0.0))
+        max_entry_mae_bps = float(entry_conf.get("max_entry_mae_bps", entry_conf["stop_bps"]))
+        opportunity = opportunity & (merged["actual_plan_net_edge_bps"] >= min_plan_net_edge_bps) & (merged["mae_bps"] <= max_entry_mae_bps)
+    merged["entry_target"] = opportunity.astype("int8")
     merged["price_plan_id"] = plan["pricePlanId"]
     merged["price_plan_hash"] = plan["pricePlanConfigHash"]
     merged["cost_bps"] = cost_bps
-    merged["label_method"] = ENTRY_LABEL_METHOD
+    merged["label_method"] = target_method
     merged["label_version"] = LABEL_VERSION
     out = merged[
         [
@@ -344,10 +585,12 @@ def build_entry_labels(args: Any) -> None:
             "ambiguous_hit",
             "time_to_target_ms",
             "time_to_stop_ms",
+            "time_to_exit_ms",
             "gross_edge_bps",
             "future_return_bps",
             "mfe_bps",
             "mae_bps",
+            "actual_plan_net_edge_bps",
             "max_achievable_gross_edge_bps",
             "max_achievable_net_edge_bps",
             "cost_bps",
@@ -393,7 +636,18 @@ def build_continue_exit_risk_labels(args: Any) -> None:
     horizon = int(labels["continue"]["horizon_minutes"])
     target_bps = float(plan["targetDistanceBps"])
     stop_bps = float(plan["stopDistanceBps"])
-    stats = _build_path_stats(replay, horizon, target_bps, stop_bps)
+    plan_config = {
+        "plan_method": plan.get("entryPlanMethod", labels["entry"].get("plan_method", "FIXED_TARGET_STOP_V1")),
+        "partial_take_1_ratio": plan.get("partialTake1Ratio", labels["entry"].get("partial_take_1_ratio", 0.50)),
+        "partial_take_2_ratio": plan.get("partialTake2Ratio", labels["entry"].get("partial_take_2_ratio", 0.25)),
+        "second_target_bps": plan.get("secondTargetBps", labels["entry"].get("second_target_bps", target_bps * 2.0)),
+        "trailing_stop_bps": plan.get("trailingStopBps", labels["entry"].get("trailing_stop_bps", stop_bps)),
+        "breakeven_after_first_target": plan.get(
+            "breakevenAfterFirstTarget",
+            labels["entry"].get("breakeven_after_first_target", True),
+        ),
+    }
+    stats = _build_path_stats(replay, horizon, target_bps, stop_bps, plan_config=plan_config)
     long_stats = stats[stats["side"] == "LONG"].drop(columns=["side"]).add_prefix("long_")
     short_stats = stats[stats["side"] == "SHORT"].drop(columns=["side"]).add_prefix("short_")
     long_stats = long_stats.rename(columns={"long_symbol": "symbol", "long_open_time_ms": "open_time_ms"})
@@ -432,6 +686,7 @@ def build_continue_exit_risk_labels(args: Any) -> None:
 
     long_edge = merged["long_gross_edge_bps"] - cost_bps
     short_edge = merged["short_gross_edge_bps"] - cost_bps
+    dynamic_plan = str(plan_config.get("plan_method")) == "DYNAMIC_TRAILING_V1"
     path_risk = np.maximum(merged["long_mae_bps"], merged["short_mae_bps"])
     max_path_move = np.maximum.reduce([merged["long_mfe_bps"], merged["short_mfe_bps"], path_risk])
     if "ret_15m_bps" in merged.columns:
@@ -449,8 +704,8 @@ def build_continue_exit_risk_labels(args: Any) -> None:
             "sample_id": merged["sample_id"],
             "symbol": merged["symbol"],
             "event_time": merged["event_time"],
-            "long_continue_target": ((long_edge >= min_continue) & (merged["long_stop_hit"] == 0)).astype("int8"),
-            "short_continue_target": ((short_edge >= min_continue) & (merged["short_stop_hit"] == 0)).astype("int8"),
+            "long_continue_target": ((long_edge >= min_continue) & ((merged["long_stop_hit"] == 0) | dynamic_plan)).astype("int8"),
+            "short_continue_target": ((short_edge >= min_continue) & ((merged["short_stop_hit"] == 0) | dynamic_plan)).astype("int8"),
             "long_expected_continue_edge_bps": long_edge,
             "short_expected_continue_edge_bps": short_edge,
             "split_id": merged["split_id"],

training/trader_training/ofi_feature_experiment.py

@@ -0,0 +1,890 @@
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+import pandas as pd
+from sklearn.linear_model import HuberRegressor, LogisticRegression
+from sklearn.metrics import accuracy_score, brier_score_loss, log_loss, mean_absolute_error, roc_auc_score
+from sklearn.preprocessing import StandardScaler
+
+from trader_training.io_utils import (
+    DEFAULT_RAW_ROOT,
+    read_json,
+    read_parquet,
+    run_root,
+    sha256_json,
+    to_utc_series,
+    write_json,
+    write_parquet,
+    write_text,
+)
+from trader_training.schemas import FEATURE_ORDER, FIT_SPLIT, LATEST_STRESS_SPLIT, TUNE_SPLIT, VALIDATION_LOCKED_SPLIT
+
+
+OFI_METHOD = "SNAPSHOT_DIFF_V1"
+BOOK_OFI_FEATURES = [
+    "ofi_l1_1m",
+    "ofi_l1_3m",
+    "ofi_l1_5m",
+    "ofi_l1_15m",
+    "mlofi_l5_1m",
+    "mlofi_l5_5m",
+    "mlofi_l20_1m",
+    "mlofi_l20_5m",
+    "mlofi_l5_l20_gap_1m",
+    "microprice_basis_change_1m_bps",
+    "microprice_basis_change_5m_bps",
+    "ofi_l1_5m_zscore_240m",
+    "mlofi_l20_5m_zscore_240m",
+]
+CROSS_OFI_FEATURES = [
+    "ofi_l1_5m_clipped",
+    "ofi_l1_taker_5m",
+    "ofi_l1_spread_rank_5m",
+]
+OFI_FEATURES = [*BOOK_OFI_FEATURES, *CROSS_OFI_FEATURES]
+
+META_COLUMNS = [
+    "sample_id",
+    "symbol",
+    "event_time",
+    "open_time_ms",
+    "split_id",
+    "walk_forward_fold",
+    "data_quality_flag",
+]
+ALL_SPLITS = (FIT_SPLIT, TUNE_SPLIT, VALIDATION_LOCKED_SPLIT, LATEST_STRESS_SPLIT)
+EVAL_SPLITS = (TUNE_SPLIT, VALIDATION_LOCKED_SPLIT, LATEST_STRESS_SPLIT)
+
+
+def run_ofi_feature_experiment(args: Any) -> None:
+    root = run_root(args)
+    baseline_root = args.data_root / "trader-v4" / "runs" / args.baseline_run_id
+    out_dir = root / "experiments" / "ofi_b2_mlofi"
+    raw_root = Path(args.raw_root or DEFAULT_RAW_ROOT)
+    logging.info(
+        "trader.training.ofi_experiment_started runId=%s baselineRunId=%s rawRoot=%s",
+        args.run_id,
+        args.baseline_run_id,
+        raw_root,
+    )
+
+    feature = _load_baseline_feature_frame(baseline_root)
+    ofi_delta = build_snapshot_diff_l1_ofi_features(raw_root, feature[["symbol", "event_time", "open_time_ms"]])
+    dataset = _merge_feature_delta(feature, ofi_delta)
+    if args.max_rows_per_split:
+        dataset = _cap_rows_per_split(dataset, int(args.max_rows_per_split))
+    delta_hash = write_parquet(out_dir / "ofi_feature_delta.parquet", ofi_delta)
+    dataset_hash = write_parquet(out_dir / "ofi_experiment_feature_frame.parquet", dataset)
+    write_json(out_dir / "ofi_feature_order.json", OFI_FEATURES)
+    write_json(out_dir / "ofi_feature_schema.json", _ofi_feature_schema())
+    write_json(out_dir / "experiment_manifest.json", _experiment_manifest(args, baseline_root, raw_root, ofi_delta, dataset, delta_hash, dataset_hash))
+    write_text(out_dir / "feature_delta_report.md", _feature_delta_report(ofi_delta, dataset))
+
+    direction = _load_direction_dataset(baseline_root, dataset)
+    entry = _load_entry_dataset(baseline_root, dataset)
+    results: dict[str, Any] = {}
+    prediction_frames: list[pd.DataFrame] = []
+    for feature_set_name, columns in _feature_sets().items():
+        direction_result, direction_predictions = _train_direction(direction, columns)
+        entry_result, entry_predictions = _train_entry(entry, columns)
+        results[feature_set_name] = {"DIRECTION": direction_result, "ENTRY": entry_result}
+        direction_predictions["model"] = "DIRECTION"
+        direction_predictions["feature_set"] = feature_set_name
+        entry_predictions["model"] = "ENTRY"
+        entry_predictions["feature_set"] = feature_set_name
+        prediction_frames.extend([direction_predictions, entry_predictions])
+        logging.info(
+            "trader.training.ofi_feature_set_trained runId=%s featureSet=%s directionTuneShortAuc=%s entryTuneShortAuc=%s",
+            args.run_id,
+            feature_set_name,
+            direction_result.get(TUNE_SPLIT, {}).get("short_auc"),
+            entry_result.get("short_entry_prob", {}).get(TUNE_SPLIT, {}).get("auc"),
+        )
+
+    predictions = pd.concat(prediction_frames, ignore_index=True) if prediction_frames else pd.DataFrame()
+    write_parquet(out_dir / "direction_entry_predictions.parquet", predictions)
+    write_json(out_dir / "ofi_experiment_result.json", results)
+    write_text(out_dir / "model_compare_to_run10.md", _model_compare_report(args, baseline_root, results, dataset))
+    write_text(out_dir / "backtest_compare_to_run10.md", _backtest_placeholder_report(args, baseline_root))
+    write_text(out_dir / "contract_change_report.md", _contract_change_report())
+    write_text(out_dir / "failure_cases_compare.md", _failure_case_placeholder_report(args))
+    logging.info("trader.training.ofi_experiment_finished runId=%s report=%s", args.run_id, out_dir / "model_compare_to_run10.md")
+
+
+def _load_baseline_feature_frame(baseline_root: Path) -> pd.DataFrame:
+    frame = read_parquet(baseline_root / "feature" / "feature_frame.parquet")
+    required = set(META_COLUMNS + FEATURE_ORDER)
+    missing = sorted(required.difference(frame.columns))
+    if missing:
+        raise ValueError(f"baseline feature frame missing columns: {missing}")
+    frame = frame[frame["data_quality_flag"].isin(["OK", "PARTIAL_OPTIONAL"])].copy()
+    frame = frame[frame["split_id"].isin(ALL_SPLITS)].copy()
+    frame["event_time"] = to_utc_series(frame["event_time"])
+    logging.info("trader.training.ofi_baseline_feature_loaded rowCount=%s splitCounts=%s", len(frame), frame["split_id"].value_counts().to_dict())
+    return frame
+
+
+def build_snapshot_diff_l1_ofi_features(raw_root: Path, replay_keys: pd.DataFrame) -> pd.DataFrame:
+    if replay_keys.empty:
+        return pd.DataFrame(columns=["symbol", "open_time_ms", *BOOK_OFI_FEATURES])
+    keys = replay_keys.copy()
+    keys["event_time"] = to_utc_series(keys["event_time"])
+    keys["event_date"] = keys["event_time"].dt.strftime("%Y-%m-%d")
+    frames: list[pd.DataFrame] = []
+    for (symbol, event_date), _ in keys.groupby(["symbol", "event_date"], sort=True, observed=False):
+        path = raw_root / "table=book" / "exchange=BINANCE_FUTURES" / f"symbol={symbol}" / f"dt={event_date}" / "data.parquet"
+        if not path.is_file():
+            logging.warning("trader.training.ofi_book_partition_missing symbol=%s eventDate=%s path=%s", symbol, event_date, path)
+            continue
+        day = _read_l1_book_day(path, symbol)
+        if not day.empty:
+            frames.append(day)
+        logging.info("trader.training.ofi_book_partition_loaded symbol=%s eventDate=%s minuteRows=%s path=%s", symbol, event_date, len(day), path)
+    if not frames:
+        raise ValueError(f"no book partitions loaded from {raw_root}")
+    minute_book = pd.concat(frames, ignore_index=True).sort_values(["symbol", "open_time_ms"])
+    minute_book = minute_book.drop_duplicates(["symbol", "open_time_ms"], keep="last").reset_index(drop=True)
+    feature_frames = []
+    for symbol, group in minute_book.groupby("symbol", sort=False, observed=False):
+        feature_frames.append(_compute_l1_ofi_for_symbol(group, str(symbol)))
+    features = pd.concat(feature_frames, ignore_index=True) if feature_frames else pd.DataFrame(columns=["symbol", "open_time_ms", *BOOK_OFI_FEATURES])
+    wanted = keys[["symbol", "open_time_ms"]].drop_duplicates()
+    out = wanted.merge(features, on=["symbol", "open_time_ms"], how="left")
+    logging.info(
+        "trader.training.ofi_features_built wantedRows=%s matchedRows=%s featureRows=%s",
+        len(wanted),
+        int(out[BOOK_OFI_FEATURES].notna().all(axis=1).sum()),
+        len(features),
+    )
+    return out
+
+
+def _read_l1_book_day(path: Path, symbol: str) -> pd.DataFrame:
+    columns = ["origin_time"]
+    for side in ("bid", "ask"):
+        for level in range(20):
+            columns.extend([f"{side}_{level}_price", f"{side}_{level}_size"])
+    book = pd.read_parquet(path, columns=columns)
+    if book.empty:
+        return pd.DataFrame(columns=["symbol", "open_time_ms", *columns[1:]])
+    required = ["origin_time", "bid_0_price", "bid_0_size", "ask_0_price", "ask_0_size"]
+    book = book.dropna(subset=required).copy()
+    book["origin_time"] = to_utc_series(book["origin_time"])
+    book["minute"] = book["origin_time"].dt.floor("min")
+    book = book.sort_values("origin_time").drop_duplicates("minute", keep="last")
+    if book.empty:
+        return pd.DataFrame(columns=["symbol", "open_time_ms", *columns[1:]])
+    out = pd.DataFrame({"symbol": symbol, "open_time_ms": (book["minute"].astype("int64") // 1_000_000).astype("int64")})
+    for column in columns[1:]:
+        out[column] = pd.to_numeric(book[column], errors="coerce").astype("float64")
+    return out
+
+
+def _compute_l1_ofi_for_symbol(book: pd.DataFrame, symbol: str) -> pd.DataFrame:
+    group = book.sort_values("open_time_ms").reset_index(drop=True).copy()
+    gap = group["open_time_ms"].astype("int64").diff().ne(60_000)
+    segment = gap.cumsum()
+    level_ofi = []
+    for level in range(20):
+        prev_bid_price = group.groupby(segment, sort=False)[f"bid_{level}_price"].shift(1)
+        prev_bid_size = group.groupby(segment, sort=False)[f"bid_{level}_size"].shift(1)
+        prev_ask_price = group.groupby(segment, sort=False)[f"ask_{level}_price"].shift(1)
+        prev_ask_size = group.groupby(segment, sort=False)[f"ask_{level}_size"].shift(1)
+        bid_part, ask_part = l1_snapshot_diff_ofi_quote(
+            group[f"bid_{level}_price"],
+            group[f"bid_{level}_size"],
+            group[f"ask_{level}_price"],
+            group[f"ask_{level}_size"],
+            prev_bid_price,
+            prev_bid_size,
+            prev_ask_price,
+            prev_ask_size,
+        )
+        level_ofi.append((bid_part + ask_part).rename(f"ofi_level_{level}_quote"))
+    level_frame = pd.concat(level_ofi, axis=1)
+    group["ofi_l1_event_quote"] = level_frame["ofi_level_0_quote"]
+    group["mlofi_l5_event_quote"] = level_frame[[f"ofi_level_{level}_quote" for level in range(5)]].sum(axis=1, min_count=5)
+    group["mlofi_l20_event_quote"] = level_frame.sum(axis=1, min_count=20)
+    group["l1_depth_quote"] = _depth_quote(group, 1)
+    group["l5_depth_quote"] = _depth_quote(group, 5)
+    group["l20_depth_quote"] = _depth_quote(group, 20)
+    for window in (1, 3, 5, 15):
+        group[f"ofi_l1_{window}m"] = _rolling_normalized(group, segment, "ofi_l1_event_quote", "l1_depth_quote", window)
+    for window in (1, 5):
+        group[f"mlofi_l5_{window}m"] = _rolling_normalized(group, segment, "mlofi_l5_event_quote", "l5_depth_quote", window)
+        group[f"mlofi_l20_{window}m"] = _rolling_normalized(group, segment, "mlofi_l20_event_quote", "l20_depth_quote", window)
+    group["mlofi_l5_l20_gap_1m"] = group["mlofi_l5_1m"] - group["mlofi_l20_1m"]
+    mid = (group["bid_0_price"] + group["ask_0_price"]) / 2.0
+    microprice = (group["ask_0_price"] * group["bid_0_size"] + group["bid_0_price"] * group["ask_0_size"]) / (group["bid_0_size"] + group["ask_0_size"]).clip(lower=1e-12)
+    group["microprice_basis_bps"] = (microprice / mid - 1.0) * 10000.0
+    group["microprice_basis_change_1m_bps"] = group.groupby(segment, sort=False)["microprice_basis_bps"].diff(1)
+    group["microprice_basis_change_5m_bps"] = group.groupby(segment, sort=False)["microprice_basis_bps"].diff(5)
+    group["ofi_l1_5m_zscore_240m"] = _rolling_zscore(group, segment, "ofi_l1_5m", 240)
+    group["mlofi_l20_5m_zscore_240m"] = _rolling_zscore(group, segment, "mlofi_l20_5m", 240)
+    out = group[["symbol", "open_time_ms", *BOOK_OFI_FEATURES]].replace([np.inf, -np.inf], np.nan)
+    for column in BOOK_OFI_FEATURES:
+        out[column] = pd.to_numeric(out[column], errors="coerce").astype("float32")
+    logging.info(
+        "trader.training.ofi_symbol_features_built symbol=%s minuteRows=%s fullFeatureRows=%s",
+        symbol,
+        len(out),
+        int(out[BOOK_OFI_FEATURES].notna().all(axis=1).sum()),
+    )
+    return out
+
+
+def _depth_quote(group: pd.DataFrame, level_count: int) -> pd.Series:
+    total = pd.Series(0.0, index=group.index, dtype="float64")
+    for level in range(level_count):
+        total = total + group[f"bid_{level}_price"] * group[f"bid_{level}_size"] + group[f"ask_{level}_price"] * group[f"ask_{level}_size"]
+    return total
+
+
+def _rolling_normalized(group: pd.DataFrame, segment: pd.Series, event_column: str, depth_column: str, window: int) -> pd.Series:
+    summed = group.groupby(segment, sort=False)[event_column].rolling(window, min_periods=window).sum().reset_index(level=0, drop=True)
+    averaged_depth = group.groupby(segment, sort=False)[depth_column].rolling(window, min_periods=window).mean().reset_index(level=0, drop=True)
+    return summed / averaged_depth.clip(lower=1e-12)
+
+
+def _rolling_zscore(group: pd.DataFrame, segment: pd.Series, column: str, window: int) -> pd.Series:
+    rolling = group.groupby(segment, sort=False)[column].rolling(window, min_periods=window)
+    mean = rolling.mean().reset_index(level=0, drop=True)
+    std = rolling.std().reset_index(level=0, drop=True).replace(0, np.nan)
+    return (group[column] - mean) / std
+
+
+def l1_snapshot_diff_ofi_quote(
+    bid_price: pd.Series,
+    bid_size: pd.Series,
+    ask_price: pd.Series,
+    ask_size: pd.Series,
+    prev_bid_price: pd.Series,
+    prev_bid_size: pd.Series,
+    prev_ask_price: pd.Series,
+    prev_ask_size: pd.Series,
+) -> tuple[pd.Series, pd.Series]:
+    bid = pd.to_numeric(bid_price, errors="coerce")
+    bid_sz = pd.to_numeric(bid_size, errors="coerce")
+    ask = pd.to_numeric(ask_price, errors="coerce")
+    ask_sz = pd.to_numeric(ask_size, errors="coerce")
+    prev_bid = pd.to_numeric(prev_bid_price, errors="coerce")
+    prev_bid_sz = pd.to_numeric(prev_bid_size, errors="coerce")
+    prev_ask = pd.to_numeric(prev_ask_price, errors="coerce")
+    prev_ask_sz = pd.to_numeric(prev_ask_size, errors="coerce")
+    valid = prev_bid.notna() & prev_bid_sz.notna() & prev_ask.notna() & prev_ask_sz.notna()
+
+    bid_part = pd.Series(np.nan, index=bid.index, dtype="float64")
+    ask_part = pd.Series(np.nan, index=ask.index, dtype="float64")
+    bid_up = valid & bid.gt(prev_bid)
+    bid_same = valid & bid.eq(prev_bid)
+    bid_down = valid & bid.lt(prev_bid)
+    bid_part.loc[bid_up] = bid_sz.loc[bid_up] * bid.loc[bid_up]
+    bid_part.loc[bid_same] = (bid_sz.loc[bid_same] - prev_bid_sz.loc[bid_same]) * bid.loc[bid_same]
+    bid_part.loc[bid_down] = -prev_bid_sz.loc[bid_down] * prev_bid.loc[bid_down]
+
+    ask_down = valid & ask.lt(prev_ask)
+    ask_same = valid & ask.eq(prev_ask)
+    ask_up = valid & ask.gt(prev_ask)
+    ask_part.loc[ask_down] = -ask_sz.loc[ask_down] * ask.loc[ask_down]
+    ask_part.loc[ask_same] = -(ask_sz.loc[ask_same] - prev_ask_sz.loc[ask_same]) * ask.loc[ask_same]
+    ask_part.loc[ask_up] = prev_ask_sz.loc[ask_up] * prev_ask.loc[ask_up]
+    return bid_part, ask_part
+
+
+def _merge_feature_delta(feature: pd.DataFrame, delta: pd.DataFrame) -> pd.DataFrame:
+    merged = feature.merge(delta, on=["symbol", "open_time_ms"], how="left")
+    merged["ofi_l1_5m_clipped"] = pd.to_numeric(merged["ofi_l1_5m"], errors="coerce").clip(-5.0, 5.0)
+    merged["ofi_l1_taker_5m"] = merged["ofi_l1_5m_clipped"] * pd.to_numeric(merged["taker_imbalance_5m"], errors="coerce")
+    merged["ofi_l1_spread_rank_5m"] = merged["ofi_l1_5m_clipped"] * pd.to_numeric(merged["spread_rank_24h_pct"], errors="coerce")
+    before = len(merged)
+    merged = merged.dropna(subset=OFI_FEATURES).copy()
+    logging.info(
+        "trader.training.ofi_feature_delta_merged rowBefore=%s rowAfter=%s droppedRows=%s splitCounts=%s",
+        before,
+        len(merged),
+        before - len(merged),
+        merged["split_id"].value_counts().to_dict(),
+    )
+    if merged.empty:
+        raise ValueError("OFI feature experiment has no rows after merging feature delta")
+    return merged
+
+
+def _cap_rows_per_split(frame: pd.DataFrame, max_rows_per_split: int) -> pd.DataFrame:
+    capped = []
+    for split_id, part in frame.sort_values("event_time").groupby("split_id", sort=False, observed=False):
+        if len(part) > max_rows_per_split:
+            part = part.tail(max_rows_per_split).copy()
+        capped.append(part)
+        logging.info("trader.training.ofi_split_capped splitId=%s rowCount=%s maxRows=%s", split_id, len(part), max_rows_per_split)
+    return pd.concat(capped, ignore_index=True)
+
+
+def _load_direction_dataset(baseline_root: Path, feature: pd.DataFrame) -> pd.DataFrame:
+    labels = read_parquet(baseline_root / "label" / "direction_labels.parquet")
+    required = {"sample_id", "long_target", "short_target", "neutral_target", "future_return_bps"}
+    missing = sorted(required.difference(labels.columns))
+    if missing:
+        raise ValueError(f"direction labels missing columns: {missing}")
+    dataset = feature.merge(labels[list(required)], on="sample_id", how="inner")
+    logging.info("trader.training.ofi_direction_dataset_loaded rowCount=%s", len(dataset))
+    return dataset
+
+
+def _load_entry_dataset(baseline_root: Path, feature: pd.DataFrame) -> pd.DataFrame:
+    labels = read_parquet(baseline_root / "label" / "entry_labels.parquet")
+    required = {"sample_id", "side", "entry_target", "expected_net_edge_bps"}
+    missing = sorted(required.difference(labels.columns))
+    if missing:
+        raise ValueError(f"entry labels missing columns: {missing}")
+    long = labels[labels["side"].eq("LONG")][["sample_id", "entry_target", "expected_net_edge_bps"]].rename(
+        columns={"entry_target": "long_entry_target", "expected_net_edge_bps": "long_expected_net_edge_bps"}
+    )
+    short = labels[labels["side"].eq("SHORT")][["sample_id", "entry_target", "expected_net_edge_bps"]].rename(
+        columns={"entry_target": "short_entry_target", "expected_net_edge_bps": "short_expected_net_edge_bps"}
+    )
+    pivot = long.merge(short, on="sample_id", how="inner")
+    dataset = feature.merge(pivot, on="sample_id", how="inner")
+    logging.info("trader.training.ofi_entry_dataset_loaded rowCount=%s", len(dataset))
+    return dataset
+
+
+def _feature_sets() -> dict[str, list[str]]:
+    return {
+        "market_only": FEATURE_ORDER,
+        "market_plus_ofi": [*FEATURE_ORDER, *OFI_FEATURES],
+    }
+
+
+def _train_direction(frame: pd.DataFrame, feature_columns: list[str]) -> tuple[dict[str, Any], pd.DataFrame]:
+    train = frame[frame["split_id"].eq(FIT_SPLIT)].copy()
+    if train.empty:
+        raise ValueError("direction experiment has no fit_inner rows")
+    scaler = StandardScaler()
+    x_train = scaler.fit_transform(train[feature_columns].astype("float32"))
+    y_train = train[["long_target", "short_target", "neutral_target"]].to_numpy().argmax(axis=1)
+    model = LogisticRegression(max_iter=500)
+    model.fit(x_train, y_train)
+    train_prior = train[["long_target", "short_target", "neutral_target"]].to_numpy(dtype=float).mean(axis=0)
+
+    metrics: dict[str, Any] = {"feature_count": len(feature_columns), "feature_hash": sha256_json(feature_columns)}
+    prediction_frames = []
+    for split_id in ALL_SPLITS:
+        part = frame[frame["split_id"].eq(split_id)].copy()
+        if part.empty:
+            continue
+        x = scaler.transform(part[feature_columns].astype("float32"))
+        proba = model.predict_proba(x)
+        y = part[["long_target", "short_target", "neutral_target"]].to_numpy().argmax(axis=1)
+        metrics[split_id] = _direction_metrics(y, proba, train_prior)
+        pred = part[["sample_id", "symbol", "event_time", "split_id"]].copy()
+        pred["long_prob"] = proba[:, 0].astype("float32")
+        pred["short_prob"] = proba[:, 1].astype("float32")
+        pred["neutral_prob"] = proba[:, 2].astype("float32")
+        pred["label_long"] = (y == 0).astype("int8")
+        pred["label_short"] = (y == 1).astype("int8")
+        pred["label_neutral"] = (y == 2).astype("int8")
+        prediction_frames.append(pred)
+    return metrics, pd.concat(prediction_frames, ignore_index=True)
+
+
+def _train_entry(frame: pd.DataFrame, feature_columns: list[str]) -> tuple[dict[str, Any], pd.DataFrame]:
+    train = frame[frame["split_id"].eq(FIT_SPLIT)].copy()
+    if train.empty:
+        raise ValueError("entry experiment has no fit_inner rows")
+    scaler = StandardScaler()
+    x_train = scaler.fit_transform(train[feature_columns].astype("float32"))
+    x_by_split = {
+        split_id: scaler.transform(frame[frame["split_id"].eq(split_id)][feature_columns].astype("float32"))
+        for split_id in ALL_SPLITS
+        if not frame[frame["split_id"].eq(split_id)].empty
+    }
+    specs = [
+        ("long_entry_prob", "binary", "long_entry_target"),
+        ("short_entry_prob", "binary", "short_entry_target"),
+        ("long_expected_net_edge_bps", "regression", "long_expected_net_edge_bps"),
+        ("short_expected_net_edge_bps", "regression", "short_expected_net_edge_bps"),
+    ]
+    results: dict[str, Any] = {"feature_count": len(feature_columns), "feature_hash": sha256_json(feature_columns)}
+    split_predictions: dict[str, pd.DataFrame] = {
+        split_id: frame[frame["split_id"].eq(split_id)][["sample_id", "symbol", "event_time", "split_id"]].copy().reset_index(drop=True)
+        for split_id in x_by_split
+    }
+    for name, kind, target in specs:
+        y_train = pd.to_numeric(train[target], errors="coerce").fillna(0.0).to_numpy()
+        if kind == "binary":
+            model = LogisticRegression(max_iter=500)
+            model.fit(x_train, y_train.astype(int))
+        else:
+            model = HuberRegressor(alpha=0.001, epsilon=1.35, max_iter=500)
+            model.fit(x_train, y_train.astype(float))
+        results[name] = {}
+        for split_id, x in x_by_split.items():
+            part = frame[frame["split_id"].eq(split_id)].copy()
+            y = pd.to_numeric(part[target], errors="coerce").fillna(0.0).to_numpy()
+            if kind == "binary":
+                pred = model.predict_proba(x)[:, 1]
+                results[name][split_id] = _binary_metrics(y_train.astype(int), y.astype(int), pred)
+            else:
+                pred = model.predict(x)
+                results[name][split_id] = _regression_metrics(y_train.astype(float), y.astype(float), pred)
+            split_predictions[split_id][name] = pred.astype("float32")
+            split_predictions[split_id][f"label_{target}"] = y
+    return results, pd.concat(split_predictions.values(), ignore_index=True)
+
+
+def _direction_metrics(y: np.ndarray, proba: np.ndarray, train_prior: np.ndarray) -> dict[str, Any]:
+    labels = [0, 1, 2]
+    train_prior = np.asarray(train_prior, dtype=float)
+    train_prior = train_prior / train_prior.sum() if train_prior.sum() > 0 else np.full(3, 1.0 / 3.0)
+    constant = np.tile(train_prior.reshape(1, -1), (len(y), 1))
+    one_hot = np.eye(3, dtype=float)[y]
+    clipped = _clip_normalize(proba)
+    constant_clipped = _clip_normalize(constant)
+    out: dict[str, Any] = {
+        "row_count": int(len(y)),
+        "accuracy": float(accuracy_score(y, proba.argmax(axis=1))),
+        "logloss": float(log_loss(y, clipped, labels=labels)),
+        "constant_logloss": float(log_loss(y, constant_clipped, labels=labels)),
+        "brier_multiclass": float(np.mean(np.sum((one_hot - proba) ** 2, axis=1))),
+        "constant_brier_multiclass": float(np.mean(np.sum((one_hot - constant) ** 2, axis=1))),
+    }
+    for idx, name in enumerate(("long", "short", "neutral")):
+        target = (y == idx).astype(int)
+        if target.sum() >= 200 and (len(target) - target.sum()) >= 200:
+            out[f"{name}_auc"] = float(roc_auc_score(target, proba[:, idx]))
+    top_count = max(1, int(len(y) * 0.10))
+    top_idx = np.argsort(proba.max(axis=1))[-top_count:]
+    out["top10_hit_rate"] = float((proba.argmax(axis=1)[top_idx] == y[top_idx]).mean())
+    out["all_hit_rate"] = float((proba.argmax(axis=1) == y).mean())
+    out["logloss_vs_constant_ratio"] = float(out["logloss"] / out["constant_logloss"]) if out["constant_logloss"] > 0 else None
+    out["brier_vs_constant_ratio"] = float(out["brier_multiclass"] / out["constant_brier_multiclass"]) if out["constant_brier_multiclass"] > 0 else None
+    return out
+
+
+def _binary_metrics(y_train: np.ndarray, y: np.ndarray, proba: np.ndarray) -> dict[str, Any]:
+    train_rate = float(np.mean(y_train))
+    constant = np.full(len(y), train_rate)
+    out: dict[str, Any] = {
+        "row_count": int(len(y)),
+        "positive_rate": float(np.mean(y)) if len(y) else 0.0,
+        "brier": float(brier_score_loss(y, proba)),
+        "constant_brier": float(brier_score_loss(y, constant)),
+    }
+    if len(np.unique(y)) == 2:
+        out["auc"] = float(roc_auc_score(y, proba))
+    top_count = max(1, int(len(y) * 0.10))
+    top_idx = np.argsort(proba)[-top_count:]
+    out["top10_hit_rate"] = float(np.mean(y[top_idx]))
+    out["all_hit_rate"] = float(np.mean(y))
+    out["brier_vs_constant_ratio"] = float(out["brier"] / out["constant_brier"]) if out["constant_brier"] > 0 else None
+    return out
+
+
+def _regression_metrics(y_train: np.ndarray, y: np.ndarray, pred: np.ndarray) -> dict[str, Any]:
+    train_median = float(np.median(y_train)) if len(y_train) else 0.0
+    constant = np.full(len(y), train_median)
+    mae = float(mean_absolute_error(y, pred))
+    constant_mae = float(mean_absolute_error(y, constant))
+    return {
+        "row_count": int(len(y)),
+        "mae": mae,
+        "constant_mae": constant_mae,
+        "mae_vs_constant_ratio": float(mae / constant_mae) if constant_mae > 0 else None,
+        "train_target_median": train_median,
+    }
+
+
+def _clip_normalize(values: np.ndarray) -> np.ndarray:
+    values = np.clip(np.asarray(values, dtype=float), 1e-6, 1.0)
+    return values / values.sum(axis=1, keepdims=True)
+
+
+def _experiment_manifest(
+    args: Any,
+    baseline_root: Path,
+    raw_root: Path,
+    ofi_delta: pd.DataFrame,
+    dataset: pd.DataFrame,
+    delta_hash: str,
+    dataset_hash: str,
+) -> dict[str, Any]:
+    return {
+        "experiment": "ofi_l1_microprice_diagnostic_v1",
+        "run_id": args.run_id,
+        "baseline_run_id": args.baseline_run_id,
+        "baseline_root": str(baseline_root),
+        "raw_root": str(raw_root),
+        "ofi_method": OFI_METHOD,
+        "uses_event_stream_ofi": False,
+        "normalization": "quote_notional_over_average_l1_depth_quote",
+        "new_features": OFI_FEATURES,
+        "formal_model_contract_changed": False,
+        "java_contract_changed": False,
+        "label_changed": False,
+        "pm_threshold_changed": False,
+        "delta_row_count": int(len(ofi_delta)),
+        "trainable_row_count": int(len(dataset)),
+        "split_counts": dataset["split_id"].value_counts().to_dict(),
+        "ofi_delta_hash_sha256": delta_hash,
+        "ofi_experiment_feature_frame_hash_sha256": dataset_hash,
+    }
+
+
+def _ofi_feature_schema() -> list[dict[str, Any]]:
+    rows = [
+        {
+            "name": "ofi_l1_1m",
+            "meaning": "买一卖一盘口变化强度，1分钟窗口",
+            "unit": "ratio",
+            "source": "Crypto Lake book snapshot",
+            "window": "1m",
+            "formula": "sum(l1 snapshot-diff quote OFI) / avg(l1 depth quote)",
+            "ofi_method": OFI_METHOD,
+            "null_handling": "first minute or gap warmup -> drop in experiment",
+            "order": 1,
+        },
+        {
+            "name": "ofi_l1_3m",
+            "meaning": "买一卖一盘口变化强度，3分钟窗口",
+            "unit": "ratio",
+            "source": "Crypto Lake book snapshot",
+            "window": "3m",
+            "formula": "sum(l1 snapshot-diff quote OFI over last 3 closed minutes) / avg(l1 depth quote)",
+            "ofi_method": OFI_METHOD,
+            "null_handling": "window warmup or gap warmup -> drop in experiment",
+            "order": 2,
+        },
+        {
+            "name": "ofi_l1_5m",
+            "meaning": "买一卖一盘口变化强度，5分钟窗口",
+            "unit": "ratio",
+            "source": "Crypto Lake book snapshot",
+            "window": "5m",
+            "formula": "sum(l1 snapshot-diff quote OFI over last 5 closed minutes) / avg(l1 depth quote)",
+            "ofi_method": OFI_METHOD,
+            "null_handling": "window warmup or gap warmup -> drop in experiment",
+            "order": 3,
+        },
+        {
+            "name": "ofi_l1_15m",
+            "meaning": "买一卖一盘口变化强度，15分钟窗口",
+            "unit": "ratio",
+            "source": "Crypto Lake book snapshot",
+            "window": "15m",
+            "formula": "sum(l1 snapshot-diff quote OFI over last 15 closed minutes) / avg(l1 depth quote)",
+            "ofi_method": OFI_METHOD,
+            "null_handling": "window warmup or gap warmup -> drop in experiment",
+            "order": 4,
+        },
+        {
+            "name": "microprice_basis_change_1m_bps",
+            "meaning": "微价格偏离值的1分钟变化",
+            "unit": "bps",
+            "source": "Crypto Lake book snapshot",
+            "window": "1m",
+            "formula": "microprice_basis_bps(t) - microprice_basis_bps(t-1m)",
+            "ofi_method": OFI_METHOD,
+            "null_handling": "first minute or gap warmup -> drop in experiment",
+            "order": 5,
+        },
+        {
+            "name": "microprice_basis_change_5m_bps",
+            "meaning": "微价格偏离值的5分钟变化",
+            "unit": "bps",
+            "source": "Crypto Lake book snapshot",
+            "window": "5m",
+            "formula": "microprice_basis_bps(t) - microprice_basis_bps(t-5m)",
+            "ofi_method": OFI_METHOD,
+            "null_handling": "window warmup or gap warmup -> drop in experiment",
+            "order": 6,
+        },
+    ]
+    rows.extend(
+        [
+            {
+                "name": "mlofi_l5_1m",
+                "meaning": "前5档盘口变化强度，1分钟窗口",
+                "unit": "ratio",
+                "source": "Crypto Lake book snapshot",
+                "window": "1m",
+                "formula": "sum(level0..4 snapshot-diff quote OFI) / avg(level0..4 depth quote)",
+                "ofi_method": OFI_METHOD,
+                "null_handling": "first minute or gap warmup -> drop in experiment",
+                "order": 7,
+            },
+            {
+                "name": "mlofi_l5_5m",
+                "meaning": "前5档盘口变化强度，5分钟窗口",
+                "unit": "ratio",
+                "source": "Crypto Lake book snapshot",
+                "window": "5m",
+                "formula": "sum(level0..4 snapshot-diff quote OFI over last 5 closed minutes) / avg(level0..4 depth quote)",
+                "ofi_method": OFI_METHOD,
+                "null_handling": "window warmup or gap warmup -> drop in experiment",
+                "order": 8,
+            },
+            {
+                "name": "mlofi_l20_1m",
+                "meaning": "前20档盘口变化强度，1分钟窗口",
+                "unit": "ratio",
+                "source": "Crypto Lake book snapshot",
+                "window": "1m",
+                "formula": "sum(level0..19 snapshot-diff quote OFI) / avg(level0..19 depth quote)",
+                "ofi_method": OFI_METHOD,
+                "null_handling": "first minute or gap warmup -> drop in experiment",
+                "order": 9,
+            },
+            {
+                "name": "mlofi_l20_5m",
+                "meaning": "前20档盘口变化强度，5分钟窗口",
+                "unit": "ratio",
+                "source": "Crypto Lake book snapshot",
+                "window": "5m",
+                "formula": "sum(level0..19 snapshot-diff quote OFI over last 5 closed minutes) / avg(level0..19 depth quote)",
+                "ofi_method": OFI_METHOD,
+                "null_handling": "window warmup or gap warmup -> drop in experiment",
+                "order": 10,
+            },
+            {
+                "name": "mlofi_l5_l20_gap_1m",
+                "meaning": "近档盘口变化和深档盘口变化的差",
+                "unit": "ratio",
+                "source": "Crypto Lake book snapshot",
+                "window": "1m",
+                "formula": "mlofi_l5_1m - mlofi_l20_1m",
+                "ofi_method": OFI_METHOD,
+                "null_handling": "dependency missing -> drop in experiment",
+                "order": 11,
+            },
+            {
+                "name": "ofi_l1_5m_zscore_240m",
+                "meaning": "L1 OFI 5分钟值相对最近240分钟的异常程度",
+                "unit": "zscore",
+                "source": "Crypto Lake book snapshot",
+                "window": "240m",
+                "formula": "(ofi_l1_5m - rolling_mean_240m) / rolling_std_240m",
+                "ofi_method": OFI_METHOD,
+                "null_handling": "window warmup or gap warmup -> drop in experiment",
+                "order": 12,
+            },
+            {
+                "name": "mlofi_l20_5m_zscore_240m",
+                "meaning": "L20 多层 OFI 5分钟值相对最近240分钟的异常程度",
+                "unit": "zscore",
+                "source": "Crypto Lake book snapshot",
+                "window": "240m",
+                "formula": "(mlofi_l20_5m - rolling_mean_240m) / rolling_std_240m",
+                "ofi_method": OFI_METHOD,
+                "null_handling": "window warmup or gap warmup -> drop in experiment",
+                "order": 13,
+            },
+            {
+                "name": "ofi_l1_5m_clipped",
+                "meaning": "截尾后的 L1 OFI 5分钟值",
+                "unit": "ratio",
+                "source": "derived from ofi_l1_5m",
+                "window": "5m",
+                "formula": "clip(ofi_l1_5m, -5, 5)",
+                "ofi_method": OFI_METHOD,
+                "null_handling": "dependency missing -> drop in experiment",
+                "order": 14,
+            },
+            {
+                "name": "ofi_l1_taker_5m",
+                "meaning": "L1 OFI 和5分钟主动成交是否同向",
+                "unit": "ratio",
+                "source": "book + trades",
+                "window": "5m",
+                "formula": "ofi_l1_5m_clipped * taker_imbalance_5m",
+                "ofi_method": OFI_METHOD,
+                "null_handling": "dependency missing -> drop in experiment",
+                "order": 15,
+            },
+            {
+                "name": "ofi_l1_spread_rank_5m",
+                "meaning": "L1 OFI 在高价差环境下的强度",
+                "unit": "ratio",
+                "source": "book + level_1",
+                "window": "5m",
+                "formula": "ofi_l1_5m_clipped * spread_rank_24h_pct",
+                "ofi_method": OFI_METHOD,
+                "null_handling": "dependency missing -> drop in experiment",
+                "order": 16,
+            },
+        ]
+    )
+    return rows
+
+
+def _feature_delta_report(ofi_delta: pd.DataFrame, dataset: pd.DataFrame) -> str:
+    rows = []
+    for feature in OFI_FEATURES:
+        series = pd.to_numeric(dataset[feature], errors="coerce")
+        q = series.quantile([0.01, 0.5, 0.99])
+        rows.append(
+            {
+                "feature": feature,
+                "null_in_delta": int(ofi_delta[feature].isna().sum()) if feature in ofi_delta.columns else "derived_after_merge",
+                "trainable_null": int(series.isna().sum()),
+                "p01": round(float(q.loc[0.01]), 6),
+                "p50": round(float(q.loc[0.5]), 6),
+                "p99": round(float(q.loc[0.99]), 6),
+            }
+        )
+    lines = [
+        "# OFI Feature Delta Report",
+        "",
+        f"- ofi_method: `{OFI_METHOD}`",
+        f"- delta_rows: `{len(ofi_delta)}`",
+        f"- trainable_rows_after_drop: `{len(dataset)}`",
+        f"- split_counts: `{dataset['split_id'].value_counts().to_dict()}`",
+        "",
+        "| feature | null_in_delta | trainable_null | p01 | p50 | p99 |",
+        "| --- | ---: | ---: | ---: | ---: | ---: |",
+    ]
+    for row in rows:
+        lines.append(f"| {row['feature']} | {row['null_in_delta']} | {row['trainable_null']} | {row['p01']} | {row['p50']} | {row['p99']} |")
+    lines.extend(
+        [
+            "",
+            "## Leakage Check",
+            "",
+            "- 只用当前分钟及之前的 book 快照。",
+            "- 第一条快照和断档后的窗口不补 0，直接作为 warmup 丢掉。",
+            "- 分子和分母都使用 quote 金额口径。",
+            "",
+        ]
+    )
+    return "\n".join(lines)
+
+
+def _model_compare_report(args: Any, baseline_root: Path, results: dict[str, Any], dataset: pd.DataFrame) -> str:
+    baseline = read_json(baseline_root / "model" / "model_train_manifest.json")
+    baseline_direction = baseline["DIRECTION"]["metrics"]["direction"]
+    baseline_entry = baseline["ENTRY"]["metrics"]
+    lines = [
+        "# OFI Model Compare To Run10",
+        "",
+        f"- run_id: `{args.run_id}`",
+        f"- baseline_run_id: `{args.baseline_run_id}`",
+        f"- ofi_method: `{OFI_METHOD}`",
+        f"- rows: `{len(dataset)}`",
+        "",
+        "## Run10 Tune Baseline",
+        "",
+        "| model | metric | value |",
+        "| --- | --- | ---: |",
+        f"| Direction | long_auc | {baseline_direction.get('long_auc')} |",
+        f"| Direction | short_auc | {baseline_direction.get('short_auc')} |",
+        f"| Direction | neutral_auc | {baseline_direction.get('neutral_auc')} |",
+        f"| Entry | long_auc | {baseline_entry['long_entry_prob'].get('auc')} |",
+        f"| Entry | short_auc | {baseline_entry['short_entry_prob'].get('auc')} |",
+        f"| Entry | long_edge_mae_ratio | {baseline_entry['long_expected_net_edge_bps'].get('mae_vs_constant_ratio')} |",
+        f"| Entry | short_edge_mae_ratio | {baseline_entry['short_expected_net_edge_bps'].get('mae_vs_constant_ratio')} |",
+        "",
+        "## Diagnostic Direction Result",
+        "",
+        "| feature_set | split | long_auc | short_auc | neutral_auc | logloss_ratio | top10_hit_rate |",
+        "| --- | --- | ---: | ---: | ---: | ---: | ---: |",
+    ]
+    for feature_set_name, payload in results.items():
+        direction = payload["DIRECTION"]
+        for split_id in EVAL_SPLITS:
+            metric = direction.get(split_id, {})
+            lines.append(
+                f"| {feature_set_name} | {split_id} | {metric.get('long_auc')} | {metric.get('short_auc')} | {metric.get('neutral_auc')} | {metric.get('logloss_vs_constant_ratio')} | {metric.get('top10_hit_rate')} |"
+            )
+    lines.extend(
+        [
+            "",
+            "## Diagnostic Entry Result",
+            "",
+            "| head | feature_set | split | auc/mae_ratio | brier_ratio | top10_hit_rate |",
+            "| --- | --- | --- | ---: | ---: | ---: |",
+        ]
+    )
+    for feature_set_name, payload in results.items():
+        entry = payload["ENTRY"]
+        for head in ("long_entry_prob", "short_entry_prob"):
+            for split_id in EVAL_SPLITS:
+                metric = entry.get(head, {}).get(split_id, {})
+                lines.append(
+                    f"| {head} | {feature_set_name} | {split_id} | {metric.get('auc')} | {metric.get('brier_vs_constant_ratio')} | {metric.get('top10_hit_rate')} |"
+                )
+        for head in ("long_expected_net_edge_bps", "short_expected_net_edge_bps"):
+            for split_id in EVAL_SPLITS:
+                metric = entry.get(head, {}).get(split_id, {})
+                lines.append(f"| {head} | {feature_set_name} | {split_id} | {metric.get('mae_vs_constant_ratio')} |  |  |")
+    lines.extend(
+        [
+            "",
+            "## Verdict Rule",
+            "",
+            "只有 `market_plus_ofi` 在 validation_locked 和 latest_stress 上同时好过 `market_only`，才进入正式特征链路。",
+            "",
+        ]
+    )
+    return "\n".join(lines)
+
+
+def _backtest_placeholder_report(args: Any, baseline_root: Path) -> str:
+    return "\n".join(
+        [
+            "# Backtest Compare To Run10",
+            "",
+            f"- run_id: `{args.run_id}`",
+            f"- baseline_run_id: `{args.baseline_run_id}`",
+            "",
+            "本轮是 Direction / Entry 特征诊断，没有导出正式 ONNX，也没有改 PM 阈值，所以不跑组合回测。",
+            "",
+            "如果诊断指标通过，下一步才把 OFI 特征纳入正式 `feature_schema.json`、导出模型包，再做 validation_locked 和 latest_stress 的完整回测。",
+            "",
+            f"- run10_baseline_root: `{baseline_root}`",
+            "",
+        ]
+    )
+
+
+def _contract_change_report() -> str:
+    return "\n".join(
+        [
+            "# Contract Change Report",
+            "",
+            "| 项 | 结论 |",
+            "| --- | --- |",
+            "| 正式 ONNX 输入 | 未改变 |",
+            "| Java SHADOW 输入契约 | 未改变 |",
+            "| 模型输出字段 | 未改变 |",
+            "| 标签口径 | 未改变 |",
+            "| PM 阈值 | 未改变 |",
+            "",
+            "原因：本轮只做旁路诊断。只有验证通过后，才会进入正式特征表和 Java 契约同步。",
+            "",
+        ]
+    )
+
+
+def _failure_case_placeholder_report(args: Any) -> str:
+    return "\n".join(
+        [
+            "# Failure Cases Compare",
+            "",
+            f"- run_id: `{args.run_id}`",
+            "",
+            "本轮没有产生正式交易决策，因此没有最差交易样本可比。",
+            "",
+            "下一步如果 OFI 进入正式模型包，必须用完整回测交易明细比较：",
+            "",
+            "1. validation_locked 最大亏损交易。",
+            "2. latest_stress 最大亏损交易。",
+            "3. 连续亏损段。",
+            "4. 高 OFI 但反向亏损样本。",
+            "",
+        ]
+    )

training/trader_training/pm.py

@@ -11,6 +11,13 @@ from trader_training.io_utils import read_json, read_parquet, run_root, sha256_j
 from trader_training.schemas import LATEST_STRESS_SPLIT, PM_CONFIG_VERSION, TUNE_SPLIT, VALIDATION_LOCKED_SPLIT
 
 
+DEFAULT_BACKTEST_PRICE_PLAN = {
+    "stopDistanceBps": 35.0,
+    "costBps": 4.0,
+    "maxHoldMinutes": 45,
+}
+
+
 def default_pm_config() -> dict:
     return {
         "pmConfigVersion": PM_CONFIG_VERSION,
@@ -70,6 +77,7 @@ def default_pm_config() -> dict:
 def search_pm_thresholds(args: Any) -> None:
     root = run_root(args)
     frame = _pm_tune_frame(root)
+    price_plan = _price_plan_context(root)
     candidate_rows: list[dict[str, Any]] = []
     best_score = -float("inf")
     best_thresholds: dict[str, float] | None = None
@@ -77,7 +85,8 @@ def search_pm_thresholds(args: Any) -> None:
     best_trades = pd.DataFrame()
 
     for thresholds in _threshold_candidates():
-        trades = _simulate_open_trades(frame, thresholds)
+        config = _pm_config_from_thresholds(thresholds)
+        trades = _simulate_open_trades(frame, thresholds, config, price_plan)
         metrics = _trade_metrics(trades)
         score = _score_thresholds(metrics)
         candidate_rows.append({**thresholds, **metrics, "score": score})
@@ -134,11 +143,27 @@ def integrated_backtest(args: Any) -> None:
     trades_path = root / "pm-search" / "pm_backtest_trades.parquet"
     # PM search is allowed to use tune_inner, but final acceptance must be
     # measured on the sealed validation_locked and latest_stress splits.
-    tune_trades = read_parquet(trades_path) if trades_path.is_file() else _simulate_open_trades(_pm_tune_frame(root), _thresholds_from_config(pm_payload["config"]))
+    price_plan = _price_plan_context(root)
+    tune_trades = read_parquet(trades_path) if trades_path.is_file() else _simulate_open_trades(
+        _pm_tune_frame(root),
+        _thresholds_from_config(pm_payload["config"]),
+        pm_payload["config"],
+        price_plan,
+    )
     tune_trades["eval_split"] = TUNE_SPLIT
-    validation_locked_trades = _simulate_open_trades(_pm_frame(root, VALIDATION_LOCKED_SPLIT), _thresholds_from_config(pm_payload["config"]))
+    validation_locked_trades = _simulate_open_trades(
+        _pm_frame(root, VALIDATION_LOCKED_SPLIT),
+        _thresholds_from_config(pm_payload["config"]),
+        pm_payload["config"],
+        price_plan,
+    )
     validation_locked_trades["eval_split"] = VALIDATION_LOCKED_SPLIT
-    stress_trades = _simulate_open_trades(_pm_frame(root, LATEST_STRESS_SPLIT), _thresholds_from_config(pm_payload["config"]))
+    stress_trades = _simulate_open_trades(
+        _pm_frame(root, LATEST_STRESS_SPLIT),
+        _thresholds_from_config(pm_payload["config"]),
+        pm_payload["config"],
+        price_plan,
+    )
     stress_trades["eval_split"] = LATEST_STRESS_SPLIT
     trades = pd.concat([tune_trades, validation_locked_trades, stress_trades], ignore_index=True)
     metrics = {
@@ -154,6 +179,8 @@ def integrated_backtest(args: Any) -> None:
         "backtest_manifest_id": f"backtest-{args.run_id}",
         "mode": "VALIDATION_PM_BACKTEST",
         "pm_config_hash_sha256": pm_payload["config_hash_sha256"],
+        "price_plan_id": price_plan.get("pricePlanId"),
+        "price_plan_config_hash": price_plan.get("pricePlanConfigHash"),
         "metrics": metrics,
         "status_reasons": status_reasons,
         "status": status,
@@ -179,6 +206,14 @@ def _pm_tune_frame(root) -> pd.DataFrame:
     return _pm_frame(root, TUNE_SPLIT)
 
 
+def _price_plan_context(root) -> dict[str, Any]:
+    path = root / "label" / "price_plan_context.json"
+    if path.is_file():
+        return read_json(path)
+    logging.warning("trader.training.price_plan_missing_for_pm path=%s usingDefault=%s", path, DEFAULT_BACKTEST_PRICE_PLAN)
+    return DEFAULT_BACKTEST_PRICE_PLAN.copy()
+
+
 def _pm_frame(root, split_id: str) -> pd.DataFrame:
     prediction_files = {
         TUNE_SPLIT: "tune_predictions.parquet",
@@ -194,12 +229,14 @@ def _pm_frame(root, split_id: str) -> pd.DataFrame:
         }
     )
     risk = read_parquet(root / "model" / "risk" / prediction_file)
+    price_plan = _price_plan_context(root)
     entry_dataset = read_parquet(root / "dataset" / "entry_train.parquet").rename(
         columns={
             "long_expected_net_edge_bps": "actual_long_expected_net_edge_bps",
             "short_expected_net_edge_bps": "actual_short_expected_net_edge_bps",
         }
     )
+    entry_plan_outcome = _entry_plan_outcome_frame(root)
     entry_cols = [
         "sample_id",
         "long_entry_prob",
@@ -214,26 +251,92 @@ def _pm_frame(root, split_id: str) -> pd.DataFrame:
         .merge(entry[entry_cols], on="sample_id", how="inner")
         .merge(risk[risk_cols], on="sample_id", how="inner")
         .merge(entry_dataset[actual_cols], on="sample_id", how="inner")
+        .merge(entry_plan_outcome, on="sample_id", how="inner")
     )
     if frame.empty:
         raise ValueError(f"PM frame is empty for {split_id}; check model predictions and entry dataset")
+    frame["model_pred_long_expected_net_edge_bps"] = frame["pred_long_expected_net_edge_bps"]
+    frame["model_pred_short_expected_net_edge_bps"] = frame["pred_short_expected_net_edge_bps"]
+    edge_mode = "MODEL_EXPECTED_NET_EDGE"
+    if price_plan.get("entryTargetMethod") not in {"OPPORTUNITY_MFE_V1", "OPPORTUNITY_QUALITY_V1"}:
+        frame["pred_long_expected_net_edge_bps"] = _probability_implied_edge(frame["long_entry_prob"], price_plan)
+        frame["pred_short_expected_net_edge_bps"] = _probability_implied_edge(frame["short_entry_prob"], price_plan)
+        edge_mode = "ENTRY_PROBABILITY_PAYOFF"
     logging.info(
-        "trader.training.pm_frame_loaded splitId=%s rowCount=%s splitCounts=%s",
+        "trader.training.pm_frame_loaded splitId=%s rowCount=%s splitCounts=%s edgeMode=%s",
         split_id,
         len(frame),
         frame["split_id"].value_counts().to_dict(),
+        edge_mode,
     )
     return frame
 
 
+def _probability_implied_edge(entry_prob: pd.Series, price_plan: dict[str, Any]) -> pd.Series:
+    target_net_bps = float(price_plan.get("targetDistanceBps", 0.0)) - float(price_plan.get("costBps", 0.0))
+    stop_net_bps = -float(price_plan.get("stopDistanceBps", DEFAULT_BACKTEST_PRICE_PLAN["stopDistanceBps"])) - float(
+        price_plan.get("costBps", DEFAULT_BACKTEST_PRICE_PLAN["costBps"])
+    )
+    probability = pd.to_numeric(entry_prob, errors="coerce").fillna(0.0).clip(lower=0.0, upper=1.0)
+    # Entry 的概率头比收益回归头稳定。这里用固定止盈止损的盈亏比把概率换成期望收益，
+    # 让低命中、高赔率计划也能被 PM 正常搜索；真实结果仍由标签里的实际路径收益评估。
+    return probability * target_net_bps + (1.0 - probability) * stop_net_bps
+
+
+def _entry_plan_outcome_frame(root) -> pd.DataFrame:
+    labels = read_parquet(root / "label" / "entry_labels.parquet").copy()
+    required = {
+        "sample_id",
+        "side",
+        "gross_edge_bps",
+        "cost_bps",
+        "target_hit",
+        "stop_hit",
+        "time_to_target_ms",
+        "time_to_stop_ms",
+        "time_to_exit_ms",
+    }
+    missing = sorted(required - set(labels.columns))
+    if missing:
+        raise ValueError(f"entry_labels is missing PM outcome columns: {missing}")
+    labels["trade_net_edge_bps"] = pd.to_numeric(labels["gross_edge_bps"], errors="coerce").fillna(0.0) - pd.to_numeric(
+        labels["cost_bps"], errors="coerce"
+    ).fillna(0.0)
+
+    def side_frame(side: str, prefix: str) -> pd.DataFrame:
+        return labels[labels["side"].eq(side)][
+            [
+                "sample_id",
+                "trade_net_edge_bps",
+                "target_hit",
+                "stop_hit",
+                "time_to_target_ms",
+                "time_to_stop_ms",
+                "time_to_exit_ms",
+            ]
+        ].rename(
+            columns={
+                "trade_net_edge_bps": f"{prefix}_trade_net_edge_bps",
+                "target_hit": f"{prefix}_target_hit",
+                "stop_hit": f"{prefix}_stop_hit",
+                "time_to_target_ms": f"{prefix}_time_to_target_ms",
+                "time_to_stop_ms": f"{prefix}_time_to_stop_ms",
+                "time_to_exit_ms": f"{prefix}_time_to_exit_ms",
+            }
+        )
+
+    return side_frame("LONG", "long").merge(side_frame("SHORT", "short"), on="sample_id", how="inner")
+
+
 def _threshold_candidates() -> list[dict[str, float]]:
+    # 1.01 表示这一侧不开仓，用来检查“只做多”或“只做空”是否更稳。
     values = itertools.product(
-        [0.50, 0.52, 0.54, 0.56, 0.58],
-        [0.50, 0.52, 0.54, 0.56, 0.58],
-        [0.10, 0.12, 0.14, 0.16, 0.20, 0.30, 0.50],
-        [0.55, 0.75, 0.90, 1.00],
-        [-8.0, -4.0, 0.0, 1.0, 3.0],
-        [0.00, 0.01, 0.02, 0.05],
+        [0.50, 0.60, 0.70, 1.01],
+        [0.50, 0.60, 0.70, 1.01],
+        [0.03, 0.50, 0.70, 0.85],
+        [0.45, 0.65, 0.85],
+        [0.0, 8.0, 15.0, 25.0],
+        [0.02, 0.06, 0.10],
     )
     return [
         {
@@ -248,23 +351,39 @@ def _threshold_candidates() -> list[dict[str, float]]:
     ]
 
 
-def _simulate_open_trades(frame: pd.DataFrame, thresholds: dict[str, float]) -> pd.DataFrame:
+def _simulate_open_trades(
+    frame: pd.DataFrame,
+    thresholds: dict[str, float],
+    pm_config: dict[str, Any] | None = None,
+    price_plan: dict[str, Any] | None = None,
+) -> pd.DataFrame:
+    direction_margin = (frame["long_prob"] - frame["short_prob"]).abs()
     long_mask = (
-        (frame["long_prob"] >= thresholds["long_open_prob"])
-        & ((frame["long_prob"] - frame["short_prob"]) >= thresholds["min_direction_margin"])
-        & (frame["long_entry_prob"] >= thresholds["min_entry_prob"])
-        & (frame["market_risk_prob"] <= thresholds["max_market_risk_prob"])
-        & (frame["pred_long_expected_net_edge_bps"] >= thresholds["min_expected_edge_bps"])
+        (frame["long_prob"] > thresholds["long_open_prob"])
+        & (direction_margin > thresholds["min_direction_margin"])
+        & (frame["long_entry_prob"] > thresholds["min_entry_prob"])
+        & (frame["market_risk_prob"] < thresholds["max_market_risk_prob"])
+        & (frame["pred_long_expected_net_edge_bps"] > thresholds["min_expected_edge_bps"])
     )
     short_mask = (
-        (frame["short_prob"] >= thresholds["short_open_prob"])
-        & ((frame["short_prob"] - frame["long_prob"]) >= thresholds["min_direction_margin"])
-        & (frame["short_entry_prob"] >= thresholds["min_entry_prob"])
-        & (frame["market_risk_prob"] <= thresholds["max_market_risk_prob"])
-        & (frame["pred_short_expected_net_edge_bps"] >= thresholds["min_expected_edge_bps"])
+        (frame["short_prob"] > thresholds["short_open_prob"])
+        & (direction_margin > thresholds["min_direction_margin"])
+        & (frame["short_entry_prob"] > thresholds["min_entry_prob"])
+        & (frame["market_risk_prob"] < thresholds["max_market_risk_prob"])
+        & (frame["pred_short_expected_net_edge_bps"] > thresholds["min_expected_edge_bps"])
+    )
+    long_score = (
+        frame["pred_long_expected_net_edge_bps"].clip(lower=0.0)
+        * frame["long_prob"]
+        * frame["long_entry_prob"]
+        * (1.0 - frame["market_risk_prob"].clip(lower=0.0, upper=1.0))
+    )
+    short_score = (
+        frame["pred_short_expected_net_edge_bps"].clip(lower=0.0)
+        * frame["short_prob"]
+        * frame["short_entry_prob"]
+        * (1.0 - frame["market_risk_prob"].clip(lower=0.0, upper=1.0))
     )
-    long_score = frame["pred_long_expected_net_edge_bps"] + (frame["long_prob"] - frame["short_prob"]) * 10.0
-    short_score = frame["pred_short_expected_net_edge_bps"] + (frame["short_prob"] - frame["long_prob"]) * 10.0
     side = np.where(long_mask & (~short_mask | (long_score >= short_score)), "LONG", np.where(short_mask, "SHORT", ""))
     trades = frame.loc[side != ""].copy().reset_index(drop=True)
     if trades.empty:
@@ -274,11 +393,19 @@ def _simulate_open_trades(frame: pd.DataFrame, thresholds: dict[str, float]) ->
     trades["direction_prob"] = np.where(is_long, trades["long_prob"], trades["short_prob"])
     trades["entry_prob"] = np.where(is_long, trades["long_entry_prob"], trades["short_entry_prob"])
     trades["predicted_edge_bps"] = np.where(is_long, trades["pred_long_expected_net_edge_bps"], trades["pred_short_expected_net_edge_bps"])
-    trades["actual_edge_bps"] = np.where(is_long, trades["actual_long_expected_net_edge_bps"], trades["actual_short_expected_net_edge_bps"])
+    trades["actual_edge_bps"] = np.where(is_long, trades["long_trade_net_edge_bps"], trades["short_trade_net_edge_bps"])
+    trades["label_max_edge_bps"] = np.where(is_long, trades["actual_long_expected_net_edge_bps"], trades["actual_short_expected_net_edge_bps"])
     trades["entry_target"] = np.where(is_long, trades["long_entry_target"], trades["short_entry_target"])
-    trades["planned_ratio"] = _planned_ratio(trades["predicted_edge_bps"], trades["market_risk_prob"], thresholds["min_expected_edge_bps"])
+    effective_pm_config = pm_config or _pm_config_from_thresholds(thresholds)
+    effective_price_plan = price_plan or DEFAULT_BACKTEST_PRICE_PLAN
+    trades["time_to_exit_ms"] = _time_to_exit_ms(trades, is_long, effective_price_plan)
+    trades["planned_ratio"] = _planned_ratio_like_position_manager(trades, effective_pm_config["sizing"], effective_price_plan)
+    trades = trades[trades["planned_ratio"] > 0].copy()
+    if trades.empty:
+        return _empty_trade_frame()
     trades["weighted_edge_bps"] = trades["actual_edge_bps"] * trades["planned_ratio"]
     trades["threshold_hash"] = sha256_json(thresholds)[:16]
+    trades = _enforce_non_overlapping_entries(trades, effective_pm_config, effective_price_plan)
     return trades[
         [
             "sample_id",
@@ -290,8 +417,10 @@ def _simulate_open_trades(frame: pd.DataFrame, thresholds: dict[str, float]) ->
             "entry_prob",
             "market_risk_prob",
             "predicted_edge_bps",
+            "label_max_edge_bps",
             "actual_edge_bps",
             "entry_target",
+            "time_to_exit_ms",
             "planned_ratio",
             "weighted_edge_bps",
             "threshold_hash",
@@ -311,8 +440,10 @@ def _empty_trade_frame() -> pd.DataFrame:
             "entry_prob",
             "market_risk_prob",
             "predicted_edge_bps",
+            "label_max_edge_bps",
             "actual_edge_bps",
             "entry_target",
+            "time_to_exit_ms",
             "planned_ratio",
             "weighted_edge_bps",
             "threshold_hash",
@@ -320,10 +451,78 @@ def _empty_trade_frame() -> pd.DataFrame:
     )
 
 
-def _planned_ratio(predicted_edge: pd.Series, market_risk: pd.Series, min_edge: float) -> np.ndarray:
-    edge_strength = ((predicted_edge.astype(float) - min_edge) / 20.0).clip(lower=0.0, upper=1.5)
-    risk_discount = (1.0 - market_risk.astype(float)).clip(lower=0.0, upper=1.0)
-    return (edge_strength * risk_discount).clip(lower=0.05, upper=1.0).to_numpy()
+def _time_to_exit_ms(trades: pd.DataFrame, is_long: pd.Series, price_plan: dict[str, Any]) -> np.ndarray:
+    max_hold_ms = int(price_plan.get("maxHoldMinutes", DEFAULT_BACKTEST_PRICE_PLAN["maxHoldMinutes"])) * 60_000
+    long_exit_col = "long_time_to_exit_ms"
+    short_exit_col = "short_time_to_exit_ms"
+    if long_exit_col in trades.columns and short_exit_col in trades.columns:
+        label_exit_ms = np.where(is_long, trades[long_exit_col], trades[short_exit_col]).astype("float64")
+        return np.where(np.isfinite(label_exit_ms) & (label_exit_ms > 0), label_exit_ms, max_hold_ms)
+    target_hit = np.where(is_long, trades["long_target_hit"], trades["short_target_hit"])
+    stop_hit = np.where(is_long, trades["long_stop_hit"], trades["short_stop_hit"])
+    target_ms = np.where(is_long, trades["long_time_to_target_ms"], trades["short_time_to_target_ms"]).astype("float64")
+    stop_ms = np.where(is_long, trades["long_time_to_stop_ms"], trades["short_time_to_stop_ms"]).astype("float64")
+    return np.where((target_hit == 1) & (target_ms >= 0), target_ms, np.where((stop_hit == 1) & (stop_ms >= 0), stop_ms, max_hold_ms))
+
+
+def _enforce_non_overlapping_entries(trades: pd.DataFrame, pm_config: dict[str, Any], price_plan: dict[str, Any]) -> pd.DataFrame:
+    if trades.empty:
+        return trades
+    cooldown_ms = int(pm_config.get("add", {}).get("cooldownMinutes", 0)) * 60_000
+    max_hold_ms = int(price_plan.get("maxHoldMinutes", DEFAULT_BACKTEST_PRICE_PLAN["maxHoldMinutes"])) * 60_000
+    sort_columns = ["symbol", "event_time", "predicted_edge_bps"]
+    sorted_keys = trades[["symbol", "event_time", "predicted_edge_bps", "time_to_exit_ms"]].sort_values(
+        sort_columns,
+        ascending=[True, True, False],
+    )
+    event_ns = pd.to_datetime(sorted_keys["event_time"], utc=True).astype("int64").to_numpy()
+    symbols = sorted_keys["symbol"].astype(str).to_numpy()
+    exit_delay_values = pd.to_numeric(sorted_keys["time_to_exit_ms"], errors="coerce").fillna(max_hold_ms).to_numpy(dtype="float64")
+    original_indices = sorted_keys.index.to_numpy()
+
+    next_available_ns_by_symbol: dict[str, int] = {}
+    keep: list[int] = []
+    for index, symbol, event_time_ns, exit_delay_ms in zip(original_indices, symbols, event_ns, exit_delay_values):
+        next_available_ns = next_available_ns_by_symbol.get(symbol)
+        if next_available_ns is not None and event_time_ns < next_available_ns:
+            continue
+        keep.append(index)
+        if not np.isfinite(exit_delay_ms) or exit_delay_ms <= 0:
+            exit_delay_ms = max_hold_ms
+        next_available_ns_by_symbol[symbol] = int(event_time_ns + (exit_delay_ms + cooldown_ms) * 1_000_000)
+    return trades.loc[keep].sort_values("event_time").reset_index(drop=True)
+
+
+def _planned_ratio_like_position_manager(trades: pd.DataFrame, sizing: dict[str, Any], price_plan: dict[str, Any]) -> np.ndarray:
+    expected_edge = trades["predicted_edge_bps"].astype(float).clip(lower=0.0)
+    direction_strength = trades["direction_prob"].astype(float).clip(lower=0.0, upper=1.0)
+    entry_prob = trades["entry_prob"].astype(float).clip(lower=0.0, upper=1.0)
+    market_risk = trades["market_risk_prob"].astype(float).clip(lower=0.0, upper=1.0)
+
+    min_edge = float(sizing["minEdgeBps"])
+    stop_loss_budget = max(
+        float(price_plan.get("stopDistanceBps", DEFAULT_BACKTEST_PRICE_PLAN["stopDistanceBps"]))
+        + float(price_plan.get("costBps", DEFAULT_BACKTEST_PRICE_PLAN["costBps"])),
+        1.0,
+    )
+    raw = (
+        float(sizing["baseRatio"])
+        * (expected_edge / stop_loss_budget)
+        * direction_strength
+        * entry_prob
+        * (1.0 - market_risk)
+    )
+    hard_cap = min(
+        float(sizing["maxSingleLegRatio"]),
+        float(sizing["maxTotalPositionRatio"]),
+        float(sizing["maxLiquidityUsageRatio"]),
+        float(sizing["maxLossPerTradeBps"]) / stop_loss_budget,
+    )
+    min_ratio = float(sizing["minInitialRatio"])
+    if hard_cap < min_ratio:
+        return np.zeros(len(trades), dtype="float64")
+    ratio = raw.clip(lower=min_ratio, upper=hard_cap)
+    return np.where(expected_edge >= min_edge, ratio, 0.0)
 
 
 def _trade_metrics(trades: pd.DataFrame) -> dict[str, Any]:
@@ -336,6 +535,9 @@ def _trade_metrics(trades: pd.DataFrame) -> dict[str, Any]:
             "total_weighted_edge_bps": 0.0,
             "max_drawdown_bps": 0.0,
             "avg_planned_ratio": 0.0,
+            "min_planned_ratio": 0.0,
+            "p50_planned_ratio": 0.0,
+            "max_planned_ratio": 0.0,
             "profit_factor": 0.0,
             "max_consecutive_losses": 0,
         }
@@ -351,6 +553,9 @@ def _trade_metrics(trades: pd.DataFrame) -> dict[str, Any]:
         "total_weighted_edge_bps": float(equity.iloc[-1]),
         "max_drawdown_bps": float(drawdown.max()),
         "avg_planned_ratio": float(trades["planned_ratio"].astype(float).mean()),
+        "min_planned_ratio": float(trades["planned_ratio"].astype(float).min()),
+        "p50_planned_ratio": float(trades["planned_ratio"].astype(float).median()),
+        "max_planned_ratio": float(trades["planned_ratio"].astype(float).max()),
         "profit_factor": float(gains / losses) if losses > 0 else float("inf"),
         "max_consecutive_losses": _max_consecutive_losses(trades["weighted_edge_bps"].astype(float).to_numpy()),
     }
@@ -382,6 +587,8 @@ def _backtest_status(metrics: dict[str, dict[str, Any]]) -> tuple[str, list[str]
         reasons.append("validation_locked_avg_trade_edge_not_positive")
     if validation_locked["max_consecutive_losses"] > 8:
         reasons.append("validation_locked_max_consecutive_losses_above_8")
+    if validation_locked["trade_count"] > 0 and validation_locked["max_planned_ratio"] <= 0.050001:
+        reasons.append("validation_locked_sizing_collapsed_to_min_initial")
     if stress["trade_count"] < 20:
         reasons.append("latest_stress_trade_count_below_20")
     if stress["profit_factor"] < 1.0:
@@ -427,6 +634,7 @@ def _pm_config_from_thresholds(thresholds: dict[str, float]) -> dict:
         }
     )
     config["add"]["maxMarketRiskProb"] = thresholds["max_market_risk_prob"]
+    config["add"]["minEntryProb"] = thresholds["min_entry_prob"]
     config["add"]["minExpectedEdgeBps"] = thresholds["min_expected_edge_bps"]
     config["sizing"]["minEdgeBps"] = thresholds["min_expected_edge_bps"]
     config["sizing"]["maxSingleLegRatio"] = 1.0
@@ -479,7 +687,7 @@ def _write_pm_report(path, candidates: pd.DataFrame, best_thresholds: dict[str,
     lines = [
         "# PM Threshold Report",
         "",
-        "本次不是固定写死阈值，而是在验证集上试一组可复现的阈值，选择净收益、回撤、交易数量综合更好的那组。",
+        "本次不是固定写死阈值，而是在调参集上试一组可复现的阈值。PM 回测使用固定止盈止损后的真实净收益，并且开仓后按持仓结束时间加冷却时间阻止重叠开仓。",
         "",
         "## Best Thresholds",
         "",
@@ -505,7 +713,7 @@ def _write_backtest_report(path, result: dict[str, Any]) -> None:
     lines = [
         "# Integrated Backtest Report",
         "",
-        "这里用验证集模型输出和 PM 阈值生成交易明细，统计净收益、胜率、回撤和分段表现。",
+        "这里用验证集模型输出和 PM 阈值生成交易明细，统计净收益、胜率、回撤和分段表现。收益按固定止盈止损计划的真实净收益计算，不使用窗口内最大可拿收益。",
         "",
         "```json",
         str(result).replace("'", '"'),

training/trader_training/price_plan_search.py

@@ -194,7 +194,7 @@ def _plan_side_rows(
     gross = np.where(target_first, target_bps, np.where(stop_first, -stop_bps, timeout_return))
     price_plan_net = gross - cost_bps
     expected_net = max_achievable_gross - cost_bps
-    positive = expected_net >= min_expected_edge_bps
+    positive = price_plan_net >= min_expected_edge_bps
     ambiguous = target_any & stop_any & (target_index == stop_index)
 
     rows: list[dict[str, Any]] = []
@@ -264,21 +264,25 @@ def _plan_summary(rows: pd.DataFrame) -> pd.DataFrame:
     split_rows["avg_price_plan_edge_eval"] = split_rows[
         [f"avg_price_plan_net_edge_bps_{split}" for split in (TUNE_SPLIT, VALIDATION_LOCKED_SPLIT, LATEST_STRESS_SPLIT)]
     ].mean(axis=1)
+    split_rows["min_price_plan_edge_eval"] = split_rows[
+        [f"avg_price_plan_net_edge_bps_{split}" for split in (TUNE_SPLIT, VALIDATION_LOCKED_SPLIT, LATEST_STRESS_SPLIT)]
+    ].min(axis=1)
     split_rows["min_margin_eval"] = split_rows[
         [f"target_rate_margin_{split}" for split in (TUNE_SPLIT, VALIDATION_LOCKED_SPLIT, LATEST_STRESS_SPLIT)]
     ].min(axis=1)
-    # The search score is not an上线门槛. It only chooses the next experiment:
-    # enough positive samples, less negative average edge, and stable behavior
-    # across tune/validation/stress.
+    # The search score is not an上线门槛. It only chooses the next experiment.
+    # Fixed price-plan net edge is weighted most heavily; max future edge is
+    # retained only as a weak tie-breaker because it can be optimistic.
     positive_rate_penalty = (
-        (0.08 - split_rows["min_positive_label_rate_eval"]).clip(lower=0.0) * 80.0
-        + (split_rows["max_positive_label_rate_eval"] - 0.45).clip(lower=0.0) * 30.0
+        (0.03 - split_rows["min_positive_label_rate_eval"]).clip(lower=0.0) * 120.0
+        + (split_rows["max_positive_label_rate_eval"] - 0.55).clip(lower=0.0) * 20.0
     )
     spread_bonus = np.log1p((split_rows["target_bps"] - split_rows["stop_bps"]).clip(lower=0.0))
     split_rows["score"] = (
-        split_rows["avg_edge_eval"]
-        + split_rows["avg_price_plan_edge_eval"] * 0.5
-        + split_rows["min_margin_eval"] * 20.0
+        split_rows["avg_price_plan_edge_eval"] * 8.0
+        + split_rows["min_price_plan_edge_eval"] * 3.0
+        + split_rows["min_margin_eval"] * 80.0
+        + split_rows["avg_edge_eval"] * 0.05
         - positive_rate_penalty
         + spread_bonus
     )
@@ -287,8 +291,10 @@ def _plan_summary(rows: pd.DataFrame) -> pd.DataFrame:
 
 def _select_best_plan(summary: pd.DataFrame) -> dict[str, Any]:
     candidates = summary[
-        (summary["min_positive_label_rate_eval"] >= 0.08)
-        & (summary["max_positive_label_rate_eval"] <= 0.45)
+        (summary["min_positive_label_rate_eval"] >= 0.03)
+        & (summary["max_positive_label_rate_eval"] <= 0.55)
+        & (summary["avg_price_plan_edge_eval"] > 0.0)
+        & (summary["min_price_plan_edge_eval"] > -1.0)
         & (summary["target_bps"] > summary["stop_bps"])
     ]
     if candidates.empty:
@@ -305,6 +311,7 @@ def _select_best_plan(summary: pd.DataFrame) -> dict[str, Any]:
         "score": float(row["score"]),
         "avg_edge_eval": float(row["avg_edge_eval"]),
         "avg_price_plan_edge_eval": float(row["avg_price_plan_edge_eval"]),
+        "min_price_plan_edge_eval": float(row["min_price_plan_edge_eval"]),
         "min_margin_eval": float(row["min_margin_eval"]),
         "min_positive_label_rate_eval": float(row["min_positive_label_rate_eval"]),
         "max_positive_label_rate_eval": float(row["max_positive_label_rate_eval"]),
@@ -332,7 +339,7 @@ def _markdown_report(payload: dict[str, Any], summary: pd.DataFrame) -> str:
         "",
         _markdown_table(top),
         "",
-        "说明：positive_label_rate 和 avg_expected_net_edge_bps 按“未来窗口最大可拿净收益”统计；target_hit_rate、stop_hit_rate、avg_price_plan_net_edge_bps 只用来检查固定止盈止损计划是否顺手。这里选的是下一轮实验用的价格计划，不是上线结论。真正能不能上线仍然看模型训练、PM 搜索、validation_locked 和 latest_stress 回测。",
+        "说明：positive_label_rate 和 avg_price_plan_net_edge_bps 按固定止盈止损计划统计；avg_expected_net_edge_bps 只是辅助观察未来最大可拿空间，不能单独决定价格计划。这里选的是下一轮实验用的价格计划，不是上线结论。真正能不能上线仍然看模型训练、PM 搜索、validation_locked 和 latest_stress 回测。",
         "",
     ]
     return "\n".join(lines)

training/trader_training/schemas.py

@@ -5,7 +5,7 @@ from typing import Any
 
 
 FEATURE_VERSION = "feature-v4-p2-book-cross"
-LABEL_VERSION = "label-v4-p1-max-edge"
+LABEL_VERSION = "label-v4-p5-entry-quality"
 SPLIT_VERSION = "split-v4-p0"
 MODEL_BUNDLE_VERSION = "trader-v4-btc-p0"
 CALIBRATION_BUNDLE_VERSION = "cal-v4-btc-p0"