stonks-oracle/tests/test_pbt_risk_tier_controller.py

"""Property-based tests for the Risk Tier Controller.

Feature: autonomous-trading-engine

Tests Property 12 from the design specification: risk tier auto-adjustment
conditions. Verifies downgrade, upgrade, and no-change behaviour across all
three starting tiers with randomly generated performance metrics.

**Validates: Requirements 5.3, 5.4**
"""
from __future__ import annotations

from datetime import datetime

from hypothesis import given, settings, assume
from hypothesis import strategies as st

from services.trading.risk_tier_controller import RiskTierController, TIER_ORDER
from services.trading.models import PerformanceMetrics


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _build_metrics(
    *,
    win_rate: float,
    current_drawdown_pct: float,
) -> PerformanceMetrics:
    """Build a PerformanceMetrics instance with the given win_rate and drawdown.

    All other fields are set to neutral defaults that do not affect tier
    evaluation logic.
    """
    return PerformanceMetrics(
        total_portfolio_value=10_000.0,
        active_pool=8_000.0,
        reserve_pool=2_000.0,
        unrealized_pnl=0.0,
        realized_pnl=0.0,
        daily_pnl=0.0,
        win_count=0,
        loss_count=0,
        win_rate=win_rate,
        avg_win=0.0,
        avg_loss=0.0,
        profit_factor=1.0,
        sharpe_ratio=0.0,
        max_drawdown=0.0,
        current_drawdown_pct=current_drawdown_pct,
        portfolio_heat=0.0,
    )


# ---------------------------------------------------------------------------
# Hypothesis strategies
# ---------------------------------------------------------------------------

_win_rate_st = st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False)
_drawdown_st = st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False)
_reserve_pct_st = st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False)
_tier_st = st.sampled_from(TIER_ORDER)


# ---------------------------------------------------------------------------
# Property 12: Risk tier auto-adjustment conditions
# ---------------------------------------------------------------------------


@settings(max_examples=100)
@given(
    current_tier=_tier_st,
    win_rate=st.floats(min_value=0.0, max_value=0.39, allow_nan=False, allow_infinity=False),
    drawdown=st.floats(min_value=0.0, max_value=0.15, allow_nan=False, allow_infinity=False),
    reserve_pct=_reserve_pct_st,
)
def test_downgrade_on_low_win_rate(
    current_tier: str,
    win_rate: float,
    drawdown: float,
    reserve_pct: float,
) -> None:
    """Tier downgrades when win rate < 40% (regardless of drawdown).

    **Validates: Requirements 5.3**
    """
    assume(win_rate < 0.40)
    controller = RiskTierController()
    metrics = _build_metrics(win_rate=win_rate, current_drawdown_pct=drawdown)
    result = controller.evaluate(current_tier, metrics, reserve_pct)

    current_index = TIER_ORDER.index(current_tier)
    if current_index > 0:
        assert result == TIER_ORDER[current_index - 1], (
            f"Expected downgrade from {current_tier} to {TIER_ORDER[current_index - 1]}, "
            f"got {result} (win_rate={win_rate})"
        )
    else:
        # Already at conservative — no change possible
        assert result is None, (
            f"Expected None (already at conservative), got {result}"
        )


@settings(max_examples=100)
@given(
    current_tier=_tier_st,
    win_rate=_win_rate_st,
    drawdown=st.floats(min_value=0.151, max_value=1.0, allow_nan=False, allow_infinity=False),
    reserve_pct=_reserve_pct_st,
)
def test_downgrade_on_high_drawdown(
    current_tier: str,
    win_rate: float,
    drawdown: float,
    reserve_pct: float,
) -> None:
    """Tier downgrades when drawdown > 15% (regardless of win rate).

    **Validates: Requirements 5.3**
    """
    assume(drawdown > 0.15)
    controller = RiskTierController()
    metrics = _build_metrics(win_rate=win_rate, current_drawdown_pct=drawdown)
    result = controller.evaluate(current_tier, metrics, reserve_pct)

    current_index = TIER_ORDER.index(current_tier)
    if current_index > 0:
        assert result == TIER_ORDER[current_index - 1], (
            f"Expected downgrade from {current_tier} to {TIER_ORDER[current_index - 1]}, "
            f"got {result} (drawdown={drawdown})"
        )
    else:
        assert result is None, (
            f"Expected None (already at conservative), got {result}"
        )


@settings(max_examples=100)
@given(
    current_tier=_tier_st,
    win_rate=st.floats(min_value=0.551, max_value=1.0, allow_nan=False, allow_infinity=False),
    reserve_pct=st.floats(min_value=0.201, max_value=1.0, allow_nan=False, allow_infinity=False),
    drawdown=st.floats(min_value=0.0, max_value=0.049, allow_nan=False, allow_infinity=False),
)
def test_upgrade_when_all_conditions_met(
    current_tier: str,
    win_rate: float,
    reserve_pct: float,
    drawdown: float,
) -> None:
    """Tier upgrades when win rate > 55% AND reserve > 20% AND drawdown < 5%.

    **Validates: Requirements 5.4**
    """
    assume(win_rate > 0.55)
    assume(reserve_pct > 0.20)
    assume(drawdown < 0.05)
    controller = RiskTierController()
    metrics = _build_metrics(win_rate=win_rate, current_drawdown_pct=drawdown)
    result = controller.evaluate(current_tier, metrics, reserve_pct)

    current_index = TIER_ORDER.index(current_tier)
    if current_index < len(TIER_ORDER) - 1:
        assert result == TIER_ORDER[current_index + 1], (
            f"Expected upgrade from {current_tier} to {TIER_ORDER[current_index + 1]}, "
            f"got {result} (win_rate={win_rate}, reserve_pct={reserve_pct}, drawdown={drawdown})"
        )
    else:
        # Already at aggressive — no change possible
        assert result is None, (
            f"Expected None (already at aggressive), got {result}"
        )


@settings(max_examples=100)
@given(
    current_tier=_tier_st,
    win_rate=st.floats(min_value=0.40, max_value=0.55, allow_nan=False, allow_infinity=False),
    drawdown=st.floats(min_value=0.0, max_value=0.15, allow_nan=False, allow_infinity=False),
    reserve_pct=_reserve_pct_st,
)
def test_no_change_when_neither_condition_met(
    current_tier: str,
    win_rate: float,
    drawdown: float,
    reserve_pct: float,
) -> None:
    """Tier stays the same when neither downgrade nor upgrade conditions are met.

    The "neutral zone" is: win_rate in [0.40, 0.55] AND drawdown in [0.0, 0.15].
    In this zone, upgrade conditions cannot all be satisfied (win_rate <= 0.55),
    and downgrade conditions are not met (win_rate >= 0.40 AND drawdown <= 0.15).

    **Validates: Requirements 5.3, 5.4**
    """
    assume(win_rate >= 0.40)
    assume(win_rate <= 0.55)
    assume(drawdown <= 0.15)
    controller = RiskTierController()
    metrics = _build_metrics(win_rate=win_rate, current_drawdown_pct=drawdown)
    result = controller.evaluate(current_tier, metrics, reserve_pct)

    assert result is None, (
        f"Expected no change (None) for {current_tier}, "
        f"got {result} (win_rate={win_rate}, drawdown={drawdown}, reserve_pct={reserve_pct})"
    )


@settings(max_examples=100)
@given(
    win_rate=_win_rate_st,
    drawdown=_drawdown_st,
    reserve_pct=_reserve_pct_st,
)
def test_tier_never_below_conservative(
    win_rate: float,
    drawdown: float,
    reserve_pct: float,
) -> None:
    """Starting from conservative, the tier never goes below conservative.

    **Validates: Requirements 5.3, 5.4**
    """
    controller = RiskTierController()
    metrics = _build_metrics(win_rate=win_rate, current_drawdown_pct=drawdown)
    result = controller.evaluate("conservative", metrics, reserve_pct)

    if result is not None:
        assert result in TIER_ORDER, f"Unknown tier: {result}"
        assert TIER_ORDER.index(result) >= 0, (
            f"Tier went below conservative: {result}"
        )


@settings(max_examples=100)
@given(
    win_rate=_win_rate_st,
    drawdown=_drawdown_st,
    reserve_pct=_reserve_pct_st,
)
def test_tier_never_above_aggressive(
    win_rate: float,
    drawdown: float,
    reserve_pct: float,
) -> None:
    """Starting from aggressive, the tier never goes above aggressive.

    **Validates: Requirements 5.3, 5.4**
    """
    controller = RiskTierController()
    metrics = _build_metrics(win_rate=win_rate, current_drawdown_pct=drawdown)
    result = controller.evaluate("aggressive", metrics, reserve_pct)

    if result is not None:
        assert result in TIER_ORDER, f"Unknown tier: {result}"
        assert TIER_ORDER.index(result) <= len(TIER_ORDER) - 1, (
            f"Tier went above aggressive: {result}"
        )


@settings(max_examples=100)
@given(
    current_tier=_tier_st,
    win_rate=_win_rate_st,
    drawdown=_drawdown_st,
    reserve_pct=_reserve_pct_st,
)
def test_result_is_always_valid_tier_or_none(
    current_tier: str,
    win_rate: float,
    drawdown: float,
    reserve_pct: float,
) -> None:
    """The evaluate result is always None or a valid tier name from TIER_ORDER.

    **Validates: Requirements 5.3, 5.4**
    """
    controller = RiskTierController()
    metrics = _build_metrics(win_rate=win_rate, current_drawdown_pct=drawdown)
    result = controller.evaluate(current_tier, metrics, reserve_pct)

    assert result is None or result in TIER_ORDER, (
        f"Invalid result: {result} (expected None or one of {TIER_ORDER})"
    )


@settings(max_examples=100)
@given(
    current_tier=_tier_st,
    win_rate=_win_rate_st,
    drawdown=_drawdown_st,
    reserve_pct=_reserve_pct_st,
)
def test_tier_changes_by_at_most_one_level(
    current_tier: str,
    win_rate: float,
    drawdown: float,
    reserve_pct: float,
) -> None:
    """A single evaluation can only move the tier by at most one level.

    **Validates: Requirements 5.3, 5.4**
    """
    controller = RiskTierController()
    metrics = _build_metrics(win_rate=win_rate, current_drawdown_pct=drawdown)
    result = controller.evaluate(current_tier, metrics, reserve_pct)

    if result is not None:
        current_index = TIER_ORDER.index(current_tier)
        new_index = TIER_ORDER.index(result)
        assert abs(new_index - current_index) == 1, (
            f"Tier jumped more than one level: {current_tier} → {result}"
        )