Loading...
Loading...
Compare original and translation side by side
from scipy import stats
import numpy as np
def adf_test(series: np.ndarray, max_lag: int = 0) -> dict:
"""Run ADF test. Reject null (p < 0.05) → stationary → mean-reverting."""
# See references/statistical_tests.md for full implementation
# Use statsmodels.tsa.stattools.adfuller for production
passfrom scipy import stats
import numpy as np
def adf_test(series: np.ndarray, max_lag: int = 0) -> dict:
"""Run ADF test. Reject null (p < 0.05) → stationary → mean-reverting."""
# See references/statistical_tests.md for full implementation
# Use statsmodels.tsa.stattools.adfuller for production
pass| Hurst Value | Interpretation | Trading Implication |
|---|---|---|
| H < 0.5 | Mean-reverting | Trade mean reversion |
| H = 0.5 | Random walk | No edge |
| H > 0.5 | Trending | Trade momentum |
def hurst_exponent(series: np.ndarray) -> float:
"""Compute Hurst exponent via R/S method. H < 0.5 → mean-reverting."""
# See references/statistical_tests.md for full R/S algorithm
pass| Hurst值 | 解释 | 交易启示 |
|---|---|---|
| H < 0.5 | 均值回归型 | 采用均值回归策略 |
| H = 0.5 | 随机游走 | 无交易优势 |
| H > 0.5 | 趋势型 | 采用动量策略 |
def hurst_exponent(series: np.ndarray) -> float:
"""Compute Hurst exponent via R/S method. H < 0.5 → mean-reverting."""
# See references/statistical_tests.md for full R/S algorithm
passdef variance_ratio(series: np.ndarray, q: int = 5) -> float:
"""Compute variance ratio at horizon q. VR < 1 → mean-reverting."""
returns = np.diff(np.log(series))
var_1 = np.var(returns)
returns_q = np.diff(np.log(series[::q]))
var_q = np.var(returns_q)
return var_q / (q * var_1)references/statistical_tests.mddef variance_ratio(series: np.ndarray, q: int = 5) -> float:
"""Compute variance ratio at horizon q. VR < 1 → mean-reverting."""
returns = np.diff(np.log(series))
var_1 = np.var(returns)
returns_q = np.diff(np.log(series[::q]))
var_q = np.var(returns_q)
return var_q / (q * var_1)references/statistical_tests.mddelta_X_t = alpha + beta * X_{t-1} + epsilondef half_life(series: np.ndarray) -> float:
"""Estimate mean-reversion half-life from AR(1) regression.
Returns:
Half-life in periods. Negative means non-mean-reverting.
"""
y = np.diff(series)
x = series[:-1]
x = np.column_stack([np.ones(len(x)), x])
beta = np.linalg.lstsq(x, y, rcond=None)[0][1]
if beta >= 0:
return -1.0 # Not mean-reverting
return -np.log(2) / np.log(1 + beta)delta_X_t = alpha + beta * X_{t-1} + epsilondef half_life(series: np.ndarray) -> float:
"""Estimate mean-reversion half-life from AR(1) regression.
Returns:
Half-life in periods. Negative means non-mean-reverting.
"""
y = np.diff(series)
x = series[:-1]
x = np.column_stack([np.ones(len(x)), x])
beta = np.linalg.lstsq(x, y, rcond=None)[0][1]
if beta >= 0:
return -1.0 # Not mean-reverting
return -np.log(2) / np.log(1 + beta)| Parameter | Rule of Thumb |
|---|---|
| Lookback window | 2x half-life |
| Holding period | 1x half-life |
| Maximum hold | 3x half-life (stop) |
| Signal recalc | 0.5x half-life |
| 参数 | 经验法则 |
|---|---|
| 回溯窗口 | 2倍半衰期 |
| 持有周期 | 1倍半衰期 |
| 最大持有时间 | 3倍半衰期(止损) |
| 信号重算周期 | 0.5倍半衰期 |
z = (price - rolling_mean) / rolling_stdz = (price - rolling_mean) / rolling_std| Condition | Signal | Action |
|---|---|---|
| z < -2.0 | Buy | Enter long (price below mean) |
| z > +2.0 | Sell | Enter short (price above mean) |
| z crosses 0 | Exit | Close position (returned to mean) |
| abs(z) > 3.0 | Stop | Close position (reversion failed) |
| 条件 | 信号 | 操作 |
|---|---|---|
| z < -2.0 | 买入 | 做多(价格低于均值) |
| z > +2.0 | 卖出 | 做空(价格高于均值) |
| z穿越0 | 离场 | 平仓(回归至均值) |
| abs(z) > 3.0 | 止损 | 平仓(回归失败) |
def z_score_signals(
prices: np.ndarray,
lookback: int,
entry_z: float = 2.0,
exit_z: float = 0.0,
stop_z: float = 3.0,
) -> np.ndarray:
"""Generate z-score-based mean-reversion signals.
Returns:
Array of signals: 1 (long), -1 (short), 0 (flat).
"""
rolling_mean = pd.Series(prices).rolling(lookback).mean().values
rolling_std = pd.Series(prices).rolling(lookback).std().values
z = (prices - rolling_mean) / rolling_std
# See scripts/mean_reversion_test.py for full signal generation
...def z_score_signals(
prices: np.ndarray,
lookback: int,
entry_z: float = 2.0,
exit_z: float = 0.0,
stop_z: float = 3.0,
) -> np.ndarray:
"""Generate z-score-based mean-reversion signals.
Returns:
Array of signals: 1 (long), -1 (short), 0 (flat).
"""
rolling_mean = pd.Series(prices).rolling(lookback).mean().values
rolling_std = pd.Series(prices).rolling(lookback).std().values
z = (prices - rolling_mean) / rolling_std
# See scripts/mean_reversion_test.py for full signal generation
...size = base_size * min(abs(z) / entry_threshold, max_scale)references/strategy_design.mdsize = base_size * min(abs(z) / entry_threshold, max_scale)references/strategy_design.mddX = theta * (mu - X) * dt + sigma * dW| Parameter | Meaning | Estimation |
|---|---|---|
| theta | Speed of mean reversion | From AR(1) beta: theta = -ln(1+beta)/dt |
| mu | Long-run mean | From AR(1) intercept: mu = -alpha/beta |
| sigma | Volatility of innovations | Residual std from AR(1) |
dX = theta * (mu - X) * dt + sigma * dW| 参数 | 含义 | 估计方法 |
|---|---|---|
| theta | 均值回归速度 | 由AR(1)的beta计算:theta = -ln(1+beta)/dt |
| mu | 长期均值 | 由AR(1)的截距计算:mu = -alpha/beta |
| sigma | 创新项波动率 | AR(1)的残差标准差 |
def estimate_ou_params(series: np.ndarray, dt: float = 1.0) -> dict:
"""Estimate OU process parameters from observed series.
Returns:
Dict with keys: theta, mu, sigma, half_life.
"""
y = np.diff(series)
x = series[:-1]
x_with_const = np.column_stack([np.ones(len(x)), x])
params = np.linalg.lstsq(x_with_const, y, rcond=None)[0]
alpha, beta = params[0], params[1]
theta = -np.log(1 + beta) / dt
mu = -alpha / beta if beta != 0 else np.mean(series)
residuals = y - (alpha + beta * x)
sigma = np.std(residuals) * np.sqrt(2 * theta / (1 - np.exp(-2 * theta * dt)))
return {
"theta": theta,
"mu": mu,
"sigma": sigma,
"half_life": np.log(2) / theta if theta > 0 else -1,
}def estimate_ou_params(series: np.ndarray, dt: float = 1.0) -> dict:
"""Estimate OU process parameters from observed series.
Returns:
Dict with keys: theta, mu, sigma, half_life.
"""
y = np.diff(series)
x = series[:-1]
x_with_const = np.column_stack([np.ones(len(x)), x])
params = np.linalg.lstsq(x_with_const, y, rcond=None)[0]
alpha, beta = params[0], params[1]
theta = -np.log(1 + beta) / dt
mu = -alpha / beta if beta != 0 else np.mean(series)
residuals = y - (alpha + beta * x)
sigma = np.std(residuals) * np.sqrt(2 * theta / (1 - np.exp(-2 * theta * dt)))
return {
"theta": theta,
"mu": mu,
"sigma": sigma,
"half_life": np.log(2) / theta if theta > 0 else -1,
}cointegration-analysisS = Y - beta * Xcointegration-analysisS = Y - beta * Xregime-detectionregime-detection| Skill | Integration |
|---|---|
| Find cointegrated pairs for pairs trading |
| RSI, Bollinger Bands as mean-reversion indicators |
| Filter: only trade MR in ranging regimes |
| Backtest mean-reversion strategies |
| Estimate sigma for OU model |
| Factor execution costs into P&L estimates |
| Size positions using Kelly + z-score scaling |
| 技能 | 集成方式 |
|---|---|
| 寻找适合配对交易的协整资产对 |
| 将RSI、布林带作为均值回归指标 |
| 过滤:仅在区间震荡市场中进行均值回归交易 |
| 回测均值回归策略 |
| 为OU模型估计sigma参数 |
| 将执行成本纳入损益估算 |
| 结合Kelly准则与z-score缩放进行仓位管理 |
references/statistical_tests.mdreferences/strategy_design.mdreferences/statistical_tests.mdreferences/strategy_design.mdscripts/mean_reversion_test.pyscripts/pairs_scanner.pyscripts/mean_reversion_test.pyscripts/pairs_scanner.pyundefinedundefined
*This skill provides analytical tools and information only. It does not constitute financial advice or trading recommendations.*
*本技能仅提供分析工具及信息,不构成财务建议或交易推荐。*