PSAR on RL10 in Python

Python

import numpy as np


def rl_series(values, n=10):
    """RL10: linear-regression endpoint at the last bar of each n-bar window.

    Returns a float array the same length as `values`; the first n-1 entries
    are NaN (insufficient lookback). Vectorized via np.correlate.
    Canonical: edge-transform CoreIndicators.rl_series (contract HR-007).
    """
    values = np.asarray(values, dtype=float)
    window = int(n)
    length = len(values)
    out = np.full(length, np.nan, dtype=float)
    if window <= 0 or length < window:
        return out
    x = np.arange(window, dtype=float)
    x_mean = (window - 1) / 2.0
    x_diff = x - x_mean
    denom = float(np.sum(x_diff ** 2))
    if denom == 0:
        return out
    # Fixed weights whose dot-product over the window equals the regression
    # endpoint value (slope*(n-1) + intercept) at x = n-1.
    w = 1.0 / window + x_mean * x_diff / denom
    out[window - 1:] = np.correlate(values, w, mode="valid")
    return out


def psar_walk_series(values, af=0.02, m=0.12):
    """Wilder Parabolic SAR forward walk over a 1-D series.

    NaN warmup bars are skipped; the walk starts at the first index i where
    values[i] and values[i+1] are both finite. Returns the SAR array.
    Canonical: edge-transform CoreIndicators.psar_walk_series (contract HR-006).
    """
    n = len(values)
    psar_vals = np.full(n, np.nan)
    if n < 2:
        return psar_vals

    start = -1
    for i in range(n - 1):
        if not np.isnan(values[i]) and not np.isnan(values[i + 1]):
            start = i
            break
    if start < 0:
        return psar_vals

    tr = 1 if values[start + 1] >= values[start] else -1
    ep = float(values[start + 1])          # extreme point
    accel = af                              # acceleration factor
    psar_vals[start] = psar_vals[start + 1] = float(values[start])

    for i in range(start + 2, n):
        if np.isnan(values[i]):
            continue
        prev = psar_vals[i - 1]
        if tr == 1:
            ps = prev + accel * (ep - prev)
            # SAR may not penetrate the prior one/two bars.
            if not np.isnan(values[i - 1]):
                ps = min(ps, float(values[i - 1]))
            if i >= 3 and not np.isnan(values[i - 2]):
                ps = min(ps, float(values[i - 2]))
            if values[i] < ps:                # flip to short
                tr = -1
                ps = ep
                ep = float(values[i])
                accel = af
            elif values[i] > ep:              # new high -> step accel
                ep = float(values[i])
                accel = min(accel + af, m)
        else:
            ps = prev + accel * (ep - prev)
            if not np.isnan(values[i - 1]):
                ps = max(ps, float(values[i - 1]))
            if i >= 3 and not np.isnan(values[i - 2]):
                ps = max(ps, float(values[i - 2]))
            if values[i] > ps:                # flip to long
                tr = 1
                ps = ep
                ep = float(values[i])
                accel = af
            elif values[i] < ep:              # new low -> step accel
                ep = float(values[i])
                accel = min(accel + af, m)
        psar_vals[i] = ps
    return psar_vals


def psar_on_rl10(close, af=0.02, m=0.12):
    """Parabolic SAR computed over the RL10 line of `close`.

    1. Smooth price into the RL10 regression-endpoint series.
    2. Walk a Wilder PSAR (af=0.02, max=0.12) over that smoothed series.
    Returns a NaN-padded SAR array aligned to `close`.
    """
    rl = rl_series(close, n=10)
    return psar_walk_series(rl, af=af, m=m)