custom-indicator

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Create a custom technical indicator with Numba JIT compilation for production-grade speed.

使用Numba JIT编译创建自定义技术指标，实现生产级别的运行速度。

Arguments

参数

```
$0
```
= indicator name (e.g., zscore, squeeze, vwap-bands, custom-rsi, mean-reversion). Required.

If no arguments, ask the user what indicator they want to build.

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

如果未提供参数，请询问用户想要构建何种指标。

Instructions

操作步骤

Read the indicator-expert rules, especially:
- ```
rules/custom-indicators.md
```
  — Numba patterns and templates
- ```
rules/numba-optimization.md
```
  — Performance best practices
- ```
rules/indicator-catalog.md
```
  — Check if indicator already exists in openalgo.ta
Check first: If the indicator already exists in
```
openalgo.ta
```
, tell the user and show the existing API
Create
```
custom_indicators/{indicator_name}/
```
directory (on-demand)
Create
```
{indicator_name}.py
```
with:

阅读指标专家规则，尤其是：
- ```
rules/custom-indicators.md
```
  — Numba模式与模板
- ```
rules/numba-optimization.md
```
  — 性能最佳实践
- ```
rules/indicator-catalog.md
```
  — 检查该指标是否已存在于openalgo.ta中
优先检查：如果指标已存在于
```
openalgo.ta
```
中，告知用户并展示现有的API
按需创建
```
custom_indicators/{indicator_name}/
```
目录
创建
```
{indicator_name}.py
```
文件，文件结构如下：

File Structure

文件结构

python

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

python

"""
{指标名称} — 自定义指标
描述：{该指标的衡量内容}
类别：{趋势/动量/波动率/成交量/震荡指标}
"""
import numpy as np
from numba import njit
import pandas as pd

--- Core Computation (Numba JIT) ---

--- 核心计算（Numba JIT） ---

@njit(cache=True, nogil=True) def compute{name}(data: np.ndarray, period: int) -> np.ndarray: """Numba-compiled core computation.""" n = len(data) result = np.full(n, np.nan) # ... O(n) algorithm ... return result

@njit(cache=True, nogil=True) def compute{name}(data: np.ndarray, period: int) -> np.ndarray: """经Numba编译的核心计算函数。""" n = len(data) result = np.full(n, np.nan) # ... O(n) 算法实现 ... return result

--- Public API ---

--- 公开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. Create `chart.py` for visualization:

```python
"""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}

def {name}(data, period=20): """ {指标名称}

参数:
    data: 收盘价数据（numpy数组、pandas Series或列表）
    period: 回溯周期（默认值：20）

返回:
    与输入类型一致的指标值数组/序列
"""
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`用于可视化：

```python
"""使用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. Create `benchmark.py` for performance testing:

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


6. 创建`benchmark.py`用于性能测试：

```python
"""对自定义指标进行基准测试。"""
import numpy as np
import time
from {indicator_name} import {name}

Warmup

预热

data = np.random.randn(1000) _ = {name}(data, 20)

Benchmark on different sizes

针对不同数据规模进行基准测试

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

undefined

undefined

Numba Rules (CRITICAL)

Numba规则（至关重要）

MUST DO

必须遵守

```
@njit(cache=True, nogil=True)
```
on all compute functions
```
np.full(n, np.nan)
```
to initialize output arrays
Use
```
np.isnan()
```
for NaN checks
Explicit
```
for
```
loops (Numba compiles to machine code)
O(n) algorithms: rolling sum, EMA recursion, deque-based extrema
Float64 for all numeric arrays

所有计算函数都要添加
```
@njit(cache=True, nogil=True)
```
装饰器
使用
```
np.full(n, np.nan)
```
初始化输出数组
使用
```
np.isnan()
```
进行NaN值检查
使用显式
```
for
```
循环（Numba会将其编译为机器码）
采用O(n)算法：滚动求和、EMA递归、基于双端队列的极值计算
所有数值数组使用Float64类型

MUST NOT

禁止操作

Never
```
fastmath=True
```
(breaks
```
np.isnan()
```
)
Never use pandas inside
```
@njit
```
Never use try/except, dicts, sets, strings inside
```
@njit
```
Never call non-jitted functions from inside
```
@njit
```

绝不能使用
```
fastmath=True
```
（会破坏
```
np.isnan()
```
的功能）
绝不能在
```
@njit
```
装饰的函数内部使用pandas
绝不能在
```
@njit
```
装饰的函数内部使用try/except、字典、集合、字符串
绝不能在
```
@njit
```
装饰的函数内部调用未经过JIT编译的函数

Available Building Blocks

可用构建模块

These existing functions can be called inside

@njit

python

from openalgo.indicators.utils import (
    sma, ema, ema_wilder, stdev, true_range, atr_wilder,
    highest, lowest, rolling_sum, crossover, crossunder
)

以下现有函数可在

@njit

装饰的函数内部调用：

python

from openalgo.indicators.utils import (
    sma, ema, ema_wilder, stdev, true_range, atr_wilder,
    highest, lowest, rolling_sum, crossover, crossunder
)

Common Custom Indicator Patterns

常见自定义指标模式

Pattern	Implementation
Z-Score	`(value - rolling_mean) / rolling_stdev`
Squeeze	Bollinger inside Keltner channel
VWAP Bands	VWAP + N * rolling stdev of (close - vwap)
Momentum Score	Weighted sum of RSI + MACD + ADX conditions
Mean Reversion	Distance from SMA as % + threshold
Range Filter	ATR-based dynamic filter on close
Trend Strength	ADX + directional movement composite

模式	实现方式
Z分数	`(value - 滚动均值) / 滚动标准差`
挤压指标	布林带处于肯特纳通道内部
VWAP带宽	VWAP + N * (收盘价 - VWAP)的滚动标准差
动量评分	RSI + MACD + ADX条件的加权和
均值回归	与SMA的距离百分比 + 阈值
范围过滤器	基于ATR的收盘价动态过滤器
趋势强度	ADX + 方向运动复合指标

Example Usage

示例用法

/custom-indicator zscore

/custom-indicator squeeze-momentum

/custom-indicator vwap-bands

/custom-indicator range-filter

/custom-indicator zscore

/custom-indicator squeeze-momentum

/custom-indicator vwap-bands

/custom-indicator range-filter