mymt5opp/scripts/diag_size_after_warmup.py

"""Diagnostic: print first 10 trades' lots / SL / entry price for finalist #1
on the IS window, with indicator warmup applied (same code path as
reeval_finalist_forward.py). Compare lots vs the MT5 first trade
(2025.01.02 01:55 buy 0.16 lots @ 2624.05).
"""
from __future__ import annotations
import sys
from pathlib import Path
PROJECT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT))

import optuna
import pandas as pd

from shared.core.engine import SizingInputs
from shared.data.loaders import load_bars
from shared.optimizer.selector import select_diverse_topn
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, SEARCH_SPACE
from strategies.gold_scalper_pro.signals import build_signals


IS_START = pd.Timestamp("2025-01-01 00:00:00")
IS_END   = pd.Timestamp("2026-01-01 00:00:00")


def main() -> int:
    db = PROJECT / "studies" / "optuna" / "gold_scalper_pro_is2025.db"
    study = optuna.load_study(
        study_name="gold_scalper_pro_is2025",
        storage=f"sqlite:///{db}",
    )
    finalists = select_diverse_topn(study, n=3, ranges=SEARCH_SPACE)
    f1 = finalists[0]
    merged = {**FROZEN_BASELINE, **f1.params}
    print(f"finalist #1  trial #{f1.number}")
    print(f"  InpAtrSLMult={merged.get('InpAtrSLMult')}  InpAtrPeriod={merged.get('InpAtrPeriod')}")
    print(f"  InpRiskPercent={merged.get('InpRiskPercent')}  InpSizingMode={merged.get('InpSizingMode')}")

    full_m5 = load_bars(PROJECT / "data" / "XAUUSD_M5_2024-06-26_2026-06-26.parquet")
    full_m1 = load_bars(PROJECT / "data" / "XAUUSD_M1_2024-06-26_2026-06-26.parquet")

    pack = build_signals(merged, full_m5, XAUUSD_REAL)
    ts = pd.to_datetime(full_m5["timestamp"].to_numpy())
    lo = int(ts.searchsorted(IS_START, side="left"))
    hi = int(ts.searchsorted(IS_END,   side="left"))
    win_bars  = full_m5.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(full_m1["timestamp"].to_numpy())
    m1_lo     = int(m1_ts.searchsorted(IS_START, side="left"))
    m1_hi     = int(m1_ts.searchsorted(IS_END,   side="left"))
    win_m1    = full_m1.iloc[m1_lo:m1_hi].reset_index(drop=True)

    engine = ScalperEngine()
    result = engine.run(
        win_bars, sig_long, sig_short, sl_p, tp_p,
        XAUUSD_REAL, SizingInputs(), 1000.0,
        m1_bars=win_m1,
        **engine_kwargs_from_params(merged),
    )

    print(f"\n  total trades: {len(result.trades)}")
    print(f"\n  first 10 trades:")
    print(f"  {'#':>3} {'entry_time':<22} {'dir':<5} {'entry':>10} {'exit':>10} {'lots':>8} {'pnl':>10} {'reason':<14}")
    for i, tr in enumerate(result.trades[:10], 1):
        d = "LONG" if tr.direction.name == "LONG" else "SHORT"
        print(f"  {i:>3} {tr.entry_time.isoformat():<22} {d:<5} "
              f"{tr.entry_price:>10.2f} {tr.exit_price:>10.2f} "
              f"{tr.lots:>8.4f} {tr.pnl:>10.2f} {tr.exit_reason:<14}")

    print(f"\n  MT5 first trade (from report): 2025.01.02 01:55 buy 0.16 lots @ 2624.05")
    if result.trades:
        t0 = result.trades[0]
        print(f"  Python first trade           : {t0.entry_time.isoformat()} "
              f"{'LONG' if t0.direction.name=='LONG' else 'SHORT'} "
              f"{t0.lots:.4f} lots @ {t0.entry_price:.2f}")

    # Per-trade lot histogram: are most trades at the min lot (sizing bug) or
    # distributed across reasonable values (sizing working)?
    lots_arr = [t.lots for t in result.trades]
    if lots_arr:
        import numpy as np
        la = np.array(lots_arr)
        print(f"\n  lots stats  : min={la.min():.4f}  p25={np.percentile(la,25):.4f}  "
              f"median={np.median(la):.4f}  p75={np.percentile(la,75):.4f}  max={la.max():.4f}")
        print(f"  lots=0.01   : {(la==0.01).sum()}/{len(la)} ({(la==0.01).mean():.1%})")
        print(f"  lots>0.10   : {(la>0.10).sum()}/{len(la)} ({(la>0.10).mean():.1%})")
        print(f"  lots>1.00   : {(la>1.00).sum()}/{len(la)} ({(la>1.00).mean():.1%})")

    # Win/loss breakdown + exit reason distribution.
    pnls = np.array([t.pnl for t in result.trades])
    wins = (pnls > 0).sum()
    losses = (pnls < 0).sum()
    flats = (pnls == 0).sum()
    print(f"\n  win/loss    : wins={wins} ({wins/len(pnls):.1%})  "
          f"losses={losses} ({losses/len(pnls):.1%})  flat={flats}")
    print(f"  PnL sum     : ${pnls.sum():.2f}  avg=${pnls.mean():.3f}  "
          f"win_avg=${pnls[pnls>0].mean():.3f}  loss_avg=${pnls[pnls<0].mean():.3f}")
    gross_profit = pnls[pnls > 0].sum()
    gross_loss = -pnls[pnls < 0].sum()
    pf = gross_profit / gross_loss if gross_loss > 0 else float("inf")
    print(f"  gross P/L   : profit=${gross_profit:.2f}  loss=${gross_loss:.2f}  PF={pf:.4f}")

    # Exit reason distribution.
    from collections import Counter
    reasons = Counter(t.exit_reason for t in result.trades)
    print(f"\n  exit reasons:")
    for r, n in reasons.most_common():
        avg_pnl = np.mean([t.pnl for t in result.trades if t.exit_reason == r])
        print(f"    {r:<20} {n:>5} ({n/len(result.trades):.1%})  avg_pnl=${avg_pnl:.3f}")

    # Equity growth: how much does equity compound over the IS window?
    eq = result.equity_curve
    if len(eq):
        print(f"\n  equity curve : start=${eq['equity'].iloc[0]:.2f}  "
              f"end=${eq['equity'].iloc[-1]:.2f}  "
              f"peak=${eq['equity'].max():.2f}  "
              f"final=${result.final_balance:.2f}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())