QuanTAlib/lib/statistics/acf/acf.pine

// Licensed under the Apache License, Version 2.0
// © mihakralj
//@version=6
indicator("Autocorrelation Function (ACF)", "ACF", overlay=false)

//@function Calculates autocorrelation at a specified lag using a circular buffer
//@param src series float Input data series
//@param len simple int Lookback period for calculation
//@param lag simple int Lag order for autocorrelation (default 1)
//@returns float Autocorrelation coefficient between -1 and 1
//@optimized for performance using running sums and biased estimator (divide by n)
acf(series float src, simple int len, simple int lag) =>
    if len <= 0
        runtime.error("Period must be greater than 0")
    if lag < 1
        runtime.error("Lag must be at least 1")
    if len <= lag + 1
        runtime.error("Period must be greater than lag + 1")
    var int p = math.max(1, len)
    var array<float> buffer = array.new_float(p, na)
    var int head = 0, var int count = 0
    float oldest = array.get(buffer, head)
    if not na(oldest)
        count -= 1
    if not na(src)
        array.set(buffer, head, src)
        count += 1
    else
        array.set(buffer, head, na)
    head := (head + 1) % p
    if count <= lag
        na
    else
        // Calculate mean
        float sum = 0.0
        int validN = 0
        for i = 0 to p - 1
            float val = array.get(buffer, i)
            if not na(val)
                sum += val
                validN += 1
        if validN <= lag
            na
        else
            float mean = sum / validN

            // Calculate variance (population): Σ(x - mean)² / n
            float variance = 0.0
            for i = 0 to p - 1
                float val = array.get(buffer, i)
                if not na(val)
                    float diff = val - mean
                    variance += diff * diff
            variance /= validN

            if variance <= 0
                0.0
            else
                // Calculate autocovariance at lag k (biased: divide by n)
                // Need to iterate in temporal order through the circular buffer
                int startIdx = count < p ? 0 : head
                float autocovariance = 0.0
                for t = lag to count - 1
                    int currentIdx = (startIdx + t) % p
                    int laggedIdx = (startIdx + t - lag) % p
                    float xt = array.get(buffer, currentIdx)
                    float xtk = array.get(buffer, laggedIdx)
                    if not na(xt) and not na(xtk)
                        autocovariance += (xt - mean) * (xtk - mean)
                autocovariance /= validN

                // ACF = γ_k / γ_0, clamped to [-1, 1]
                float result = autocovariance / variance
                math.max(-1.0, math.min(1.0, result))

// ---------- Main loop ----------

// Inputs
i_source = input.source(close, "Source")
i_period = input.int(20, "Period", minval=3)
i_lag = input.int(1, "Lag", minval=1)

// Calculation
acf_value = acf(i_source, i_period, i_lag)

// Plot
plot(acf_value, "ACF", color=color.yellow, linewidth=2)