Spaces:
Paused
Paused
File size: 6,208 Bytes
90dc098 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """
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())) |