custom-indicator

通过组合 openalgo 的 Rust 核心 ta 基元与向量化 NumPy，创建自定义技术指标。

参数

• $0 = 指标名称（例如 zscore、squeeze、vwap-bands、custom-rsi、mean-reversion）。必填。

如果没有参数，请询问用户想要构建什么指标。

指令

1. 阅读 indicator-expert 规则，尤其是：
   • rules/custom-indicators.md — NumPy + ta 基元模式和模板
   • rules/performance.md — Rust 核心性能、O(n) 保证、基准测试
   • rules/indicator-catalog.md — 检查 openalgo.ta 中是否已有该指标
2. 先检查：如果该指标已存在于 openalgo.ta（100+ 个指标），告知用户并展示现有 API
3. 创建 custom_indicators/{indicator_name}/ 目录（按需创建）
4. 创建 {indicator_name}.py，内容如下：

文件结构

"""
{Indicator Name} — Custom Indicator
Description: {what it measures}
Category: {trend/momentum/volatility/volume/oscillator}
"""
import numpy as np
import pandas as pd
from openalgo import ta

# --- Core Computation (vectorized NumPy on ta primitives) ---
def _compute_{name}(arr: np.ndarray, period: int) -> np.ndarray:
    """Vectorized core computation built on Rust-core primitives."""
    n = len(arr)
    result = np.full(n, np.nan)
    # Compose from ta primitives (they run in Rust):
    # mean = ta.sma(arr, period); std = ta.stdev(arr, period); ...
    # then combine with NumPy array math (np.where, masks)
    return result

# --- Public API ---
def {name}(data, period=20):
    """
    {Indicator Name}

    Args:
        data: Close prices (numpy array, pandas Series, or list)
        period: Lookback period (default: 20)

    Returns:
        Same type as input with indicator values
    """
    if isinstance(data, pd.Series):
        idx = data.index
        result = _compute_{name}(data.values.astype(np.float64), period)
        return pd.Series(result, index=idx, name="{Name}({period})")
    arr = np.asarray(data, dtype=np.float64)
    return _compute_{name}(arr, period)

5. 创建用于可视化的 chart.py：

"""Chart the custom indicator with Plotly."""
import os
from pathlib import Path
from datetime import datetime, timedelta
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from {indicator_name} import {name}

# ... fetch data, compute indicator, create chart ...

6. 创建用于性能测试的 benchmark.py（无需预热 — Rust 核心从首次调用起就以全速运行）：

"""Benchmark the custom indicator."""
import numpy as np
import time
from {indicator_name} import {name}

for size in [10_000, 100_000, 500_000]:
    data = np.random.randn(size).cumsum() + 1000
    t0 = time.perf_counter()
    _ = {name}(data, 20)
    elapsed = (time.perf_counter() - t0) * 1000
    print(f"{size:>10,} bars: {elapsed:>8.2f}ms")

NumPy 规则（重要）

必须做到

• 尽可能组合使用 ta 基元 — 它们在 Rust 核心中运行
• 使用 np.full(n, np.nan) 初始化输出数组
• 使用数组表达式、np.where 和布尔掩码进行向量化
• 对除法进行保护：使用 np.errstate(invalid="ignore", divide="ignore")，并配合安全分母掩码
• 遵循基元的 NaN 预热期（对 ~np.isnan(...) 使用掩码）
• 所有数值数组均使用 Float64
• 仅使用 O(n) 算法

禁止事项

• 绝不重新实现 openalgo.ta 中已有的指标
• 绝不对大型数组使用逐条 Python 循环 — 应改用向量化
• 分母为零或 NaN 时，绝不进行除法
• 如果指标确实依赖路径（需要基元无法覆盖的顺序状态），先检查 ta.ema / Wilder 风格基元是否已提供递归计算；纯 Python 循环是最后手段 — 保持 O(n)，并说明这种权衡

可用构建模块

在 Rust 核心中运行的公开 ta 方法：

from openalgo import ta

# Rolling math:    ta.sma, ta.ema, ta.wma, ta.stdev, ta.highest, ta.lowest
# Price action:    ta.true_range, ta.atr, ta.change, ta.roc
# Bands/channels:  ta.bbands, ta.keltner, ta.donchian
# Signals:         ta.crossover, ta.crossunder, ta.exrem, ta.rising, ta.falling

常见自定义指标模式

模式	实现
Z-Score	(value - rolling_mean) / rolling_stdev
Squeeze	布林带位于肯特纳通道内
VWAP Bands	VWAP + N * (close - vwap) 的滚动标准差
Momentum Score	RSI + MACD + ADX 条件的加权和
Mean Reversion	与 SMA 的距离百分比 + 阈值
Range Filter	基于 ATR 对 close 进行动态过滤
Trend Strength	ADX + 方向运动复合指标

用法示例

/custom-indicator zscore /custom-indicator squeeze-momentum /custom-indicator vwap-bands /custom-indicator range-filter