15分钟KDJ+布林+30分钟趋势做T，统计意义上靠不靠谱？

今天服务器出了状况，没有数据环境，所以改用 workbuddy 验证一个 UP 主分享的 KDJ 日内交易思路：30分钟MA20+方向定环境，15分钟KDJ交叉找时机，布林轨打位置标签。

下面是我整理的 AI 提示词，具体参数都是我自己定的，实际用之前都需要反复测试信号的有效性。

AI 提示词

你是一名严谨的量化事件研究助理。请对我指定的“15分钟KDJ+布林七轨+30分钟趋势过滤”交易思路做信号层面验证，而不是做策略回测。不要输出净值曲线，不要优化参数，不要用未来信息。若数据不足，先列出缺失字段和需要我补充的数据；若能执行代码，请直接计算并输出表格；若不能执行代码，请输出可运行的 Python/pandas 脚本和空表模板。

【标的】 【填写】
【样本区间】 【填写】
【数据频率】 15分钟K线为主，30分钟K线用于趋势判定
【数据字段】 时间、开盘、最高、最低、收盘、成交量；若已有布林轨值请一并提供

一、事件定义（每根15分钟K线只判定一次，且必须在K线收盘后确认）

1. 买信号：CROSS(K,D) 且 K < 20。CROSS(K,D) 指当前 K > D 且前一根 K <= D。  
2. 强延续信号：CROSS(K,D) 且 50 <= K <= 70。该信号作为独立事件，不与买信号混合。  
3. 卖信号A：CROSS(D,K) 且 K > 80。  
4. 卖信号B：CROSS(D,K) 且 K > 50 且 30分钟趋势 = 下降。注意：卖信号A和B可能嵌套，必须分开统计，不得合并。  
5. KDJ参数：RSV = (CLOSE - LLV(LOW,9)) / (HHV(HIGH,9) - LLV(LOW,9)) * 100；K = SMA(RSV,3,1)；D = SMA(K,3,1)。  
6. 布林七轨：中轨 = MA(C,20)，上下轨 = 中轨 ± 1/2/3 倍 STD(C,20)。信号发生时，用当时已计算的轨值给价格打位置标签：下3轨—下2轨、下2轨—下1轨、下1轨—中轨、中轨—上1轨、上1轨—上2轨、上2轨—上3轨、上3轨以上。  
7. 30分钟趋势三态：只用已完成30分钟K判定。上升 = 收盘 > MA20且MA20上行；下降 = 收盘 < MA20且MA20下行；其余 = 震荡。禁止用事后走势划分。

二、可成交口径与防未来函数

1. 信号确认时点：15分钟K收盘后。  
2. 主口径入场：信号后下一根15分钟K开盘。  
3. 前向收益：下一根开盘到第1、2、4、8、16、32根15分钟K收盘的累计收益。  
4. 辅助口径：信号K收盘到第n根收盘，仅作对照，不作为可成交收益。  
5. 卖信号的前向收益解释为“信号后下跌/回撤”，收益越低说明卖信号越有效；ETF不能做空时，卖信号验证的是避损能力，不是做空收益。

三、基准与有效性度量

1. 基准必须逐事件配对：对每个信号事件，取同一标的、同一时间起点、同一期限的无条件收益作为基准；不得用全样本同期限平均收益直接替代。  
2. 超额胜率 = 信号前向收益 > 同起点同期限基准收益的比例，不是收益 > 0 的比例。  
3. 每个信号类型、每个前向期输出：样本数、独立交易日数、平均收益、中位数、胜率、超额胜率、5%/25%/50%/75%/95%分位数、95%置信区间、扣除单边0.05%和0.1%成本后的平均收益。  
4. 前向窗口重叠处理：按事件日期聚类bootstrap，或设冷却期，同一信号类型32根K内只取第一次。必须同时报告原始样本数和独立事件数/独立交易日数。  
5. 分环境拆分：先全样本单变量，再按30分钟趋势三态分层，再按布林位置分层，最后做趋势×位置交叉。样本数 < 30 的组只描述，不下结论。  
6. 多重比较：这是探索性验证，不要只挑显著格子。报告置信区间，并说明未做多重比较校正。  
7. 参数敏感性：对KDJ 9/3/3、布林20/2、阈值20/80/50各微调一档，检查结论是否稳定。  
8. 分段验证：按年份或前后半段拆分，避免过拟合。

四、输出物

请输出 Markdown 表格，不要净值曲线。至少包括：
表1 数据校验与样本概况；
表2 事件研究总表：信号类型×前向期×样本数×独立日数×平均收益×中位数×胜率×超额胜率×95%置信区间；
表3 分趋势环境表；
表4 分布林位置表；
表5 趋势×位置交叉表；
表6 成本敏感性与参数敏感性；
表7 卖信号避损效果表：后续收益、最大回撤、是否出现反向买信号；
最后给出：结论、局限、需要补充的数据、可复现代码或脚本。

五、禁止事项

不要使用未来函数；不要用信号K收盘价作为入场价；不要把强延续信号混入低吸买信号；不要用全样本平均收益冒充配对基准；不要只报告显著结果；不要输出净值曲线；不要给投资建议。结论必须区分“统计显著”和“经济意义显著”。

参考代码

# -*- coding: utf-8 -*-
"""
“15分钟KDJ+布林七轨+30分钟趋势过滤”信号层面事件研究脚本
=========================================================
用途：信号统计验证（非策略回测、非净值曲线、非参数寻优）。

口径要点：
  * 所有指标只用截至当前15分钟K收盘的信息
  * 主口径入场价 = 信号确认后下一根15分钟K开盘价
  * 前向收益 = 下一根开盘 -> 第 n 根15分钟K收盘
  * 辅助口径 = 信号K收盘 -> 第 n 根收盘
  * 基准 += 每个事件按同一标的、同一时间起点、同一期限的无条件收益逐事件配对
  * 超额胜率 = 信号前向收益 > 对应配对基准收益 的比例
  * 重叠处理：同一信号类型32根K内只取第一次
  * 置信区间：按事件日期聚类的 bootstrap

依赖：pandas numpy tabulate
用法：
    python kdj_boll_event_study.py --m15 m15.csv --m30 m30.csv --out report.md
    --m30 可省略；--symbol、--start、--end 可选
输出：report.md 内含 表1~表7 + 结论框架
"""

import argparse
import itertools
import os

import numpy as np
import pandas as pd

HORIZONS = [1, 2, 4, 8, 16, 32]
COOLDOWN = 32  # 15分钟K根数
COSTS = [0.0, 0.0005, 0.001]  # 单边
BOOT_B = 2000
SEED = 42
MIN_GROUP_N = 30

COL_MAP = {
    "datetime": "dt", "时间": "dt", "date": "dt", "time": "dt", "日期时间": "dt",
    "open": "open", "开盘": "open", "开盘价": "open",
    "high": "high", "最高": "high", "最高价": "high",
    "low": "low", "最低": "low", "最低价": "low",
    "close": "close", "收盘": "close", "收盘价": "close",
    "volume": "volume", "成交量": "volume", "vol": "volume", "量": "volume",
}

def load_kline(path: str) -> pd.DataFrame:
    df = pd.read_csv(path)
    df.columns = [COL_MAP.get(str(c).strip().lower(), str(c).strip().lower()) for c in df.columns]
    need = ["dt", "open", "high", "low", "close"]
    miss = [c for c in need if c not in df.columns]
    if miss:
        raise ValueError(f"缺少字段 {miss}，请检查CSV表头。现有列：{list(df.columns)}")

    if "volume" not in df.columns:
        df["volume"] = np.nan

    df["dt"] = pd.to_datetime(df["dt"])
    df = df.sort_values("dt").drop_duplicates("dt").reset_index(drop=True)

    for c in ["open", "high", "low", "close", "volume"]:
        df[c] = pd.to_numeric(df[c], errors="coerce")

    bad = df[["open", "high", "low", "close"]].isna().any(axis=1).sum()
    if bad:
        print(f"[警告] {path} 有 {bad} 行 OHLC 缺失，已剔除")
        df = df.dropna(subset=["open", "high", "low", "close"]).reset_index(drop=True)

    df["date"] = df["dt"].dt.date
    return df

def kdj(df: pd.DataFrame, n=9, m1=3, m2=3) -> pd.DataFrame:
    """标准通达信口径：RSV -> SMA(m1,1) -> SMA(m2,1)"""
    low_n = df["low"].rolling(n, min_periods=n).min()
    high_n = df["high"].rolling(n, min_periods=n).max()
    rsv = (df["close"] - low_n) / (high_n - low_n) * 100
    k = rsv.ewm(alpha=1 / m1, adjust=False).mean()  # SMA(X,3,1) 等价
    d = k.ewm(alpha=1 / m2, adjust=False).mean()

    out = df[["dt", "date", "close"]].copy()
    out["K"] = k
    out["D"] = d
    out["J"] = 3 * k - 2 * d
    return out

def boll7(df: pd.DataFrame, n=20) -> pd.DataFrame:
    """布林七轨：中轨 = MA(C,20)，上下轨 = 中轨 ± 1/2/3 倍 STD(C,20)。"""
    mid = df["close"].rolling(n, min_periods=n).mean()
    std = df["close"].rolling(n, min_periods=n).std(ddof=0)

    out = df[["dt"]].copy()
    out["boll_mid"] = mid
    out["std20"] = std
    for i in (1, 2, 3):
        out[f"up{i}"] = mid + i * std
        out[f"dn{i}"] = mid - i * std
    return out

def boll_position(close: float, r: pd.Series) -> str:
    """七轨位置标签（信号发生时用当时已计算轨值）"""
    keys = ["up1", "up2", "up3", "dn1", "dn2", "dn3", "boll_mid"]
    if any(pd.isna(r[c]) for c in keys):
        return "不足轨值"
    if close >= r["up3"]:
        return "上3轨以上"
    if close >= r["up2"]:
        return "上2轨—上3轨"
    if close >= r["up1"]:
        return "上1轨—上2轨"
    if close >= r["boll_mid"]:
        return "中轨—上1轨"
    if close >= r["dn1"]:
        return "下1轨—中轨"
    if close >= r["dn2"]:
        return "下2轨—下1轨"
    if close >= r["dn3"]:
        return "下3轨—下2轨"
    return "下3轨以下"

def m30_trend(m30: pd.DataFrame, n=20) -> pd.DataFrame:
    """30分钟趋势三态：只用已完成30分钟K。上升=收>MA20且MA20上行；下降反之。"""
    d = m30.copy()
    d["ma20"] = d["close"].rolling(n, min_periods=n).mean()
    d["ma20_up"] = d["ma20"] > d["ma20"].shift(1)
    d["ma20_dn"] = d["ma20"] < d["ma20"].shift(1)

    cond_up = (d["close"] > d["ma20"]) & d["ma20_up"]
    cond_dn = (d["close"] < d["ma20"]) & d["ma20_dn"]
    d["trend"] = np.where(cond_up, "上升", np.where(cond_dn, "下降", "震荡"))
    d.loc[d["ma20"].isna(), "trend"] = "趋势样本不足"

    # 关键防未来：15分钟时点 t 只能使用“15分钟K收盘前已收盘”的30分钟K
    # 30分钟K的 dt 取该K的结束时间
    return d[["dt", "trend"]].rename(columns={"dt": "m30_end"})

def resample_m30(m15: pd.DataFrame) -> pd.DataFrame:
    """由15分钟重采样30分钟；bar的dt=该30分钟K的收盘（结束）时刻。"""
    g = m15.set_index("dt").resample("30min", label="right", closed="right")
    d = pd.DataFrame({
        "open": g["open"].first(),
        "high": g["high"].max(),
        "low": g["low"].min(),
        "close": g["close"].last(),
        "volume": g["volume"].sum(),
    }).dropna(subset=["close"]).reset_index()
    return d

def forward_returns(m15: pd.DataFrame) -> pd.DataFrame:
    """
    对每一根15分钟K计算：
      ret_main[n] = close(t+n) / open(t+1) - 1   主口径，可成交
      ret_aux[n]  = close(t+n) / close(t) - 1    辅助口径，仅对照
    均为事后收益，仅用于统计描述，不构成可执行净值。
    """
    out = m15[["dt", "date", "close", "open"]].copy()
    nxt_open = m15["open"].shift(-1)

    for n in HORIZONS:
        out[f"ret_main_{n}"] = m15["close"].shift(-n) / nxt_open - 1
        out[f"ret_aux_{n}"] = m15["close"].shift(-n) / m15["close"] - 1
    return out

def detect_events(kd: pd.DataFrame, boll: pd.DataFrame, tr: pd.DataFrame,
                  params: dict) -> pd.DataFrame:
    """在15分钟K收盘确认信号。阈值/参数可调（用于敏感性）。"""
    k_lo, k_hi = params["k_lo"], params["k_hi"]
    mid_th = params["mid_th"]

    df = kd.merge(boll, on="dt", how="left")
    df = df.sort_values("dt").reset_index(drop=True)

    # 30分钟趋势对齐：取15分钟K收盘时刻之前已收盘的最近一根30分钟K
    tr_sorted = tr.sort_values("m30_end").reset_index(drop=True)
    raw_idx = np.searchsorted(tr_sorted["m30_end"].values, df["dt"].values, side="right") - 1
    valid = raw_idx >= 0
    idx = np.clip(raw_idx, 0, len(tr_sorted) - 1)
    df["trend"] = np.where(valid, tr_sorted["trend"].values[idx], "趋势样本不足")

    K, D = df["K"], df["D"]
    Kp, Dp = df["K"].shift(1), df["D"].shift(1)
    cross_kd = (K > D) & (Kp <= Dp)
    cross_dk = (D > K) & (Dp >= Kp)

    df["signal"] = np.select(
        [
            cross_kd & (K < k_lo),
            cross_kd & (K >= mid_th) & (K <= k_hi),
            cross_dk & (K > k_hi),
            cross_dk & (K > mid_th) & (df["trend"] == "下降"),
        ],
        ["买信号", "强延续信号", "卖信号A", "卖信号B"],
        default="",
    )

    df["K_at_signal"] = np.where(df["signal"] != "", K, np.nan)
    df["boll_pos"] = [
        boll_position(c, r) if s else ""
        for c, s, r in zip(df["close"], df["signal"], df.to_dict("records"))
    ]
    return df

def apply_cooldown(df: pd.DataFrame) -> pd.DataFrame:
    """同一信号类型32根K内只取第一次；记录原始与独立事件。"""
    df = df.copy()
    df["event_id"] = np.arange(len(df))
    df["is_independent"] = False
    last_idx = {}

    for i, (s, eid) in enumerate(zip(df["signal"], df["event_id"])):
        if s == "":
            continue
        if s not in last_idx or i - last_idx[s] > COOLDOWN:
            df.at[i, "is_independent"] = True
            last_idx[s] = i

    return df

def cluster_boot_ci(x: np.ndarray, dates: np.ndarray, b=BOOT_B, seed=SEED):
    """按事件日期聚类的bootstrap 95%CI（对日均值做百分比法）"""
    rng = np.random.default_rng(seed)
    if len(x) == 0:
        return (np.nan, np.nan)

    tmp = pd.DataFrame({"x": x, "d": dates}).dropna()
    if len(tmp) < 3:
        return (np.nan, np.nan)

    groups = {d: g["x"].values for d, g in tmp.groupby("d")}
    keys = list(groups.keys())
    means = np.empty(b)

    for j in range(b):
        pick = rng.choice(keys, size=len(keys), replace=True)
        means[j] = np.concatenate([groups[k] for k in pick]).mean()

    return tuple(np.percentile(means, [2.5, 97.5]))

def paired_benchmark(ev_rows: pd.DataFrame, base: pd.DataFrame, n: int) -> pd.DataFrame:
    """
    逐事件配对基准：同一标的、同一时间起点、同一期限的无条件收益。
    实现：对每个信号事件，基准 = 同交易日内、剔除事件冷却窗后其余K线 ret_main_n 的均值。
    若当日可用点 < 3 个，回退全样本均值，并写 fallback 标记。
    """
    ret_col = f"ret_main_{n}"
    ev = ev_rows.copy()
    overall = base[ret_col].mean()
    b = base[["dt", "date", ret_col]].copy()

    ev_dt_pos = {d: i for i, d in enumerate(pd.DatetimeIndex(base["dt"]))}
    bm = []
    fb = []

    for _, r in ev.iterrows():
        day_mask = b["date"] == r["date"]
        i = ev_dt_pos.get(r["dt"], -1)
        mask = day_mask.copy()

        if i >= 0:
            lo = max(0, i - 2)
            hi = min(len(base), i + COOLDOWN + 1)
            pos_idx = np.arange(len(base))[day_mask.values]
            sel = set(pos_idx) & set(range(lo, hi))
            mask.iloc[list(sel)] = False

        vals = b.loc[mask, ret_col]
        if len(vals) >= 3:
            bm.append(vals.mean())
            fb.append(False)
        else:
            bm.append(overall)
            fb.append(True)

    ev["benchmark"] = bm
    ev["bench_fallback"] = fb
    return ev

def stat_block(rows: pd.DataFrame, n: int, col: str = None) -> dict:
    col = col or f"ret_main_{n}"
    x = rows[col].values.astype(float)
    dts = rows["date"].values

    mean = np.nanmean(x) if len(x) else np.nan
    win = np.nanmean(x > 0) if len(x) else np.nan
    ex_win = np.nanmean(x > rows["benchmark"].values) if len(x) else np.nan
    lo, hi = cluster_boot_ci(x, dts)
    q = np.nanpercentile(x, [5, 25, 50, 75, 95]) if len(x) else [np.nan] * 5

    return {
        "样本数": len(x),
        "独立日数": rows["date"].nunique(),
        "平均收益%": round(mean * 100, 4),
        "中位数%": round(q[2] * 100, 4),
        "胜率%": round(win * 100, 2),
        "超额胜率%": round(ex_win * 100, 2),
        "q5%": round(q[0] * 100, 3),
        "q25%": round(q[1] * 100, 3),
        "q75%": round(q[3] * 100, 3),
        "q95%": round(q[4] * 100, 3),
        "CI下限%": round(lo * 100, 4) if np.isfinite(lo) else None,
        "CI上限%": round(hi * 100, 4) if np.isfinite(hi) else None,
    }

def md_table(rows, cols=None) -> str:
    df = pd.DataFrame(rows)
    if cols:
        df = df[[c for c in cols if c in df.columns]]
    return df.to_markdown(index=False)

def run(m15_path, m30_path=None, symbol="未填写", start=None, end=None, out="report.md"):
    m15 = load_kline(m15_path)
    m15 = m15[m15["dt"] >= pd.to_datetime(start)] if start else m15
    m15 = m15[m15["dt"] <= pd.to_datetime(end)] if end else m15

    if m30_path:
        m30 = load_kline(m30_path)
        m30 = m30[m30["dt"] >= m15["dt"].min()]
        m30 = m30[m30["dt"] <= m15["dt"].max()]
    else:
        m30 = resample_m30(m15)

    tr = m30_trend(m30)
    kd = kdj(m15)
    bl = boll7(m15)
    fr = forward_returns(m15)

    ev = detect_events(kd, bl, tr, {"k_lo": 20, "k_hi": 80, "mid_th": 50})
    ev = ev.merge(fr, on=["dt", "date", "close"], suffixes=("", "_f"))
    ev = apply_cooldown(ev)

    lines = [f"# 信号层面事件研究报告（{symbol}）", ""]

    # ---------------- 表1 数据校验
    chk = {
        "15分钟K线数": len(m15),
        "30分钟K线数": len(m30),
        "起止": f"{m15['dt'].min()} ~ {m15['dt'].max()}",
        "OHLC越界行": int(((m15["high"] < m15[["open", "close"]].max(axis=1)) |
                           (m15["low"] > m15[["open", "close"]].min(axis=1))).sum()),
        "成交量为空占比%": round(m15["volume"].isna().mean() * 100, 2),
        "事件原始数(含嵌套)": int((ev["signal"] != "").sum()),
        "独立事件数(冷却32根)": int(ev["is_independent"].sum()),
    }
    lines += ["## 表1 数据校验与样本概况", md_table([chk]), ""]

    for s in ["买信号", "强延续信号", "卖信号A", "卖信号B"]:
        sub = ev[ev["signal"] == s]
        info = f"{s}: 原始={len(sub)} / 独立={int(sub['is_independent'].sum())} / 交易日={sub[sub['is_independent']]['date'].nunique()}" if len(sub) else f"{s}: 无事件"
        lines += [f"- {info}"]

    lines.append("")

    # ---------------- 表2 事件研究总表
    rows2 = []
    for s in ["买信号", "强延续信号", "卖信号A", "卖信号B"]:
        for n in HORIZONS:
            sub = ev[(ev["signal"] == s) & (ev["is_independent"])].copy()
            if len(sub) == 0:
                continue
            sub = paired_benchmark(sub, fr, n)
            r = {"信号": s, "前向期(根15min)": n}
            r.update(stat_block(sub, n))
            rows2.append(r)

    for r in rows2:
        for c in COSTS[1:]:
            r[f"扣单边{c*100:.2f}%净收益%"] = round(r["平均收益%"] - 2 * c * 100, 4)

    rows2 = [{k: v for k, v in r.items() if not k.startswith("扣单边0.00")} for r in rows2]
    lines += [
        "## 表2 事件研究总表（主口径：下一根15分钟开盘入场；配对基准；冷却32根；CI为按日聚类bootstrap）",
        "> 注：未做多重比较校正，为探索性结果。", "",
        md_table(rows2), "",
    ]

    # ---------------- 表3 分趋势环境
    rows3 = []
    for s in ["买信号", "强延续信号", "卖信号A", "卖信号B"]:
        for trd in ["上升", "震荡", "下降", "趋势样本不足"]:
            for n in [2, 8, 32]:
                sub = ev[(ev["signal"] == s) & (ev["trend"] == trd) & (ev["is_independent"])].copy()
                if len(sub) == 0:
                    continue
                sub = paired_benchmark(sub, fr, n)
                r = {"信号": s, "趋势": trd, "前向期": n}
                r.update(stat_block(sub, n))
                if len(sub) < MIN_GROUP_N:
                    r["备注"] = "样本<30，仅描述不下结论"
                rows3.append(r)

    lines += ["## 表3 分30分钟趋势环境表", md_table(rows3), ""]

    # ---------------- 表4 分布林位置
    rows4 = []
    pos_order = ["下3轨以下", "下3轨—下2轨", "下2轨—下1轨", "下1轨—中轨",
                 "中轨—上1轨", "上1轨—上2轨", "上2轨—上3轨", "上3轨以上"]
    for s in ["买信号", "强延续信号", "卖信号A", "卖信号B"]:
        for p in pos_order:
            for n in [2, 8, 32]:
                sub = ev[(ev["signal"] == s) & (ev["boll_pos"] == p) & (ev["is_independent"])].copy()
                if len(sub) == 0:
                    continue
                sub = paired_benchmark(sub, fr, n)
                r = {"信号": s, "布林位置": p, "前向期": n}
                r.update(stat_block(sub, n))
                if len(sub) < MIN_GROUP_N:
                    r["备注"] = "样本<30，仅描述不下结论"
                rows4.append(r)

    lines += ["## 表4 分布林位置表", md_table(rows4), ""]

    # ---------------- 表5 趋势×位置交叉
    rows5 = []
    for s in ["买信号", "强延续信号"]:
        for trd, p in itertools.product(["上升", "震荡", "下降"], pos_order):
            n = 8
            sub = ev[(ev["signal"] == s) & (ev["trend"] == trd) & (ev["boll_pos"] == p)
                     & (ev["is_independent"])].copy()
            if len(sub) == 0:
                continue
            sub = paired_benchmark(sub, fr, n)
            r = {"信号": s, "趋势": trd, "布林位置": p, "前向期": n}
            r.update(stat_block(sub, n))
            if len(sub) < MIN_GROUP_N:
                r["备注"] = "样本<30，仅描述不下结论"
            rows5.append(r)

    lines += ["## 表5 趋势×布林位置交叉表（前向期=8根）", md_table(rows5), ""]

    # ---------------- 表6 参数敏感性
    rows6 = []
    base_p = {"k_lo": 20, "k_hi": 80, "mid_th": 50}
    sens = []
    sens.append(("基准 KDJ9/3/3 阈20/80/50", base_p))
    sens.append(("KDJ 12/3/3", {"k_lo": 20, "k_hi": 80, "mid_th": 50, "kdj_n": 12}))
    sens.append(("KDJ 6/3/3", {"k_lo": 20, "k_hi": 80, "mid_th": 50, "kdj_n": 6}))
    sens.append(("阈值 10/90", {"k_lo": 10, "k_hi": 90, "mid_th": 50}))
    sens.append(("阈值 25/75", {"k_lo": 25, "k_hi": 75, "mid_th": 50}))
    sens.append(("强延续带 45~75", {"k_lo": 20, "k_hi": 80, "mid_th": 45, "k_hi2": 75}))

    for name, p in sens:
        p_full = dict(base_p)
        p_full.update(p)
        n_kdj = p_full.pop("kdj_n", 9)

        kd2 = kdj(m15, n=n_kdj)
        ev2 = detect_events(kd2, bl, tr, p_full)

        if "k_hi2" in p:
            K, D = ev2["K"], ev2["D"]
            cross_kd = (K > D) & (K.shift(1) <= D.shift(1))
            ev2["signal"] = np.where(cross_kd & (K >= p_full["mid_th"]) & (K <= p_full["k_hi2"]),
                                     "强延续信号", ev2["signal"])

        ev2 = apply_cooldown(ev2)
        ev2 = ev2.merge(fr, on=["dt", "date", "close"], suffixes=("", "_f"))

        for s in ["买信号", "卖信号A", "卖信号B", "强延续信号"]:
            for n in [2, 8]:
                sub = ev2[(ev2["signal"] == s) & (ev2["is_independent"])].copy()
                if len(sub) == 0:
                    continue
                sub = paired_benchmark(sub, fr, n)
                r = {"参数组": name, "信号": s, "前向期": n}
                r.update(stat_block(sub, n))
                rows6.append(r)

    lines += ["## 表6 参数敏感性（仅微调一档；结论稳定性看均值/超额胜率符号与区间是否跨零）",
              md_table(rows6), ""]

    # ---------------- 表6b 成本敏感性
    rows6b = []
    for s in ["买信号", "强延续信号", "卖信号A", "卖信号B"]:
        for n in [2, 8]:
            sub = ev[(ev["signal"] == s) & (ev["is_independent"])].copy()
            if len(sub) == 0:
                continue
            sub = paired_benchmark(sub, fr, n)
            st = stat_block(sub, n)
            r = {"信号": s, "前向期": n}
            r.update({f"净收益(单边{c*100:.2f}%)%": round(st["平均收益%"] - 2 * c * 100, 4) for c in COSTS})
            rows6b.append(r)

    lines += ["## 表6b 成本敏感性（双边合计=2×单边）", md_table(rows6b), ""]

    # ---------------- 表6c 分段验证
    half = sorted(ev[ev["is_independent"]]["date"].unique())
    mid = half[len(half) // 2]
    rows6c = []
    for s in ["买信号", "卖信号A", "卖信号B", "强延续信号"]:
        for tag, cond in [("前半段", lambda d: d <= mid), ("后半段", lambda d: d > mid)]:
            for n in [2, 8]:
                sub = ev[(ev["signal"] == s) & (ev["is_independent"]) & ev["date"].map(cond)].copy()
                if len(sub) == 0:
                    continue
                sub = paired_benchmark(sub, fr, n)
                r = {"信号": s, "分段": tag, "前向期": n}
                r.update(stat_block(sub, n))
                rows6c.append(r)

    lines += ["## 表6c 分段验证（按时间中点前后半段）", md_table(rows6c), ""]

    # ---------------- 表7 卖信号避损
    rows7 = []
    for s in ["卖信号A", "卖信号B"]:
        sub = ev[(ev["signal"] == s) & (ev["is_independent"])].copy()
        for _, r0 in sub.iterrows():
            i = r0["event_id"]
            win = m15.iloc[i + 1: i + 1 + 32]
            if len(win) == 0:
                continue

            entry = m15["open"].iloc[i + 1] if i + 1 < len(m15) else np.nan
            path = win["close"] / entry - 1

            if len(path) == 32:
                peak = (1 + path).cummax()
                drawdown = (1 + path) / peak - 1
                max_dd = -drawdown.min()
                cum_ret = path.iloc[-1]
                reverse_buy = bool((ev["signal"].iloc[i + 1: i + 33] == "买信号").any())
            else:
                max_dd = np.nan
                cum_ret = np.nan
                reverse_buy = False

            rows7.append({
                "信号": s,
                "dt": r0["dt"],
                "趋势": r0["trend"],
                "布林位置": r0["boll_pos"],
                "后32根累计收益%": round(cum_ret * 100, 3) if np.isfinite(cum_ret) else np.nan,
                "后32根最大回撤%": round(max_dd * 100, 3) if np.isfinite(max_dd) else np.nan,
                "期间出现买信号": reverse_buy,
            })

    lines += ["## 表7 卖信号避损效果表（事件明细+汇总；ETF无法做空时此表衡量避损能力）", ""]

    if rows7:
        det = pd.DataFrame(rows7)
        summ = det.groupby("信号").agg(
            事件数=("后32根累计收益%", "size"),
            平均后32根收益=("后32根累计收益%", "mean"),
            中位后32根收益=("后32根累计收益%", "median"),
            平均最大回撤=("后32根最大回撤%", "mean"),
            期间出现买信号比例=("期间出现买信号", "mean"),
        ).round(3)
        lines += ["**汇总**", "", summ.to_markdown(), "",
                  "**明细（前50条）**", "", det.head(50).to_markdown(index=False), ""]
    else:
        lines += ["（无卖信号事件）", ""]

    lines += [
        "## 结论框架（数据填入后填写）",
        "1. 统计显著：均值95%CI是否不含0；超额胜率CI是否不含50%。",
        "2. 经济意义显著：扣除双边成本后均值是否仍>0（买）/仍为正避损（卖），幅度是否超过滑点量级。",
        "3. 区分：统计显著≠经济显著；样本<30的组仅描述。",
        "4. 局限：未做多重比较校正；冷却期选择影响独立事件数；ETF不能做空时卖信号结论仅限避损；配对基准在日内非事件点不足时回退全样本均值（已在代码中标记）。",
        "## 需要补充的数据",
        "- 标的与区间（本次【填写】处为空）；15分钟K线CSV；30分钟K线CSV（可选，缺省自动重采样）；",
        "- 若已有布林轨值可提供，但脚本默认自行按 C,20,+/-1/2/3×STD 计算。",
    ]

    report = "\n".join(lines)
    with open(out, "w", encoding="utf-8") as f:
        f.write(report)

    print(f"报告已写出: {out}")
    return report

if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--m15", required=True, help="15分钟K线CSV")
    ap.add_argument("--m30", default=None, help="30分钟K线CSV（可选）")
    ap.add_argument("--symbol", default="未填写")
    ap.add_argument("--start", default=None)
    ap.add_argument("--end", default=None)
    ap.add_argument("--out", default="report.md")
    a = ap.parse_args()
    run(a.m15, a.m30, a.symbol, a.start, a.end, a.out)

因为 workbuddy 数据接口只获取了 800 个 15 分钟K，差不多两个多月，样本偏少，跑出来结果统计不显著，所以分析结果这次就先不放了。