"""Build machine-learning-ready feature parquet datasets. Two datasets are produced for the current finalist #1 (trial #324, the one in ``registry/``): 1. ``studies/features/trade_features_gold_scalper_pro_is2025.parquet`` (+ .csv) — one row per closed trade, with the indicator + market state at entry time. Used for "which entry conditions predict winning trades" classification / feature analysis. 2. ``studies/features/trial_features_gold_scalper_pro_is2025.parquet`` (+ .csv) — one row per Optuna trial, with all params + the objective's reported metrics. Used for parameter- sensitivity analysis, parameter importance, and meta-learning. Both are written as Parquet (binary, typed) and CSV (human-readable) so you can ``pd.read_parquet`` for ML or open the CSV in Excel. Usage: python scripts/build_feature_datasets.py """ from __future__ import annotations import sys from pathlib import Path PROJECT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(PROJECT)) import json import optuna import numpy as np import pandas as pd from shared.core.engine import SizingInputs from shared.data.loaders import load_bars from shared.indicators.base import atr, ema, rsi from strategies.gold_scalper_pro.instruments import XAUUSD_REAL from strategies.gold_scalper_pro.scalper_engine import ( ScalperEngine, engine_kwargs_from_params, ) from strategies.gold_scalper_pro.search_space import FROZEN_BASELINE from strategies.gold_scalper_pro.signals import build_signals # ───────────────────────────────────────────────────────────────────────────── # Trade-level features # ───────────────────────────────────────────────────────────────────────────── TRADE_FEATURE_COLUMNS = [ # identity "trade_id", # timing "entry_time", "exit_time", "duration_minutes", "hour_of_day", "day_of_week", # trade "direction", "entry_price", "exit_price", "lots", "pnl", "swap", "exit_reason", "is_win", "pnl_pct", # sizing context "equity_at_entry", "risk_percent", "sl_distance", "sl_distance_pct", # indicators at entry (computed on the SIGNAL bar, i.e. one bar before fill) "atr_at_entry", "atr_pct_of_close", "rsi_at_entry", "fast_ema_at_entry", "slow_ema_at_entry", "dist_to_fast", "dist_to_fast_atr", "dist_to_slow", "fast_minus_slow", "trend_up", "close_at_entry", "high_at_entry", "low_at_entry", "spread_at_entry", "spread_atr_ratio", # ML target candidates (the user can pick) "label_win", # binary 0/1 — classification target "label_pnl_zscore", # z-score of pnl across all trades — regression target ] def build_trade_features( bars: pd.DataFrame, m1_bars: pd.DataFrame, params: dict, is_start: pd.Timestamp, is_end: pd.Timestamp, ) -> pd.DataFrame: """Run finalist #1 with warmup, then build a per-trade feature table.""" # Warmup pattern: signals on full bars, slice to IS window for engine. pack = build_signals(params, bars, XAUUSD_REAL) ts = pd.to_datetime(bars["timestamp"].to_numpy()) lo = int(ts.searchsorted(is_start, side="left")) hi = int(ts.searchsorted(is_end, side="left")) bars_is = bars.iloc[lo:hi].reset_index(drop=True) sig_long = pack.signals_long[lo:hi] sig_short = pack.signals_short[lo:hi] sl_p = pack.sl_prices[lo:hi] tp_p = pack.tp_prices[lo:hi] m1_ts = pd.to_datetime(m1_bars["timestamp"].to_numpy()) m1_lo = int(m1_ts.searchsorted(is_start, side="left")) m1_hi = int(m1_ts.searchsorted(is_end, side="left")) m1_is = m1_bars.iloc[m1_lo:m1_hi].reset_index(drop=True) engine = ScalperEngine() result = engine.run( bars_is, sig_long, sig_short, sl_p, tp_p, XAUUSD_REAL, SizingInputs(), 1000.0, m1_bars=m1_is, **engine_kwargs_from_params(params), ) trades = result.trades if not trades: return pd.DataFrame(columns=TRADE_FEATURE_COLUMNS) # Recompute indicator arrays on the full bars (same as build_signals), # then index by each trade's entry_time to get the at-entry state. close = bars["close"].to_numpy(dtype=float) high = bars["high"].to_numpy(dtype=float) low = bars["low"].to_numpy(dtype=float) atr_arr = atr(high, low, close, int(params["InpAtrPeriod"])) rsi_arr = rsi(close, int(params["InpRsiPeriod"])) fast_e = ema(close, int(params["InpFastEmaPeriod"])) slow_e = ema(close, int(params["InpSlowEmaPeriod"])) spread_pts = bars["spread"].to_numpy(dtype=float) if "spread" in bars else np.zeros(len(bars)) spread_px = spread_pts * XAUUSD_REAL.point bars_ts = pd.to_datetime(bars["timestamp"].to_numpy()) # Pre-build a ts → idx lookup so per-trade search is O(log n). # Each trade's entry_time is the bar AFTER the signal bar (the engine fills # at next-bar open), so we look up the bar index for entry_time, then take # idx-1 as the signal bar (where indicators are read). bar_idx_at = pd.Index(bars_ts) def signal_idx(entry_time: pd.Timestamp) -> int: # The engine records entry_time as the fill bar's timestamp. We want # the PREVIOUS bar (the signal bar where indicators were ready). pos = bar_idx_at.get_indexer([entry_time], method="pad")[0] return int(pos) - 1 if pos > 0 else 0 rows = [] risk_pct = float(params["InpRiskPercent"]) atr_sl_mult = float(params["InpAtrSLMult"]) for i, tr in enumerate(trades): sig_i = signal_idx(tr.entry_time) if sig_i < 0 or sig_i >= len(close): continue c_sig = close[sig_i] atr_sig = atr_arr[sig_i] rsi_sig = rsi_arr[sig_i] fast_sig = fast_e[sig_i] slow_sig = slow_e[sig_i] sp_sig = spread_px[sig_i] sl_dist = atr_sl_mult * atr_sig duration_min = (tr.exit_time - tr.entry_time).total_seconds() / 60.0 pnl_pct = (tr.pnl / max(tr.entry_price * tr.lots * XAUUSD_REAL.contract_size, 1e-9)) * 100.0 rows.append({ "trade_id": i + 1, "entry_time": tr.entry_time, "exit_time": tr.exit_time, "duration_minutes": duration_min, "hour_of_day": int(tr.entry_time.hour), "day_of_week": int(tr.entry_time.dayofweek), "direction": tr.direction.name, "entry_price": tr.entry_price, "exit_price": tr.exit_price, "lots": tr.lots, "pnl": tr.pnl, "swap": tr.swap, "exit_reason": tr.exit_reason, "is_win": bool(tr.pnl > 0), "pnl_pct": pnl_pct, "equity_at_entry": float("nan"), # filled below from equity curve "risk_percent": risk_pct, "sl_distance": sl_dist, "sl_distance_pct": (sl_dist / c_sig) * 100.0, "atr_at_entry": atr_sig, "atr_pct_of_close": (atr_sig / c_sig) * 100.0, "rsi_at_entry": rsi_sig, "fast_ema_at_entry": fast_sig, "slow_ema_at_entry": slow_sig, "dist_to_fast": abs(c_sig - fast_sig), "dist_to_fast_atr": abs(c_sig - fast_sig) / atr_sig if atr_sig > 0 else float("nan"), "dist_to_slow": abs(c_sig - slow_sig), "fast_minus_slow": fast_sig - slow_sig, "trend_up": bool(fast_sig > slow_sig and c_sig > slow_sig), "close_at_entry": c_sig, "high_at_entry": high[sig_i], "low_at_entry": low[sig_i], "spread_at_entry": sp_sig, "spread_atr_ratio": sp_sig / atr_sig if atr_sig > 0 else float("nan"), "label_win": 1 if tr.pnl > 0 else 0, "label_pnl_zscore": float("nan"), # filled below }) df = pd.DataFrame(rows) if df.empty: return df # Approximate equity-at-entry from the equity curve (the engine samples # periodically; the closest sample before entry_time is a fair proxy). ec = result.equity_curve if not ec.empty and "equity" in ec.columns: ec_ts = pd.to_datetime(ec["timestamp"].to_numpy()) ec_eq = ec["equity"].to_numpy(dtype=float) ec_idx = pd.Index(ec_ts) positions = ec_idx.get_indexer(df["entry_time"].to_numpy(), method="pad") positions = np.where(positions < 0, 0, positions) df["equity_at_entry"] = ec_eq[positions] # Z-score of pnl across all trades — a regression-style label that # normalizes for the strategy's overall edge. if df["pnl"].std() > 0: df["label_pnl_zscore"] = (df["pnl"] - df["pnl"].mean()) / df["pnl"].std() return df[TRADE_FEATURE_COLUMNS] # ───────────────────────────────────────────────────────────────────────────── # Trial-level features # ───────────────────────────────────────────────────────────────────────────── TRIAL_FEATURE_COLUMNS = [ "trial_number", "state", # searched params (SEARCH_SPACE keys) "InpFastEmaPeriod", "InpSlowEmaPeriod", "InpRsiPeriod", "InpRsiBuyLevel", "InpRsiSellLevel", "InpPullbackAtrMult", "InpAtrPeriod", "InpMaxSpreadAtrPct", "InpRiskPercent", "InpAtrSLMult", "InpAtrTPMult", "InpBreakEvenPoints", "InpBreakEvenLock", "InpTrailStartPoints", "InpTrailStepPoints", "InpMaxTradesPerDay", "InpDailyLossLimit", "InpMinSecondsBetween", # objective output "score", # user_attrs metrics (written by objective on completion) "net_profit", "profit_factor", "total_trades", "max_equity_dd", "max_equity_dd_pct", "win_rate", "sharpe", # finalist tagging "is_finalist", "finalist_rank", # error info "error_message", ] def build_trial_features(study: optuna.Study, finalists_json: dict | None) -> pd.DataFrame: """One row per Optuna trial with params + metrics + finalist tag.""" # finalist map: trial_number → rank (0/1/2) finalist_map: dict[int, int] = {} if finalists_json: for rank, fl in enumerate(finalists_json.get("finalists", [])): tn = fl.get("trial_number") if tn is not None: finalist_map[int(tn)] = rank rows = [] for t in study.trials: # Skip RUNNING / WAITING trials — no metrics yet. if t.state == optuna.trial.TrialState.COMPLETE: state = "COMPLETE" elif t.state == optuna.trial.TrialState.PRUNED: state = "PRUNED" elif t.state == optuna.trial.TrialState.FAIL: state = "FAIL" else: continue # RUNNING / WAITING: skip ua = t.user_attrs or {} row = { "trial_number": t.number, "state": state, } # Fill params (None for missing → preserves column type). for p in TRIAL_FEATURE_COLUMNS: if p in ("trial_number", "state", "is_finalist", "finalist_rank", "error_message", "score"): continue if p in ("net_profit", "profit_factor", "total_trades", "max_equity_dd", "max_equity_dd_pct", "win_rate", "sharpe"): row[p] = ua.get(p) continue # param row[p] = t.params.get(p) row["score"] = t.value row["is_finalist"] = t.number in finalist_map row["finalist_rank"] = finalist_map.get(t.number) row["error_message"] = (ua.get("error") if state == "FAIL" else None) rows.append(row) return pd.DataFrame(rows, columns=TRIAL_FEATURE_COLUMNS) # ───────────────────────────────────────────────────────────────────────────── # Main # ───────────────────────────────────────────────────────────────────────────── def main() -> int: IS_START = pd.Timestamp("2025-01-01 00:00:00") IS_END = pd.Timestamp("2026-01-01 00:00:00") # ── Trial features ────────────────────────────────────────────────────── db = PROJECT / "studies" / "optuna" / "gold_scalper_pro_is2025.db" finalists_json_path = PROJECT / "studies" / "finalists" / "gold_scalper_pro_is2025-2026.json" print(f"=== trial-level features ===") print(f" study: gold_scalper_pro_is2025 ({db.relative_to(PROJECT)})") study = optuna.load_study( study_name="gold_scalper_pro_is2025", storage=f"sqlite:///{db}", ) finalists_json = None if finalists_json_path.exists(): import json finalists_json = json.loads(finalists_json_path.read_text(encoding="utf-8")) print(f" finalists JSON: {len(finalists_json.get('finalists', []))} entries") trial_df = build_trial_features(study, finalists_json) trial_out_parquet = PROJECT / "studies" / "features" / "trial_features_gold_scalper_pro_is2025.parquet" trial_out_csv = PROJECT / "studies" / "features" / "trial_features_gold_scalper_pro_is2025.csv" trial_df.to_parquet(trial_out_parquet, index=False) trial_df.to_csv(trial_out_csv, index=False) print(f" → {trial_out_parquet.relative_to(PROJECT)} ({len(trial_df):,} rows)") print(f" → {trial_out_csv.relative_to(PROJECT)}") complete = trial_df[trial_df["state"] == "COMPLETE"] print(f" complete: {len(complete):,} finalists: {trial_df['is_finalist'].sum()}") # ── Trade features (finalist #1 only — the registered one) ────────────── print(f"\n=== trade-level features (finalist #1) ===") if finalists_json is None or not finalists_json.get("finalists"): print(" ERROR: finalists JSON missing — run reeval_finalist_forward.py first") return 1 f1 = finalists_json["finalists"][0] f1_trial = study.trials[f1["trial_number"]] params = {**FROZEN_BASELINE, **f1["params"]} print(f" finalist #1: trial #{f1_trial.number} score={f1['score']:.4f}") bars = load_bars(PROJECT / "data" / "XAUUSD_M5_2024-06-26_2026-06-26.parquet") m1 = load_bars(PROJECT / "data" / "XAUUSD_M1_2024-06-26_2026-06-26.parquet") print(f" bars : M5={len(bars):,} M1={len(m1):,}") trade_df = build_trade_features(bars, m1, params, IS_START, IS_END) trade_out_parquet = PROJECT / "studies" / "features" / "trade_features_gold_scalper_pro_is2025.parquet" trade_out_csv = PROJECT / "studies" / "features" / "trade_features_gold_scalper_pro_is2025.csv" trade_df.to_parquet(trade_out_parquet, index=False) trade_df.to_csv(trade_out_csv, index=False) print(f" → {trade_out_parquet.relative_to(PROJECT)} ({len(trade_df):,} rows)") print(f" → {trade_out_csv.relative_to(PROJECT)}") if not trade_df.empty: wins = trade_df["label_win"].sum() print(f" trades: {len(trade_df):,} wins: {wins} ({wins/len(trade_df):.1%}) " f"avg pnl: ${trade_df['pnl'].mean():.3f}") print(f" exit reasons:") for r, n in trade_df["exit_reason"].value_counts().items(): sub = trade_df[trade_df["exit_reason"] == r] print(f" {r:<14} {n:>5} ({n/len(trade_df):.1%}) " f"avg_pnl=${sub['pnl'].mean():.3f} win_rate={sub['label_win'].mean():.1%}") return 0 if __name__ == "__main__": raise SystemExit(main())