using System.Runtime.CompilerServices; using static System.Math; namespace QuanTAlib; /// /// Cointegration: Measures the statistical equilibrium relationship between two price series /// using the Engle-Granger two-step method with Augmented Dickey-Fuller test. /// /// /// Cointegration tests whether two non-stationary time series have a long-run equilibrium /// relationship. The indicator returns the ADF test statistic for the regression residuals. /// /// Algorithm: /// 1. Estimate linear regression: A = α + β*B + ε /// - β = correlation(A,B) × (σA/σB) /// - α = mean(A) - β × mean(B) /// 2. Calculate residuals: ε = A - (α + β×B) /// 3. Run ADF test on residuals: /// - Δε_t = γ × ε_{t-1} + u_t /// - ADF statistic = γ / SE(γ) /// /// Interpretation: /// - More negative ADF values indicate stronger evidence of cointegration /// - Critical values (approx): -3.43 (1%), -2.86 (5%), -2.57 (10%) /// - Values more negative than critical values reject null hypothesis of no cointegration /// /// Uses Kahan compensated summation for numerical stability over long streams. /// [SkipLocalsInit] public sealed class Cointegration : AbstractBase { private readonly RingBuffer _bufferA; private readonly RingBuffer _bufferB; // Running sums for O(1) statistics private double _sumA, _sumB; private double _sumA2, _sumB2; private double _sumAB; // Kahan compensation for main sums private double _sumAComp, _sumBComp; private double _sumA2Comp, _sumB2Comp; private double _sumABComp; // Residual tracking private double _prevResidual; private double _p_prevResidual; private bool _hasPrevResidual; private bool _p_hasPrevResidual; // ADF regression running sums (period-1 window) private readonly RingBuffer _deltaResiduals; private readonly RingBuffer _laggedResiduals; private double _sumDeltaLagged, _sumLagged2, _sumDelta2; // Kahan compensation for ADF sums private double _sumDeltaLaggedComp, _sumLagged2Comp, _sumDelta2Comp; // Previous compensation state for rollback private double _p_sumAComp, _p_sumBComp; private double _p_sumA2Comp, _p_sumB2Comp; private double _p_sumABComp; private double _p_sumDeltaLaggedComp, _p_sumLagged2Comp, _p_sumDelta2Comp; // Last valid values for NaN handling private double _lastValidA, _lastValidB; private double _p_lastValidA, _p_lastValidB; private const double Epsilon = 1e-10; /// public override bool IsHot => _bufferA.IsFull && _hasPrevResidual; /// /// Creates a new Cointegration indicator. /// /// Lookback period for regression and ADF test (must be > 1) public Cointegration(int period = 20) { if (period <= 1) { throw new ArgumentException("Period must be greater than 1", nameof(period)); } _bufferA = new RingBuffer(period); _bufferB = new RingBuffer(period); _deltaResiduals = new RingBuffer(period - 1); _laggedResiduals = new RingBuffer(period - 1); Name = $"Cointegration({period})"; WarmupPeriod = period + 1; // Need extra bar for first delta } /// /// Updates the Cointegration indicator with new values from both series. /// /// First series value (dependent variable) /// Second series value (independent variable) /// Whether this is a new bar /// The ADF test statistic (more negative = stronger cointegration) [MethodImpl(MethodImplOptions.AggressiveInlining)] public TValue Update(TValue seriesA, TValue seriesB, bool isNew = true) { double a = SanitizeA(seriesA.Value); double b = SanitizeB(seriesB.Value); if (isNew) { ProcessNewBar(a, b); } else { ProcessBarCorrection(a, b); } double adfStat = CalculateAdfStatistic(); Last = new TValue(seriesA.Time, adfStat); PubEvent(Last); return Last; } /// /// Updates with raw double values. /// /// /// Stamps both inputs with DateTime.UtcNow as their timestamp. For /// deterministic or replay-safe sequences use /// with explicit timestamps instead. /// [MethodImpl(MethodImplOptions.AggressiveInlining)] public TValue Update(double seriesA, double seriesB, bool isNew = true) { DateTime now = DateTime.UtcNow; return Update(new TValue(now, seriesA), new TValue(now, seriesB), isNew); } /// Not supported. This indicator requires two inputs; use instead. /// Not supported for bi-input indicator. Use Update(seriesA, seriesB) instead. public override TValue Update(TValue input, bool isNew = true) { throw new NotSupportedException("Cointegration requires two inputs (seriesA and seriesB). Use Update(seriesA, seriesB)."); } /// Not supported. This indicator requires two inputs; use instead. /// Not supported for bi-input indicator. Use Calculate(seriesA, seriesB, period) instead. public override TSeries Update(TSeries source) { throw new NotSupportedException("Cointegration requires two inputs. Use Batch(seriesA, seriesB, period)."); } [MethodImpl(MethodImplOptions.AggressiveInlining)] private double SanitizeA(double value) { if (double.IsFinite(value)) { _lastValidA = value; return value; } return double.IsFinite(_lastValidA) ? _lastValidA : 0.0; } [MethodImpl(MethodImplOptions.AggressiveInlining)] private double SanitizeB(double value) { if (double.IsFinite(value)) { _lastValidB = value; return value; } return double.IsFinite(_lastValidB) ? _lastValidB : 0.0; } [MethodImpl(MethodImplOptions.AggressiveInlining)] private void ProcessNewBar(double a, double b) { // Save state for bar correction _p_lastValidA = _lastValidA; _p_lastValidB = _lastValidB; _p_prevResidual = _prevResidual; _p_hasPrevResidual = _hasPrevResidual; _p_sumAComp = _sumAComp; _p_sumBComp = _sumBComp; _p_sumA2Comp = _sumA2Comp; _p_sumB2Comp = _sumB2Comp; _p_sumABComp = _sumABComp; _p_sumDeltaLaggedComp = _sumDeltaLaggedComp; _p_sumLagged2Comp = _sumLagged2Comp; _p_sumDelta2Comp = _sumDelta2Comp; // Update main buffers if (_bufferA.IsFull) { double oldA = _bufferA.Oldest; double oldB = _bufferB.Oldest; // Kahan subtract oldA from _sumA { double y = -oldA - _sumAComp; double t = _sumA + y; _sumAComp = (t - _sumA) - y; _sumA = t; } // Kahan subtract oldB from _sumB { double y = -oldB - _sumBComp; double t = _sumB + y; _sumBComp = (t - _sumB) - y; _sumB = t; } // Kahan subtract oldA² from _sumA2 { double y = -(oldA * oldA) - _sumA2Comp; double t = _sumA2 + y; _sumA2Comp = (t - _sumA2) - y; _sumA2 = t; } // Kahan subtract oldB² from _sumB2 { double y = -(oldB * oldB) - _sumB2Comp; double t = _sumB2 + y; _sumB2Comp = (t - _sumB2) - y; _sumB2 = t; } // Kahan subtract oldA*oldB from _sumAB { double y = -(oldA * oldB) - _sumABComp; double t = _sumAB + y; _sumABComp = (t - _sumAB) - y; _sumAB = t; } } _bufferA.Add(a); _bufferB.Add(b); // Kahan add a to _sumA { double y = a - _sumAComp; double t = _sumA + y; _sumAComp = (t - _sumA) - y; _sumA = t; } // Kahan add b to _sumB { double y = b - _sumBComp; double t = _sumB + y; _sumBComp = (t - _sumB) - y; _sumB = t; } // Kahan add a² to _sumA2 { double y = (a * a) - _sumA2Comp; double t = _sumA2 + y; _sumA2Comp = (t - _sumA2) - y; _sumA2 = t; } // Kahan add b² to _sumB2 { double y = (b * b) - _sumB2Comp; double t = _sumB2 + y; _sumB2Comp = (t - _sumB2) - y; _sumB2 = t; } // Kahan add a*b to _sumAB { double y = (a * b) - _sumABComp; double t = _sumAB + y; _sumABComp = (t - _sumAB) - y; _sumAB = t; } // Calculate current residual double residual = CalculateResidual(a, b); // Update ADF regression buffers if (_hasPrevResidual) { double delta = residual - _prevResidual; double lagged = _prevResidual; if (_deltaResiduals.IsFull) { double oldDelta = _deltaResiduals.Oldest; double oldLagged = _laggedResiduals.Oldest; // Kahan subtract from ADF sums { double y = -(oldDelta * oldLagged) - _sumDeltaLaggedComp; double t = _sumDeltaLagged + y; _sumDeltaLaggedComp = (t - _sumDeltaLagged) - y; _sumDeltaLagged = t; } { double y = -(oldLagged * oldLagged) - _sumLagged2Comp; double t = _sumLagged2 + y; _sumLagged2Comp = (t - _sumLagged2) - y; _sumLagged2 = t; } { double y = -(oldDelta * oldDelta) - _sumDelta2Comp; double t = _sumDelta2 + y; _sumDelta2Comp = (t - _sumDelta2) - y; _sumDelta2 = t; } } _deltaResiduals.Add(delta); _laggedResiduals.Add(lagged); // Kahan add to ADF sums { double y = (delta * lagged) - _sumDeltaLaggedComp; double t = _sumDeltaLagged + y; _sumDeltaLaggedComp = (t - _sumDeltaLagged) - y; _sumDeltaLagged = t; } { double y = (lagged * lagged) - _sumLagged2Comp; double t = _sumLagged2 + y; _sumLagged2Comp = (t - _sumLagged2) - y; _sumLagged2 = t; } { double y = (delta * delta) - _sumDelta2Comp; double t = _sumDelta2 + y; _sumDelta2Comp = (t - _sumDelta2) - y; _sumDelta2 = t; } } _prevResidual = residual; _hasPrevResidual = true; } [MethodImpl(MethodImplOptions.AggressiveInlining)] private void ProcessBarCorrection(double a, double b) { // Restore state _lastValidA = _p_lastValidA; _lastValidB = _p_lastValidB; _prevResidual = _p_prevResidual; _hasPrevResidual = _p_hasPrevResidual; _sumAComp = _p_sumAComp; _sumBComp = _p_sumBComp; _sumA2Comp = _p_sumA2Comp; _sumB2Comp = _p_sumB2Comp; _sumABComp = _p_sumABComp; _sumDeltaLaggedComp = _p_sumDeltaLaggedComp; _sumLagged2Comp = _p_sumLagged2Comp; _sumDelta2Comp = _p_sumDelta2Comp; // Update newest values in main buffers if (_bufferA.Count == 0) { // Nothing to correct yet; no current bar exists return; } double oldA = _bufferA.Newest; double oldB = _bufferB.Newest; // Kahan subtract old + add new for main sums { double y = (-oldA + a) - _sumAComp; double t = _sumA + y; _sumAComp = (t - _sumA) - y; _sumA = t; } { double y = (-oldB + b) - _sumBComp; double t = _sumB + y; _sumBComp = (t - _sumB) - y; _sumB = t; } { double y = (-(oldA * oldA) + (a * a)) - _sumA2Comp; double t = _sumA2 + y; _sumA2Comp = (t - _sumA2) - y; _sumA2 = t; } { double y = (-(oldB * oldB) + (b * b)) - _sumB2Comp; double t = _sumB2 + y; _sumB2Comp = (t - _sumB2) - y; _sumB2 = t; } { double y = (-(oldA * oldB) + (a * b)) - _sumABComp; double t = _sumAB + y; _sumABComp = (t - _sumAB) - y; _sumAB = t; } _bufferA.UpdateNewest(a); _bufferB.UpdateNewest(b); // Calculate current residual double residual = CalculateResidual(a, b); // Update ADF regression buffers if (_hasPrevResidual) { double delta = residual - _prevResidual; double lagged = _prevResidual; if (_deltaResiduals.Count == 0) { // Nothing to correct yet in ADF buffers; no current entry exists return; } double oldDelta = _deltaResiduals.Newest; double oldLagged = _laggedResiduals.Newest; // Kahan subtract old + add new for ADF sums { double y = (-(oldDelta * oldLagged) + (delta * lagged)) - _sumDeltaLaggedComp; double t = _sumDeltaLagged + y; _sumDeltaLaggedComp = (t - _sumDeltaLagged) - y; _sumDeltaLagged = t; } { double y = (-(oldLagged * oldLagged) + (lagged * lagged)) - _sumLagged2Comp; double t = _sumLagged2 + y; _sumLagged2Comp = (t - _sumLagged2) - y; _sumLagged2 = t; } { double y = (-(oldDelta * oldDelta) + (delta * delta)) - _sumDelta2Comp; double t = _sumDelta2 + y; _sumDelta2Comp = (t - _sumDelta2) - y; _sumDelta2 = t; } _deltaResiduals.UpdateNewest(delta); _laggedResiduals.UpdateNewest(lagged); } _prevResidual = residual; _hasPrevResidual = true; } [MethodImpl(MethodImplOptions.AggressiveInlining)] private double CalculateResidual(double a, double b) { int n = _bufferA.Count; if (n < 2) { return 0.0; } // Calculate means double meanA = _sumA / n; double meanB = _sumB / n; // Calculate variance of B and covariance double varB = Max(0.0, (_sumB2 / n) - (meanB * meanB)); double cov = (_sumAB / n) - (meanA * meanB); // Calculate beta and alpha double beta = 0.0; if (varB > Epsilon) { beta = cov / varB; } double alpha = meanA - (beta * meanB); // Calculate residual return a - (alpha + (beta * b)); } [MethodImpl(MethodImplOptions.AggressiveInlining)] private double CalculateAdfStatistic() { int n = _deltaResiduals.Count; if (n < 2) { return double.NaN; } if (_sumLagged2 < Epsilon) { return double.NaN; } // No-intercept ADF regression: Δε_t = γ × ε_{t-1} + u_t double gamma = _sumDeltaLagged / _sumLagged2; // Calculate sum of squared regression errors in O(1) // Sum((Δε_t - γ ε_{t-1})^2) = Sum(Δε_t^2) - 2γ Sum(Δε_t ε_{t-1}) + γ^2 Sum(ε_{t-1}^2) double sumErrorSq = _sumDelta2 - (2.0 * gamma * _sumDeltaLagged) + (gamma * gamma * _sumLagged2); // Ensure non-negative due to floating point errors sumErrorSq = Max(0.0, sumErrorSq); double varError = sumErrorSq / (n - 1); double seGammaSq = varError / _sumLagged2; if (seGammaSq <= 0 || !double.IsFinite(seGammaSq)) { return double.NaN; } double seGamma = Sqrt(seGammaSq); if (seGamma < Epsilon) { return double.NaN; } return gamma / seGamma; } /// Not supported. This indicator requires two input spans. public override void Prime(ReadOnlySpan source, TimeSpan? step = null) { throw new NotSupportedException("Cointegration requires two inputs."); } /// public override void Reset() { _bufferA.Clear(); _bufferB.Clear(); _deltaResiduals.Clear(); _laggedResiduals.Clear(); _sumA = 0; _sumB = 0; _sumA2 = 0; _sumB2 = 0; _sumAB = 0; _sumAComp = 0; _sumBComp = 0; _sumA2Comp = 0; _sumB2Comp = 0; _sumABComp = 0; _sumDeltaLagged = 0; _sumLagged2 = 0; _sumDelta2 = 0; _sumDeltaLaggedComp = 0; _sumLagged2Comp = 0; _sumDelta2Comp = 0; _prevResidual = 0; _p_prevResidual = 0; _hasPrevResidual = false; _p_hasPrevResidual = false; _lastValidA = 0; _lastValidB = 0; _p_lastValidA = 0; _p_lastValidB = 0; Last = default; } /// /// Calculates cointegration for two time series. /// public static TSeries Batch(TSeries seriesA, TSeries seriesB, int period = 20) => Calculate(seriesA, seriesB, period).Results; /// /// Static batch calculation for span-based processing. /// public static void Batch( ReadOnlySpan seriesA, ReadOnlySpan seriesB, Span output, int period = 20) { if (seriesA.Length != seriesB.Length) { throw new ArgumentException("Series must have the same length", nameof(seriesB)); } if (seriesA.Length != output.Length) { throw new ArgumentException("Output must have the same length as input", nameof(output)); } if (period <= 1) { throw new ArgumentException("Period must be greater than 1", nameof(period)); } var indicator = new Cointegration(period); for (int i = 0; i < seriesA.Length; i++) { var result = indicator.Update(seriesA[i], seriesB[i], isNew: true); output[i] = result.Value; } } /// /// Calculates the ADF cointegration statistic for two time series and returns both the result series and the live indicator instance. /// public static (TSeries Results, Cointegration Indicator) Calculate(TSeries seriesA, TSeries seriesB, int period = 20) { if (seriesA.Count != seriesB.Count) { throw new ArgumentException("Series must have the same length", nameof(seriesB)); } var indicator = new Cointegration(period); var result = new TSeries(seriesA.Count); var timesA = seriesA.Times; var valuesA = seriesA.Values; var valuesB = seriesB.Values; for (int i = 0; i < seriesA.Count; i++) { result.Add(indicator.Update(new TValue(timesA[i], valuesA[i]), new TValue(timesA[i], valuesB[i]), isNew: true)); } return (result, indicator); } }