理解一个指标的数值怎样得来,才更容易进一步改进。哪些数据参与计算?什么会保留到下一根?换一种平均方式或价格来源会怎样?源码能回答这些问题。

PyneCore 的 ta 库使用 Pyne code 编写。其中的 Persistent 变量、函数调用和条件都遵循本书学过的机制。完整源码可用,因此你能像跟踪自己程序一样理解现成指标。

本章面向进阶学习,重点是阅读源码和实验方法。不需要一次弄懂整个库。先选一个短函数,再沿着其依赖调用继续阅读。

打开实际实现

  1. 在 PyneIDE 打开此前的 moving_averages.py。
  2. 在 ta.ema(close, 10) 一行,macOS 按住 Command、Windows 或 Linux 按住 Ctrl,点击 ema。也可使用右键 Go to Definition。
  3. 编辑器会打开已安装 PyneCore 的 ta.py,定位到 ema 定义。
  4. 先读函数说明,再读函数体。若调用其他函数,也可以继续跳到其定义。
PyneIDE 编辑器显示 ta.py 中 ema 的定义及关于初始值的说明。
跳转到定义会打开已安装 ta.py 中的 EMA 函数,自己的程序保留在另一标签页。

你在编写代码时获取帮助中已经尝试查看参数和跳转定义。同样操作也能找到 RSI、MACD 及其余库函数源码。完整 ta 参考可帮助查找函数并理解参数。

SMA:公式背后的求和

ta.sma 的两行计算:

length = int(length)
return lib_math.sum(source, length) / length

平均值就是窗口总和除以元素数。长度转成整数后,求和交给 lib_math.sum。状态不在 sma 函数体中,而由调用的求和计算维护。继续追踪,会经过 math.py 到 _math_stateful.py 的实现。

这是阅读源码的重要习惯:函数短,并不意味着没有状态。被调用函数的行为也是计算的一部分。手写移动平均用循环累加的,正是同一窗口中的值。

EMA 与 RMA:更新保留的平均值

预热结束后,EMA 执行:

last_val = last_val + alpha * (source - last_val)
return last_val

last_val 是上一平均值,source 是当前值。用两者差值的一部分更新旧平均。EMA 的 alpha 为 2 / (length + 1)。长度十时,约将新价格与旧均值差的 18.18% 加到旧值上。

源码中 alpha 和 last_val 都是 Persistent。last_val 最初缺失,所以先尝试用同长度 SMA 初始化:

if not (last_val == last_val):
    last_val = sma(source, length)
    return last_val

SMA 数据也不够时,结果仍缺失。有了初始平均,之后才更新旧值。not (last_val == last_val) 是缺失检查,因为缺失值不等于自身。你在自己的示例中用 na(value) 达到同样目的。

RMA 即 Wilder 平滑,起点相近,更新公式不同:

last_val = (last_val * (length - 1) + source) / length
return last_val

长度十时,新值权重为十分之一。EMA 在同长度下赋予新值更大权重,所以二者面对同一价格序列反应不同。

完整源码还包含缺失输入处理和长度为一的特殊处理。即便使用相同公式,若初值不同、缺失处理不同,或没有在同一组 K 线上执行,自己的版本仍可能得到不同结果。浮点运算顺序也有影响,精确比较时需要保持一致。

RSI:分别平滑上涨与下跌

RSI 将数据源的上涨和下跌分别处理。PyneCore 中的核心计算如下:

rma_u = rma(builtins.max(source - prev_src, 0.0), length)
rma_d = rma(builtins.max(prev_src - source, 0.0), length)
prev_src = source

prev_src 是前一个有效源值,由 Persistent 保存。价格从 100 涨到 103,上涨变化为 3、下跌为 0;从 103 跌到 101,则分别为 0 和 2。两个 RMA 分别平滑这两组变化,各调用位置拥有独立状态。

两项差值计算完成后才更新 prev_src,与用持久变量保存前值的进阶练习顺序相同。

两项平滑变化的比值得到 RSI:

if rma_d <= 1e-10:
    return 100.0
if rma_u <= 1e-10:
    return 0.0

return 100 - 100 / (1 + rma_u / rma_d)

没有下跌时返回 100,没有上涨时返回 0。源码也处理极小值,并先检查下跌一侧:若两侧都低于阈值,结果为 100。只看简写公式,不容易注意这些细节。

第一个有效价格还不能构成变化。RSI 先保存它,等下一值才启动两项平滑,因此预热由变化数量决定。RSI 不是概率:70 不表示上涨概率为 70%。

MACD:从均线组合新指标

MACD 先计算两个 EMA 之差,再对差值计算 EMA。省略缺失检查后的核心步骤:

fast = ema(source, fastlen)
slow = ema(source, slowlen)
macd_val = fast - slow
signal = ema(macd_val, siglen)

三个返回值:

return macd_val, signal, macd_val - signal

MACD 线是快慢均线之差,信号线是其平滑版本,柱状值是两线之差。所以两线相等时,柱状值为零。

一次调用返回三个结果,可用三个变量接收:

macd_line, signal_line, histogram = ta.macd(close, 12, 26, 9)

顺序很重要:MACD 线、信号线、柱状值。两条价格均线尚未就绪时,差值也缺失;之后信号线仍有自己的预热。因此,组合指标开头的缺值也来自内部计算。

ATR:跳空也计入区间

ATR 实现很短:

return rma(tr(True), length)

你已熟悉 RMA。这里它平滑的是 True Range(真实波幅),即以下三项的最大值:

  • 当前最高与最低之差;
  • 当前最高与上一收盘的距离;
  • 当前最低与上一收盘的距离。

若上一收盘为 100,当前最低 108、最高 110,本根区间只有 2。但真实波幅是 max(2, 10, 8),即 10,把上一收盘以来的跳空也纳入。

第一根缺上一收盘时,tr(True) 使用 high - low。源码从执行器提供的内部数据读取上一收盘。ATR 用价格单位表示运动幅度,不表示方向。

布林带:平均值与标准差

ta.bb 中轨为 SMA,上下轨与它相距标准差的指定倍数。源码先算标准差与平均值:

std_dev = stdev(series, length)
middle = sma(series, length)

中值有效时,返回部分为:

std_dev *= mult
return middle, middle + std_dev, middle - std_dev

标准差描述窗口值围绕平均值分散的程度。中值 100、标准差 3、倍数 2 时,上轨 106,下轨 94。同样返回三个值,但顺序与 MACD 不同:中轨、上轨、下轨。

自己的版本:先重现原版

创建 bands_workshop.py。自定义函数用库中较基础的部分组合出布林带,同时计算现成 ta.bb 的上轨作为对照。

"""
@pyne edge
"""
from pynecore.lib import close, color, input, plot, script, ta


def custom_bands(source: float, length: int, multiplier: float) -> tuple[float, float, float]:
    middle = ta.sma(source, length)
    deviation = ta.stdev(source, length)
    upper = middle + multiplier * deviation
    lower = middle - multiplier * deviation
    return middle, upper, lower


@script.indicator("Bands Workshop", overlay=True)
def main(
    length: int = input.int(20, title="Length", minval=2, maxval=200),
    multiplier: float = input.float(2.0, title="Deviation multiplier", minval=0.1, maxval=5.0),
):
    builtin_middle, builtin_upper, builtin_lower = ta.bb(close, length, multiplier)
    own_middle, own_upper, own_lower = custom_bands(close, length, multiplier)
    plot(builtin_upper, "Built-in upper", color=color.blue, linewidth=3)
    plot(own_upper, "Custom upper", color=color.orange)
    plot(own_middle, "Custom middle", color=color.green)
    plot(own_lower, "Custom lower", color=color.gray)

tuple[float, float, float] 表示由三个数构成的返回类型。return middle, upper, lower 返回这三个值,调用前的三个变量分别接收。builtin_middle 与 builtin_lower 暂不绘制,之后也可用于检查。

用演示数据运行。自定义上轨与内置上轨重合,细橙线压在较粗蓝线上。鼠标移到一根 K 线,比较 Built-in upper 和 Custom upper。长度二十时,最前十九根还没有完整窗口。

bands_workshop.py 自定义函数旁的图表显示 Built-in upper 与 Custom upper 数值相同,同时显示自己的中轨和下轨。
选中 K 线的内置上轨和自定义上轨都为 216.85,两条线重合。

要更精确检查,通过图表工具栏 Open plot output CSV 打开输出值。验证自定义版本时,既检查首个有效点,也检查后续值。肉眼相同的线仍可能有微小差别。

自己试一试:EMA 中轨

只修改 custom_bands 的中轨,用 EMA 替代 SMA。保留标准差计算和内置 ta.bb 作为比较。

运行前预测:自定义上轨还会始终与内置重合吗?突然的价格变化后会怎样?

参考答案

自定义函数第一行计算改为:

middle = ta.ema(source, length)

上下边界从这个 EMA 向外计算,ta.stdev 保持不变。自己的中轨及上下轨通常会偏离原版。两条上轨之差等于 EMA 与 SMA 的差,因为加上的标准差和倍数相同。

这已经是你自己的实验版本。线条不同本身不说明交易规则会更好;应在相同数据和设置下另用策略验证。

如何继续阅读库

选一个指标,用自己的话说明它接收什么、保留什么、返回什么。跟踪内部调用,直到理解计算。例如 RSI 的 rsi → rma → sma,会从差值一路追到平滑和初始值。

也要读缺失值、预热和边界情况的分支。做自定义版本时,先只改一个细节,与原版比较结果,才能区分有意变化与计算错误。

其他库函数的源码也能用相同的 Command 或 Ctrl 点击访问。较长指标可能需要更多时间,但阅读方法相同:输入、保留数据、计算、结果。