optimize.py

"""Parameter sweep for the momentum / breakout strategy.

`optimize` grid-searches StrategyParams and RiskParams for *risk-adjusted*
performance. The score, pooled across the whole watchlist, is

    score = (profit_factor - 1) - max_drawdown

so a combo is rewarded for a profit factor above 1 and penalised for drawdown.

Out-of-sample guard
-------------------
History is split into a TRAIN slice (the older 70%) and a TEST slice (the most
recent 30%). Every combo is scored on TRAIN; the finalists are re-scored on
TEST, and the winner is the finalist with the best *worst-of-the-two* score.
A combo that only shines on the data it was tuned on therefore loses -- this is
a deliberate hedge against curve-fitting, not a promise of future returns.

The search runs in two stages to stay tractable: first the signal parameters
(risk held at a sensible fixed setting), then the risk parameters (signal
parameters held at the stage-1 winner).
"""
from __future__ import annotations

import itertools
from dataclasses import replace

import config
import marketdata
from backtest import run_backtest, summarize

INTERVAL = "5m"
PERIOD = "59d"
TRAIN_FRAC = 0.70
MIN_TRADES_TRAIN = 40   # combos trading less than this across the watchlist are out
MIN_TRADES_TEST = 12    # ...and the same, lower, floor on the shorter test slice
TOP_K = 25              # finalists carried from the TRAIN ranking into the TEST check


def _score(stats: dict, min_trades: int) -> float:
    """Risk-adjusted score -- high profit factor, low drawdown. Higher is better."""
    if stats["trades"] < min_trades:
        return float("-inf")
    pf = min(stats["profit_factor"], 4.0)  # cap: a tiny sample can post a freak PF
    return (pf - 1.0) - stats["max_drawdown"]


def _pooled(symbol_dfs, strategy_params, risk) -> dict:
    """Backtest every ticker and pool the trades into one watchlist-wide stat block."""
    trades = []
    for sym, df in symbol_dfs:
        if len(df) < 250:            # too little history to mean anything
            continue
        t, _ = run_backtest(df, sym, mode="underlying",
                            strategy_params=strategy_params, risk=risk)
        trades.extend(t)
    trades.sort(key=lambda tr: tr.exit_time)   # a realistic pooled equity curve
    return summarize("WATCHLIST", "underlying", trades, risk.starting_equity)


def _evaluate(combos, train_dfs, test_dfs, make_params):
    """Score combos on TRAIN, carry the finalists to TEST, return them ranked.

    `make_params(combo)` -> (strategy_params, risk). The returned list of dicts
    is sorted by the out-of-sample (worst-of-two) score, best first.
    """
    scored = []
    for n, combo in enumerate(combos, 1):
        sp, risk = make_params(combo)
        train = _pooled(train_dfs, sp, risk)
        scored.append({"sp": sp, "risk": risk, "train": train,
                       "train_score": _score(train, MIN_TRADES_TRAIN)})
        if n % 50 == 0:
            print(f"    ...scored {n}/{len(combos)} combos", flush=True)

    scored.sort(key=lambda r: r["train_score"], reverse=True)
    finalists = scored[:TOP_K]
    for r in finalists:
        r["test"] = _pooled(test_dfs, r["sp"], r["risk"])
        r["test_score"] = _score(r["test"], MIN_TRADES_TEST)
        r["robust_score"] = min(r["train_score"], r["test_score"])
    finalists.sort(key=lambda r: r["robust_score"], reverse=True)
    return finalists


def _fmt(stats: dict) -> str:
    pf = "inf" if stats["profit_factor"] == float("inf") else f"{stats['profit_factor']:.2f}"
    return (f"{stats['trades']:>4} trades  win {stats['win_rate'] * 100:>5.1f}%  "
            f"PF {pf:>5}  return {stats['return_pct'] * 100:>8.2f}%  "
            f"maxDD {stats['max_drawdown'] * 100:>5.1f}%")


def _params_line(sp) -> str:
    return (f"lookback={sp.breakout_lookback} vol_mult={sp.volume_multiplier} "
            f"buffer={sp.breakout_atr_buffer} trend_ema={sp.trend_filter_ema} "
            f"skip_open={sp.skip_open_bars} short={sp.enable_short}")


def _print_config_block(sp, risk) -> None:
    print("Recommended config.py values (StrategyParams / RiskParams):")
    print(f"    breakout_lookback   = {sp.breakout_lookback}")
    print(f"    volume_multiplier   = {sp.volume_multiplier}")
    print(f"    breakout_atr_buffer = {sp.breakout_atr_buffer}")
    print(f"    trend_filter_ema    = {sp.trend_filter_ema}")
    print(f"    skip_open_bars      = {sp.skip_open_bars}")
    print(f"    enable_short        = {sp.enable_short}")
    print(f"    stop_atr_mult       = {risk.stop_atr_mult}")
    print(f"    target_atr_mult     = {risk.target_atr_mult}")
    print(f"    use_trailing_stop   = {risk.use_trailing_stop}")
    print(f"    trail_atr_mult      = {risk.trail_atr_mult}")
    print(f"    max_hold_bars       = {risk.max_hold_bars}")


def run(interval: str = INTERVAL, period: str = PERIOD, source: str = "auto") -> None:
    """Fetch the watchlist, run the two-stage sweep and print the winner."""
    resolved = marketdata.resolve_source(source)
    print(f"Optimising momentum-breakout | interval={interval} | period={period} "
          f"| source={resolved}")
    print(f"Watchlist: {', '.join(config.WATCHLIST)}")
    print("Fetching history...", flush=True)

    raw = []
    for ticker in config.WATCHLIST:
        try:
            df = marketdata.intraday(ticker, interval=interval, period=period,
                                     source=source)
        except Exception as exc:  # noqa: BLE001 - one bad ticker shouldn't abort
            print(f"  {ticker}: skipped ({exc})")
            continue
        raw.append((ticker, df))
    if not raw:
        print("No data fetched -- aborting.")
        return

    train_dfs, test_dfs = [], []
    for sym, df in raw:
        cut = int(len(df) * TRAIN_FRAC)
        train_dfs.append((sym, df.iloc[:cut]))
        test_dfs.append((sym, df.iloc[cut:]))
    bars = sum(len(df) for _, df in raw)
    print(f"  {len(raw)} tickers, {bars:,} bars "
          f"(train {TRAIN_FRAC:.0%} / test {1 - TRAIN_FRAC:.0%})")

    base_train = _pooled(train_dfs, config.STRATEGY, config.RISK)
    base_test = _pooled(test_dfs, config.STRATEGY, config.RISK)
    print()
    print("Baseline (current config.py):")
    print(f"  train: {_fmt(base_train)}")
    print(f"  test : {_fmt(base_test)}")

    # ---- stage 1: signal parameters -------------------------------------
    print()
    print("Stage 1/2: signal parameters (StrategyParams)...")
    fixed_risk = replace(config.RISK, stop_atr_mult=1.5, target_atr_mult=3.0,
                         use_trailing_stop=False, max_hold_bars=24)
    strat_combos = list(itertools.product(
        [12, 20, 30, 48],          # breakout_lookback
        [1.0, 1.5, 2.0],           # volume_multiplier
        [0.0, 0.25, 0.5],          # breakout_atr_buffer
        [0, 50, 100, 200],         # trend_filter_ema
        [0, 6],                    # skip_open_bars
        [True, False],             # enable_short
    ))
    print(f"  {len(strat_combos)} combos x {len(raw)} tickers")

    def _make_strategy(c):
        bl, vm, buf, ema, skip, shrt = c
        sp = replace(config.STRATEGY, breakout_lookback=bl, volume_multiplier=vm,
                     breakout_atr_buffer=buf, trend_filter_ema=ema,
                     skip_open_bars=skip, enable_short=shrt)
        return sp, fixed_risk

    strat_finalists = _evaluate(strat_combos, train_dfs, test_dfs, _make_strategy)
    best_strategy = strat_finalists[0]["sp"]
    print(f"  winner: {_params_line(best_strategy)}")

    # ---- stage 2: risk parameters ---------------------------------------
    print()
    print("Stage 2/2: risk parameters (RiskParams)...")
    risk_combos = list(itertools.product(
        [1.0, 1.5, 2.0, 2.5],                                    # stop_atr_mult
        [2.0, 3.0, 4.0, 6.0],                                    # target_atr_mult
        [(False, 3.0), (True, 2.0), (True, 3.0), (True, 4.0)],   # (trailing, mult)
        [12, 24, 48],                                            # max_hold_bars
    ))
    print(f"  {len(risk_combos)} combos x {len(raw)} tickers")

    def _make_risk(c):
        stop, tgt, (trail_on, trail_m), hold = c
        risk = replace(config.RISK, stop_atr_mult=stop, target_atr_mult=tgt,
                       use_trailing_stop=trail_on, trail_atr_mult=trail_m,
                       max_hold_bars=hold)
        return best_strategy, risk

    risk_finalists = _evaluate(risk_combos, train_dfs, test_dfs, _make_risk)
    best = risk_finalists[0]

    # ---- report ---------------------------------------------------------
    print()
    print("=" * 72)
    print("OPTIMISED PARAMETERS")
    print("=" * 72)
    print(f"  baseline train: {_fmt(base_train)}")
    print(f"  baseline test : {_fmt(base_test)}")
    print(f"  tuned    train: {_fmt(best['train'])}")
    print(f"  tuned    test : {_fmt(best['test'])}")
    print()
    _print_config_block(best["sp"], best["risk"])
    print()
    print("Note: 'test' is the held-out recent 30% of history -- it is the more "
          "honest read. Past performance does not predict future returns.")


if __name__ == "__main__":
    run()