Chan/research/lib/breakout.py

"""突破跟随的事件驱动回测。

趋势跟随是低胜率高赔率，固定持有期会把大赢利截断、把小亏损放大，
必须用止损/止盈/时间三重出场才能测出真实期望。
"""
from __future__ import annotations

from dataclasses import dataclass

import numpy as np
import pandas as pd

FEE = 0.0008  # 双边


@dataclass
class Trade:
    entry_idx: int
    exit_idx: int
    direction: int
    entry: float
    exit: float
    ret: float          # 已扣费
    reason: str         # sl / tp / time
    bars_held: int
    gross: float        # 未扣费，便于事后做费率 what-if
    risk_pct: float     # 止损距离占入场价的比例，用于反推名义仓位
    boosted: bool = False   # 持仓期间是否等到了更大级别的同向确认


def run_trades(
    df: pd.DataFrame,
    entries: list[tuple[int, int]],
    sl_atr: float = 1.5,
    tp_atr: float = 3.0,
    max_bars: int = 48,
    trail: bool = False,
    fee: float = FEE,
    entry_delay: int = 0,
    slippage: float = 0.0,
    boost_dir: np.ndarray | None = None,
    tp_boost: float = 2.0,
    boost_breakeven: bool = False,
) -> pd.DataFrame:
    """按 (entry_idx, direction) 逐笔模拟。

    出场优先级：同一根内若同时触及止损与止盈，保守地判为止损。
    entry_delay=1 表示信号次根开盘成交，用来检验「收盘价入场」是否过于乐观。

    boost_dir 给出每根K线上「趋势被更大级别确认」的方向（+1/-1/0）。持仓期间
    一旦等到同向确认，就把止盈目标放大 tp_boost 倍；boost_breakeven 同时把
    止损收到成本价。用来检验「新证据出现后该不该改单」。
    """
    high = df["high"].to_numpy(dtype=float)
    low = df["low"].to_numpy(dtype=float)
    close = df["close"].to_numpy(dtype=float)
    open_ = df["open"].to_numpy(dtype=float) if "open" in df.columns else close
    atr = (
        df["atr"].to_numpy(dtype=float)
        if "atr" in df.columns
        else pd.Series(close).rolling(14).std().bfill().to_numpy()
    )
    n = len(df)
    out: list[Trade] = []

    for e_idx, d in entries:
        sig_idx = e_idx
        e_idx = e_idx + entry_delay
        if e_idx >= n - 1:
            continue
        entry = open_[e_idx] if entry_delay else close[e_idx]
        a = atr[sig_idx]
        if not np.isfinite(a) or a <= 0:
            continue
        sl = entry - d * sl_atr * a
        tp = entry + d * tp_atr * a
        best = entry
        exit_idx, exit_px, reason = None, None, "time"
        boosted = False

        for j in range(e_idx + 1, min(e_idx + max_bars + 1, n)):
            if boost_dir is not None and not boosted and boost_dir[j] == d:
                tp = entry + d * tp_atr * tp_boost * a
                if boost_breakeven:
                    sl = max(sl, entry) if d == 1 else min(sl, entry)
                boosted = True
            if trail:
                best = max(best, high[j]) if d == 1 else min(best, low[j])
                sl = max(sl, best - sl_atr * a) if d == 1 else min(sl, best + sl_atr * a)
            hit_sl = low[j] <= sl if d == 1 else high[j] >= sl
            hit_tp = high[j] >= tp if d == 1 else low[j] <= tp
            if hit_sl:
                exit_idx, exit_px, reason = j, sl, "sl"
                break
            if hit_tp:
                exit_idx, exit_px, reason = j, tp, "tp"
                break
        if exit_idx is None:
            exit_idx = min(e_idx + max_bars, n - 1)
            exit_px = close[exit_idx]

        gross = d * (exit_px - entry) / entry
        out.append(Trade(sig_idx, exit_idx, d, entry, float(exit_px),
                         gross - fee - slippage, reason, exit_idx - e_idx,
                         gross, sl_atr * a / entry, boosted))

    return pd.DataFrame([t.__dict__ for t in out])


def run_trades_dynamic(
    df: pd.DataFrame,
    entries: list[tuple[int, int]],
    upgrades: dict[int, int],
    sl_atr: float = 1.5,
    tp_atr: float = 3.0,
    tp_atr_up: float = 6.0,
    max_bars: int = 48,
    max_bars_up: int = 96,
    lock_breakeven: bool = True,
    fee: float = FEE,
    entry_delay: int = 0,
    slippage: float = 0.0,
) -> pd.DataFrame:
    """持仓中若出现更大级别的同向确认，就把目标放远、并把止损收到成本价。

    upgrades: {K线索引: 方向}，表示该根出现了大级别同向三买/中枢突破。
    对应的交易逻辑是「小级别进场、大级别接力」——趋势被更高级别确认后，
    原本 3 ATR 的目标就过早了，但同时不该再让这笔回到亏损。
    """
    high = df["high"].to_numpy(dtype=float)
    low = df["low"].to_numpy(dtype=float)
    close = df["close"].to_numpy(dtype=float)
    open_ = df["open"].to_numpy(dtype=float) if "open" in df.columns else close
    atr = (
        df["atr"].to_numpy(dtype=float)
        if "atr" in df.columns
        else pd.Series(close).rolling(14).std().bfill().to_numpy()
    )
    n = len(df)
    out: list[dict] = []

    for e_idx, d in entries:
        sig_idx = e_idx
        e_idx = e_idx + entry_delay
        if e_idx >= n - 1:
            continue
        entry = open_[e_idx] if entry_delay else close[e_idx]
        a = atr[sig_idx]
        if not np.isfinite(a) or a <= 0:
            continue
        sl = entry - d * sl_atr * a
        tp = entry + d * tp_atr * a
        limit = max_bars
        upgraded = False
        exit_idx, exit_px, reason = None, None, "time"

        j = e_idx + 1
        while j < min(e_idx + limit + 1, n):
            if not upgraded and upgrades.get(j) == d:
                upgraded = True
                tp = entry + d * tp_atr_up * a
                limit = max_bars_up
                if lock_breakeven:
                    sl = max(sl, entry) if d == 1 else min(sl, entry)
            hit_sl = low[j] <= sl if d == 1 else high[j] >= sl
            hit_tp = high[j] >= tp if d == 1 else low[j] <= tp
            if hit_sl:
                exit_idx, exit_px, reason = j, sl, "be" if upgraded and sl == entry else "sl"
                break
            if hit_tp:
                exit_idx, exit_px, reason = j, tp, "tp_up" if upgraded else "tp"
                break
            j += 1
        if exit_idx is None:
            exit_idx = min(e_idx + limit, n - 1)
            exit_px = close[exit_idx]

        gross = d * (exit_px - entry) / entry
        out.append({
            "entry_idx": sig_idx, "exit_idx": exit_idx, "direction": d,
            "entry": entry, "exit": float(exit_px),
            "ret": gross - fee - slippage, "reason": reason,
            "bars_held": exit_idx - e_idx, "gross": gross,
            "risk_pct": sl_atr * a / entry, "upgraded": upgraded,
        })

    return pd.DataFrame(out)


def summarize_trades(tr: pd.DataFrame, label: str) -> dict:
    if tr.empty:
        return {"策略": label, "笔数": 0}
    r = tr["ret"].to_numpy()
    win = r[r > 0]
    loss = r[r <= 0]
    pf = win.sum() / abs(loss.sum()) if len(loss) and loss.sum() != 0 else np.inf
    sd = r.std(ddof=1)
    eq = np.cumprod(1 + r)
    dd = float((1 - eq / np.maximum.accumulate(eq)).max()) if len(eq) else 0.0
    return {
        "策略": label,
        "笔数": len(r),
        "胜率": f"{(r > 0).mean() * 100:.1f}%",
        "均收益": f"{r.mean() * 100:+.3f}%",
        "赔率": f"{(win.mean() / abs(loss.mean())):.2f}" if len(win) and len(loss) else "—",
        "盈亏比PF": f"{pf:.2f}",
        "总收益": f"{(eq[-1] - 1) * 100:+.1f}%",
        "最大回撤": f"{dd * 100:.1f}%",
        "t值": f"{r.mean() / (sd / np.sqrt(len(r))):+.2f}" if sd else "—",
        "均持有": f"{tr['bars_held'].mean():.0f}",
    }


def find_breakout_entries(
    sig: pd.DataFrame, df: pd.DataFrame, window: int = 10, mode: str = "fail"
) -> list[tuple[int, int]]:
    """分型突破入场点。

    mode="fail"    分型失败 -> 顺势跟随：顶分型被向上突破则做多。
    mode="reverse" 传统反转 -> 分型成立方向：顶分型做空（作为对照）。
    """
    close = df["close"].to_numpy(dtype=float)
    n = len(df)
    entries: list[tuple[int, int]] = []
    for _, r in sig.iterrows():
        d_fx = int(r["direction"])       # +1 底分型 / -1 顶分型
        lvl = float(r["price"])
        c0 = int(r["confirm_idx"])
        if mode == "reverse":
            entries.append((c0, d_fx))
            continue
        # 突破方向与分型指向相反：顶分型(-1)被向上(+1)突破
        d_bo = -d_fx
        for j in range(c0 + 1, min(c0 + window + 1, n)):
            broken = close[j] > lvl if d_bo == 1 else close[j] < lvl
            if broken:
                entries.append((j, d_bo))
                break
    return entries