""" 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()))