quant-trader-service/training/trader_training/pm.py

from __future__ import annotations

import itertools
import logging
from typing import Any

import numpy as np
import pandas as pd

from trader_training.io_utils import read_json, read_parquet, run_root, sha256_json, write_json, write_parquet, write_text
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,
        "open": {
            "longOpenProb": 0.58,
            "shortOpenProb": 0.58,
            "minLongEntryProb": 0.55,
            "minShortEntryProb": 0.55,
            "maxMarketRiskProb": 0.45,
            "minExpectedEdgeBps": 3.0,
            "minDirectionMargin": 0.03,
            "minLiquidityCapacityRatio": 0.10,
            "maxOodScore": 0.80,
        },
        "add": {
            "minLongProb": 0.60,
            "minShortProb": 0.60,
            "minContinueProb": 0.58,
            "minEntryProb": 0.55,
            "maxExitProb": 0.45,
            "maxMarketRiskProb": 0.45,
            "maxPositionRiskProb": 0.50,
            "minExpectedEdgeBps": 3.0,
            "minContinueVsExitEdgeBps": 0.0,
            "minLiquidityCapacityRatio": 0.10,
            "minPostTradeLiquidationBufferBps": 500.0,
            "maxAddCount": 3,
            "cooldownMinutes": 5,
        },
        "exit": {
            "closeExitProb": 0.70,
            "closePositionRiskProb": 0.70,
            "closeMarketRiskProb": 0.70,
            "closeContinueMax": 0.25,
            "reduceAdverseMoveProb": 0.62,
            "reduceContinueMin": 0.35,
            "reduceContinueMax": 0.70,
            "minProfitForReduceBps": 5.0,
            "maxPositionPathRiskBps": 80.0,
        },
        "sizing": {
            "baseRatio": 0.80,
            "minInitialRatio": 0.05,
            "maxSingleLegRatio": 1.0,
            "minAddRatio": 0.02,
            "maxAddRatio": 0.25,
            "maxTotalPositionRatio": 1.0,
            "minEdgeBps": 3.0,
            "maxLossPerTradeBps": 80.0,
            "maxLiquidityUsageRatio": 0.20,
            "uncertaintyPenaltyMultiplier": 0.50,
            "minPostTradeLiquidationBufferBps": 500.0,
        },
    }


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
    best_metrics: dict[str, Any] | None = None
    best_trades = pd.DataFrame()

    for thresholds in _threshold_candidates():
        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})
        if score > best_score:
            best_score = score
            best_thresholds = thresholds
            best_metrics = metrics
            best_trades = trades

    if best_thresholds is None or best_metrics is None:
        raise ValueError("PM threshold search did not evaluate any candidate")

    config = _pm_config_from_thresholds(best_thresholds)
    threshold_stability = {
        "source": "tune_predictions_and_entry_labels",
        "method": "deterministic_grid_search_v1",
        "candidate_count": len(candidate_rows),
        "best_score": best_score,
        "best_metrics": best_metrics,
    }
    payload = {
        "pm_config_version": PM_CONFIG_VERSION,
        "config": config,
        "config_hash_sha256": sha256_json(config),
        "threshold_stability_json": threshold_stability,
    }
    candidate_frame = pd.DataFrame(candidate_rows).sort_values("score", ascending=False).reset_index(drop=True)
    equity_curve = _equity_curve(best_trades)
    regime_metrics = _regime_metrics(best_trades)
    write_json(root / "pm-search" / "position_manager_config.json", payload)
    write_json(root / "pm-search" / "pm_threshold_config.json", payload)
    write_text(root / "pm-search" / "pm_search_candidates.csv", candidate_frame.to_csv(index=False))
    write_parquet(root / "pm-search" / "pm_backtest_trades.parquet", best_trades)
    write_text(root / "pm-search" / "pm_equity_curve.csv", equity_curve.to_csv(index=False))
    write_text(root / "pm-search" / "pm_regime_metrics.csv", regime_metrics.to_csv(index=False))
    _write_pm_report(root / "pm-search" / "pm_threshold_report.md", candidate_frame, best_thresholds, best_metrics)
    _write_pm_report(root / "pm-search" / "pm_search_report.md", candidate_frame, best_thresholds, best_metrics)
    logging.info(
        "trader.training.pm_thresholds_searched runId=%s candidateCount=%s bestScore=%.6f tradeCount=%s totalWeightedEdgeBps=%.6f",
        args.run_id,
        len(candidate_rows),
        best_score,
        best_metrics["trade_count"],
        best_metrics["total_weighted_edge_bps"],
    )


def integrated_backtest(args: Any) -> None:
    root = run_root(args)
    config_path = root / "pm-search" / "position_manager_config.json"
    if not config_path.is_file():
        raise FileNotFoundError(f"PM config is required before backtest: {config_path}")
    pm_payload = read_json(config_path)
    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.
    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"]),
        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"]),
        pm_payload["config"],
        price_plan,
    )
    stress_trades["eval_split"] = LATEST_STRESS_SPLIT
    trade_parts = [part for part in (tune_trades, validation_locked_trades, stress_trades) if not part.empty]
    trades = pd.concat(trade_parts, ignore_index=True) if trade_parts else _empty_trade_frame()
    metrics = {
        TUNE_SPLIT: _trade_metrics(tune_trades),
        VALIDATION_LOCKED_SPLIT: _trade_metrics(validation_locked_trades),
        LATEST_STRESS_SPLIT: _trade_metrics(stress_trades),
        "combined": _trade_metrics(trades),
    }
    status, status_reasons = _backtest_status(metrics)
    equity_curve = _equity_curve(trades)
    regime_metrics = _regime_metrics(trades)
    result = {
        "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,
    }
    write_json(root / "backtest" / "backtest_manifest.json", result)
    write_parquet(root / "backtest" / "backtest_trades.parquet", trades)
    write_text(root / "backtest" / "equity_curve.csv", equity_curve.to_csv(index=False))
    write_text(root / "backtest" / "regime_metrics.csv", regime_metrics.to_csv(index=False))
    _write_backtest_report(root / "backtest" / "backtest_report.md", result)
    _write_failure_cases(root / "backtest" / "failure_cases.md", trades)
    _write_no_baseline_ablation(root / "backtest" / "direction_ablation_backtest_report.md")
    logging.info(
        "trader.training.backtest_written runId=%s status=%s tradeCount=%s totalWeightedEdgeBps=%.6f maxDrawdownBps=%.6f",
        args.run_id,
        status,
        metrics[VALIDATION_LOCKED_SPLIT]["trade_count"],
        metrics[VALIDATION_LOCKED_SPLIT]["total_weighted_edge_bps"],
        metrics[VALIDATION_LOCKED_SPLIT]["max_drawdown_bps"],
    )


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",
        VALIDATION_LOCKED_SPLIT: "validation_locked_predictions.parquet",
        LATEST_STRESS_SPLIT: "latest_stress_predictions.parquet",
    }
    prediction_file = prediction_files[split_id]
    direction = read_parquet(root / "model" / "direction" / prediction_file)
    entry = read_parquet(root / "model" / "entry" / prediction_file).rename(
        columns={
            "long_expected_net_edge_bps": "pred_long_expected_net_edge_bps",
            "short_expected_net_edge_bps": "pred_short_expected_net_edge_bps",
        }
    )
    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_actual_plan_net_edge_bps": "actual_long_plan_edge_bps",
            "short_actual_plan_net_edge_bps": "actual_short_plan_edge_bps",
        }
    )
    entry_plan_outcome = _entry_plan_outcome_frame(root)
    entry_cols = [
        "sample_id",
        "long_entry_prob",
        "short_entry_prob",
        "pred_long_expected_net_edge_bps",
        "pred_short_expected_net_edge_bps",
    ]
    risk_cols = ["sample_id", "market_risk_prob", "long_position_risk_prob", "short_position_risk_prob"]
    actual_cols = ["sample_id", "actual_long_plan_edge_bps", "actual_short_plan_edge_bps", "long_entry_target", "short_entry_target"]
    missing_actual_cols = sorted(set(actual_cols) - set(entry_dataset.columns))
    if missing_actual_cols:
        raise ValueError(f"entry_train is missing actual plan edge columns for PM: {missing_actual_cols}")
    frame = (
        direction[["sample_id", "symbol", "event_time", "split_id", "long_prob", "short_prob", "neutral_prob"]]
        .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_ACTUAL_PLAN_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 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.60, 0.70, 1.01],
        [0.50, 0.60, 0.70, 1.01],
        [0.30, 0.50, 0.70, 0.85],
        [0.45, 0.65],
        [3.0, 8.0, 15.0, 25.0],
        [0.02, 0.06, 0.10],
    )
    return [
        {
            "long_open_prob": long_prob,
            "short_open_prob": short_prob,
            "min_entry_prob": entry_prob,
            "max_market_risk_prob": risk_prob,
            "min_expected_edge_bps": edge_bps,
            "min_direction_margin": margin,
        }
        for long_prob, short_prob, entry_prob, risk_prob, edge_bps, margin in values
    ]


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"])
        & (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"])
        & (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))
    )
    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:
        return _empty_trade_frame()
    trades["side"] = side[side != ""]
    is_long = trades["side"].eq("LONG")
    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["long_trade_net_edge_bps"], trades["short_trade_net_edge_bps"])
    trades["label_actual_plan_edge_bps"] = np.where(is_long, trades["actual_long_plan_edge_bps"], trades["actual_short_plan_edge_bps"])
    trades["entry_target"] = np.where(is_long, trades["long_entry_target"], trades["short_entry_target"])
    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",
            "symbol",
            "event_time",
            "split_id",
            "side",
            "direction_prob",
            "entry_prob",
            "market_risk_prob",
            "predicted_edge_bps",
            "label_actual_plan_edge_bps",
            "actual_edge_bps",
            "entry_target",
            "time_to_exit_ms",
            "planned_ratio",
            "weighted_edge_bps",
            "threshold_hash",
        ]
    ].sort_values("event_time")


def _empty_trade_frame() -> pd.DataFrame:
    return pd.DataFrame(
        columns=[
            "sample_id",
            "symbol",
            "event_time",
            "split_id",
            "side",
            "direction_prob",
            "entry_prob",
            "market_risk_prob",
            "predicted_edge_bps",
            "label_actual_plan_edge_bps",
            "actual_edge_bps",
            "entry_target",
            "time_to_exit_ms",
            "planned_ratio",
            "weighted_edge_bps",
            "threshold_hash",
        ]
    )


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]:
    if trades.empty:
        return {
            "trade_count": 0,
            "win_rate": 0.0,
            "avg_actual_edge_bps": 0.0,
            "avg_weighted_edge_bps": 0.0,
            "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,
        }
    equity = trades["weighted_edge_bps"].astype(float).cumsum()
    drawdown = equity.cummax() - equity
    gains = trades.loc[trades["weighted_edge_bps"] > 0, "weighted_edge_bps"].astype(float).sum()
    losses = -trades.loc[trades["weighted_edge_bps"] < 0, "weighted_edge_bps"].astype(float).sum()
    return {
        "trade_count": int(len(trades)),
        "win_rate": float((trades["actual_edge_bps"].astype(float) > 0).mean()),
        "avg_actual_edge_bps": float(trades["actual_edge_bps"].astype(float).mean()),
        "avg_weighted_edge_bps": float(trades["weighted_edge_bps"].astype(float).mean()),
        "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()),
    }


def _max_consecutive_losses(values: np.ndarray) -> int:
    max_count = 0
    current = 0
    for value in values:
        if value < 0:
            current += 1
            max_count = max(max_count, current)
        else:
            current = 0
    return max_count


def _backtest_status(metrics: dict[str, dict[str, Any]]) -> tuple[str, list[str]]:
    reasons: list[str] = []
    validation_locked = metrics[VALIDATION_LOCKED_SPLIT]
    stress = metrics[LATEST_STRESS_SPLIT]
    if validation_locked["total_weighted_edge_bps"] <= 0:
        reasons.append("validation_locked_net_edge_not_positive")
    if validation_locked["trade_count"] < 80:
        reasons.append("validation_locked_trade_count_below_80")
    if validation_locked["profit_factor"] < 1.15:
        reasons.append("validation_locked_profit_factor_below_1.15")
    if validation_locked["avg_weighted_edge_bps"] <= 0:
        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:
        reasons.append("latest_stress_profit_factor_below_1.0")
    if stress["avg_weighted_edge_bps"] < -3.0:
        reasons.append("latest_stress_avg_trade_edge_below_minus_3")
    if stress["max_consecutive_losses"] > 10:
        reasons.append("latest_stress_max_consecutive_losses_above_10")
    if validation_locked["total_weighted_edge_bps"] > 0 and stress["total_weighted_edge_bps"] < -0.5 * validation_locked["total_weighted_edge_bps"]:
        reasons.append("latest_stress_loss_too_large_vs_validation")
    return ("REJECTED", reasons) if reasons else ("PASS", [])


def _score_thresholds(metrics: dict[str, Any]) -> float:
    if metrics["trade_count"] == 0:
        return -1_000_000.0
    # 最终上线门槛要求 validation_locked 至少 80 笔；调参区如果只挑几十笔，
    # 很容易是运气好，不是稳定规则，所以这里提前惩罚小样本阈值。
    low_sample_penalty = max(0, 120 - int(metrics["trade_count"])) * 1.5
    profit_factor_penalty = max(0.0, 1.15 - float(metrics["profit_factor"])) * 20.0
    negative_edge_penalty = max(0.0, -float(metrics["avg_weighted_edge_bps"])) * 40.0
    return (
        metrics["avg_weighted_edge_bps"] * np.sqrt(metrics["trade_count"])
        + metrics["total_weighted_edge_bps"] * 0.05
        - metrics["max_drawdown_bps"] * 0.25
        - low_sample_penalty
        - profit_factor_penalty
        - negative_edge_penalty
    )


def _pm_config_from_thresholds(thresholds: dict[str, float]) -> dict:
    config = default_pm_config()
    config["open"].update(
        {
            "longOpenProb": thresholds["long_open_prob"],
            "shortOpenProb": thresholds["short_open_prob"],
            "minLongEntryProb": thresholds["min_entry_prob"],
            "minShortEntryProb": thresholds["min_entry_prob"],
            "maxMarketRiskProb": thresholds["max_market_risk_prob"],
            "minExpectedEdgeBps": thresholds["min_expected_edge_bps"],
            "minDirectionMargin": thresholds["min_direction_margin"],
        }
    )
    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
    return config


def _thresholds_from_config(config: dict) -> dict[str, float]:
    open_config = config["open"]
    return {
        "long_open_prob": float(open_config["longOpenProb"]),
        "short_open_prob": float(open_config["shortOpenProb"]),
        "min_entry_prob": float(min(open_config["minLongEntryProb"], open_config["minShortEntryProb"])),
        "max_market_risk_prob": float(open_config["maxMarketRiskProb"]),
        "min_expected_edge_bps": float(open_config["minExpectedEdgeBps"]),
        "min_direction_margin": float(open_config["minDirectionMargin"]),
    }


def _equity_curve(trades: pd.DataFrame) -> pd.DataFrame:
    if trades.empty:
        return pd.DataFrame(columns=["event_time", "trade_index", "weighted_edge_bps", "equity_bps", "drawdown_bps"])
    curve = trades[["event_time", "weighted_edge_bps"]].copy().reset_index(drop=True)
    curve["trade_index"] = np.arange(1, len(curve) + 1)
    curve["equity_bps"] = curve["weighted_edge_bps"].astype(float).cumsum()
    curve["drawdown_bps"] = curve["equity_bps"].cummax() - curve["equity_bps"]
    return curve[["event_time", "trade_index", "weighted_edge_bps", "equity_bps", "drawdown_bps"]]


def _regime_metrics(trades: pd.DataFrame) -> pd.DataFrame:
    if trades.empty:
        return pd.DataFrame(columns=["split_id", "side", "trade_count", "win_rate", "avg_actual_edge_bps", "total_weighted_edge_bps"])
    rows = []
    for (split_id, side), group in trades.groupby(["split_id", "side"], sort=True):
        metrics = _trade_metrics(group)
        rows.append(
            {
                "split_id": split_id,
                "side": side,
                "trade_count": metrics["trade_count"],
                "win_rate": metrics["win_rate"],
                "avg_actual_edge_bps": metrics["avg_actual_edge_bps"],
                "total_weighted_edge_bps": metrics["total_weighted_edge_bps"],
            }
        )
    return pd.DataFrame(rows)


def _write_pm_report(path, candidates: pd.DataFrame, best_thresholds: dict[str, float], best_metrics: dict[str, Any]) -> None:
    top = candidates.head(10)
    lines = [
        "# PM Threshold Report",
        "",
        "本次不是固定写死阈值，而是在调参集上试一组可复现的阈值。PM 回测使用当前价格计划的真实净收益，并且开仓后按持仓结束时间加冷却时间阻止重叠开仓。",
        "",
        "## Best Thresholds",
        "",
        "```json",
        str(best_thresholds).replace("'", '"'),
        "```",
        "",
        "## Best Metrics",
        "",
        "```json",
        str(best_metrics).replace("'", '"'),
        "```",
        "",
        "## Top Candidates",
        "",
        _markdown_table(top.to_dict("records"), list(top.columns)),
        "",
    ]
    write_text(path, "\n".join(lines))


def _write_backtest_report(path, result: dict[str, Any]) -> None:
    lines = [
        "# Integrated Backtest Report",
        "",
        "这里用验证集模型输出和 PM 阈值生成交易明细，统计净收益、胜率、回撤和分段表现。收益按当前价格计划的真实净收益计算，不使用窗口内最大可拿收益。",
        "",
        "```json",
        str(result).replace("'", '"'),
        "```",
        "",
    ]
    write_text(path, "\n".join(lines))


def _write_failure_cases(path, trades: pd.DataFrame) -> None:
    worst = trades.sort_values("weighted_edge_bps").head(20) if not trades.empty else trades
    lines = [
        "# Backtest Failure Cases",
        "",
        "按加权净收益从差到好列出最差样本，方便回看特征、标签和阈值。",
        "",
        _markdown_table(worst.to_dict("records"), list(worst.columns)) if not worst.empty else "无交易样本。",
        "",
    ]
    write_text(path, "\n".join(lines))


def _write_no_baseline_ablation(path) -> None:
    lines = [
        "# Direction Ablation Backtest Report",
        "",
        "- status: NO_BASELINE",
        "- reason: 当前 run 目录没有旧 Direction 基准模型包，所以首版不能做只替换 Direction 的消融回测。",
        "- action: 后续版本必须拿上一版 ACTIVE 包做 baseline，再比较新 Direction 是否真的提升。",
        "",
    ]
    write_text(path, "\n".join(lines))


def _markdown_table(rows: list[dict[str, Any]], columns: list[str]) -> str:
    lines = ["| " + " | ".join(columns) + " |", "| " + " | ".join("---" for _ in columns) + " |"]
    for row in rows:
        lines.append("| " + " | ".join(str(row.get(column, "")) for column in columns) + " |")
    return "\n".join(lines)