custom-indicator

openalgo の Rust コアの ta プリミティブとベクトル化された NumPy を組み合わせて、カスタムテクニカル指標を作成します。

引数

• $0 = 指標名（例：zscore、squeeze、vwap-bands、custom-rsi、mean-reversion）。必須です。

引数がない場合は、作成したい指標をユーザーに尋ねます。

手順

1. インジケーターエキスパートのルールを読みます。特に以下を確認してください。
   • 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 またはワイルダー方式のプリミティブで再帰処理を実現できるか確認してください。通常の 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 スコア	(value - rolling_mean) / rolling_stdev
スクイーズ	ケルトナーチャネル内のボリンジャーバンド
VWAP バンド	VWAP + N * (close - vwap) のローリング標準偏差
モメンタムスコア	RSI + MACD + ADX 条件の加重合計
平均回帰	SMA からの乖離率 + 閾値
レンジフィルター	close に対する ATR ベースの動的フィルター
トレンド強度	ADX + 方向性変動の複合指標

使用例

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