Spectral Analysis of Markets: Fourier Transform Trading
Markets exhibit cyclical behavior—earnings cycles, economic cycles, seasonal patterns, and technical cycles all influence price movements. Spectral analysis, powered by the Fourier transform, provides a mathematical framework for identifying these periodicities by decomposing price series into constituent frequency components. By moving from the time domain to the frequency domain, traders gain insights into dominant cycles, hidden periodicities, and optimal trading horizons.Understanding Spectral Analysis
Spectral analysis reveals which frequencies (cycles) are present in a time series and how much power (variance) each frequency contributes.The Fourier Transform
The Discrete Fourier Transform (DFT) converts a time series x[n] into frequency domain X[k]: ``
X[k] = Σ(n=0 to N-1) x[n] e^(-i2πkn/N)
`
Where:
- k: frequency index (0 to N-1)
- N: number of samples
- i: imaginary unit
The inverse transform reconstructs the original signal:
`
x[n] = (1/N) Σ(k=0 to N-1) X[k] e^(i2πkn/N)
`
Power Spectral Density
The power spectral density (PSD) shows how power is distributed across frequencies:
`
P[k] = |X[k]|² / N
`
Peaks in the PSD indicate dominant cycles in the data.
Periodogram
The periodogram is an estimate of the PSD:
`
I(f) = (1/N) |Σ x[n] e^(-i2πfn)|²
`
Key Takeaways
- Spectral analysis identifies dominant market cycles and periodicities
- Transforms time-domain data to frequency domain for cycle analysis
- Power spectral density reveals which cycles contribute most to returns
- Enables filtering, cycle extraction, and harmonic trading
- Fast Fourier Transform (FFT) makes computation efficient even for large datasets
- Works best for stationary or weakly stationary data
Why Spectral Analysis Matters for Trading
1. Cycle Identification
Discover hidden periodicities that drive market movements:
`python
import numpy as np
from scipy import signal
import matplotlib.pyplot as plt
def identify_market_cycles(prices, sampling_freq=252):
"""
Identify dominant cycles using spectral analysis
sampling_freq: trading days per year (252 for daily data)
"""
# Calculate returns
returns = np.log(prices / prices.shift(1)).dropna()
# Compute power spectral density
frequencies, psd = signal.periodogram(returns, fs=sampling_freq)
# Find peaks (dominant cycles)
peaks, properties = signal.find_peaks(psd, height=np.percentile(psd, 90))
# Convert frequencies to periods (in days)
dominant_periods = sampling_freq / frequencies[peaks]
# Create results DataFrame
cycles = pd.DataFrame({
'frequency': frequencies[peaks],
'period_days': dominant_periods,
'power': psd[peaks]
}).sort_values('power', ascending=False)
return cycles, frequencies, psd
Example usage
cycles, freqs, psd = identify_market_cycles(spy_prices)
print("Top 5 dominant cycles:")
print(cycles.head())
Might show: 252-day (annual), 126-day (semi-annual), 21-day (monthly), etc.
`
2. Seasonality Detection
Quantify seasonal effects using spectral peaks:
`python
def detect_seasonality(prices, expected_periods=[21, 63, 126, 252]):
"""
Test for seasonality at expected frequencies
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Spectral density
frequencies, psd = signal.periodogram(returns, fs=252)
# Check power at expected seasonal frequencies
results = {}
for period in expected_periods:
target_freq = 252 / period
# Find closest frequency in spectrum
idx = np.argmin(np.abs(frequencies - target_freq))
# Significance test: is this peak significantly above baseline?
baseline = np.median(psd)
peak_power = psd[idx]
significance = peak_power / baseline
results[f'{period}d'] = {
'frequency': frequencies[idx],
'period': period,
'power': peak_power,
'significance': significance,
'strong_signal': significance > 3.0
}
return pd.DataFrame(results).T
`
3. Optimal Trading Horizon
Determine which time scales contain the most information:
`python
def optimal_trading_horizons(prices, min_period=5, max_period=100):
"""
Find time horizons with maximum spectral power
"""
returns = np.log(prices / prices.shift(1)).dropna()
frequencies, psd = signal.periodogram(returns, fs=252)
# Convert to periods
periods = 252 / frequencies
# Filter to range of interest
mask = (periods >= min_period) & (periods <= max_period)
periods_filt = periods[mask]
psd_filt = psd[mask]
# Find top horizons
top_indices = np.argsort(psd_filt)[-10:]
optimal_horizons = pd.DataFrame({
'period_days': periods_filt[top_indices],
'power': psd_filt[top_indices],
'relative_power': psd_filt[top_indices] / np.sum(psd_filt)
}).sort_values('power', ascending=False)
return optimal_horizons
`
Practical Trading Applications
Band-Pass Filtering for Cycle Trading
Extract specific cycles and trade them:
`python
def extract_cycle_component(prices, target_period, bandwidth=0.2):
"""
Extract specific cycle using band-pass filter
target_period: desired cycle period in days
bandwidth: width of pass band as fraction of target
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Design band-pass filter
sampling_freq = 252
center_freq = sampling_freq / target_period
low_freq = center_freq * (1 - bandwidth)
high_freq = center_freq * (1 + bandwidth)
# Butterworth band-pass filter
sos = signal.butter(4, [low_freq, high_freq],
btype='band', fs=sampling_freq, output='sos')
# Apply filter
filtered = signal.sosfilt(sos, returns)
return pd.Series(filtered, index=returns.index)
Example: extract 21-day (monthly) cycle
monthly_cycle = extract_cycle_component(spy_prices, target_period=21)
Trading signal: phase of cycle
signals = np.sign(monthly_cycle) # Long when cycle is positive
`
Spectral Momentum Strategy
Trade based on shifting spectral power:
`python
def spectral_momentum(prices, window=252, top_n=3):
"""
Identify momentum using changing spectral characteristics
"""
returns = np.log(prices / prices.shift(1)).dropna()
momentum_scores = []
for i in range(window, len(returns)):
# Current window
current = returns.iloc[i-window:i]
# Previous window
previous = returns.iloc[i-2*window:i-window]
# Spectral densities
f_curr, psd_curr = signal.periodogram(current, fs=252)
f_prev, psd_prev = signal.periodogram(previous, fs=252)
# Find top frequencies in current period
top_indices = np.argsort(psd_curr)[-top_n:]
# Compare power in these frequencies to previous period
power_change = np.mean(psd_curr[top_indices] - psd_prev[top_indices])
momentum_scores.append(power_change)
return pd.Series(momentum_scores, index=returns.index[window:])
`
Cross-Spectral Analysis for Pairs
Identify pairs with coherent cycles:
`python
def spectral_coherence(prices1, prices2):
"""
Measure frequency-domain coherence between two assets
"""
returns1 = np.log(prices1 / prices1.shift(1)).dropna()
returns2 = np.log(prices2 / prices2.shift(1)).dropna()
# Align series
common_idx = returns1.index.intersection(returns2.index)
r1 = returns1.loc[common_idx]
r2 = returns2.loc[common_idx]
# Coherence: frequency-domain correlation
frequencies, coherence = signal.coherence(r1, r2, fs=252)
# Find frequencies with high coherence
high_coherence = coherence > 0.7
coherent_periods = 252 / frequencies[high_coherence]
results = pd.DataFrame({
'frequency': frequencies[high_coherence],
'period_days': coherent_periods,
'coherence': coherence[high_coherence]
}).sort_values('coherence', ascending=False)
return results, frequencies, coherence
`
Adaptive Cycle Period Detection
Dynamically track dominant cycle period:
`python
def adaptive_cycle_period(prices, window=126, freq_range=(5, 100)):
"""
Track time-varying dominant cycle period
"""
returns = np.log(prices / prices.shift(1)).dropna()
dominant_periods = []
for i in range(window, len(returns)):
window_data = returns.iloc[i-window:i]
# Spectral analysis
frequencies, psd = signal.periodogram(window_data, fs=252)
periods = 252 / frequencies
# Filter to range
mask = (periods >= freq_range[0]) & (periods <= freq_range[1])
if mask.sum() > 0:
# Find dominant period in range
dominant_idx = np.argmax(psd[mask])
dominant_period = periods[mask][dominant_idx]
else:
dominant_period = np.nan
dominant_periods.append(dominant_period)
return pd.Series(dominant_periods, index=returns.index[window:])
Use adaptive period for dynamic moving average
adaptive_periods = adaptive_cycle_period(spy_prices)
Adjust MA window based on current dominant cycle
`
Advanced Techniques
Welch's Method for Smoother Spectra
Reduce variance in spectral estimates:
`python
def welch_spectral_estimate(prices, nperseg=256):
"""
Welch's method: average periodograms of overlapping segments
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Welch's method
frequencies, psd = signal.welch(returns, fs=252, nperseg=nperseg,
scaling='density', detrend='constant')
return frequencies, psd
Smoother than standard periodogram, better for noisy data
`
Multitaper Spectral Estimation
Reduce spectral leakage:
`python
from spectrum import pmtm
def multitaper_spectrum(prices, NW=4, k=7):
"""
Multitaper method for reduced leakage
NW: time-bandwidth product
k: number of tapers (typically 2*NW - 1)
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Multitaper spectral estimate
psd, freq = pmtm(returns, NW=NW, k=k, show=False)
# Convert to periods
periods = 252 / np.array(freq)
return periods, psd
`
Short-Time Fourier Transform
Analyze time-varying spectra:
`python
def time_frequency_analysis(prices, window_size=63, overlap=0.5):
"""
STFT: spectral analysis in sliding windows
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Short-time Fourier transform
noverlap = int(window_size * overlap)
frequencies, times, Zxx = signal.stft(
returns,
fs=252,
nperseg=window_size,
noverlap=noverlap
)
# Power spectrogram
power = np.abs(Zxx) ** 2
# Plot
plt.figure(figsize=(15, 8))
plt.pcolormesh(times, frequencies, power, shading='gouraud', cmap='jet')
plt.ylabel('Frequency [Hz]')
plt.xlabel('Time [days]')
plt.title('Time-Frequency Spectrogram')
plt.colorbar(label='Power')
return frequencies, times, power
`
Singular Spectrum Analysis
Decompose series into trend, cycles, and noise:
`python
def singular_spectrum_analysis(prices, window=50, n_components=10):
"""
SSA: data-adaptive spectral decomposition
"""
returns = np.log(prices / prices.shift(1)).dropna().values
N = len(returns)
# Embedding
K = N - window + 1
trajectory = np.zeros((K, window))
for i in range(K):
trajectory[i, :] = returns[i:i+window]
# SVD
U, s, Vt = np.linalg.svd(trajectory, full_matrices=False)
# Reconstruct components
components = []
for i in range(n_components):
comp = s[i] * np.outer(U[:, i], Vt[i, :])
# Diagonal averaging to get time series
reconstructed = np.zeros(N)
for j in range(N):
indices = [(k, j-k) for k in range(window)
if 0 <= j-k < window and k < K]
reconstructed[j] = np.mean([comp[k, m] for k, m in indices])
components.append(reconstructed)
return components, s
`
Harmonic Regression
Model prices as sum of sinusoids:
`python
def harmonic_regression(prices, periods=[21, 63, 252]):
"""
Fit harmonic components at specified periods
"""
returns = np.log(prices / prices.shift(1)).dropna()
t = np.arange(len(returns))
# Design matrix: sin and cos for each period
X = np.ones((len(returns), 1)) # Constant
for period in periods:
freq = 2 * np.pi / period
X = np.column_stack([
X,
np.sin(freq * t),
np.cos(freq * t)
])
# Least squares fit
beta = np.linalg.lstsq(X, returns.values, rcond=None)[0]
# Fitted values
fitted = X @ beta
# Decompose by component
components = {}
components['constant'] = beta[0] * np.ones(len(returns))
idx = 1
for period in periods:
freq = 2 * np.pi / period
sin_coef = beta[idx]
cos_coef = beta[idx + 1]
amplitude = np.sqrt(sin_coef2 + cos_coef2)
phase = np.arctan2(cos_coef, sin_coef)
components[f'{period}d'] = amplitude np.sin(freq t + phase)
idx += 2
return components, beta, fitted
`
Best Practices
1. Detrending
Remove trends before spectral analysis:
`python
from scipy.signal import detrend
def detrend_before_spectral(prices, method='linear'):
"""
Detrend to avoid spectral leakage
method: 'linear' or 'constant'
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Detrend
if method == 'linear':
detrended = detrend(returns, type='linear')
else:
detrended = returns - returns.mean()
# Now apply spectral analysis
frequencies, psd = signal.periodogram(detrended, fs=252)
return frequencies, psd
`
2. Windowing
Apply window functions to reduce spectral leakage:
`python
def windowed_spectrum(prices, window_type='hann'):
"""
Apply window before FFT
window_type: 'hann', 'hamming', 'blackman', 'bartlett'
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Apply window
window = signal.get_window(window_type, len(returns))
windowed_data = returns * window
# Spectral analysis
frequencies, psd = signal.periodogram(windowed_data, fs=252,
scaling='spectrum')
return frequencies, psd
`
3. Significance Testing
Test if detected cycles are significant:
`python
def test_cycle_significance(prices, period, n_simulations=1000):
"""
Permutation test for cycle significance
"""
returns = np.log(prices / prices.shift(1)).dropna()
# Observed power at target frequency
frequencies, psd = signal.periodogram(returns, fs=252)
target_freq = 252 / period
target_idx = np.argmin(np.abs(frequencies - target_freq))
observed_power = psd[target_idx]
# Null distribution: random permutations
null_powers = []
for _ in range(n_simulations):
shuffled = np.random.permutation(returns)
_, psd_null = signal.periodogram(shuffled, fs=252)
null_powers.append(psd_null[target_idx])
# P-value
p_value = np.mean(null_powers >= observed_power)
return {
'period': period,
'observed_power': observed_power,
'p_value': p_value,
'significant': p_value < 0.05
}
`
Real-World Case Study
Complete Spectral Trading System
`python
class SpectralTradingSystem:
def __init__(self, top_n_cycles=3, min_period=10, max_period=100):
self.top_n_cycles = top_n_cycles
self.min_period = min_period
self.max_period = max_period
self.dominant_periods = None
def identify_cycles(self, prices):
"""Identify dominant cycles"""
returns = np.log(prices / prices.shift(1)).dropna()
# Spectral analysis
frequencies, psd = signal.periodogram(returns, fs=252)
periods = 252 / frequencies
# Filter to range
mask = (periods >= self.min_period) & (periods <= self.max_period)
periods_filt = periods[mask]
psd_filt = psd[mask]
# Top cycles
top_indices = np.argsort(psd_filt)[-self.top_n_cycles:]
self.dominant_periods = periods_filt[top_indices]
return self.dominant_periods
def extract_cycle_signals(self, prices):
"""Extract signals from each dominant cycle"""
if self.dominant_periods is None:
self.identify_cycles(prices)
returns = np.log(prices / prices.shift(1)).dropna()
signals = []
for period in self.dominant_periods:
# Band-pass filter for this cycle
center_freq = 252 / period
bandwidth = 0.3
low_freq = center_freq * (1 - bandwidth)
high_freq = center_freq * (1 + bandwidth)
sos = signal.butter(4, [low_freq, high_freq],
btype='band', fs=252, output='sos')
filtered = signal.sosfilt(sos, returns)
# Signal: phase of cycle (sign)
cycle_signal = np.sign(filtered)
signals.append(cycle_signal)
# Average signals
combined = np.mean(signals, axis=0)
return pd.Series(combined, index=returns.index)
def backtest(self, prices, refit_freq=63):
"""Backtest spectral strategy"""
returns = prices.pct_change()
portfolio_returns = []
for i in range(252, len(prices)):
# Refit periodically
if i % refit_freq == 0:
history = prices.iloc[:i]
self.identify_cycles(history)
# Generate signal
history = prices.iloc[:i]
signals = self.extract_cycle_signals(history)
# Current signal (last value)
current_signal = signals.iloc[-1] if len(signals) > 0 else 0
# Portfolio return
port_return = current_signal * returns.iloc[i]
portfolio_returns.append(port_return)
return pd.Series(portfolio_returns, index=returns.index[252:])
Usage
system = SpectralTradingSystem(top_n_cycles=3)
backtest_returns = system.backtest(spy_prices)
Performance
sharpe = backtest_returns.mean() / backtest_returns.std() * np.sqrt(252)
cumulative = (1 + backtest_returns).cumprod()
print(f"Sharpe Ratio: {sharpe:.2f}")
print(f"Total Return: {cumulative.iloc[-1] - 1:.2%}")
``