8.自定义指标与性能优化

交易世界里，技术指标就像厨师手中的刀具，内置的固然锋利，但总有需要定制的时候。Jesse 提供了相当丰富的内置指标库，可一旦策略需要独特的市场视角，就得自己动手写指标了。这一章我们深入探讨如何开发自定义指标，以及让计算飞起来的性能优化技巧。

开发自定义指标的过程并不复杂，核心在于理解 Jesse 的数据结构和计算范式。整个框架基于 Numpy 构建，所有 K 线数据都以数组形式流转，这为批量计算创造了绝佳条件，但也要求我们对数组操作有基本认知。

开发自定义指标

动手之前，先确认一件事：这个指标真的需要从零写起吗？GitHub 上藏着大量现成的 Python 实现，搜索时把指标名加上 high、low、open、close 或 price 这类关键词，筛选 Python 或 Jupyter Notebook 类型，往往能找到可用代码。找到后别照搬，理解逻辑再改造，才能融入 Jesse 的生态系统。

以 Elliott Wave Oscillator 为例，这个指标在 TradingView 上的 Pine Script 实现只有几行：

//@version=3
study("Elliott Wave Oscillator")
s2=ema(close, 5) - ema(close, 34)
c_color=s2 <= 0 ? red : lime
plot(s2, color=c_color, style=histogram, linewidth=2)

逻辑很直白：5 周期 EMA 减去 34 周期 EMA，正值绿柱，负值红柱。把它搬到 Jesse 需要几个步骤。

先在项目根目录创建 custom_indicators 文件夹，里面放 __init__.py 文件。然后新建 ewo.py 存放具体逻辑。目录结构会变成这样：

├── storage
├── strategies
└── custom_indicators
    ├── __init__.py
    └── ewo.py

__init__.py 里导入指标函数：

from .ewo import ewo

真正的计算逻辑写在 ewo.py 中。Jesse 的指标函数有固定签名：第一个参数是 candles 数组，接着是指标参数，最后必须是 sequential 布尔值控制返回类型。看具体实现：

import numpy as np
from typing import Union
from jesse.helpers import get_candle_source, slice_candles

def ewo(candles: np.ndarray, short_period: int = 5, long_period: int = 34, source_type="close", sequential = False) -> Union[float, np.ndarray]:
    """
    Elliott Wave Oscillator
    :param candles: np.ndarray
    :param short_period: int - default: 5
    :param long_period: int - default: 34
    :param source_type: str - default: close
    :param sequential: bool - default: False
    :return: Union[float, np.ndarray]
    """
    candles = slice_candles(candles, sequential)

    src = get_candle_source(candles, source_type)
    from jesse.indicators import ema
    ewo = np.subtract(ema(src, period=short_period, sequential=True), ema(src, period=long_period, sequential=True))

    if sequential:
        return ewo
    else:
        return ewo[-1]

策略里调用时，把 custom_indicators 当成模块导入：

from jesse.strategies import Strategy
import custom_indicators as cta

class Strategy01(Strategy):
    @property
    def ewo(self):
        return cta.ewo(self.candles, short_period=5, long_period=34, source_type="close", sequential=True)

这里用 @property 装饰器缓存计算结果，避免重复调用。Jesse 0.21.3 版本引入了 @cached 装饰器，性能提升更明显，后面会详细说。

Numpy 数组操作

Jesse 的 K 线数据是二维 Numpy 数组，每行一根 K 线，列的顺序固定为 [timestamp, open, close, high, low, volume]。记住这个顺序，后面操作数组时不会迷路。

访问特定列用切片语法。比如提取收盘价：

candles_close = candles[:, 2]  # 第2列是close

各列索引对应关系：

• candles[:, 1] 是 open
• candles[:, 2] 是 close
• candles[:, 3] 是 high
• candles[:, 4] 是 low
• candles[:, 5] 是 volume

常用价格类型可以直接计算：

candles_hl2 = (candles[:, 3] + candles[:, 4]) / 2
candles_hlc3 = (candles[:, 3] + candles[:, 4] + candles[:, 2]) / 3
candles_ohlc4 = (candles[:, 1] + candles[:, 3] + candles[:, 4] + candles[:, 2]) / 4

get_candle_source 函数封装了这些逻辑，支持字符串参数 "open"、"close"、"high"、"low"、"volume"、"hl2"、"hlc3"、"ohlc4"，返回对应数据。自己写指标时，直接操作数组往往更快。

数组长度一致性是隐藏的大坑。Numpy 的向量化运算要求参与计算的数组形状相同，长度不一就会报错。更麻烦的是，某些指标需要历史数据预热，前 N 根 K 线算不出有效值，这时必须用 np.nan 填充，而不是 0。

为什么不能用 0？假设策略逻辑是 indicator_value < price，如果指标值因数据不足无法计算，用 0 填充可能导致条件意外成立，触发错误信号。用 np.nan 则安全得多，任何与 np.nan 的比较都会返回 False，避免误判。

处理长度不一致的数组，可以手动填充：

def make_same_length(arr: np.ndarray, target_length: int) -> np.ndarray:
    if len(arr) < target_length:
        padding = np.full(target_length - len(arr), np.nan)
        return np.concatenate([padding, arr])
    return arr[-target_length:]

Numba 性能优化

回测速度是生命线，尤其参数优化时，指标计算可能执行数万次。Jesse 内部用 Numba 加速，自定义指标同样可以。

Numba 是 Python 的即时编译器，把 Python 代码翻译成机器码，对循环和 Numpy 函数效果显著。使用方法简单，给函数加上 @njit 装饰器：

from numba import njit
import numpy as np

@njit
def fast_calculation(data: np.ndarray, period: int) -> np.ndarray:
    result = np.full_like(data, np.nan)
    for i in range(period, len(data)):
        result[i] = np.mean(data[i-period:i])
    return result

注意，Numba 不支持所有 Python 特性，函数里避免使用列表推导式、字典等动态类型。纯数值计算和 Numpy 操作最稳妥。

Jesse 0.21.3 版本在 Strategy 类中加入了 @cached 装饰器，对频繁调用的指标缓存结果。策略里可以这样用：

from jesse.strategies import Strategy
from jesse.helpers import cached

class MyStrategy(Strategy):
    @cached
    def my_indicator(self):
        # 复杂计算只执行一次
        return complex_calculation(self.candles)

缓存机制避免重复计算，某些策略能获得巨大性能提升。

自定义指标最佳实践

切片优化

回测时，策略通常只需要最近 N 根 K 线的指标值，计算全部历史数据是浪费。slice_candles 函数负责这件事：

def slice_candles(candles: np.ndarray, sequential: bool) -> np.ndarray:
    warmup_candles_num = get_config('env.data.warmup_candles_num', 240)
    if not sequential and len(candles) > warmup_candles_num:
        candles = candles[-warmup_candles_num:]
    return candles

sequential=False 时，自动截取最后 240 根 K 线（可配置）。sequential=True 时，Jesse 不知道你需要多少历史数据，所以不自动切片，这时得手动控制：

def my_indicator(candles: np.ndarray, period: int = 14, sequential=False):
    if sequential:
        # 需要多少就切多少
        candles = candles[-500:]
    else:
        candles = slice_candles(candles, sequential)
    # 后续计算...

避免循环

循环是性能杀手，双重循环更是灾难。Numpy 和 Scipy 提供了大量向量化函数，能替代绝大多数循环操作。比如计算 10 周期价格差：

# 错误示范：纯循环
close = candles[:, 2]
result = np.full_like(close, np.nan)
for i in range(10, len(close)):
    result[i] = close[i] - close[i-10]

# 正确示范：向量化
close = candles[:, 2]
result = close - np_shift(close, 10)

np_shift 是自定义的位移函数，实现如下：

def np_shift(arr: np.ndarray, num: int, fill_value=np.nan) -> np.ndarray:
    result = np.empty_like(arr)
    if num > 0:
        result[:num] = fill_value
        result[num:] = arr[:-num]
    elif num < 0:
        result[num:] = fill_value
        result[:num] = arr[-num:]
    else:
        result[:] = arr
    return result

这个函数把数组向前或向后移动，空位用 np.nan 填充，避免了显式循环。

前向填充

某些指标计算中可能出现 np.nan，后续运算需要把前面的有效值向前填充。Numpy 没有内置函数，但可以这样实现：

def np_ffill(arr: np.ndarray, axis: int = 0) -> np.ndarray:
    idx_shape = tuple([slice(None)] + [np.newaxis] * (len(arr.shape) - axis - 1))
    idx = np.where(~np.isnan(arr), np.arange(arr.shape[axis])[idx_shape], 0)
    np.maximum.accumulate(idx, axis=axis, out=idx)
    slc = [np.arange(k)[tuple([slice(None) if dim == i else np.newaxis
                               for dim in range(len(arr.shape))])]
           for i, k in enumerate(arr.shape)]
    slc[axis] = idx
    return arr[tuple(slc)]

这个函数会沿着指定轴，用最后一个有效值填充后续的 np.nan，直到遇到下一个有效值。

滑动窗口

计算滚动指标时，滑动窗口是常用模式。Numpy 的 stride_tricks 可以高效实现：

from numpy.lib.stride_tricks import sliding_window_view

def rolling_mean(data: np.ndarray, window: int) -> np.ndarray:
    # 创建滑动窗口视图
    windows = sliding_window_view(data, window)
    # 计算每窗口均值
    means = np.mean(windows, axis=1)
    # 前 window-1 个值填充 nan
    result = np.full_like(data, np.nan)
    result[window-1:] = means
    return result

这种方式比循环快几个数量级，内存占用也合理。

预热期配置

指标有记忆效应，计算结果依赖历史数据长度。Jesse 的 warmup_candles_num 配置就是为此设计。如果指标周期很大，比如 200 周期的均线，默认的 240 根预热 K 线可能不够，需要在 config.py 中增加：

config = {
    'env': {
        'data': {
            'warmup_candles_num': 500  # 增加到500根
        }
    }
}

预热不足会导致指标值与其他平台（如 TradingView）对不上，这不是 bug，是数据长度不够导致的计算差异。

内置指标的使用

自定义指标不必从零开始，可以组合内置指标。Jesse 内置了 100 多个指标，都从 jesse.indicators 导入：

from jesse.indicators import ema, rsi, bollinger_bands

# 单值模式
current_rsi = rsi(self.candles, period=14)

# 序列模式
rsi_series = rsi(self.candles, period=14, sequential=True)

# 多返回值
bb = bollinger_bands(self.candles, period=20)
upper = bb.upperband
middle = bb.middleband
lower = bb.lowerband

内置指标已经优化过，直接调用比自己实现更可靠。组合现有指标能快速构建新指标，比如把 RSI 和布林带结合，创建波动率加权 RSI。

类型注解与文档

写指标时加上类型注解和文档字符串，方便后续维护。Jesse 的指标签名有固定模式：

def indicator_name(
    candles: np.ndarray,
    param1: int = 默认值,
    param2: float = 默认值,
    source_type: str = "close",
    sequential: bool = False
) -> Union[float, np.ndarray]:
    """
    指标描述
    
    :param candles: np.ndarray - K线数据
    :param param1: int - 参数描述
    :param param2: float - 参数描述
    :param source_type: str - 数据源类型
    :param sequential: bool - 是否返回序列
    :return: Union[float, np.ndarray] - 单值或序列
    """

统一的接口让策略调用更一致，也便于自动生成文档。

调试技巧

指标开发中，调试是个挑战。建议在 Jupyter Notebook 里单独测试：

from jesse import research
import jesse.indicators as ta
import numpy as np

# 获取测试数据
_, candles = research.get_candles(
    'Binance Spot',
    'BTC-USDT',
    '1h',
    start_timestamp,
    end_timestamp
)

# 测试自定义指标
from custom_indicators import my_indicator
values = my_indicator(candles, sequential=True)

# 可视化
import matplotlib.pyplot as plt
plt.plot(values)
plt.show()

可视化能快速发现问题，比如 np.nan 处理不当、数值异常等。确认无误后再集成到策略中。

性能对比

优化前后，性能差异可能惊人。以简单均线为例：

# 未优化版本
def slow_sma(data, period):
    result = []
    for i in range(len(data)):
        if i < period:
            result.append(np.nan)
        else:
            result.append(np.mean(data[i-period:i]))
    return np.array(result)

# 优化版本
@njit
def fast_sma(data: np.ndarray, period: int) -> np.ndarray:
    result = np.full_like(data, np.nan)
    for i in range(period, len(data)):
        result[i] = np.mean(data[i-period:i])
    return result

# 向量化版本
def vectorized_sma(data: np.ndarray, period: int) -> np.ndarray:
    windows = sliding_window_view(data, period)
    means = np.mean(windows, axis=1)
    result = np.full_like(data, np.nan)
    result[period-1:] = means
    return result

在 10 万根 K 线上测试，未优化版本可能耗时数秒，Numba 加速后降到毫秒级，向量化版本更快。参数优化时，这种差距会被放大数千倍。

版本兼容性

Jesse 持续迭代，指标 API 可能变化。比如 0.19.2 版本引入了杠杆支持，config.py 的 margin 配置改为 futures。0.20.0 增加了多个新指标和时间周期。0.21.3 优化了字符串处理，用 f-string 替代 format，提升了性能。

写指标时，尽量用稳定的 API，避免依赖内部实现。如果必须用到新特性，注明最低版本要求：

# 需要 Jesse >= 0.21.3
from jesse.helpers import cached

这样别人使用时能提前知道兼容性。

总结

自定义指标是策略开发的进阶技能。掌握 Numpy 数组操作、Numba 加速、最佳实践，能写出既快又稳的指标。记住几个核心点：切片减少计算、NaN 保证安全、向量化替代循环、缓存避免重复。

下一章我们将进入仓位与风险管理，看看如何把指标信号转化为实际交易，并控制回撤。技术指标是武器，风险管理才是铠甲，两者结合才能在市场中长久生存。