Indicators reference.py

"""
Reference Python port of two LuxAlgo Pine v5 indicators, for verifying
Claude Code's implementation against.

1) Nadaraya-Watson Envelope [LuxAlgo]  -- NON-REPAINTING (endpoint) mode only.
   The repainting mode recomputes history on the last bar and is NOT valid
   for backtesting. Endpoint mode matches `repaint = false` in the Pine code.

2) RSI Multi Length [LuxAlgo]

Port notes (Pine -> Python semantics):
- Pine `src[i]` = value i bars back. Endpoint NWE uses a one-sided gaussian
  kernel over the last 500 bars (i = 0..499).
- Pine `ta.sma(x, 499)` = simple mean of last 499 values.
- Pine `ta.crossunder(a, b)` on bar t: a[t] < b[t] and a[t-1] >= b[t-1].
- Pine `nz(x, y)` = y if x is na else x. The RSI ma seeds with avg_rsi on
  the first bar via nz(..., avg_rsi).
- All outputs are NaN until enough history exists (500 bars warm-up for NWE).
"""

import numpy as np
import pandas as pd

WINDOW = 500          # max_bars_back in the Pine script
MAE_LEN = 499         # ta.sma(..., 499)


# ----------------------------------------------------------------------------
# 1) Nadaraya-Watson Envelope, endpoint (non-repainting) mode
# ----------------------------------------------------------------------------
def nwe_endpoint(src: pd.Series, h: float = 8.0, mult: float = 3.0) -> pd.DataFrame:
    """Returns DataFrame[out, mae, upper, lower] indexed like src.

    out[t]  = sum_{i=0..499} src[t-i] * exp(-i^2 / (2 h^2)) / sum(weights)
    mae[t]  = SMA_499(|src - out|) * mult
    upper   = out + mae ; lower = out - mae
    """
    s = src.astype(float).to_numpy()
    n = len(s)
    i = np.arange(WINDOW)
    w = np.exp(-(i ** 2) / (2.0 * h * h))
    den = w.sum()

    out = np.full(n, np.nan)
    # exact convolution, one-sided kernel; valid from bar WINDOW-1 on
    if n >= WINDOW:
        # np.convolve(s, w)[k] = sum_j s[j] * w[k-j]; slice 'valid' region
        conv = np.convolve(s, w, mode="full")[WINDOW - 1 : n]
        out[WINDOW - 1 :] = conv / den

    out_s = pd.Series(out, index=src.index)
    abs_err = (src - out_s).abs()
    mae = abs_err.rolling(MAE_LEN).mean() * mult
    upper = out_s + mae
    lower = out_s - mae
    return pd.DataFrame({"out": out_s, "mae": mae, "upper": upper, "lower": lower})


def nwe_signals(close: pd.Series, upper: pd.Series, lower: pd.Series) -> pd.DataFrame:
    """Green ▲ = ta.crossunder(close, lower); Red ▼ = ta.crossover(close, upper)."""
    c, c1 = close, close.shift(1)
    up_sig = (c < lower) & (c1 >= lower.shift(1))      # ▲ (buy-side/reversal-up)
    dn_sig = (c > upper) & (c1 <= upper.shift(1))      # ▼
    return pd.DataFrame({"sig_up": up_sig.fillna(False),
                         "sig_dn": dn_sig.fillna(False)})


# ----------------------------------------------------------------------------
# 2) RSI Multi Length
# ----------------------------------------------------------------------------
def rsi_multi_length(src: pd.Series, min_len: int = 10, max_len: int = 20,
                     overbought: float = 70.0, oversold: float = 30.0) -> pd.DataFrame:
    """Returns DataFrame[avg_rsi, overbuy_pct, oversell_pct, buy_rsi_ma, sell_rsi_ma].

    For each length L in [min_len, max_len]:
        num_L = RMA(diff, L) ; den_L = RMA(|diff|, L)   (alpha = 1/L, seeded at 0)
        rsi_L = 50 * num_L / den_L + 50
    avg_rsi      = mean over lengths
    overbuy_pct  = % of lengths with rsi > overbought   (green area in Pine)
    oversell_pct = % of lengths with rsi < oversold     (RED area / "red spike")
    buy/sell_rsi_ma: adaptive channels, seeded with avg_rsi on the first bar.
    """
    s = src.astype(float).to_numpy()
    n = len(s)
    diff = np.zeros(n)
    diff[1:] = s[1:] - s[:-1]                      # nz(src - src[1]) -> 0 on bar 0

    lengths = np.arange(min_len, max_len + 1)
    N = len(lengths)
    alpha = 1.0 / lengths                          # vector over lengths
    num = np.zeros(N)
    den = np.zeros(N)

    avg_rsi = np.full(n, np.nan)
    overbuy_pct = np.full(n, np.nan)
    oversell_pct = np.full(n, np.nan)
    buy_ma = np.full(n, np.nan)
    sell_ma = np.full(n, np.nan)

    for t in range(n):
        num = alpha * diff[t] + (1 - alpha) * num
        den = alpha * abs(diff[t]) + (1 - alpha) * den
        with np.errstate(divide="ignore", invalid="ignore"):
            rsi = 50.0 * num / den + 50.0
        rsi = np.where(den == 0, np.nan, rsi)
        a = np.nanmean(rsi) if not np.all(np.isnan(rsi)) else np.nan
        ob = np.nansum(rsi > overbought)
        os_ = np.nansum(rsi < oversold)
        avg_rsi[t] = a
        overbuy_pct[t] = ob / N * 100.0
        oversell_pct[t] = os_ / N * 100.0
        prev_b = buy_ma[t - 1] if t > 0 and not np.isnan(buy_ma[t - 1]) else a
        prev_s = sell_ma[t - 1] if t > 0 and not np.isnan(sell_ma[t - 1]) else a
        buy_ma[t] = prev_b + (ob / N) * (a - prev_b)
        sell_ma[t] = prev_s + (os_ / N) * (a - prev_s)

    idx = src.index
    return pd.DataFrame({"avg_rsi": avg_rsi, "overbuy_pct": overbuy_pct,
                         "oversell_pct": oversell_pct, "buy_rsi_ma": buy_ma,
                         "sell_rsi_ma": sell_ma}, index=idx)


# ----------------------------------------------------------------------------
# Golden test-vector generation (deterministic synthetic series)
# ----------------------------------------------------------------------------
def make_test_vectors(n_bars: int = 1600, seed: int = 42) -> pd.DataFrame:
    rng = np.random.default_rng(seed)
    steps = rng.normal(0, 1.0, n_bars)
    close = pd.Series(100 + np.cumsum(steps) + 5 * np.sin(np.arange(n_bars) / 25),
                      name="close")
    nwe = nwe_endpoint(close, h=8.0, mult=3.0)
    sig = nwe_signals(close, nwe["upper"], nwe["lower"])
    rsi = rsi_multi_length(close, 10, 20, 70.0, 30.0)
    df = pd.concat([close, nwe, sig, rsi], axis=1)
    df.index.name = "bar"
    return df


if __name__ == "__main__":
    df = make_test_vectors()
    df.to_csv("nwe_rsi_test_vectors.csv", float_format="%.10f")
    tail = df.dropna().tail(3)
    print(tail[["close", "out", "upper", "lower", "avg_rsi", "oversell_pct"]])
    print("up signals:", int(df.sig_up.sum()), "| dn signals:", int(df.sig_dn.sum()))