mymt5opp/shared/optimizer/objective.py

"""Optuna objective function and scoring (doc 06 §2).

Structure of one objective call (doc 06 §2):

    1. sampled  = suggest_params(trial)               # from SEARCH_SPACE
    2. params   = build_params(sampled, frozen_baseline)
    3. result   = engine.run(params, bars, instrument, deposit)
       metrics  = compute_metrics(result)
    4. score, violations = score_metrics(metrics, sampled)
       for k, v in metrics.items(): trial.set_user_attr(k, v)

Score design balances reward against risk; constraints are hard rejections
implemented as a large penalty so violators sort to the bottom (not soft
nudges). A path-independent fragility guard rejects martingale configs that
would blow up on a path the backtest never sampled.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Callable, Optional

import numpy as np
import pandas as pd

from ..core.engine import Engine, SizingInputs
from ..core.metrics import Metrics, compute_metrics
from ..instruments.config import InstrumentConfig
from .search_space import SearchSpace, suggest_params

# Penalty applied per hard-constraint violation (must dominate any real score).
VIOLATION_PENALTY: float = 1.0e6


@dataclass
class Constraints:
    """Hard constraints — violators are rejected (doc 06 §2).

    A trial that violates any constraint is pushed to the bottom of the study
    by subtracting ``VIOLATION_PENALTY`` per violation. Set a field to
    ``None`` to skip that check.
    """

    min_trades: Optional[int] = 25          # ≥ ~25–30 active trades per year
    min_profit_factor: Optional[float] = 1.3
    max_equity_dd: Optional[float] = None   # hard drawdown cap in account ccy
    max_equity_dd_pct: Optional[float] = 0.35  # or as a fraction of deposit
    # Path-independent fragility guard (doc 06 §2): reject martingale configs
    # that survive the backtest but would blow up on a straight adverse path.
    fragility_check: Optional[Callable[[Metrics, dict], list[str]]] = None


@dataclass
class ObjectiveConfig:
    """Everything the objective closure needs, captured once.

    The objective is rebuilt per iteration (each iteration owns its own
    snapshot — doc 04 Rule 5), so this config is the single source the run
    script assembles before calling :func:`build_objective`.
    """

    engine: Engine
    bars: pd.DataFrame
    instrument: InstrumentConfig
    sizing: SizingInputs
    initial_deposit: float
    search_space: SearchSpace
    int_params: set[str] = field(default_factory=set)
    frozen_baseline: dict = field(default_factory=dict)
    constraints: Constraints = field(default_factory=Constraints)
    dd_weight: float = 1.0                  # score = Net − DD_WEIGHT × EquityDD
    # Strategy-specific hook: turn sampled params into engine-ready params +
    # signal arrays + sl/tp arrays. The snapshot owns this so the optimizer
    # stays strategy-agnostic.
    build_signals: Optional[Callable[[dict, pd.DataFrame, InstrumentConfig], "SignalPack"]] = None
    # Strategy-specific engine kwargs builder: returns a dict of extra kwargs
    # to pass to engine.run (e.g. {"scalper_cfg": ScalperConfig(...)}). Lets a
    # strategy pipe tunable point values into its engine without the optimizer
    # knowing about strategy-specific config objects. None → no extra kwargs.
    build_engine_kwargs: Optional[Callable[[dict], dict]] = None
    # M1 bars for tick-level exit simulation. MANDATORY if the EA moves its
    # SL intra-trade (break-even / trailing / basket trailing) — bar-level
    # exit simulation produces a −40% to −50% net gap on those EAs (doc 03 §7
    # measured failure mode). Leave None only for clean-directional setups
    # that don't move the SL. The bars must cover the same window as ``bars``.
    m1_bars: Optional[pd.DataFrame] = None
    # Indicator-warmup bars (doc 03 §8 / reeval_finalist_forward.py pattern).
    # When set, the objective builds signals on this FULL history (so EMA/RSI/
    # ATR are already stable before the eval window starts — matching MT5
    # Strategy Tester's pre-test chart-history warmup), then slices the signal
    # arrays to ``bars``' time range before running the engine. Must be a
    # contiguous superset of ``bars`` (same OHLC source, same timestamps).
    # Leave None to build signals directly on ``bars`` (legacy behaviour).
    signals_full_bars: Optional[pd.DataFrame] = None


@dataclass
class SignalPack:
    """Output of the strategy's ``build_signals`` hook.

    Bundles the pre-computed arrays the engine consumes. Kept here (not in
    ``core``) so the engine contract stays free of optimizer concerns.
    """

    params: dict
    signals_long: np.ndarray
    signals_short: np.ndarray
    sl_prices: np.ndarray
    tp_prices: np.ndarray


def score_metrics(
    metrics: Metrics,
    sampled: dict,
    constraints: Constraints,
    *,
    dd_weight: float = 1.0,
) -> tuple[float, list[str]]:
    """Compute the scalar score + list of violation reasons.

    ``score = Net − dd_weight × EquityDD`` minus a huge penalty per violation.
    Violators therefore sort to the bottom of the study, not just docked.
    """
    violations: list[str] = []
    if constraints.min_trades is not None and metrics.total_trades < constraints.min_trades:
        violations.append(f"too few trades ({metrics.total_trades} < {constraints.min_trades})")
    if constraints.min_profit_factor is not None and metrics.profit_factor < constraints.min_profit_factor:
        violations.append(f"PF too low ({metrics.profit_factor:.2f} < {constraints.min_profit_factor})")
    if constraints.max_equity_dd is not None and metrics.max_equity_dd > constraints.max_equity_dd:
        violations.append(f"DD over cap ({metrics.max_equity_dd:.2f} > {constraints.max_equity_dd})")
    if constraints.max_equity_dd_pct is not None and metrics.max_equity_dd_pct > constraints.max_equity_dd_pct:
        violations.append(
            f"DD% over cap ({metrics.max_equity_dd_pct:.2%} > {constraints.max_equity_dd_pct:.2%})"
        )
    if constraints.fragility_check is not None:
        violations.extend(constraints.fragility_check(metrics, sampled))

    base = metrics.net_profit - dd_weight * metrics.max_equity_dd
    if violations:
        return base - VIOLATION_PENALTY * len(violations), violations
    return base, violations


def build_objective(cfg: ObjectiveConfig):
    """Return an Optuna-compatible objective closure for ``cfg``.

    The closure samples params, builds signals (via ``cfg.build_signals``),
    runs the engine, scores, and stashes metrics on the trial as user attrs.
    """
    if cfg.build_signals is None:
        raise ValueError(
            "ObjectiveConfig.build_signals must be set — the optimizer must "
            "not encode strategy logic itself."
        )

    def objective(trial) -> float:
        sampled = suggest_params(trial, cfg.search_space, cfg.int_params)
        merged = {**cfg.frozen_baseline, **sampled}
        if cfg.signals_full_bars is not None:
            # Warmup mode (doc 03 §8 / reeval_finalist_forward.py pattern):
            # build signals on the FULL bars so indicators are already stable
            # at the eval window start — mirrors MT5 Strategy Tester's pre-test
            # chart-history warmup. Then slice the signal arrays to cfg.bars'
            # time range. cfg.bars must be a contiguous sub-range of
            # signals_full_bars (same OHLC source, same timestamps).
            pack = cfg.build_signals(merged, cfg.signals_full_bars, cfg.instrument)
            eval_start = pd.Timestamp(cfg.bars["timestamp"].iloc[0])
            full_ts = pd.to_datetime(cfg.signals_full_bars["timestamp"].to_numpy())
            lo = int(full_ts.searchsorted(eval_start, side="left"))
            hi = lo + len(cfg.bars)
            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]
        else:
            # Legacy mode: build signals directly on the (already-trimmed)
            # cfg.bars. Indicators warm up at the eval window start — fine for
            # short-period indicators but produces ~14h of EMA-stabilization
            # noise at the start of long-period EMA strategies.
            pack = cfg.build_signals(merged, cfg.bars, cfg.instrument)
            sig_long  = pack.signals_long
            sig_short = pack.signals_short
            sl_p      = pack.sl_prices
            tp_p      = pack.tp_prices
        extra = cfg.build_engine_kwargs(merged) if cfg.build_engine_kwargs else {}
        # If M1 bars are wired up, pass them so the engine switches to tick-
        # level exit simulation (mandatory for trailing/BE EAs — doc 03 §7).
        if cfg.m1_bars is not None:
            extra = {**extra, "m1_bars": cfg.m1_bars}
        result = cfg.engine.run(
            cfg.bars,
            sig_long,
            sig_short,
            sl_p,
            tp_p,
            cfg.instrument,
            cfg.sizing,
            cfg.initial_deposit,
            **extra,
        )
        metrics = compute_metrics(result)
        score, violations = score_metrics(
            metrics, sampled, cfg.constraints, dd_weight=cfg.dd_weight
        )
        # Stash metrics for later inspection + the diverse selector.
        trial.set_user_attr("net_profit", metrics.net_profit)
        trial.set_user_attr("profit_factor", metrics.profit_factor)
        trial.set_user_attr("total_trades", metrics.total_trades)
        trial.set_user_attr("max_equity_dd", metrics.max_equity_dd)
        trial.set_user_attr("max_equity_dd_pct", metrics.max_equity_dd_pct)
        trial.set_user_attr("win_rate", metrics.win_rate)
        trial.set_user_attr("sharpe", metrics.sharpe)
        trial.set_user_attr("violations", violations)
        trial.set_user_attr("params", merged)
        return score

    return objective