from __future__ import annotations import logging from typing import Any import numpy as np import pandas as pd from numpy.lib.stride_tricks import sliding_window_view from trader_training.io_utils import ( manifest, read_json, read_parquet, require_columns, run_root, sha256_json, to_utc_series, write_json, write_parquet, write_text, ) 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, "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": { "horizon_minutes": 45, "market_drawdown_bps": 60.0, "position_path_risk_bps": 35.0, "vol_expansion_ratio": 1.8, "spike_bps": 80.0, }, } DEFAULT_COST_CONFIG = { "fee_bps": 4.0, "slippage_bps": 2.0, "funding_cost_bps": 0.5, } ENTRY_LABEL_METHOD = "PRICE_PLAN_OUTCOME_V1" def _load_config(path, default): if path is None: return default value = read_json(path) merged = default.copy() for key, item in value.items(): if isinstance(item, dict) and isinstance(merged.get(key), dict): merged[key] = {**merged[key], **item} else: merged[key] = item return merged def _config_number(config: dict[str, Any], keys: tuple[str, ...], default: float) -> float: for key in keys: if key in config: return float(config[key]) return default def _base_frames(args: Any) -> tuple[pd.DataFrame, pd.DataFrame]: root = run_root(args) feature_path = args.feature_path or root / "feature" / "feature_frame.parquet" replay_path = args.replay_path or root / "replay" / "replay_1m.parquet" features = read_parquet(feature_path) replay = read_parquet(replay_path) require_columns(features, ("sample_id", "symbol", "event_time", "open_time_ms", "split_id", "walk_forward_fold", "data_quality_flag"), "feature_frame") require_columns(replay, ("symbol", "event_time", "open_time_ms", "open", "high", "low", "close", "spread_bps"), "replay_1m") features = features.copy() replay = replay.copy() features["event_time"] = to_utc_series(features["event_time"]) replay["event_time"] = to_utc_series(replay["event_time"]) replay = replay.sort_values(["symbol", "event_time"]).reset_index(drop=True) return features, replay PATH_STAT_COLUMNS = [ "symbol", "open_time_ms", "side", "target_hit", "stop_hit", "timeout_hit", "ambiguous_hit", "time_to_target_ms", "time_to_stop_ms", "time_to_exit_ms", "gross_edge_bps", "future_return_bps", "mfe_bps", "mae_bps", "future_spread_p80", "future_realized_vol_bps", ] def _empty_path_stats_frame() -> pd.DataFrame: return pd.DataFrame(columns=PATH_STAT_COLUMNS) def _first_hit_index(hit_window: np.ndarray) -> tuple[np.ndarray, np.ndarray]: hit_any = hit_window.any(axis=1) first_idx = np.argmax(hit_window, axis=1) first_idx = np.where(hit_any, first_idx, hit_window.shape[1] + 1) return hit_any, first_idx 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() grouped = group.sort_values("event_time").reset_index(drop=True) open_ms = grouped["open_time_ms"].astype("int64").to_numpy() close = grouped["close"].astype("float64").to_numpy() high = grouped["high"].astype("float64").to_numpy() low = grouped["low"].astype("float64").to_numpy() spread = grouped["spread_bps"].astype("float64").to_numpy() entry = close[:-horizon] exit_price = close[horizon:] current_open_ms = open_ms[:-horizon] bad_gap = (np.diff(open_ms) != 60_000).astype("int64") gap_cumsum = np.concatenate(([0], np.cumsum(bad_gap))) contiguous = (gap_cumsum[horizon:] - gap_cumsum[:-horizon]) == 0 finite = np.isfinite(entry) & np.isfinite(exit_price) valid = contiguous & finite future_high = sliding_window_view(high[1:], horizon) future_low = sliding_window_view(low[1:], horizon) future_spread = sliding_window_view(spread[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) if horizon > 1: log_close = np.log(np.clip(close, 1e-12, None)) log_return = np.diff(log_close) future_log_return = sliding_window_view(log_return[1:], horizon - 1) with np.errstate(all="ignore"): realized_vol_bps = np.nanstd(future_log_return, axis=1, ddof=1) * 10000.0 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) target_window = future_high >= target_price[:, None] stop_window = future_low <= stop_price[:, None] future_return_bps = (exit_price / entry - 1.0) * 10000.0 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: target_price = entry * (1.0 - target_bps / 10000.0) stop_price = entry * (1.0 + stop_bps / 10000.0) target_window = future_low <= target_price[:, None] stop_window = future_high >= stop_price[:, None] future_return_bps = (entry / exit_price - 1.0) * 10000.0 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) target_any, first_target_idx = _first_hit_index(target_window) stop_any, first_stop_idx = _first_hit_index(stop_window) ambiguous_hit = target_any & stop_any & (first_target_idx == first_stop_idx) target_hit = target_any & (first_target_idx < first_stop_idx) 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( { "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, (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"), "mae_bps": mae_bps.astype("float64"), "future_spread_p80": spread_p80.astype("float64"), "future_realized_vol_bps": realized_vol_bps.astype("float64"), } ) return out.loc[valid, PATH_STAT_COLUMNS].reset_index(drop=True) 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 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, plan_config=plan_config) frames.append(stats) logging.info( "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 planMethod=%s rowCount=%s", horizon, (plan_config or {}).get("plan_method", "FIXED_TARGET_STOP_V1"), len(out), ) return out def write_price_plan_context(args: Any) -> None: root = run_root(args) cost = _load_config(args.cost_config_path, DEFAULT_COST_CONFIG) labels = _load_config(args.label_config_path, DEFAULT_LABEL_CONFIG) entry = labels["entry"] cost_bps = float(cost["fee_bps"]) + float(cost["slippage_bps"]) + float(cost["funding_cost_bps"]) context = { "pricePlanId": args.price_plan_id, "pricePlanConfigHash": sha256_json({"entry": entry, "cost": cost, "entry_label_method": ENTRY_LABEL_METHOD}), "stopDistanceBps": float(entry["stop_bps"]), "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) frame = pd.DataFrame([{ "price_plan_id": context["pricePlanId"], "price_plan_hash": context["pricePlanConfigHash"], "target_bps": context["targetDistanceBps"], "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) def build_direction_labels(args: Any) -> None: root = run_root(args) config = _load_config(args.label_config_path, DEFAULT_LABEL_CONFIG)["direction"] features, replay = _base_frames(args) horizon = int(config["horizon_minutes"]) replay = replay[["symbol", "event_time", "open_time_ms", "close"]].copy() future = replay[["symbol", "open_time_ms", "close"]].copy() future["open_time_ms"] = future["open_time_ms"].astype("int64") - horizon * 60_000 future = future.rename(columns={"close": "future_close"}) merged = features.merge(replay[["symbol", "open_time_ms", "close"]], on=["symbol", "open_time_ms"], how="left") merged = merged.merge(future, on=["symbol", "open_time_ms"], how="left") merged["future_return_bps"] = (merged["future_close"] / merged["close"] - 1.0) * 10000.0 merged["direction_label"] = np.select( [merged["future_return_bps"] >= float(config["long_threshold_bps"]), merged["future_return_bps"] <= float(config["short_threshold_bps"])], ["LONG", "SHORT"], default="NEUTRAL", ) out = pd.DataFrame( { "sample_id": merged["sample_id"], "symbol": merged["symbol"], "event_time": merged["event_time"], "horizon_minutes": horizon, "future_return_bps": merged["future_return_bps"], "direction_label": merged["direction_label"], "long_target": merged["direction_label"].eq("LONG").astype("int8"), "short_target": merged["direction_label"].eq("SHORT").astype("int8"), "neutral_target": merged["direction_label"].eq("NEUTRAL").astype("int8"), "split_id": merged["split_id"], "walk_forward_fold": merged["walk_forward_fold"], "label_version": LABEL_VERSION, } ).dropna(subset=["future_return_bps"]) path = root / "label" / "direction_labels.parquet" data_hash = write_parquet(path, out) _write_label_manifest(root / "label" / "direction_labels.manifest.json", path, out, data_hash) _write_distribution_report(root / "label" / "direction_label_report.md", out, "direction_label") logging.info("trader.training.direction_labels_written runId=%s rowCount=%s", args.run_id, len(out)) def build_entry_labels(args: Any) -> None: root = run_root(args) labels = _load_config(args.label_config_path, DEFAULT_LABEL_CONFIG) cost = _load_config(args.cost_config_path, DEFAULT_COST_CONFIG) plan_path = args.price_plan_context_path or root / "label" / "price_plan_context.json" plan = read_json(plan_path) features, replay = _base_frames(args) entry_conf = labels["entry"] cost_bps = float(cost["fee_bps"]) + float(cost["slippage_bps"]) + float(cost["funding_cost_bps"]) stats = _build_path_stats( replay, int(entry_conf["max_hold_minutes"]), float(entry_conf["target_bps"]), float(entry_conf["stop_bps"]), plan_config=entry_conf, ) feature_columns = [ "sample_id", "symbol", "event_time", "open_time_ms", "split_id", "walk_forward_fold", "spread_bps", "spread_rank_24h_pct", "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 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"] = target_method merged["label_version"] = LABEL_VERSION out = merged[ [ "sample_id", "symbol", "event_time", "side", "price_plan_id", "price_plan_hash", "target_hit", "stop_hit", "timeout_hit", "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", "expected_net_edge_bps", "entry_target", "label_method", "split_id", "walk_forward_fold", "label_version", ] ].copy() path = root / "label" / "entry_labels.parquet" data_hash = write_parquet(path, out) _write_label_manifest(root / "label" / "entry_labels.manifest.json", path, out, data_hash) _write_distribution_report(root / "label" / "entry_label_report.md", out, "entry_target") logging.info("trader.training.entry_labels_written runId=%s rowCount=%s", args.run_id, len(out)) def build_position_state_samples(args: Any) -> None: root = run_root(args) entry_path = args.entry_label_path or root / "label" / "entry_labels.parquet" entry = read_parquet(entry_path) if entry.empty: raise ValueError("entry labels are required before building position samples") samples = entry[entry["entry_target"] == 1].copy() samples["position_age_minutes"] = 0 samples["unrealized_pnl_bps"] = 0.0 samples["mfe_bps"] = pd.to_numeric(samples["mfe_bps"], errors="coerce").fillna(0.0).clip(lower=0) samples["mae_bps"] = pd.to_numeric(samples["mae_bps"], errors="coerce").fillna(0.0).clip(lower=0) path = root / "label" / "position_state_samples.parquet" data_hash = write_parquet(path, samples) write_json(root / "label" / "position_state_samples.manifest.json", manifest(path, {"row_count": len(samples), "data_hash_sha256": data_hash})) logging.info("trader.training.position_samples_written runId=%s rowCount=%s", args.run_id, len(samples)) def build_continue_exit_risk_labels(args: Any) -> None: root = run_root(args) labels = _load_config(args.label_config_path, DEFAULT_LABEL_CONFIG) cost = _load_config(args.cost_config_path, DEFAULT_COST_CONFIG) plan = read_json(args.price_plan_context_path or root / "label" / "price_plan_context.json") features, replay = _base_frames(args) cost_bps = float(cost["fee_bps"]) + float(cost["slippage_bps"]) + float(cost["funding_cost_bps"]) horizon = int(labels["continue"]["horizon_minutes"]) target_bps = float(plan["targetDistanceBps"]) stop_bps = float(plan["stopDistanceBps"]) 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"}) short_stats = short_stats.rename(columns={"short_symbol": "symbol", "short_open_time_ms": "open_time_ms"}) feature_columns = [ "sample_id", "symbol", "event_time", "open_time_ms", "split_id", "walk_forward_fold", "spread_bps", "spread_rank_24h_pct", "realized_vol_15m_bps", ] if "ret_15m_bps" in features.columns: feature_columns.append("ret_15m_bps") merged = features[feature_columns].merge(long_stats, on=["symbol", "open_time_ms"], how="inner") merged = merged.merge(short_stats, on=["symbol", "open_time_ms"], how="inner") min_continue = float(labels["continue"]["min_expected_continue_edge_bps"]) adverse_threshold = float(labels["exit"]["adverse_move_bps"]) current_vol = merged["realized_vol_15m_bps"].astype(float).fillna(0.0).clip(lower=1.0) risk_config = labels["risk"] market_risk_threshold = _config_number( risk_config, ("market_path_risk_threshold_bps", "market_drawdown_bps"), 60.0, ) position_risk_threshold = _config_number( risk_config, ("position_path_risk_threshold_bps", "position_path_risk_bps"), 35.0, ) spike_threshold = _config_number(risk_config, ("spike_1m_threshold_bps", "spike_bps"), 80.0) vol_expansion_ratio = _config_number(risk_config, ("vol_expansion_ratio",), 1.8) 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: reversal = (np.sign(merged["long_future_return_bps"]) != np.sign(merged["ret_15m_bps"])).astype("int8") else: reversal = pd.Series(0, index=merged.index, dtype="int8") future_vol = merged["long_future_realized_vol_bps"].fillna(0.0) volatility_expansion = future_vol >= current_vol * vol_expansion_ratio spike = max_path_move >= spike_threshold market_risk = (path_risk >= market_risk_threshold) | spike | volatility_expansion liquidity_deterioration = merged["spread_rank_24h_pct"].astype(float).fillna(0.0) >= 0.90 rows_continue = pd.DataFrame( { "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) | 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"], "walk_forward_fold": merged["walk_forward_fold"], "label_version": LABEL_VERSION, } ) rows_exit = pd.DataFrame( { "sample_id": merged["sample_id"], "symbol": merged["symbol"], "event_time": merged["event_time"], "long_exit_target": ((merged["long_stop_hit"] == 1) | (merged["long_mae_bps"] >= adverse_threshold)).astype("int8"), "short_exit_target": ((merged["short_stop_hit"] == 1) | (merged["short_mae_bps"] >= adverse_threshold)).astype("int8"), "long_adverse_move_bps": merged["long_mae_bps"], "short_adverse_move_bps": merged["short_mae_bps"], "adverse_move_prob_label": (path_risk >= adverse_threshold).astype("int8"), "reversal_prob_label": reversal, "stop_hit_prob_label": ((merged["long_stop_hit"] == 1) | (merged["short_stop_hit"] == 1)).astype("int8"), "stagnation_prob_label": (merged["long_future_return_bps"].abs() <= float(labels["exit"]["stagnation_abs_return_bps"])).astype("int8"), "split_id": merged["split_id"], "walk_forward_fold": merged["walk_forward_fold"], "label_version": LABEL_VERSION, } ) rows_risk = pd.DataFrame( { "sample_id": merged["sample_id"], "symbol": merged["symbol"], "event_time": merged["event_time"], "market_risk_target": market_risk.astype("int8"), "market_path_risk_bps": path_risk, "long_position_path_risk_bps": merged["long_mae_bps"], "short_position_path_risk_bps": merged["short_mae_bps"], "long_position_risk_target": ((merged["long_mae_bps"] >= position_risk_threshold) | (merged["long_stop_hit"] == 1)).astype("int8"), "short_position_risk_target": ((merged["short_mae_bps"] >= position_risk_threshold) | (merged["short_stop_hit"] == 1)).astype("int8"), "market_drawdown_prob_label": (path_risk >= market_risk_threshold).astype("int8"), "volatility_expansion_prob_label": volatility_expansion.astype("int8"), "spike_prob_label": spike.astype("int8"), "liquidity_deterioration_prob_label": liquidity_deterioration.astype("int8"), "position_drawdown_prob_label": (path_risk >= position_risk_threshold).astype("int8"), "split_id": merged["split_id"], "walk_forward_fold": merged["walk_forward_fold"], "label_version": LABEL_VERSION, } ) outputs = [ ("continue", pd.DataFrame(rows_continue), "long_continue_target"), ("exit", pd.DataFrame(rows_exit), "long_exit_target"), ("risk", pd.DataFrame(rows_risk), "market_risk_target"), ] report_parts = ["# Continue Exit Risk Label Report", ""] report_parts.extend( [ "## Risk Thresholds", "", str( { "market_risk_threshold_bps": market_risk_threshold, "position_risk_threshold_bps": position_risk_threshold, "spike_threshold_bps": spike_threshold, "vol_expansion_ratio": vol_expansion_ratio, } ), "", ] ) for name, frame, target in outputs: path = root / "label" / f"{name}_labels.parquet" data_hash = write_parquet(path, frame) _write_label_manifest(root / "label" / f"{name}_labels.manifest.json", path, frame, data_hash) report_parts.append(f"## {name}") report_parts.append("") report_parts.append(str(frame[target].value_counts(dropna=False).to_dict() if not frame.empty else {})) report_parts.append("") logging.info("trader.training.%s_labels_written runId=%s rowCount=%s", name, args.run_id, len(frame)) write_text(root / "label" / "continue_exit_risk_label_report.md", "\n".join(report_parts) + "\n") def _write_label_manifest(path, parquet_path, frame: pd.DataFrame, data_hash: str) -> None: write_json(path, manifest(parquet_path, {"row_count": len(frame), "label_version": LABEL_VERSION, "data_hash_sha256": data_hash})) def _write_distribution_report(path, frame: pd.DataFrame, column: str) -> None: counts = frame[column].value_counts(dropna=False).to_dict() if not frame.empty else {} lines = ["# Label Report", "", f"- row_count: {len(frame)}", f"- target_column: {column}", f"- distribution: {counts}", ""] write_text(path, "\n".join(lines))