160 lines
4.5 KiB
Python
160 lines
4.5 KiB
Python
"""Fractal indicators: Hurst exponent, fractal dimension, self-similarity."""
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
from scipy.stats import linregress
|
|
|
|
|
|
def hurst_exponent(series: np.ndarray, max_lag: int | None = None) -> float:
|
|
"""Compute Hurst exponent via R/S analysis.
|
|
H > 0.5: trending (persistent), H < 0.5: mean-reverting, H ≈ 0.5: random walk.
|
|
"""
|
|
n = len(series)
|
|
if n < 20:
|
|
return 0.5
|
|
|
|
if max_lag is None:
|
|
max_lag = min(n // 4, 100)
|
|
|
|
lags = range(10, max_lag + 1)
|
|
rs_values = []
|
|
lag_values = []
|
|
|
|
for lag in lags:
|
|
rs_list = []
|
|
for start in range(0, n - lag, lag):
|
|
chunk = series[start : start + lag]
|
|
if len(chunk) < lag:
|
|
continue
|
|
mean = np.mean(chunk)
|
|
deviations = np.cumsum(chunk - mean)
|
|
r = np.max(deviations) - np.min(deviations)
|
|
s = np.std(chunk, ddof=1)
|
|
if s > 0:
|
|
rs_list.append(r / s)
|
|
|
|
if rs_list:
|
|
rs_values.append(np.mean(rs_list))
|
|
lag_values.append(lag)
|
|
|
|
if len(lag_values) < 3:
|
|
return 0.5
|
|
|
|
log_lags = np.log(lag_values)
|
|
log_rs = np.log(rs_values)
|
|
slope, _, _, _, _ = linregress(log_lags, log_rs)
|
|
return float(np.clip(slope, 0, 1))
|
|
|
|
|
|
def rolling_hurst(close: np.ndarray, window: int = 100, step: int = 1) -> np.ndarray:
|
|
"""Compute rolling Hurst exponent."""
|
|
n = len(close)
|
|
result = np.full(n, 0.5)
|
|
returns = np.diff(np.log(np.where(close == 0, 1e-10, close)))
|
|
|
|
for i in range(window, n, step):
|
|
h = hurst_exponent(returns[i - window : i])
|
|
result[i] = h
|
|
for j in range(1, min(step, n - i)):
|
|
result[i + j] = h
|
|
|
|
return result
|
|
|
|
|
|
def fractal_dimension_higuchi(series: np.ndarray, k_max: int = 10) -> float:
|
|
"""Higuchi fractal dimension of a time series."""
|
|
n = len(series)
|
|
if n < k_max * 2:
|
|
return 1.5
|
|
|
|
lk = []
|
|
x = np.arange(1, k_max + 1)
|
|
|
|
for k in range(1, k_max + 1):
|
|
lm_list = []
|
|
for m in range(1, k + 1):
|
|
indices = np.arange(m - 1, n, k)
|
|
if len(indices) < 2:
|
|
continue
|
|
vals = series[indices]
|
|
length = np.sum(np.abs(np.diff(vals)))
|
|
norm = (n - 1) / (k * ((n - m) // k) * k)
|
|
lm_list.append(length * norm)
|
|
|
|
if lm_list:
|
|
lk.append(np.mean(lm_list))
|
|
|
|
if len(lk) < 3:
|
|
return 1.5
|
|
|
|
log_k = np.log(1.0 / x[: len(lk)])
|
|
log_lk = np.log(np.array(lk))
|
|
slope, _, _, _, _ = linregress(log_k, log_lk)
|
|
return float(np.clip(slope, 1.0, 2.0))
|
|
|
|
|
|
def self_similarity_score(close: np.ndarray, window: int, scales: list[int] | None = None) -> float:
|
|
"""Measure self-similarity across multiple time scales.
|
|
Higher score = more fractal (self-similar) structure.
|
|
"""
|
|
if scales is None:
|
|
scales = [2, 3, 4, 6]
|
|
|
|
if len(close) < window:
|
|
return 0.0
|
|
|
|
base = close[-window:]
|
|
base_returns = np.diff(np.log(np.where(base == 0, 1e-10, base)))
|
|
if np.std(base_returns) == 0:
|
|
return 0.0
|
|
|
|
similarities = []
|
|
for scale in scales:
|
|
scaled_window = window * scale
|
|
if scaled_window > len(close):
|
|
continue
|
|
|
|
scaled = close[-scaled_window:]
|
|
step = scale
|
|
downsampled = scaled[::step][:window]
|
|
|
|
if len(downsampled) != len(base):
|
|
downsampled = np.interp(
|
|
np.linspace(0, 1, window),
|
|
np.linspace(0, 1, len(downsampled)),
|
|
downsampled,
|
|
)
|
|
|
|
ds_returns = np.diff(np.log(np.where(downsampled == 0, 1e-10, downsampled)))
|
|
|
|
if len(ds_returns) != len(base_returns):
|
|
ds_returns = np.interp(
|
|
np.linspace(0, 1, len(base_returns)),
|
|
np.linspace(0, 1, len(ds_returns)),
|
|
ds_returns,
|
|
)
|
|
|
|
std_ds = np.std(ds_returns)
|
|
if std_ds == 0:
|
|
continue
|
|
|
|
corr = np.corrcoef(base_returns, ds_returns)[0, 1]
|
|
if np.isfinite(corr):
|
|
similarities.append(abs(corr))
|
|
|
|
if not similarities:
|
|
return 0.0
|
|
return float(np.mean(similarities))
|
|
|
|
|
|
def volatility_ratio(close: np.ndarray, fast: int = 12, slow: int = 48) -> float:
|
|
"""Ratio of short-term to long-term volatility."""
|
|
returns = np.diff(np.log(np.where(close == 0, 1e-10, close)))
|
|
if len(returns) < slow:
|
|
return 1.0
|
|
fast_vol = np.std(returns[-fast:])
|
|
slow_vol = np.std(returns[-slow:])
|
|
if slow_vol == 0:
|
|
return 1.0
|
|
return float(fast_vol / slow_vol)
|