市场水温算法设计方案：多因子热力累积模型

Market Temperature — Heat Accumulation Index (HAI)

一、背景与动机

在加密市场中存在一个"水温"现象：

• 当锚定资产（BTC）价格持续运行在高位时，ALT 资产的 Beta 系数、相关性、跟随性会发生相变
• 高温持续 → ALT 暴躁：Beta 跳变、跟随性增强、波动幅度夸张
• 低温持续 → ALT 迟钝：Beta 低迷、跟随性差

关键特征：不是瞬时温度决定行为，而是温度在高位维持足够长时间后，行情才发生质变。

本方案利用已有的 BTC OHLCV 实时流 + ALT 收益率数据，新建因子计算模块，聚合为市场级温度指标，影响所有配对的入场/出场参数。

二、算法架构

┌───────────────────────────────────────────────┐
│       BTC 5m K 线 + ALT 收益率（WebSocket）      │
└───────────────────┬───────────────────────────┘
                    │
       ┌────────────▼──────────────┐
       │   第一层：瞬时温度计        │
       │  7 因子 × Inverse-Cov 加权 │
       │  → instant_temp (0~100)   │
       └────────────┬──────────────┘
                    │
       ┌────────────▼──────────────┐
       │   第二层：Multi-Scale      │
       │   Leaky Integrator        │
       │  3 时间尺度 × 非对称衰减   │
       │  × 超线性注入              │
       │  → heat_percentile (0~100)│
       └────────────┬──────────────┘
                    │
       ┌────────────▼──────────────┐
       │   第三层：统一高斯          │
       │   Fuzzy Regime 分类       │
       │  p_cold + p_warm + p_hot=1│
       │  + Regime 转移概率矩阵     │
       │  + EMA 输出平滑           │
       │  → 策略参数 = Σpᵢ×paramᵢ │
       └────────────┬──────────────┘
                    │
       ┌────────────▼──────────────┐
       │   第四层：在线验证          │
       │  Performance Attribution  │
       │  各 regime 下交易表现统计   │
       │  → 自适应调整 regime 参数  │
       └───────────────────────────┘

与现有系统的关系：

决策流程：
  should_enter()
    ├─ ① market_temperature.regime_probs  → 市场级参数调整
    ├─ ② beta_regime.check()              → 配对级 β 拦截
    └─ ③ [未来] momentum_filter.check()   → 配对级动量过滤

三、第一层：瞬时温度计

3.1 七因子定义

所有因子统一使用 7 天滚动百分位 归一化到 0~100。

因子 6/7：相关性结构（跨资产）

def _compute_correlation_structure(
    self,
    btc_returns: deque,
    alt_returns_dict: dict[str, deque],
) -> tuple[float, float]:
    """计算 BTC-ALT Spearman 相关性的 level 和 dispersion

    Returns:
        (mean_corr, std_corr)
        - mean_corr: 平均相关系数，高值 → ALT 同步跟随 BTC → 温度高
        - std_corr: 相关系数标准差，低值 → 市场同质化 → 系统性风险高
    """
    min_window = 288  # 1 天
    if len(btc_returns) < min_window or len(alt_returns_dict) < 3:
        return 0.3, 0.3  # 数据不足返回中性值

    btc = np.array(btc_returns)[-min_window:]
    correlations = []
    for sym, alt_ret in alt_returns_dict.items():
        if len(alt_ret) < min_window:
            continue
        alt = np.array(alt_ret)[-min_window:]
        corr, _ = spearmanr(btc, alt)
        if not np.isnan(corr):
            correlations.append(corr)

    if len(correlations) < 3:
        return 0.3, 0.3

    return float(np.mean(correlations)), float(np.std(correlations))

Level + Dispersion 组合语义：

• Level 高 + Dispersion 低 = 最危险（ALT 高度同步跟随 BTC，Absorption Ratio 趋近 1）
• Level 中 + Dispersion 高 = 最安全（各 ALT 有独立行情，配对策略有效）

3.2 自适应加权：Inverse-Covariance (Ledoit-Wolf)

使用最小方差组合权重 w ∝ Σ⁻¹·1，显式处理因子间相关性（高相关因子被联合降权）。Ledoit-Wolf 收缩防止小样本协方差矩阵病态。

class InverseCovWeights:
    """Inverse-Covariance 自适应因子权重

    w = Σ⁻¹·1 / (1ᵀ·Σ⁻¹·1)
    Σ_shrunk = (1-δ)·Σ_sample + δ·diag(Σ_sample)
    """

    def __init__(self, n_factors: int = 7, vol_window: int = 288, shrinkage: float = 0.1):
        self.n_factors = n_factors
        self.vol_window = vol_window
        self.shrinkage = shrinkage
        self._buffers: list[deque] = [deque(maxlen=vol_window) for _ in range(n_factors)]
        self._default_weights = [1.0 / n_factors] * n_factors

    def update(self, scores: list[float]) -> list[float]:
        assert len(scores) == self.n_factors
        for i, s in enumerate(scores):
            self._buffers[i].append(s)

        min_samples = max(30, self.n_factors * 3)
        if any(len(buf) < min_samples for buf in self._buffers):
            return self._default_weights.copy()

        data = np.array([list(buf) for buf in self._buffers])
        cov = np.cov(data)
        diag_cov = np.diag(np.diag(cov))
        cov_shrunk = (1.0 - self.shrinkage) * cov + self.shrinkage * diag_cov

        try:
            inv_cov = np.linalg.inv(cov_shrunk)
            raw_weights = np.maximum(inv_cov @ np.ones(self.n_factors), 0.0)
            total = raw_weights.sum()
            if total < 1e-10:
                return self._default_weights.copy()
            return (raw_weights / total).tolist()
        except np.linalg.LinAlgError:
            return self._default_weights.copy()

3.3 滚动百分位计算器

from collections import deque
from sortedcontainers import SortedList

class RollingPercentile:
    """O(log n) 增量滚动百分位"""

    def __init__(self, window: int = 2016):
        self.window = window
        self.values = deque(maxlen=window)
        self.sorted_vals = SortedList()

    def update(self, value: float) -> float:
        if len(self.values) == self.window:
            self.sorted_vals.remove(self.values[0])
        self.values.append(value)
        self.sorted_vals.add(value)
        return (self.sorted_vals.bisect_right(value) / len(self.sorted_vals)) * 100.0

3.4 加权合成

def compute_instant_temp(self, scores: list[float]) -> float:
    weights = self._inv_cov_weights.update(scores)
    return sum(w * s for w, s in zip(weights, scores))  # 0~100

四、第二层：Multi-Scale Leaky Integrator

市场 regime 存在多个时间尺度：短期冲击（4_{5h）、日内趋势（11}12h）、多日牛熊（2~3d）。三个不同半衰期的漏积分器并行运行。

核心公式（单个积分器）：

excess = max(0, T(t) - T_base)
injection = g × excess^α                    (α=1.3, 超线性放大极端温度)
decay = λ_heat  若 T > T_base               (正在加热)
      = λ_cool  若 T ≤ T_base               (正在冷却，衰减更快)
h(t) = decay × h(t-1) + injection

三尺度合成：heat = 0.3×h_fast + 0.5×h_medium + 0.2×h_slow

非对称衰减：冷却速度 2~3× 快于加热（恐慌传导远快于信心建立）。

超线性注入：excess^1.3 使极端温度的影响被加速放大（excess=50 时注入量是线性的 3×）。

import math

class MultiScaleHeatAccumulator:
    def __init__(
        self,
        base_temp: float = 50.0,
        fast_decay_heat: float = 0.988,    # 半衰期 ≈ 4.8h
        fast_decay_cool: float = 0.975,    # 半衰期 ≈ 2.3h
        fast_gain: float = 0.10,
        medium_decay_heat: float = 0.995,  # 半衰期 ≈ 11.5h
        medium_decay_cool: float = 0.990,  # 半衰期 ≈ 5.8h
        medium_gain: float = 0.15,
        slow_decay_heat: float = 0.999,    # 半衰期 ≈ 57.8h
        slow_decay_cool: float = 0.997,    # 半衰期 ≈ 19.3h
        slow_gain: float = 0.05,
        injection_power: float = 1.3,
        scale_weights: tuple[float, float, float] = (0.3, 0.5, 0.2),
    ):
        self.base_temp = base_temp
        self.injection_power = injection_power
        self.scale_weights = scale_weights
        self._scales = [
            {"heat": 0.0, "decay_heat": fast_decay_heat, "decay_cool": fast_decay_cool, "gain": fast_gain},
            {"heat": 0.0, "decay_heat": medium_decay_heat, "decay_cool": medium_decay_cool, "gain": medium_gain},
            {"heat": 0.0, "decay_heat": slow_decay_heat, "decay_cool": slow_decay_cool, "gain": slow_gain},
        ]

    def update(self, instant_temp: float) -> float:
        excess = max(0.0, instant_temp - self.base_temp)
        injection_base = math.pow(excess, self.injection_power) if excess > 0 else 0.0
        is_heating = instant_temp > self.base_temp

        composite = 0.0
        for i, scale in enumerate(self._scales):
            decay = scale["decay_heat"] if is_heating else scale["decay_cool"]
            scale["heat"] = decay * scale["heat"] + scale["gain"] * injection_base
            composite += self.scale_weights[i] * scale["heat"]
        return composite

    def get_scale_heats(self) -> dict[str, float]:
        return {
            "fast": self._scales[0]["heat"],
            "medium": self._scales[1]["heat"],
            "slow": self._scales[2]["heat"],
        }

均衡分析（medium 尺度，h_eq = gain × (T-base)^α / (1-λ)）：

T=60 → h_eq ≈ 600    T=70 → h_eq ≈ 1440
T=80 → h_eq ≈ 2280   T=90 → h_eq ≈ 3450    T=100 → h_eq ≈ 4500

热量有自然上界，无需硬上限钳位。合成热量通过 RollingPercentile(window=2016) 归一化到 0~100。

五、第三层：统一高斯 Fuzzy Regime 分类

三个 regime 统一使用高斯形式 score_i = -(h-μ_i)²/(2σ_i²)，远离中心时自然衰减到零。

import math

class GaussianRegimeClassifier:
    def __init__(
        self,
        centers: tuple[float, float, float] = (0.0, 50.0, 100.0),
        widths: tuple[float, float, float] = (20.0, 18.0, 20.0),
    ):
        self.centers = centers
        self.widths = widths

    def classify(self, heat_percentile: float) -> dict[str, float]:
        h = heat_percentile
        names = ("cold", "warm", "hot")
        scores = [-((h - mu) ** 2) / (2.0 * sigma ** 2)
                  for mu, sigma in zip(self.centers, self.widths)]
        max_s = max(scores)
        exps = [math.exp(s - max_s) for s in scores]
        total = sum(exps)
        return {n: e / total for n, e in zip(names, exps)}

隶属度速查：

heat=0  → cold=0.97 warm=0.03 hot=0.00
heat=30 → cold=0.28 warm=0.72 hot=0.00
heat=50 → cold=0.03 warm=0.94 hot=0.03
heat=70 → cold=0.00 warm=0.72 hot=0.28
heat=100→ cold=0.00 warm=0.03 hot=0.97

5.1 Regime 转移概率矩阵

在线统计 regime 间的转移频次，用于前瞻性风控：当前 WARM 但 P(WARM→HOT) 升高时提前收紧。

class RegimeTransitionTracker:
    def __init__(self, smooth_count: float = 1.0):
        self.regimes = ("cold", "warm", "hot")
        self._counts = {r1: {r2: smooth_count for r2 in self.regimes} for r1 in self.regimes}
        self._prev_dominant: str | None = None

    def update(self, dominant_regime: str) -> dict[str, dict[str, float]]:
        if self._prev_dominant is not None:
            self._counts[self._prev_dominant][dominant_regime] += 1.0
        self._prev_dominant = dominant_regime
        probs = {}
        for r1 in self.regimes:
            total = sum(self._counts[r1].values())
            probs[r1] = {r2: self._counts[r1][r2] / total for r2 in self.regimes}
        return probs

    def get_transition_risk(self, current_regime: str) -> float:
        """向更热 regime 转移的概率"""
        total = sum(self._counts[current_regime].values())
        probs = {r: self._counts[current_regime][r] / total for r in self.regimes}
        if current_regime == "cold":
            return probs["warm"] + probs["hot"]
        elif current_regime == "warm":
            return probs["hot"]
        return 0.0

5.2 平滑策略参数调整

_REGIME_PARAMS = {
    "cold": {"zscore_entry_multiplier": 1.3, "cointegration_min_passed": 4,
             "position_size_multiplier": 0.7, "max_concurrent_positions": 3},
    "warm": {"zscore_entry_multiplier": 1.0, "cointegration_min_passed": 3,
             "position_size_multiplier": 1.0, "max_concurrent_positions": 3},
    "hot":  {"zscore_entry_multiplier": 1.5, "cointegration_min_passed": 5,
             "position_size_multiplier": 0.5, "max_concurrent_positions": 2},
}

def get_blended_adjustments(regime_probs: dict[str, float], transition_risk: float = 0.0) -> dict:
    result = {}
    for key in _REGIME_PARAMS["warm"]:
        result[key] = sum(regime_probs[r] * _REGIME_PARAMS[r][key] for r in ("cold", "warm", "hot"))

    # 前瞻性风控：转移风险高时预防性收紧
    if transition_risk > 0.25:
        risk_factor = 1.0 + 0.2 * transition_risk
        result["zscore_entry_multiplier"] *= risk_factor
        result["position_size_multiplier"] /= risk_factor

    dominant = max(regime_probs, key=regime_probs.get)
    result["dominant_regime"] = dominant
    result["dominant_prob"] = regime_probs[dominant]
    result["transition_risk"] = transition_risk
    return result

5.3 输出 EMA 防抖

class TemperatureOutputSmoother:
    def __init__(self, alpha: float = 0.1):
        self.alpha = alpha
        self._smoothed: dict[str, float] | None = None

    def smooth(self, regime_probs: dict[str, float]) -> dict[str, float]:
        if self._smoothed is None:
            self._smoothed = regime_probs.copy()
        else:
            for key in regime_probs:
                self._smoothed[key] = self.alpha * regime_probs[key] + (1.0 - self.alpha) * self._smoothed[key]
        total = sum(self._smoothed.values())
        return {k: v / total for k, v in self._smoothed.items()}

六、第四层：在线 Performance Attribution

统计各 regime 下的交易表现，验证 regime 判断有效性。若 HOT 下交易表现并不差 → 过度保守 → 自动放松。

class OnlinePerformanceAttribution:
    """约束：≥30 笔后才调整，幅度 ±20%，滑动窗口 200 笔"""

    def __init__(self, min_trades: int = 30, max_adjustment: float = 0.2, window: int = 200):
        self.min_trades = min_trades
        self.max_adjustment = max_adjustment
        self._trades: dict[str, deque] = {r: deque(maxlen=window) for r in ("cold", "warm", "hot")}

    def record_trade(self, dominant_regime: str, pnl_pct: float):
        self._trades[dominant_regime].append(pnl_pct)

    def get_regime_adjustment(self) -> dict[str, float]:
        """返回 {regime: adj}，adj<1.0 表示可放松，1.0 表示维持"""
        warm_stats = self._compute_stats("warm")
        adjustments = {}
        for regime in ("cold", "warm", "hot"):
            stats = self._compute_stats(regime)
            if stats is None or warm_stats is None:
                adjustments[regime] = 1.0
                continue

            perf_ratio = stats["sharpe"] / max(abs(warm_stats["sharpe"]), 1e-6)
            if regime in ("hot", "cold") and perf_ratio > 0.8:
                adjustments[regime] = max(1.0 - self.max_adjustment, perf_ratio)
            else:
                adjustments[regime] = 1.0
        return adjustments

    def _compute_stats(self, regime: str) -> dict | None:
        trades = list(self._trades[regime])
        if len(trades) < self.min_trades:
            return None
        arr = np.array(trades)
        mean, std = float(np.mean(arr)), float(np.std(arr))
        return {"sharpe": mean / max(std, 1e-6), "win_rate": float(np.mean(arr > 0)), "n": len(trades)}

七、集成设计

7.1 文件变更

新增：
  src/trading/market_temperature.py  # ~400 行，核心（7 个类 + 顶层聚合）
  src/trading/market_factors.py      # ~150 行，因子计算（RS/ER/Volume/Corr）

修改：
  src/trading/config.py              # ~25 行，新增配置
  src/trading/strategy.py            # ~50 行，集成温度判断
  src/services/realtime_kline_service_base.py  # ~30 行，驱动温度更新

新增依赖：
  sortedcontainers>=2.4.0

7.2 数据流

_on_candle_update("BTC", "5m")
  ├─ market_factors.update_btc(candle)
  │     → rs_vol, bocpd_prob, cusum_max, er, vol_ratio
  └─ market_temperature.update(7 个因子原始值)
       → 归一化 → Inverse-Cov 加权 → Multi-Scale Integrate
       → 高斯 Regime → 转移概率 → EMA 平滑

_on_candle_update("ALT", "5m")
  └─ market_factors.update_alt_returns(symbol, returns)

strategy.should_enter()
  └─ adjustments = market_temperature.get_adjustments()
       → effective_threshold = adaptive_threshold × adjustments["zscore_entry_multiplier"]
       → position_multiplier = adjustments["position_size_multiplier"]

_on_trade_closed(trade)
  └─ market_temperature.record_trade(regime, pnl_pct)

7.3 strategy.py 集成点

def _check_entry(self, key, z4h, adaptive_z, timestamp, current_above, params, latest_price=None):
    # === 第一关：市场水温（市场级） ===
    if self._market_temp and self._market_temp.enabled:
        adj = self._market_temp.get_adjustments()
        effective_zscore_threshold = params.adaptive_threshold * adj["zscore_entry_multiplier"]
        min_passed = round(adj["cointegration_min_passed"])
        position_multiplier = adj["position_size_multiplier"]
        logger.info(
            f"🌡️ 水温 | regime={adj['dominant_regime']}({adj['dominant_prob']:.0%}) "
            f"t_risk={adj['transition_risk']:.0%} | "
            f"阈值×{adj['zscore_entry_multiplier']:.2f} 仓位×{adj['position_size_multiplier']:.2f}"
        )
    else:
        effective_zscore_threshold = params.adaptive_threshold
        position_multiplier = 1.0

    # === 第二关：Beta 体制过滤（配对级） ===
    ...

7.4 配置参数

# trading/config.py → StrategyParams 新增
market_temperature_enabled: bool = True
# 第一层
market_temperature_percentile_window: int = 2016    # 百分位窗口（7天）
market_temperature_cov_window: int = 288            # 协方差窗口（1天）
market_temperature_cov_shrinkage: float = 0.1       # Ledoit-Wolf 收缩系数
market_temperature_corr_window: int = 288           # 相关性窗口（1天）
# 第二层
market_temperature_base_temp: float = 50.0
market_temperature_injection_power: float = 1.3
market_temperature_fast_decay_heat: float = 0.988
market_temperature_fast_decay_cool: float = 0.975
market_temperature_fast_gain: float = 0.10
market_temperature_medium_decay_heat: float = 0.995
market_temperature_medium_decay_cool: float = 0.990
market_temperature_medium_gain: float = 0.15
market_temperature_slow_decay_heat: float = 0.999
market_temperature_slow_decay_cool: float = 0.997
market_temperature_slow_gain: float = 0.05
market_temperature_scale_weights: tuple = (0.3, 0.5, 0.2)
# 第三层
market_temperature_regime_centers: tuple = (0.0, 50.0, 100.0)
market_temperature_regime_widths: tuple = (20.0, 18.0, 20.0)
market_temperature_smooth_alpha: float = 0.1
# 第四层
market_temperature_perf_min_trades: int = 30
market_temperature_perf_max_adjustment: float = 0.2

八、监控与告警

飞书推送（Regime 切换时）

🌡️ 市场水温变化

Regime: WARM → HOT (p=0.72)
分布: cold=0.02 warm=0.26 hot=0.72
转移: P(hot→hot)=0.87

因子 (百分位 × 权重):
  RS=91.2(w=0.16) BOCPD=78.5(w=0.13) CUSUM=85.3(w=0.11)
  ER=67.2(w=0.15) Vol=72.8(w=0.14) CorrLv=88.4(w=0.18) CorrDisp=92.1(w=0.13)

热量: fast=127.3 medium=847.5 slow=412.8 | pct=78.1
调整: 阈值×1.38 仓位×0.63 协整≥5

日志

DEBUG [MarketTemp] instant=72.3 heat_pct=78.1 scales=[127,848,413] regime=HOT(0.72) t_risk=0.11
INFO  [MarketTemp] REGIME SHIFT: WARM→HOT (p=0.72, 4.2h in WARM)

数据库存储（可选，回测用）

ALTER TABLE analysis_results
  ADD COLUMN IF NOT EXISTS market_temp DOUBLE PRECISION,
  ADD COLUMN IF NOT EXISTS market_regime VARCHAR(10),
  ADD COLUMN IF NOT EXISTS market_regime_prob DOUBLE PRECISION;

九、回测验证

# 历史回放
for candle in historical_btc_5m:
    temp.update(candle)
    adjustments = temp.get_adjustments()

# 对比：A=无水温 / B=有水温 / C=有水温+反馈
# 指标：胜率、盈亏比、最大回撤、夏普、Calmar

核心检验：

• HOT 期 B/C 回撤 < A
• WARM 期 ABC 表现一致
• 相关性飙升期拦截率提升
• C 组过度保守 < B 组（反馈放松）
• 因子权重：RS/CUSUM 相关期联合降权

十、实现优先级

P0：~435 行 | P0+P1：~840 行 | 全量：~1145 行

十一、局限与演进

演进路径：

附录：参数速查表

半衰期: t½ = ln(0.5)/ln(λ) × 5min

均衡热量: h_eq = gain × (T-50)^1.3 / (1-λ)