Denoising HFT Noise:
Causal Rolling Fourier Filters

In algorithmic finance and high-frequency trading (HFT), tick-by-tick orderbook data is plagued by rapid, artificial price fluctuations. This is known as microstructure noise, and it is primarily driven by bid-ask bouncing, order cancellations, and execution fragmentation.

Traditional linear filters, such as the Simple Moving Average (SMA) or Exponential Moving Average (EMA), attempt to smooth this noise by averaging historical prices. However, these time-domain convolutions introduce a substantial phase lag $\tau$ directly proportional to their window size:

This delay degrades the performance of execution algorithms. By the time a lagged moving average detects a structural market breakout, the optimal entry price has frequently disappeared, leading to severe slippage.

Rather than averaging prices in the time domain, we translate our price series into the frequency domain. Because microstructure bid-ask bounce occurs at rapid speeds (high frequencies) and macroscopic trends drift slowly (low frequencies), we can isolate and mute noise using a spectral mask.

The uniform price series $x_n$ of length $N$ is first decomposed into frequency coefficients $X_k$ using the Discrete Fourier Transform (DFT):

We then apply a low-pass binary frequency mask $H(f_k)$ to the coefficients, muting any signal component whose frequency exceeds our cutoff frequency $f_c$:

Finally, we map the filtered spectrum back into the time domain via the Inverse DFT (IDFT) to reconstruct our clean, denoised signal $\tilde{x}_n$:

A major challenge of applying the FFT to rolling financial price data is spectral leakage. The FFT assumes that the input window is periodic, meaning the start of the window ($x_0$) connects smoothly to the end ($x_{W-1}$). When prices drift, this boundary mismatch creates a sharp step discontinuity, which the FFT interprets as high-frequency energy. This causes severe edge distortion at the boundaries—distorting our most critical value: the current price.

The Mirrored Windowing technique achieves zero attenuation at the current price boundary. We copy the window $[x_0, x_1, \dots, x_{W-1}]$ and append its elements in reverse order:

This mirrored sequence has a length of $M = 2W-2$. When we apply a standard Hann window $w$ of length $M$, the current price $x_{W-1}$ lies at the exact center index $W-1$, where:

This eliminates the discontinuity at the outer edges (which taper smoothly to 0) while leaving the active price boundary completely unattenuated.

In many retail trading indicators, the FFT is computed over a global array (e.g., an entire dataset). While this yields a perfectly smooth, lag-free curve on historical charts, it introduces massive look-ahead bias (non-causal filtering). A filter calculating coefficients at 10:00 AM using data from 3:00 PM is useless for live trading.

For live trading production, the signal engine must operate strictly in a rolling window buffer. Each incoming tick is pushed to a buffer of size $W$, the mirroring is applied, and the forward/inverse FFT operates strictly on this historic slice. The indicator value plotted at the current millisecond is the final index of the reconstructed window, preventing future data leakage.

Below is the core implementation of the causal rolling Fourier filter from the GitHub repository, featuring Radix-2 low-pass filtering and the mirrored boundary mitigation algorithm:

import numpy as np

class FourierFilter:
    """
    A causal rolling Fast Fourier Transform (FFT) noise-reduction filter.
    """
    def __init__(self, window_size: int, cutoff_hz: float, fs: float, window_type: str = 'mirror'):
        self.window_size = window_size
        self.cutoff_hz = cutoff_hz
        self.fs = fs
        self.window_type = window_type.lower()

    def _get_lowpass_mask(self, n: int) -> np.ndarray:
        freqs = np.fft.fftfreq(n, d=1.0/self.fs)
        return np.abs(freqs) <= self.cutoff_hz

    def _filter_single_window(self, window_data: np.ndarray) -> float:
        W = len(window_data)

        if self.window_type == 'rectangular':
            fft_vals = np.fft.fft(window_data)
            mask = self._get_lowpass_mask(W)
            fft_filtered = fft_vals * mask
            return np.fft.ifft(fft_filtered).real[-1]

        elif self.window_type == 'taper':
            w = np.hamming(W)
            mean_val = np.mean(window_data)
            centered_data = window_data - mean_val
            windowed_data = centered_data * w
            
            fft_vals = np.fft.fft(windowed_data)
            mask = self._get_lowpass_mask(W)
            reconstructed = np.fft.ifft(fft_vals * mask).real
            
            scale = w[-1] if w[-1] > 1e-5 else 1e-5
            return (reconstructed[-1] / scale) + mean_val

        elif self.window_type == 'mirror':
            if W <= 2:
                return window_data[-1]

            # Symmetrically mirror the rolling window around the last element
            mirrored_part = window_data[-2:0:-1]
            z = np.concatenate([window_data, mirrored_part])
            M = len(z)
            center_idx = W - 1

            w = np.hanning(M)
            mean_val = np.mean(z)
            centered_z = z - mean_val
            windowed_z = centered_z * w

            fft_vals = np.fft.fft(windowed_z)
            mask = self._get_lowpass_mask(M)
            reconstructed = np.fft.ifft(fft_vals * mask).real

            return (reconstructed[center_idx] / w[center_idx]) + mean_val