Chan/research/step21_best_config.py

"""Step 21：最优配置的尽职调查。

Step 20 在完整数据（BTC/ETH/SOL，2172~2543 天）上给出两个候选：
  双大级别同向   2055 笔  PF 1.95  中位 +0.301%  t 11.43
  30m/2h+4h     269 笔  PF 2.24  中位 +0.915%  t  5.51

本步只做一件事：判断它们是靠尾部还是靠主体，以及扛不扛得住成本。
"""
from __future__ import annotations

import sys
import warnings
from pathlib import Path

import numpy as np
import pandas as pd

warnings.filterwarnings("ignore")
pd.set_option("display.width", 280)

HERE = Path(__file__).resolve().parent
FEE = 0.0008


def desc(r: np.ndarray, label: str) -> dict:
    if len(r) < 5:
        return {}
    win, loss = r[r > 0], r[r <= 0]
    sd = r.std(ddof=1)
    return {
        "配置": label, "笔数": len(r),
        "胜率": f"{(r > 0).mean() * 100:.1f}%",
        "均收益": f"{r.mean() * 100:+.3f}%",
        "中位": f"{np.median(r) * 100:+.3f}%",
        "赔率": f"{win.mean() / abs(loss.mean()):.2f}" if len(win) and len(loss) else "—",
        "PF": f"{win.sum() / abs(loss.sum()):.2f}" if len(loss) else "inf",
        "偏度": f"{pd.Series(r).skew():.2f}",
        "t值": f"{r.mean() / (sd / np.sqrt(len(r))):+.2f}",
    }


def trim(r: np.ndarray, label: str) -> list[dict]:
    rows = []
    for k in (0, 2, 5, 10, 20):
        v = r if k == 0 else r[r <= np.quantile(r, 1 - k / 100)]
        if len(v) < 10:
            continue
        win, loss = v[v > 0], v[v <= 0]
        sd = v.std(ddof=1)
        rows.append({
            "配置": label, "剔除最赚": f"{k}%", "笔数": len(v),
            "PF": f"{win.sum() / abs(loss.sum()):.2f}" if len(loss) else "inf",
            "中位": f"{np.median(v) * 100:+.3f}%",
            "t值": f"{v.mean() / (sd / np.sqrt(len(v))):+.2f}",
        })
    return rows


def main() -> None:
    t = pd.read_csv(HERE / "out" / "step20_main_trades.csv", parse_dates=["date"])
    a1 = t["h1_agree"] == 1
    a2 = t["h2_agree"] == 1
    both = a1 & a2

    cfgs = {
        "全部信号": t,
        "双大级别同向": t[both],
        "30m/2h+4h 全部": t[t.ltf == "30m"],
        "30m/2h+4h 双同向": t[(t.ltf == "30m") & both],
        "15m+30m+1h 双同向": t[t.ltf.isin(["15m", "30m", "1h"]) & both],
    }

    print("########## 1. 候选配置画像 ##########")
    print(pd.DataFrame([desc(g["ret"].to_numpy(), k) for k, g in cfgs.items()
                        if len(g) >= 5]).to_string(index=False))
    print("\n  中位为正 = 靠主体赚钱；偏度低 = 不依赖极端值。")

    print("\n########## 2. 尾部依赖 ##########")
    rows = []
    for k in ("双大级别同向", "30m/2h+4h 双同向", "15m+30m+1h 双同向"):
        rows += trim(cfgs[k]["ret"].to_numpy(), k)
    print(pd.DataFrame(rows).pivot(index="剔除最赚", columns="配置",
                                   values=["PF", "t值"]).to_string())

    print("\n########## 3. 成本敏感性 ##########")
    rows = []
    for mult, name in ((1, "0.08%"), (2, "0.16%"), (3, "0.24%"), (5, "0.40%")):
        for k in ("双大级别同向", "30m/2h+4h 双同向", "15m+30m+1h 双同向"):
            r = cfgs[k]["ret"].to_numpy() - (mult - 1) * FEE
            win, loss = r[r > 0], r[r <= 0]
            sd = r.std(ddof=1)
            rows.append({
                "成本": name, "配置": k,
                "PF": f"{win.sum() / abs(loss.sum()):.2f}",
                "t值": f"{r.mean() / (sd / np.sqrt(len(r))):+.2f}",
            })
    print(pd.DataFrame(rows).pivot(index="成本", columns="配置",
                                   values=["PF", "t值"]).to_string())

    print("\n########## 4. 30m 双同向：分年与分品种 ##########")
    g = cfgs["30m/2h+4h 双同向"].copy()
    g["year"] = pd.to_datetime(g["date"]).dt.year
    rows = [desc(x["ret"].to_numpy(), str(y)) for y, x in g.groupby("year") if len(x) >= 8]
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))
    rows = [desc(x["ret"].to_numpy(), s) for s, x in g.groupby("symbol") if len(x) >= 8]
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))

    print("\n########## 5. 推荐组合：15m/30m/1h 双同向，分年 ##########")
    g = cfgs["15m+30m+1h 双同向"].copy()
    g["year"] = pd.to_datetime(g["date"]).dt.year
    rows = [desc(x["ret"].to_numpy(), str(y)) for y, x in g.groupby("year") if len(x) >= 15]
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))
    pos = sum(1 for _, x in g.groupby("year") if len(x) >= 15 and x["ret"].mean() > 0)
    tot = sum(1 for _, x in g.groupby("year") if len(x) >= 15)
    print(f"\n  盈利年份 {pos}/{tot}")

    print("\n########## 6. 出场结构与持有时长 ##########")
    g = cfgs["15m+30m+1h 双同向"]
    print(g.groupby("reason").agg(
        笔数=("ret", "size"),
        占比=("ret", lambda x: f"{len(x) / len(g) * 100:.0f}%"),
        均收益=("ret", lambda x: f"{x.mean() * 100:+.3f}%"),
        均持有=("bars_held", lambda x: f"{x.mean():.0f}"),
    ).to_string())

    print("\n########## 7. 权益曲线（固定 1% 风险，按时间序）##########")
    for k in ("双大级别同向", "15m+30m+1h 双同向", "30m/2h+4h 双同向"):
        g = cfgs[k].sort_values("date")
        r = g["ret"].to_numpy()
        stop = abs(np.quantile(r, 0.05))
        sc = np.clip(r / max(stop, 1e-6) * 0.01, -0.1, 0.5)
        eq = np.cumprod(1 + sc)
        dd = (1 - eq / np.maximum.accumulate(eq)).max()
        yrs = (g["date"].max() - g["date"].min()).days / 365.25
        cagr = eq[-1] ** (1 / yrs) - 1 if yrs > 0 else np.nan
        shp = sc.mean() / sc.std(ddof=1) * np.sqrt(len(sc) / yrs) if yrs > 0 else np.nan
        print(f"  {k:>18}: 年化 {cagr * 100:+6.1f}%  最大回撤 {dd * 100:5.1f}%  "
              f"Sharpe {shp:5.2f}  频率 {len(r) / yrs:.0f} 笔/年")


if __name__ == "__main__":
    main()