4.2 自定义指标实战

4.2 自定义指标实战

本节摘要：本节走完一个自定义指标的全程：从一个可证伪的想法出发——「波动率收缩之后，往往有行情展开」——先写成输入输出规格（三个产出：ATR 叠加线、带宽分位数区间带、收缩事件点），再用纯标准库实现三件套，配一段不依赖平台的独立自测，最后接入图表与回测。四步各有产物：规格表、indicators_volatility.py、自测输出、接入配置。重点不在 ATR 本身，而在「想法→规格→实现→自测→接入」这条流水线——它是你在任何指标想法上都能复用的模板。

学习目标

• 把一句模糊的行情直觉改写成可证伪的指标规格。
• 用纯标准库实现 ATR、滚动分位数带宽与收缩事件三件套。
• 为指标编写独立于平台的自测（边界值、窗口不足、常数序列）。
• 把自测通过的指标接入图表渲染与回测消费两条路径。

一、从想法到规格

「波动率收缩后往往有行情展开」是一句直觉，直接写代码会漏掉所有关键决策。先把直觉钉死成规格——每个产出对应第 4.1 节的一种形态：

三个可证伪的细节：其一，「收缩」必须有量化定义（这里取分位数低于 20%，阈值是策略超参而非真理）；其二，收缩之后「行情展开」的验证属于回测环节（第 6 章），指标只负责标记，不负责证明；其三，所有窗口参数集中成常量，便于第 6.2 节的 walk-forward 检验它们是不是过拟合来源。

二、实现：纯标准库三件套

# indicators_volatility.py —— 波动率收缩三件套：ATR、带宽分位数、收缩事件
"""纯标准库实现，可独立运行；接入 QuantDinger 时只搬运函数体。"""

WINDOW_ATR = 14        # ATR 周期（示意值）
WINDOW_QTL = 120       # 带宽分位数滚动窗口（示意值）
QTL_LOW = 0.2          # 收缩判定分位数（示意值，属策略超参）

def true_range(bars):
    # bars: [(high, low, close), ...]；首根无前收盘，TR 取当日高低差
    out = []
    for i, (high, low, _close) in enumerate(bars):
        if i == 0:
            out.append(high - low)
        else:
            prev_close = bars[i - 1][2]
            out.append(max(high - low, abs(high - prev_close), abs(low - prev_close)))
    return out

def atr_series(bars, window=WINDOW_ATR):
    tr = true_range(bars)
    out = [None] * len(bars)
    if len(bars) >= window:
        acc = sum(tr[:window])
        out[window - 1] = acc / window              # 首个 ATR：简单平均
        for i in range(window, len(bars)):
            acc = acc - tr[i - window] + tr[i]      # 之后滚动更新
            out[i] = acc / window
    return out

def percentile(sorted_vals, q):
    # 线性插值分位数；sorted_vals 必须已升序
    if not sorted_vals:
        return None
    pos = q * (len(sorted_vals) - 1)
    lo = int(pos)
    hi = min(lo + 1, len(sorted_vals) - 1)
    frac = pos - lo
    return sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac

def bandwidth_squeeze(bars, window_q=WINDOW_QTL, qtl_low=QTL_LOW):
    # 返回 (band 序列三条参考线, squeeze 事件序列)
    atrs = atr_series(bars)
    bandwidth = [None if a is None or bars[i][2] == 0 else a / bars[i][2]
                 for i, a in enumerate(atrs)]
    bands, squeeze = [None] * len(bars), [None] * len(bars)
    for i in range(len(bars)):
        seg = [b for b in bandwidth[max(0, i - window_q + 1): i + 1] if b is not None]
        if len(seg) >= window_q // 2:               # 窗口半满才开始出值（示意）
            s = sorted(seg)
            bands[i] = (percentile(s, 0.2), percentile(s, 0.5), percentile(s, 0.8))
            squeeze[i] = 1 if bandwidth[i] is not None and bandwidth[i] <= bands[i][0] else 0
    return bands, squeeze
​

实现里有三处刻意保守的选择：输出序列与输入等长、缺值用 None 占位（下游不用猜对齐）；分位数用线性插值而不是最近邻（参考线连续、事件不抖动）；收缩事件直接复用 band 的下轨（「一处定义、两处使用」在指标内部也要成立）。

三、自测：不依赖平台的证据

第 4.1 节说过，指标要在接入前先自证。自测覆盖四类边界：窗口不足（应输出 None 而不是崩溃）、常数序列（ATR 应为 0、带宽应为 0）、单调序列（分位数单调）、已知小样本手算对照。

# test_indicators_volatility.py —— 独立自测：边界值与手算对照
from indicators_volatility import atr_series, bandwidth_squeeze, percentile

def _run_all():
    bars = [(10 + i * 0.1, 9 + i * 0.1, 9.5 + i * 0.1) for i in range(30)]  # 单调上行的合成数据

    atrs = atr_series(bars)
    assert atrs[12] is None and atrs[13] is not None        # 窗口不足→None；恰好满窗→出值
    hand = sum(max(h - l, abs(h - pc), abs(l - pc))
               for (h, l, _c), (_h, _l, pc) in zip(bars[1:14], bars[0:13])) / 14 \
           + (bars[0][0] - bars[0][1]) / 14
    assert abs(atrs[13] - hand) < 1e-9                       # 与手算一致（首窗含首根 TR）

    flat = [(10.0, 10.0, 10.0)] * 40                         # 常数序列：TR 全零
    assert all(a == 0 for a in atr_series(flat) if a is not None)

    assert percentile([1, 2, 3, 4], 0.5) == 2.5              # 线性插值中位数
    bands, squeeze = bandwidth_squeeze(bars, window_q=10, qtl_low=0.2)
    assert bands[4] is None and bands[9] is not None         # 分位数窗口同样守规矩
    assert all(s in (0, 1) for s in squeeze if s is not None)
    print("全部断言通过：窗口不足/常数/单调/手算对照")

if __name__ == "__main__":
    _run_all()
​

四、接入图表与回测

自测通过后才是接入。图表侧：把三个产出按第 4.1 节的形态注册——ATR 注册为 overlay、分位数三条线注册为 band、收缩事件注册为 marker（注册接口的写法示意见第 4.1 节，以官方文档为准）。回测侧：收缩事件就是策略的触发条件，第 5.2 节会把它写成 intent 的来源。接入后做一次「对表」：抽一段历史，图表上点开某个收缩 marker，核对回测日志同一根 K 线上是否也出现该事件——两处消费同一实现，「一处定义、两处使用」的承诺至此兑现。

接入后的数据流（本节产物在链路中的位置）

 K 线 ──▶ indicators_volatility.py ──┬──▶ 图表：overlay/band/marker
                （一处定义）          └──▶ 回测：squeeze 事件 ──▶ 第 5 章策略 intent
​

本节要点回顾

• 规格先行：直觉拆成形态、定义、消费方三列，参数显式化为常量。
• 实现三件套：ATR 滚动更新、分位数线性插值、事件复用 band 下轨。
• 自测四边界：窗口不足、常数、单调、手算对照——接入前先自证。
• 接入后对表：图表与回测消费同一实现，抽查一致再进入下一章。

指标已经能把「收缩」说清楚了，但它还不会告诉你买还是卖——把指标变成意图与仓位，是下一章 Strategy API V2 的工作。