regime-detection

市场状态检测

识别当前市场状态，以便选择合适的策略、正确设置仓位规模，并避免在震荡市场中部署趋势跟踪逻辑（反之亦然）。

为什么市场状态检测很重要

每种策略都有自己的“适用市场状态”。动量策略在明确的上涨趋势中能赚得盆满钵满，但在震荡区间中会不断亏损。均值回归网格策略在低波动盘整中表现出色，但遇到趋势突破时就会被击溃。市场状态检测会告诉你现在该使用哪套策略。

主要优势：

• 策略选择：将信号发送给适合当前环境的策略
• 仓位规模：在不利状态下减少敞口，在有利状态下增加敞口
• 止损调整：高波动状态下设置更宽的止损，低波动趋势中设置更紧的止损
• 回撤控制：在“危险区”状态（高波动 + 无趋势）中暂时离场

市场状态的核心维度

两个相互独立的轴定义了四象限市场状态模型：

	低波动	高波动
趋势	Q1：明确趋势 — 最适合趋势跟踪	Q2：高波动趋势 — 谨慎使用动量策略
震荡	Q3：平静区间 — 均值回归的理想环境	Q4：混乱震荡 — 减少交易或暂时离场

第三个维度——均值回归倾向（Hurst 指数）——可以进一步细分 Q3，告诉你价格回归的规律性有多高。

简单方法（无需 ML）

1. ATR 波动率百分位数

将当前 ATR 与其近期历史数据进行比较排名，得到 0–100 的百分位数得分。

import pandas as pd
import numpy as np

def atr_percentile(
    high: pd.Series, low: pd.Series, close: pd.Series,
    atr_period: int = 14, lookback: int = 100
) -> pd.Series:
    """ATR percentile rank over a rolling window."""
    tr = pd.concat([
        high - low,
        (high - close.shift(1)).abs(),
        (low - close.shift(1)).abs()
    ], axis=1).max(axis=1)
    atr = tr.rolling(atr_period).mean()
    return atr.rolling(lookback).apply(
        lambda x: pd.Series(x).rank(pct=True).iloc[-1], raw=False
    )

• < 第 25 百分位数 → 低波动状态
• 第 25–75 百分位数 → 正常波动
• > 第 75 百分位数 → 高波动状态

2. ADX 趋势强度

ADX 高于 25 表示市场处于趋势状态；低于 20 则表示市场处于区间状态。

def compute_adx(
    high: pd.Series, low: pd.Series, close: pd.Series,
    period: int = 14
) -> pd.Series:
    """Average Directional Index."""
    plus_dm = high.diff().clip(lower=0)
    minus_dm = (-low.diff()).clip(lower=0)
    # Zero out when the other is larger
    plus_dm[plus_dm < minus_dm] = 0
    minus_dm[minus_dm < plus_dm] = 0

    tr = pd.concat([
        high - low,
        (high - close.shift(1)).abs(),
        (low - close.shift(1)).abs()
    ], axis=1).max(axis=1)

    atr = tr.ewm(span=period, adjust=False).mean()
    plus_di = 100 * plus_dm.ewm(span=period, adjust=False).mean() / atr
    minus_di = 100 * minus_dm.ewm(span=period, adjust=False).mean() / atr
    dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di)
    return dx.ewm(span=period, adjust=False).mean()

3. EMA 斜率 + 价格位置

def trend_direction(close: pd.Series, period: int = 20) -> pd.Series:
    """Returns +1 (uptrend), -1 (downtrend), 0 (neutral)."""
    ema = close.ewm(span=period, adjust=False).mean()
    slope = ema.diff(5)  # 5-bar slope
    above = (close > ema).astype(int)
    direction = pd.Series(0, index=close.index)
    direction[(slope > 0) & (above == 1)] = 1
    direction[(slope < 0) & (above == 0)] = -1
    return direction

4. 布林带宽度百分位数

将 BB 宽度（上轨 - 下轨）/ 中轨作为波动率指标。“挤压”（低百分位数）通常预示着即将发生突破。

def bb_width_percentile(
    close: pd.Series, period: int = 20,
    std_dev: float = 2.0, lookback: int = 100
) -> pd.Series:
    """Bollinger Band width percentile."""
    sma = close.rolling(period).mean()
    std = close.rolling(period).std()
    width = (2 * std_dev * std) / sma
    return width.rolling(lookback).apply(
        lambda x: pd.Series(x).rank(pct=True).iloc[-1], raw=False
    )

统计方法

滚动 Hurst 指数

Hurst 指数 H 用于对时间序列行为进行分类：

• H < 0.4 → 均值回归（反持久性）
• 0.4 ≤ H ≤ 0.6 → 随机游走（不存在可利用的结构）
• H > 0.6 → 趋势（持久性）

使用重标极差（R/S）方法计算。完整推导请参阅 references/methodology.md。

def hurst_exponent(series: pd.Series, max_lag: int = 50) -> float:
    """Estimate Hurst exponent using R/S method."""
    lags = range(2, max_lag)
    rs_values = []
    for lag in lags:
        chunks = [series.iloc[i:i+lag] for i in range(0, len(series) - lag, lag)]
        rs_list = []
        for chunk in chunks:
            if len(chunk) < lag:
                continue
            mean_val = chunk.mean()
            devs = chunk - mean_val
            cumdev = devs.cumsum()
            r = cumdev.max() - cumdev.min()
            s = chunk.std(ddof=1)
            if s > 0:
                rs_list.append(r / s)
        if rs_list:
            rs_values.append(np.mean(rs_list))
        else:
            rs_values.append(np.nan)
    valid = [(l, r) for l, r in zip(lags, rs_values) if not np.isnan(r)]
    if len(valid) < 5:
        return 0.5
    log_lags = np.log([v[0] for v in valid])
    log_rs = np.log([v[1] for v in valid])
    coeffs = np.polyfit(log_lags, log_rs, 1)
    return coeffs[0]

变点检测（CUSUM）

检测收益率序列均值或方差的突变。

def cusum_test(
    returns: pd.Series, threshold: float = 2.0
) -> list[int]:
    """CUSUM change-point detection on returns.

    Returns indices where regime changes are detected.
    """
    mean_r = returns.mean()
    std_r = returns.std()
    if std_r == 0:
        return []
    s_pos, s_neg = 0.0, 0.0
    changes = []
    for i, r in enumerate(returns):
        z = (r - mean_r) / std_r
        s_pos = max(0, s_pos + z - 0.5)
        s_neg = max(0, s_neg - z - 0.5)
        if s_pos > threshold or s_neg > threshold:
            changes.append(i)
            s_pos, s_neg = 0.0, 0.0
    return changes

隐马尔可夫模型

使用 hmmlearn 构建 2–3 状态市场模型。此项为可选功能——所有核心功能仅依赖 numpy/pandas 即可运行。

# Optional: requires `uv pip install hmmlearn`
from hmmlearn import hmm

def fit_hmm_regimes(
    returns: np.ndarray, n_states: int = 2, n_iter: int = 100
) -> tuple[np.ndarray, object]:
    """Fit a Gaussian HMM to return series."""
    X = returns.reshape(-1, 1)
    model = hmm.GaussianHMM(
        n_components=n_states, covariance_type="full", n_iter=n_iter
    )
    model.fit(X)
    states = model.predict(X)
    return states, model

有关特征选择和状态解读的详情，请参阅 references/methodology.md。

加密货币的特殊注意事项

市场状态变化速度

加密货币市场状态的变化速度远快于股票市场：

参数	股票	加密货币（大市值）	加密货币（微市值 / PumpFun）
ATR 回看周期	100–200 根 K 线	50–100 根 K 线	20–50 根 K 线
ADX 周期	14–28	10–14	7–10
市场状态持续时间	数周至数月	数天至数周	数小时至数天
Hurst 窗口	200 根以上 K 线	100 根 K 线	50 根 K 线

成交量作为市场状态信号

在加密货币市场中，成交量可以确认市场状态的可靠性：

• 高成交量 + 趋势 → 信心强，顺势持有
• 低成交量 + 趋势 → 缓慢漂移，不可靠，减少仓位
• 高成交量 + 震荡 → 可能处于派发或吸筹阶段，留意突破
• 低成交量 + 震荡 → 市场缺乏活跃度，跳过

PumpFun 微观市场状态

新代币发行通常会经历一系列典型阶段：

1. 发行拉升（数分钟）：垂直上涨，极高波动，无均值回归
2. 首次抛售（数分钟至数小时）：获利了结，高波动，向下趋势
3. 盘整（数小时至数天）：低波动区间，可能出现均值回归
4. 第二波上涨或归零：要么再次突破（形成新趋势），要么逐渐跌至零

每种微观市场状态只持续数分钟至数小时。使用 1 分钟 K 线，窗口设为 20–50 根。

综合市场状态分类

def classify_regime(
    vol_percentile: float, adx: float, hurst: float,
    trend_dir: int
) -> dict[str, str]:
    """Classify into the 4-quadrant model."""
    vol_regime = (
        "low" if vol_percentile < 0.30
        else "high" if vol_percentile > 0.70
        else "normal"
    )
    trend_regime = (
        "trending" if adx > 25
        else "ranging" if adx < 20
        else "transitional"
    )
    direction = (
        "up" if trend_dir > 0
        else "down" if trend_dir < 0
        else "neutral"
    )
    mr_regime = (
        "mean_reverting" if hurst < 0.4
        else "trending" if hurst > 0.6
        else "random"
    )
    return {
        "volatility": vol_regime,
        "trend": trend_regime,
        "direction": direction,
        "mean_reversion": mr_regime,
        "quadrant": f"{vol_regime}_vol_{trend_regime}",
    }

策略适配

完整的市场状态—策略矩阵见 references/strategy_adaptation.md。

快速参考：

当前市场状态	操作
低波动 + 上涨趋势	使用满仓位进行趋势跟踪，设置紧密止损
高波动 + 趋势行情	使用半仓位进行动量交易，设置宽幅止损
低波动 + 区间震荡	采用均值回归 / 网格策略
高波动 + 区间震荡	减至 25% 仓位或暂时离场
市场状态转换	平仓或减至最低仓位

与其他技能集成

• pandas-ta：计算 ATR、ADX、布林带和 EMA
• volatility-modeling：进行高级波动率预测（GARCH、已实现波动率）
• strategy-framework：执行前先通过市场状态过滤器筛选信号
• position-sizing：根据市场状态波动率调整仓位规模
• risk-management：根据市场状态调整投资组合风险限额

文件

参考资料

• references/methodology.md — Hurst 指数、HMM、变点检测和波动率估算方法的详细数学说明
• references/strategy_adaptation.md — 完整的市场状态—策略矩阵，包括仓位规模、止损调整和 PumpFun 微观市场状态操作手册

脚本

• scripts/detect_regime.py — 在实时或演示数据上计算市场状态指标，并分类到四象限模型
• scripts/regime_backtest.py — 使用具有明确市场状态转换的合成数据，对比市场状态自适应策略与静态策略