472 lines
18 KiB
Python
472 lines
18 KiB
Python
"""Unit tests for v3 posterior assembly and regime classification.
|
||
|
||
Tests for compute_v3_posterior and classify_regime_v3 functions.
|
||
|
||
Requirements validated: 5.1–5.7, 6.1–6.8
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
|
||
import pytest
|
||
|
||
from services.aggregation.bayesian import V3Posterior, compute_v3_posterior
|
||
from services.aggregation.regime import (
|
||
_V3_REGIME_PARAMS,
|
||
MarketRegime,
|
||
V3RegimeClassification,
|
||
classify_regime_v3,
|
||
)
|
||
from services.aggregation.worker import EvidenceCluster
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_cluster(cluster_llr: float, n_eff: float = 1.0) -> EvidenceCluster:
|
||
"""Create a minimal EvidenceCluster with given LLR and n_eff."""
|
||
return EvidenceCluster(
|
||
cluster_id="test", units=[], llrs=[], n_eff=n_eff, cluster_llr=cluster_llr
|
||
)
|
||
|
||
|
||
def _make_regime(
|
||
regime: MarketRegime = MarketRegime.UNCERTAINTY, gamma: float = 0.80
|
||
) -> V3RegimeClassification:
|
||
"""Create a V3RegimeClassification with specified regime and gamma."""
|
||
return V3RegimeClassification(
|
||
regime=regime,
|
||
trend_z=0.0,
|
||
vol_ratio=1.0,
|
||
evidence_multiplier=gamma,
|
||
confidence_multiplier=0.85,
|
||
phi_decay=0.50,
|
||
atr_multiplier=2.0,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: Empty clusters → P_up = 0.50 (neutral prior)
|
||
# Requirements: 5.1, 5.3
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestEmptyClusters:
|
||
"""No evidence → log_odds = logit(0.50) = 0 → P_up = 0.50."""
|
||
|
||
def test_empty_clusters_gives_neutral_posterior(self):
|
||
regime = _make_regime()
|
||
result = compute_v3_posterior(clusters=[], regime=regime, p_prior=0.50)
|
||
|
||
assert isinstance(result, V3Posterior)
|
||
assert result.p_up == pytest.approx(0.50, abs=1e-9)
|
||
assert result.p_down == pytest.approx(0.50, abs=1e-9)
|
||
assert result.log_odds == pytest.approx(0.0, abs=1e-9)
|
||
assert result.strength == pytest.approx(0.0, abs=1e-9)
|
||
assert result.direction == "neutral"
|
||
assert result.n_eff_total == 0.0
|
||
|
||
def test_empty_clusters_with_default_prior(self):
|
||
regime = _make_regime()
|
||
result = compute_v3_posterior(clusters=[], regime=regime)
|
||
|
||
assert result.p_up == pytest.approx(0.50, abs=1e-9)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: All bullish clusters → P_up > 0.50
|
||
# Requirements: 5.1, 5.2
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestAllBullishClusters:
|
||
"""Positive cluster LLRs with uncertainty regime (gamma=0.80) → P_up > 0.50."""
|
||
|
||
def test_bullish_clusters_give_p_up_above_half(self):
|
||
clusters = [_make_cluster(1.0), _make_cluster(0.5)]
|
||
regime = _make_regime(MarketRegime.UNCERTAINTY, gamma=0.80)
|
||
|
||
result = compute_v3_posterior(clusters=clusters, regime=regime, p_prior=0.50)
|
||
|
||
assert result.p_up > 0.50
|
||
assert result.direction in ("bullish", "neutral") # depends on threshold
|
||
assert result.log_odds > 0.0
|
||
|
||
def test_bullish_computes_correct_log_odds(self):
|
||
"""Verify log_odds = logit(0.50) + gamma * sum(LLR_c)."""
|
||
clusters = [_make_cluster(1.0), _make_cluster(0.5)]
|
||
regime = _make_regime(MarketRegime.UNCERTAINTY, gamma=0.80)
|
||
|
||
result = compute_v3_posterior(clusters=clusters, regime=regime, p_prior=0.50)
|
||
|
||
expected_log_odds = 0.0 + 0.80 * (1.0 + 0.5) # = 1.2
|
||
assert result.log_odds == pytest.approx(expected_log_odds, abs=1e-9)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: All bearish clusters → P_up < 0.50
|
||
# Requirements: 5.1, 5.2
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestAllBearishClusters:
|
||
"""Negative cluster LLRs → P_up < 0.50."""
|
||
|
||
def test_bearish_clusters_give_p_up_below_half(self):
|
||
clusters = [_make_cluster(-1.0), _make_cluster(-0.5)]
|
||
regime = _make_regime(MarketRegime.UNCERTAINTY, gamma=0.80)
|
||
|
||
result = compute_v3_posterior(clusters=clusters, regime=regime, p_prior=0.50)
|
||
|
||
assert result.p_up < 0.50
|
||
assert result.log_odds < 0.0
|
||
|
||
def test_bearish_computes_correct_log_odds(self):
|
||
clusters = [_make_cluster(-1.0), _make_cluster(-0.5)]
|
||
regime = _make_regime(MarketRegime.UNCERTAINTY, gamma=0.80)
|
||
|
||
result = compute_v3_posterior(clusters=clusters, regime=regime, p_prior=0.50)
|
||
|
||
expected_log_odds = 0.0 + 0.80 * (-1.0 + -0.5) # = -1.2
|
||
assert result.log_odds == pytest.approx(expected_log_odds, abs=1e-9)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: Regime direction thresholds at boundary values
|
||
# Requirements: 5.5
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDirectionThresholds:
|
||
"""Test direction classification at panic regime boundaries (0.68/0.32)."""
|
||
|
||
def _compute_with_target_p_up(self, target_p_up: float) -> V3Posterior:
|
||
"""Compute posterior that results in a specific P_up value.
|
||
|
||
We reverse-engineer the cluster LLR needed to produce the target P_up
|
||
under panic regime with gamma=0.70 and p_prior=0.50.
|
||
"""
|
||
# logit(target) = logit(0.50) + gamma * cluster_llr
|
||
# logit(target) = 0.0 + 0.70 * cluster_llr
|
||
# cluster_llr = logit(target) / 0.70
|
||
logit_target = math.log(target_p_up / (1.0 - target_p_up))
|
||
cluster_llr = logit_target / 0.70
|
||
|
||
clusters = [_make_cluster(cluster_llr)]
|
||
regime = _make_regime(MarketRegime.PANIC, gamma=0.70)
|
||
return compute_v3_posterior(clusters=clusters, regime=regime, p_prior=0.50)
|
||
|
||
def test_panic_bullish_at_068(self):
|
||
"""P_up = 0.68 → bullish in panic regime (threshold is 0.68)."""
|
||
result = self._compute_with_target_p_up(0.68)
|
||
assert result.p_up == pytest.approx(0.68, abs=1e-6)
|
||
assert result.direction == "bullish"
|
||
|
||
def test_panic_neutral_at_067(self):
|
||
"""P_up = 0.67 → neutral in panic regime (below 0.68 threshold)."""
|
||
result = self._compute_with_target_p_up(0.67)
|
||
assert result.p_up == pytest.approx(0.67, abs=1e-6)
|
||
assert result.direction == "neutral"
|
||
|
||
def test_panic_bearish_at_032(self):
|
||
"""P_up = 0.32 → bearish in panic regime (threshold is 0.32)."""
|
||
result = self._compute_with_target_p_up(0.32)
|
||
assert result.p_up == pytest.approx(0.32, abs=1e-6)
|
||
assert result.direction == "bearish"
|
||
|
||
def test_panic_neutral_at_033(self):
|
||
"""P_up = 0.33 → neutral in panic regime (above 0.32 threshold)."""
|
||
result = self._compute_with_target_p_up(0.33)
|
||
assert result.p_up == pytest.approx(0.33, abs=1e-6)
|
||
assert result.direction == "neutral"
|
||
|
||
def test_trend_following_thresholds(self):
|
||
"""Trend following thresholds: bullish >= 0.60, bearish <= 0.40."""
|
||
# Bullish at 0.60
|
||
logit_target = math.log(0.60 / 0.40)
|
||
cluster_llr = logit_target / 1.10 # gamma for trend_following
|
||
clusters = [_make_cluster(cluster_llr)]
|
||
regime = _make_regime(MarketRegime.TREND_FOLLOWING, gamma=1.10)
|
||
result = compute_v3_posterior(clusters=clusters, regime=regime, p_prior=0.50)
|
||
assert result.p_up == pytest.approx(0.60, abs=1e-6)
|
||
assert result.direction == "bullish"
|
||
|
||
def test_mean_reversion_thresholds(self):
|
||
"""Mean reversion thresholds: bullish >= 0.63, bearish <= 0.37."""
|
||
# Use slightly above 0.63 to avoid floating-point boundary issues
|
||
target = 0.631
|
||
logit_target = math.log(target / (1.0 - target))
|
||
cluster_llr = logit_target / 0.90 # gamma for mean_reversion
|
||
clusters = [_make_cluster(cluster_llr)]
|
||
regime = _make_regime(MarketRegime.MEAN_REVERSION, gamma=0.90)
|
||
result = compute_v3_posterior(clusters=clusters, regime=regime, p_prior=0.50)
|
||
assert result.p_up >= 0.63
|
||
assert result.direction == "bullish"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: Prior clamp [0.40, 0.60]
|
||
# Requirements: 5.6
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPriorClamp:
|
||
"""Prior values outside [0.40, 0.60] are clamped."""
|
||
|
||
def test_prior_below_040_clamped(self):
|
||
"""p_prior=0.30 → clamped to 0.40."""
|
||
regime = _make_regime()
|
||
result = compute_v3_posterior(clusters=[], regime=regime, p_prior=0.30)
|
||
|
||
# logit(0.40) ≈ -0.4055
|
||
expected_log_odds = math.log(0.40 / 0.60)
|
||
assert result.log_odds == pytest.approx(expected_log_odds, abs=1e-9)
|
||
# P_up should be 0.40 with no evidence
|
||
assert result.p_up == pytest.approx(0.40, abs=1e-6)
|
||
|
||
def test_prior_above_060_clamped(self):
|
||
"""p_prior=0.80 → clamped to 0.60."""
|
||
regime = _make_regime()
|
||
result = compute_v3_posterior(clusters=[], regime=regime, p_prior=0.80)
|
||
|
||
expected_log_odds = math.log(0.60 / 0.40)
|
||
assert result.log_odds == pytest.approx(expected_log_odds, abs=1e-9)
|
||
assert result.p_up == pytest.approx(0.60, abs=1e-6)
|
||
|
||
def test_prior_at_040_not_clamped(self):
|
||
"""p_prior=0.40 is within bounds, no clamping."""
|
||
regime = _make_regime()
|
||
result = compute_v3_posterior(clusters=[], regime=regime, p_prior=0.40)
|
||
assert result.p_up == pytest.approx(0.40, abs=1e-6)
|
||
|
||
def test_prior_at_060_not_clamped(self):
|
||
"""p_prior=0.60 is within bounds, no clamping."""
|
||
regime = _make_regime()
|
||
result = compute_v3_posterior(clusters=[], regime=regime, p_prior=0.60)
|
||
assert result.p_up == pytest.approx(0.60, abs=1e-6)
|
||
|
||
def test_prior_at_050_standard(self):
|
||
"""p_prior=0.50 is standard neutral prior."""
|
||
regime = _make_regime()
|
||
result = compute_v3_posterior(clusters=[], regime=regime, p_prior=0.50)
|
||
assert result.p_up == pytest.approx(0.50, abs=1e-9)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: Regime classification with known inputs
|
||
# Requirements: 6.1–6.8
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestRegimeClassificationKnownInputs:
|
||
"""Test classify_regime_v3 with known inputs mapping to specific regimes."""
|
||
|
||
def _make_prices_and_returns(
|
||
self, n: int = 120
|
||
) -> tuple[list[float], list[float]]:
|
||
"""Generate flat price series and near-zero returns."""
|
||
prices = [100.0] * n
|
||
returns = [0.001] * n
|
||
return prices, returns
|
||
|
||
def test_high_vol_ratio_triggers_panic(self):
|
||
"""vol_ratio > 1.5 → panic."""
|
||
# Generate returns where sigma_20 >> sigma_100
|
||
# sigma_20 high, sigma_100 low → vol_ratio > 1.5
|
||
prices = [100.0] * 120
|
||
# Low-vol returns for sigma_100 context
|
||
returns_low = [0.001] * 80
|
||
# High-vol returns for sigma_20
|
||
returns_high = [0.05, -0.05] * 10 # stdev ≈ 0.0526
|
||
returns = returns_low + returns_high
|
||
|
||
# With flat prices, trend_z ≈ 0, but vol_ratio > 1.5 → panic
|
||
result = classify_regime_v3(prices, returns, atr_20=1.0)
|
||
assert result.regime == MarketRegime.PANIC
|
||
|
||
def test_trend_following_regime(self):
|
||
"""trend_z >= 0.75, vol_ratio < 1.3 → trend_following.
|
||
|
||
|trend_z| >= 0.75 AND vol_ratio < 1.3 → trend_following.
|
||
"""
|
||
# Trending price series: EMA_20 significantly above EMA_100
|
||
prices = [100.0 + i * 0.5 for i in range(120)]
|
||
|
||
# Varied returns with stable vol (sigma_20 ≈ sigma_100 → vol_ratio near 1.0)
|
||
returns = [0.005 + (i % 5) * 0.001 for i in range(120)]
|
||
|
||
# ATR chosen so trend_z ≈ 1.77 (well above 0.75 threshold)
|
||
result = classify_regime_v3(prices, returns, atr_20=10.0)
|
||
|
||
assert result.regime == MarketRegime.TREND_FOLLOWING
|
||
assert abs(result.trend_z) >= 0.75
|
||
assert result.vol_ratio < 1.3
|
||
|
||
def test_mean_reversion_regime(self):
|
||
"""|trend_z| < 0.50 AND vol_ratio < 1.0 → mean_reversion."""
|
||
# Flat prices → trend_z ≈ 0
|
||
prices = [100.0] * 120
|
||
|
||
# Returns with decreasing volatility (sigma_20 < sigma_100)
|
||
# High vol early, low vol recently
|
||
returns_early = [0.03, -0.03] * 40 # high vol for sigma_100
|
||
returns_recent = [0.001] * 40 # low vol for sigma_20
|
||
returns = returns_early + returns_recent
|
||
|
||
# Large ATR so trend_z stays small
|
||
result = classify_regime_v3(prices, returns, atr_20=50.0)
|
||
|
||
assert result.regime == MarketRegime.MEAN_REVERSION
|
||
assert abs(result.trend_z) < 0.50
|
||
assert result.vol_ratio < 1.0
|
||
|
||
def test_uncertainty_regime_default(self):
|
||
"""When conditions don't match any specific regime → uncertainty.
|
||
|
||
|trend_z| between 0.50 and 0.75 OR vol_ratio between 1.0 and 1.3.
|
||
"""
|
||
# Mild trend + moderate vol → uncertainty
|
||
# Slightly trending prices but not enough for trend_following
|
||
prices = [100.0 + i * 0.1 for i in range(120)]
|
||
|
||
# Uniform returns → vol_ratio ≈ 1.0
|
||
returns = [0.01] * 120
|
||
|
||
# ATR chosen so |trend_z| is between 0.50 and 0.75
|
||
# We need to find ATR such that it lands in uncertainty
|
||
# With mild trend, vol_ratio ≈ 1.0 (not < 1.0), so mean_reversion won't fire
|
||
# And trend_z might be < 0.75, so trend_following won't fire
|
||
result = classify_regime_v3(prices, returns, atr_20=5.0)
|
||
|
||
# With uniform returns, stdev is 0 → sigma_100 = 0
|
||
# We need non-trivial returns. Let's use a better approach.
|
||
# Use returns that give vol_ratio between 1.0 and 1.3
|
||
returns_varied = [0.01 + (i % 3) * 0.002 for i in range(120)]
|
||
result = classify_regime_v3(prices, returns_varied, atr_20=5.0)
|
||
|
||
# This should fall through to uncertainty since conditions are moderate
|
||
assert result.regime == MarketRegime.UNCERTAINTY
|
||
|
||
def test_data_insufficient_returns_uncertainty(self):
|
||
"""Fewer than 100 closing prices → default uncertainty."""
|
||
prices = [100.0] * 50 # < 100
|
||
returns = [0.01] * 50
|
||
result = classify_regime_v3(prices, returns, atr_20=1.0)
|
||
|
||
assert result.regime == MarketRegime.UNCERTAINTY
|
||
assert result.trend_z == 0.0
|
||
assert result.vol_ratio == 1.0
|
||
assert result.evidence_multiplier == 0.80
|
||
|
||
def test_atr_zero_returns_uncertainty(self):
|
||
"""ATR_20 <= 0 → default uncertainty (Req 6.8)."""
|
||
prices = [100.0] * 120
|
||
returns = [0.01] * 120
|
||
result = classify_regime_v3(prices, returns, atr_20=0.0)
|
||
|
||
assert result.regime == MarketRegime.UNCERTAINTY
|
||
|
||
def test_insufficient_returns_data(self):
|
||
"""Fewer than 100 daily returns → default uncertainty."""
|
||
prices = [100.0] * 120
|
||
returns = [0.01] * 50 # < 100
|
||
result = classify_regime_v3(prices, returns, atr_20=1.0)
|
||
|
||
assert result.regime == MarketRegime.UNCERTAINTY
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: Regime parameters are correctly assigned
|
||
# Requirements: 6.6, 6.7
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestRegimeParameters:
|
||
"""Verify regime parameters (gamma, conf_mult, phi, atr_mult) are assigned."""
|
||
|
||
def test_panic_parameters(self):
|
||
gamma, conf_mult, phi, atr_mult, min_edge = _V3_REGIME_PARAMS[
|
||
MarketRegime.PANIC
|
||
]
|
||
assert gamma == 0.70
|
||
assert conf_mult == 0.70
|
||
assert phi == 0.35
|
||
assert atr_mult == 2.5
|
||
assert min_edge == 0.0100
|
||
|
||
def test_trend_following_parameters(self):
|
||
gamma, conf_mult, phi, atr_mult, min_edge = _V3_REGIME_PARAMS[
|
||
MarketRegime.TREND_FOLLOWING
|
||
]
|
||
assert gamma == 1.10
|
||
assert conf_mult == 1.00
|
||
assert phi == 0.80
|
||
assert atr_mult == 1.8
|
||
assert min_edge == 0.0035
|
||
|
||
def test_mean_reversion_parameters(self):
|
||
gamma, conf_mult, phi, atr_mult, min_edge = _V3_REGIME_PARAMS[
|
||
MarketRegime.MEAN_REVERSION
|
||
]
|
||
assert gamma == 0.90
|
||
assert conf_mult == 0.95
|
||
assert phi == 0.55
|
||
assert atr_mult == 1.4
|
||
assert min_edge == 0.0050
|
||
|
||
def test_uncertainty_parameters(self):
|
||
gamma, conf_mult, phi, atr_mult, min_edge = _V3_REGIME_PARAMS[
|
||
MarketRegime.UNCERTAINTY
|
||
]
|
||
assert gamma == 0.80
|
||
assert conf_mult == 0.85
|
||
assert phi == 0.50
|
||
assert atr_mult == 2.0
|
||
assert min_edge == 0.0075
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: Posterior output fields
|
||
# Requirements: 5.4, 5.5
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPosteriorOutputFields:
|
||
"""Verify derived fields (strength, n_eff_total, regime) are correct."""
|
||
|
||
def test_strength_computed_correctly(self):
|
||
"""strength = abs(2 × P_up - 1)."""
|
||
clusters = [_make_cluster(1.5)]
|
||
regime = _make_regime(MarketRegime.UNCERTAINTY, gamma=0.80)
|
||
result = compute_v3_posterior(clusters=clusters, regime=regime, p_prior=0.50)
|
||
|
||
expected_strength = abs(2.0 * result.p_up - 1.0)
|
||
assert result.strength == pytest.approx(expected_strength, abs=1e-9)
|
||
|
||
def test_n_eff_total_sums_clusters(self):
|
||
"""n_eff_total = sum of cluster n_eff values."""
|
||
clusters = [
|
||
_make_cluster(0.5, n_eff=2.0),
|
||
_make_cluster(0.3, n_eff=1.5),
|
||
_make_cluster(-0.2, n_eff=3.0),
|
||
]
|
||
regime = _make_regime()
|
||
result = compute_v3_posterior(clusters=clusters, regime=regime)
|
||
|
||
assert result.n_eff_total == pytest.approx(6.5, abs=1e-9)
|
||
|
||
def test_regime_string_matches_input(self):
|
||
"""regime field should match the input regime's value."""
|
||
regime = _make_regime(MarketRegime.PANIC, gamma=0.70)
|
||
result = compute_v3_posterior(clusters=[], regime=regime)
|
||
assert result.regime == "panic"
|
||
|
||
def test_p_down_is_complement(self):
|
||
"""p_down = 1 - p_up."""
|
||
clusters = [_make_cluster(0.8)]
|
||
regime = _make_regime()
|
||
result = compute_v3_posterior(clusters=clusters, regime=regime)
|
||
|
||
assert result.p_down == pytest.approx(1.0 - result.p_up, abs=1e-10)
|