403 lines
15 KiB
Python
403 lines
15 KiB
Python
"""Unit tests for v3 fractional Kelly sizing and regime-aware stops.
|
||
|
||
Tests Kelly fraction computation, capacity cap enforcement, position minimum
|
||
downgrade, stop/take-profit levels, and trailing stop monotonicity.
|
||
|
||
Requirements validated: 14.1–14.7, 16.1–16.5
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from services.trading.position_sizer import (
|
||
KellySizingResult,
|
||
compute_kelly_sizing,
|
||
compute_reward_ratio,
|
||
)
|
||
from services.trading.stop_loss_manager import (
|
||
TrailingStopResult,
|
||
V3StopLevels,
|
||
compute_trailing_stop,
|
||
compute_v3_stops,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Kelly sizing: negative edge → size = 0 (Req 14.7)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestKellyNegativeEdge:
|
||
"""Negative Kelly fraction forces zero sizing and downgrade."""
|
||
|
||
def test_p_win_03_b_2_negative_edge(self):
|
||
"""p_win=0.3, b=2.0 → f_kelly = (0.3*2 - 0.7)/2 = -0.05 → downgrade."""
|
||
result = compute_kelly_sizing(
|
||
p_win=0.3,
|
||
b=2.0,
|
||
confidence=0.8,
|
||
data_quality=0.9,
|
||
contradiction=0.1,
|
||
max_position_pct=0.10,
|
||
available_caps={"sector_capacity": 0.10, "correlation_capacity": 0.10, "heat_capacity": 0.10},
|
||
)
|
||
assert isinstance(result, KellySizingResult)
|
||
assert result.f_kelly == pytest.approx(-0.05, abs=1e-9)
|
||
assert result.portfolio_pct == 0.0
|
||
assert result.downgrade is True
|
||
assert result.downgrade_reason == "negative_edge"
|
||
|
||
def test_p_win_05_b_1_zero_edge(self):
|
||
"""p_win=0.5, b=1.0 → f_kelly = (0.5*1 - 0.5)/1 = 0.0 → downgrade."""
|
||
result = compute_kelly_sizing(
|
||
p_win=0.5,
|
||
b=1.0,
|
||
confidence=0.8,
|
||
data_quality=0.9,
|
||
contradiction=0.1,
|
||
max_position_pct=0.10,
|
||
available_caps={},
|
||
)
|
||
assert result.f_kelly == pytest.approx(0.0, abs=1e-9)
|
||
assert result.portfolio_pct == 0.0
|
||
assert result.downgrade is True
|
||
assert result.downgrade_reason == "negative_edge"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Kelly sizing: positive edge → bounded size (Req 14.1–14.4)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestKellyPositiveEdge:
|
||
"""Positive Kelly fraction produces a sized position within caps."""
|
||
|
||
def test_p_win_07_b_2_positive_size(self):
|
||
"""p_win=0.7, b=2.0 → f_kelly = (1.4-0.3)/2 = 0.55 → positive size."""
|
||
confidence = 0.8
|
||
data_quality = 0.9
|
||
contradiction = 0.1
|
||
result = compute_kelly_sizing(
|
||
p_win=0.7,
|
||
b=2.0,
|
||
confidence=confidence,
|
||
data_quality=data_quality,
|
||
contradiction=contradiction,
|
||
max_position_pct=0.10,
|
||
available_caps={"sector_capacity": 0.10, "correlation_capacity": 0.10, "heat_capacity": 0.10},
|
||
)
|
||
assert result.f_kelly == pytest.approx(0.55, abs=1e-9)
|
||
# portfolio_pct = 0.55 * 0.25 * 0.8 * 0.9 * (1 - 0.1) = 0.55 * 0.25 * 0.8 * 0.9 * 0.9
|
||
expected_raw = 0.55 * 0.25 * confidence * data_quality * (1.0 - contradiction)
|
||
# Clamped to max_position_pct = 0.10
|
||
expected_pct = min(expected_raw, 0.10)
|
||
assert result.portfolio_pct == pytest.approx(expected_pct, abs=1e-9)
|
||
assert result.portfolio_pct > 0.0
|
||
assert result.portfolio_pct <= 0.10
|
||
assert result.downgrade is False
|
||
assert result.downgrade_reason == ""
|
||
|
||
def test_high_confidence_respects_max_cap(self):
|
||
"""Even with strong edge, portfolio_pct cannot exceed max_position_pct."""
|
||
result = compute_kelly_sizing(
|
||
p_win=0.9,
|
||
b=3.0,
|
||
confidence=1.0,
|
||
data_quality=1.0,
|
||
contradiction=0.0,
|
||
max_position_pct=0.05,
|
||
available_caps={},
|
||
)
|
||
assert result.portfolio_pct <= 0.05
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Cap enforcement: sector, correlation, heat (Req 14.5)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestCapEnforcement:
|
||
"""Capacity constraints cap portfolio_pct."""
|
||
|
||
def test_sector_capacity_caps(self):
|
||
"""sector_capacity=0.02 caps portfolio_pct at 0.02."""
|
||
result = compute_kelly_sizing(
|
||
p_win=0.7,
|
||
b=2.0,
|
||
confidence=0.8,
|
||
data_quality=0.9,
|
||
contradiction=0.1,
|
||
max_position_pct=0.10,
|
||
available_caps={"sector_capacity": 0.02, "correlation_capacity": 0.10, "heat_capacity": 0.10},
|
||
)
|
||
assert result.portfolio_pct <= 0.02
|
||
|
||
def test_correlation_capacity_zero_forces_zero(self):
|
||
"""correlation_capacity=0.0 (avg corr > 0.80) → portfolio_pct = 0."""
|
||
result = compute_kelly_sizing(
|
||
p_win=0.7,
|
||
b=2.0,
|
||
confidence=0.8,
|
||
data_quality=0.9,
|
||
contradiction=0.1,
|
||
max_position_pct=0.10,
|
||
available_caps={"sector_capacity": 0.10, "correlation_capacity": 0.0, "heat_capacity": 0.10},
|
||
)
|
||
# correlation_capacity=0 → min(pct, 0) = 0 → below minimum → downgrade
|
||
assert result.portfolio_pct == 0.0
|
||
assert result.downgrade is True
|
||
assert result.downgrade_reason == "position_below_minimum"
|
||
|
||
def test_heat_capacity_caps(self):
|
||
"""heat_capacity=0.01 caps portfolio_pct at 0.01."""
|
||
result = compute_kelly_sizing(
|
||
p_win=0.7,
|
||
b=2.0,
|
||
confidence=0.8,
|
||
data_quality=0.9,
|
||
contradiction=0.1,
|
||
max_position_pct=0.10,
|
||
available_caps={"sector_capacity": 0.10, "correlation_capacity": 0.10, "heat_capacity": 0.01},
|
||
)
|
||
assert result.portfolio_pct <= 0.01
|
||
|
||
def test_minimum_of_all_caps(self):
|
||
"""portfolio_pct is min of all capacity constraints."""
|
||
result = compute_kelly_sizing(
|
||
p_win=0.7,
|
||
b=2.0,
|
||
confidence=0.8,
|
||
data_quality=0.9,
|
||
contradiction=0.1,
|
||
max_position_pct=0.10,
|
||
available_caps={"sector_capacity": 0.03, "correlation_capacity": 0.05, "heat_capacity": 0.04},
|
||
)
|
||
# Minimum cap is sector_capacity=0.03
|
||
assert result.portfolio_pct <= 0.03
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Position below minimum → downgrade (Req 14.6)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPositionBelowMinimum:
|
||
"""Tiny positions below 0.005 trigger a downgrade."""
|
||
|
||
def test_tiny_position_downgrade(self):
|
||
"""Low data_quality and high contradiction → pct < 0.005 → downgrade."""
|
||
result = compute_kelly_sizing(
|
||
p_win=0.55,
|
||
b=1.5,
|
||
confidence=0.3,
|
||
data_quality=0.3,
|
||
contradiction=0.7,
|
||
max_position_pct=0.10,
|
||
available_caps={},
|
||
)
|
||
# f_kelly = (0.55*1.5 - 0.45)/1.5 = (0.825-0.45)/1.5 = 0.25
|
||
# raw = 0.25 * 0.25 * 0.3 * 0.3 * 0.3 = 0.0016875 < 0.005
|
||
assert result.portfolio_pct == 0.0
|
||
assert result.downgrade is True
|
||
assert result.downgrade_reason == "position_below_minimum"
|
||
|
||
def test_just_above_minimum_no_downgrade(self):
|
||
"""Position at or above 0.005 is not downgraded."""
|
||
result = compute_kelly_sizing(
|
||
p_win=0.7,
|
||
b=2.0,
|
||
confidence=0.8,
|
||
data_quality=0.9,
|
||
contradiction=0.1,
|
||
max_position_pct=0.10,
|
||
available_caps={},
|
||
)
|
||
# f_kelly=0.55, raw=0.55*0.25*0.8*0.9*0.9 = 0.0891 >> 0.005
|
||
assert result.portfolio_pct >= 0.005
|
||
assert result.downgrade is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Reward ratio computation (Req 14.2)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestRewardRatio:
|
||
"""Tests compute_reward_ratio clamping and formula."""
|
||
|
||
def test_reward_ratio_typical(self):
|
||
"""Typical values produce a ratio between 1.2 and 3.0."""
|
||
b = compute_reward_ratio(confidence=0.7, strength=0.5, contradiction=0.2)
|
||
# raw = 1.2 + 2.0*0.7 + 1.0*0.5 - 0.2 = 1.2 + 1.4 + 0.5 - 0.2 = 2.9
|
||
assert b == pytest.approx(2.9, abs=1e-9)
|
||
|
||
def test_reward_ratio_clamp_low(self):
|
||
"""Low confidence/strength and high contradiction → clamped to 1.2."""
|
||
b = compute_reward_ratio(confidence=0.0, strength=0.0, contradiction=1.0)
|
||
# raw = 1.2 + 0 + 0 - 1.0 = 0.2 → clamped to 1.2
|
||
assert b == pytest.approx(1.2, abs=1e-9)
|
||
|
||
def test_reward_ratio_clamp_high(self):
|
||
"""High confidence/strength → clamped to 3.0."""
|
||
b = compute_reward_ratio(confidence=1.0, strength=1.0, contradiction=0.0)
|
||
# raw = 1.2 + 2.0 + 1.0 - 0 = 4.2 → clamped to 3.0
|
||
assert b == pytest.approx(3.0, abs=1e-9)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Stop/TP computation with known inputs (Req 16.1–16.3)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestV3Stops:
|
||
"""Tests for regime-aware stop loss and take profit computation."""
|
||
|
||
def test_known_inputs(self):
|
||
"""entry=100, ATR_pct=0.02, regime_mult=2.0, sigma_h=0.04, b=2.0."""
|
||
result = compute_v3_stops(
|
||
entry_price=100.0,
|
||
atr_pct=0.02,
|
||
regime_atr_mult=2.0,
|
||
sigma_h=0.04,
|
||
reward_ratio=2.0,
|
||
)
|
||
assert isinstance(result, V3StopLevels)
|
||
# stop_distance = max(0.02*2.0, 0.04*1.25, 0.005) = max(0.04, 0.05, 0.005) = 0.05
|
||
assert result.stop_distance_pct == pytest.approx(0.05, abs=1e-9)
|
||
# stop = 100 * (1 - 0.05) = 95
|
||
assert result.stop_loss == pytest.approx(95.0, abs=1e-9)
|
||
# TP = 100 * (1 + 2.0 * 0.05) = 110
|
||
assert result.take_profit == pytest.approx(110.0, abs=1e-9)
|
||
assert result.reward_ratio == pytest.approx(2.0, abs=1e-9)
|
||
|
||
def test_min_stop_distance_enforced(self):
|
||
"""Very low ATR and sigma → min stop distance of 0.005 is enforced."""
|
||
result = compute_v3_stops(
|
||
entry_price=50.0,
|
||
atr_pct=0.001,
|
||
regime_atr_mult=1.0,
|
||
sigma_h=0.001,
|
||
reward_ratio=2.0,
|
||
)
|
||
# max(0.001*1.0, 0.001*1.25, 0.005) = 0.005
|
||
assert result.stop_distance_pct == pytest.approx(0.005, abs=1e-9)
|
||
assert result.stop_loss == pytest.approx(50.0 * (1 - 0.005), abs=1e-9)
|
||
|
||
def test_atr_dominates(self):
|
||
"""High ATR*mult dominates stop distance."""
|
||
result = compute_v3_stops(
|
||
entry_price=200.0,
|
||
atr_pct=0.05,
|
||
regime_atr_mult=2.5,
|
||
sigma_h=0.03,
|
||
reward_ratio=1.5,
|
||
)
|
||
# max(0.05*2.5, 0.03*1.25, 0.005) = max(0.125, 0.0375, 0.005) = 0.125
|
||
assert result.stop_distance_pct == pytest.approx(0.125, abs=1e-9)
|
||
assert result.stop_loss == pytest.approx(200.0 * (1 - 0.125), abs=1e-9)
|
||
assert result.take_profit == pytest.approx(200.0 * (1 + 1.5 * 0.125), abs=1e-9)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Trailing stop monotonicity (Req 16.4–16.5)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestTrailingStopMonotonicity:
|
||
"""Tests trailing stop activation and non-decreasing behavior."""
|
||
|
||
def test_not_activated_below_threshold(self):
|
||
"""Gain < 50% of TP distance → trailing not activated."""
|
||
result = compute_trailing_stop(
|
||
existing_stop=95.0,
|
||
current_price=101.0, # Gain = 1.0, TP distance = 10, 1.0 < 0.5*10
|
||
entry_price=100.0,
|
||
take_profit=110.0,
|
||
atr_pct=0.02,
|
||
trailing_atr_mult=1.5,
|
||
sigma_h=0.04,
|
||
)
|
||
assert isinstance(result, TrailingStopResult)
|
||
assert result.activated is False
|
||
assert result.trailing_stop == 95.0 # Unchanged
|
||
|
||
def test_activated_above_threshold(self):
|
||
"""Gain >= 50% of TP distance → trailing activated."""
|
||
result = compute_trailing_stop(
|
||
existing_stop=95.0,
|
||
current_price=105.0, # Gain = 5.0, TP distance = 10, 5.0 >= 0.5*10
|
||
entry_price=100.0,
|
||
take_profit=110.0,
|
||
atr_pct=0.02,
|
||
trailing_atr_mult=1.5,
|
||
sigma_h=0.04,
|
||
)
|
||
assert result.activated is True
|
||
# trailing_distance = max(0.02*1.5, 0.04*0.75) = max(0.03, 0.03) = 0.03
|
||
# candidate = 105 * (1 - 0.03) = 101.85
|
||
# max(95.0, 101.85) = 101.85
|
||
assert result.trailing_stop == pytest.approx(105.0 * (1 - 0.03), abs=1e-9)
|
||
assert result.trailing_stop > 95.0
|
||
|
||
def test_monotonicity_over_price_sequence(self):
|
||
"""Trailing stop never decreases over a price sequence."""
|
||
# Setup: entry=100, TP=110, existing_stop=95
|
||
entry = 100.0
|
||
take_profit = 110.0
|
||
atr_pct = 0.02
|
||
trailing_atr_mult = 1.5
|
||
sigma_h = 0.04
|
||
prices = [102.0, 105.0, 103.0, 108.0]
|
||
|
||
current_stop = 95.0
|
||
stops_recorded = [current_stop]
|
||
|
||
for price in prices:
|
||
result = compute_trailing_stop(
|
||
existing_stop=current_stop,
|
||
current_price=price,
|
||
entry_price=entry,
|
||
take_profit=take_profit,
|
||
atr_pct=atr_pct,
|
||
trailing_atr_mult=trailing_atr_mult,
|
||
sigma_h=sigma_h,
|
||
)
|
||
current_stop = result.trailing_stop
|
||
stops_recorded.append(current_stop)
|
||
|
||
# Verify monotonically non-decreasing
|
||
for i in range(1, len(stops_recorded)):
|
||
assert stops_recorded[i] >= stops_recorded[i - 1], (
|
||
f"Stop decreased at step {i}: {stops_recorded[i]} < {stops_recorded[i-1]}"
|
||
)
|
||
|
||
def test_trailing_stop_never_below_existing(self):
|
||
"""Even with price drop, trailing stop stays at existing_stop."""
|
||
result = compute_trailing_stop(
|
||
existing_stop=102.0,
|
||
current_price=105.0,
|
||
entry_price=100.0,
|
||
take_profit=110.0,
|
||
atr_pct=0.05, # Large ATR → candidate might be below existing
|
||
trailing_atr_mult=2.0,
|
||
sigma_h=0.08,
|
||
)
|
||
# trailing_distance = max(0.05*2.0, 0.08*0.75) = max(0.10, 0.06) = 0.10
|
||
# candidate = 105 * (1 - 0.10) = 94.5
|
||
# max(102.0, 94.5) = 102.0
|
||
assert result.trailing_stop == pytest.approx(102.0, abs=1e-9)
|
||
assert result.activated is True
|
||
|
||
def test_exact_50pct_threshold_activates(self):
|
||
"""Gain exactly at 50% of TP distance activates trailing."""
|
||
# TP distance = 10, so gain must be >= 5.0
|
||
result = compute_trailing_stop(
|
||
existing_stop=95.0,
|
||
current_price=105.0, # Gain = 5.0 = 0.50 * 10
|
||
entry_price=100.0,
|
||
take_profit=110.0,
|
||
atr_pct=0.02,
|
||
trailing_atr_mult=1.5,
|
||
sigma_h=0.04,
|
||
)
|
||
assert result.activated is True
|