"""Property-based tests for v3 Kelly sizing and regime-aware stops. Validates: - Property 11: Fractional Kelly sizing is bounded and respects negative edge - Property 18: Stop loss is below entry price and take profit is above - Property 19: Trailing stop never decreases Requirements: 14.4, 14.7, 16.2, 16.3, 16.5, 21.8, 21.9 """ from __future__ import annotations from hypothesis import given, settings from hypothesis import strategies as st 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, ) # --------------------------------------------------------------------------- # Strategies # --------------------------------------------------------------------------- # Kelly sizing inputs p_up_values = st.floats(min_value=0.01, max_value=0.99, allow_nan=False, allow_infinity=False) b_values = st.floats(min_value=1.2, max_value=3.0, allow_nan=False, allow_infinity=False) confidence_values = st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False) data_quality_values = st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False) contradiction_values = st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False) max_position_pct_values = st.floats(min_value=0.001, max_value=0.50, allow_nan=False, allow_infinity=False) # Stop loss inputs entry_price_values = st.floats(min_value=0.01, max_value=100000.0, allow_nan=False, allow_infinity=False) stop_distance_pct_values = st.floats(min_value=0.005, max_value=0.999, allow_nan=False, allow_infinity=False) reward_ratio_values = st.floats(min_value=1.2, max_value=5.0, allow_nan=False, allow_infinity=False) # Trailing stop price sequences price_values = st.floats(min_value=1.0, max_value=10000.0, allow_nan=False, allow_infinity=False) atr_pct_values = st.floats(min_value=0.005, max_value=0.20, allow_nan=False, allow_infinity=False) trailing_atr_mult_values = st.floats(min_value=0.5, max_value=3.0, allow_nan=False, allow_infinity=False) sigma_h_values = st.floats(min_value=0.005, max_value=0.50, allow_nan=False, allow_infinity=False) # --------------------------------------------------------------------------- # Feature: math-core-v3-engine, Property 11: Fractional Kelly sizing is # bounded and respects negative edge # --------------------------------------------------------------------------- # **Validates: Requirements 14.4, 14.7, 21.8, 21.9** @settings(max_examples=100) @given( p_up=p_up_values, b=b_values, confidence=confidence_values, data_quality=data_quality_values, contradiction=contradiction_values, max_position_pct=max_position_pct_values, ) def test_property_11_kelly_sizing_bounded_and_respects_negative_edge( p_up: float, b: float, confidence: float, data_quality: float, contradiction: float, max_position_pct: float, ) -> None: """Property 11: Fractional Kelly sizing is bounded and respects negative edge. For any valid inputs (P_up in (0,1), b in [1.2, 3.0], confidence in [0,1], data_quality in [0,1], contradiction in [0,1], max_position_pct > 0), the computed portfolio_pct SHALL be in [0, max_position_pct]. When f_kelly = (P_up * b - (1 - P_up)) / b <= 0, portfolio_pct SHALL be exactly 0. """ # Use generous capacity caps that don't constrain available_caps = { "sector_capacity": max_position_pct, "correlation_capacity": max_position_pct, "heat_capacity": max_position_pct, } result = compute_kelly_sizing( p_win=p_up, b=b, confidence=confidence, data_quality=data_quality, contradiction=contradiction, max_position_pct=max_position_pct, available_caps=available_caps, ) # portfolio_pct must be in [0, max_position_pct] assert 0.0 <= result.portfolio_pct <= max_position_pct, ( f"portfolio_pct={result.portfolio_pct} not in [0, {max_position_pct}] " f"for p_up={p_up}, b={b}, conf={confidence}, dq={data_quality}, " f"contra={contradiction}" ) # When f_kelly <= 0, portfolio_pct must be exactly 0 f_kelly = (p_up * b - (1.0 - p_up)) / b if f_kelly <= 0: assert result.portfolio_pct == 0.0, ( f"portfolio_pct={result.portfolio_pct} should be 0.0 when " f"f_kelly={f_kelly} <= 0 (p_up={p_up}, b={b})" ) assert result.downgrade is True, ( f"downgrade should be True when f_kelly={f_kelly} <= 0" ) assert result.downgrade_reason == "negative_edge", ( f"downgrade_reason should be 'negative_edge' when f_kelly={f_kelly} <= 0, " f"got '{result.downgrade_reason}'" ) # --------------------------------------------------------------------------- # Feature: math-core-v3-engine, Property 18: Stop loss is below entry price # and take profit is above # --------------------------------------------------------------------------- # **Validates: Requirements 16.2, 16.3** @settings(max_examples=100) @given( entry_price=entry_price_values, atr_pct=atr_pct_values, regime_atr_mult=st.floats(min_value=0.5, max_value=3.0, allow_nan=False, allow_infinity=False), sigma_h=sigma_h_values, reward_ratio=reward_ratio_values, ) def test_property_18_stop_loss_below_entry_take_profit_above( entry_price: float, atr_pct: float, regime_atr_mult: float, sigma_h: float, reward_ratio: float, ) -> None: """Property 18: Stop loss is below entry price and take profit is above. For any entry_price > 0, stop_distance_pct in [0.005, 1.0), and reward ratio b >= 1.2, the computed stop_loss SHALL be less than entry_price and take_profit SHALL be greater than entry_price. """ result = compute_v3_stops( entry_price=entry_price, atr_pct=atr_pct, regime_atr_mult=regime_atr_mult, sigma_h=sigma_h, reward_ratio=reward_ratio, ) # stop_distance_pct should be at least 0.005 (the min floor) assert result.stop_distance_pct >= 0.005, ( f"stop_distance_pct={result.stop_distance_pct} should be >= 0.005" ) # Stop loss must be strictly below entry price assert result.stop_loss < entry_price, ( f"stop_loss={result.stop_loss} should be < entry_price={entry_price} " f"(stop_distance_pct={result.stop_distance_pct})" ) # Take profit must be strictly above entry price assert result.take_profit > entry_price, ( f"take_profit={result.take_profit} should be > entry_price={entry_price} " f"(reward_ratio={reward_ratio}, stop_distance_pct={result.stop_distance_pct})" ) # --------------------------------------------------------------------------- # Feature: math-core-v3-engine, Property 19: Trailing stop never decreases # --------------------------------------------------------------------------- # **Validates: Requirements 16.5** @settings(max_examples=100) @given( entry_price=st.floats(min_value=10.0, max_value=1000.0, allow_nan=False, allow_infinity=False), atr_pct=atr_pct_values, trailing_atr_mult=trailing_atr_mult_values, sigma_h=sigma_h_values, price_moves=st.lists( st.floats(min_value=0.0, max_value=2.0, allow_nan=False, allow_infinity=False), min_size=3, max_size=20, ), ) def test_property_19_trailing_stop_never_decreases( entry_price: float, atr_pct: float, trailing_atr_mult: float, sigma_h: float, price_moves: list[float], ) -> None: """Property 19: Trailing stop never decreases. For any sequence of current prices and trailing stop computations, each new trailing_stop value SHALL be >= the previous trailing_stop value (monotonically non-decreasing). """ # Set up a take profit that's achievable reward_ratio = 2.0 stop_distance_pct = max(atr_pct * 1.5, sigma_h * 1.25, 0.005) take_profit = entry_price * (1.0 + reward_ratio * stop_distance_pct) # Start with an initial stop below entry existing_stop = entry_price * (1.0 - stop_distance_pct) # Generate a sequence of prices that move upward from entry # (scaled by price_moves multiplied by stop distance to be meaningful) previous_trailing_stop = existing_stop for move_factor in price_moves: # Price moves upward from entry by some fraction of TP distance tp_distance = take_profit - entry_price current_price = entry_price + move_factor * tp_distance result = compute_trailing_stop( existing_stop=previous_trailing_stop, current_price=current_price, entry_price=entry_price, take_profit=take_profit, atr_pct=atr_pct, trailing_atr_mult=trailing_atr_mult, sigma_h=sigma_h, ) # The trailing stop must never decrease (monotonically non-decreasing) assert result.trailing_stop >= previous_trailing_stop, ( f"trailing_stop={result.trailing_stop} decreased from " f"previous={previous_trailing_stop} at current_price={current_price}, " f"entry={entry_price}, tp={take_profit}" ) # Update for next iteration previous_trailing_stop = result.trailing_stop # --------------------------------------------------------------------------- # Additional imports for heat and tier adjustment tests # --------------------------------------------------------------------------- from services.risk.engine import ( TierMetrics, check_heat_capacity, compute_portfolio_heat, evaluate_tier_adjustment, ) # --------------------------------------------------------------------------- # Strategies for heat and tier tests # --------------------------------------------------------------------------- # Portfolio heat inputs position_value_strategy = st.floats( min_value=100.0, max_value=1_000_000.0, allow_nan=False, allow_infinity=False ) stop_distance_strategy = st.floats( min_value=0.005, max_value=0.50, allow_nan=False, allow_infinity=False ) max_heat_pct_strategy = st.floats( min_value=0.01, max_value=0.50, allow_nan=False, allow_infinity=False ) portfolio_value_strategy = st.floats( min_value=10_000.0, max_value=10_000_000.0, allow_nan=False, allow_infinity=False ) # Tier metrics inputs profit_factor_strategy = st.floats( min_value=0.0, max_value=5.0, allow_nan=False, allow_infinity=False ) drawdown_strategy = st.floats( min_value=0.0, max_value=0.50, allow_nan=False, allow_infinity=False ) calibration_error_strategy = st.floats( min_value=0.0, max_value=0.50, allow_nan=False, allow_infinity=False ) sharpe_strategy = st.floats( min_value=-3.0, max_value=5.0, allow_nan=False, allow_infinity=False ) n_trades_strategy = st.integers(min_value=0, max_value=200) reserve_pool_strategy = st.floats( min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False ) # --------------------------------------------------------------------------- # Feature: math-core-v3-engine, Property 22: Portfolio heat rejection is # correct # --------------------------------------------------------------------------- # **Validates: Requirements 15.3, 15.5** @settings(max_examples=100) @given( position_values=st.lists( position_value_strategy, min_size=1, max_size=10 ), stop_distances_list=st.lists( stop_distance_strategy, min_size=1, max_size=10 ), new_position_value=position_value_strategy, new_stop_distance=stop_distance_strategy, max_heat_pct=max_heat_pct_strategy, portfolio_value=portfolio_value_strategy, ) def test_property_22_portfolio_heat_rejection_is_correct( position_values: list[float], stop_distances_list: list[float], new_position_value: float, new_stop_distance: float, max_heat_pct: float, portfolio_value: float, ) -> None: """Property 22: Portfolio heat rejection is correct. For any set of open positions with stop distances, if the sum of (position_value × stop_distance_pct) exceeds max_portfolio_heat × portfolio_value, then new position entry SHALL be rejected. """ # Align list lengths (use shorter of the two) n = min(len(position_values), len(stop_distances_list)) position_values = position_values[:n] stop_distances_list = stop_distances_list[:n] # Build positions and stop_distances dicts positions = [] stop_distances_dict: dict[str, float] = {} for i in range(n): ticker = f"TICK{i}" positions.append({"ticker": ticker, "position_value": position_values[i]}) stop_distances_dict[ticker] = stop_distances_list[i] # Compute current portfolio heat current_heat = compute_portfolio_heat(positions, stop_distances_dict) # Compute new position risk dollars new_risk_dollars = new_position_value * new_stop_distance # Check heat capacity allowed = check_heat_capacity( current_heat=current_heat, new_risk_dollars=new_risk_dollars, max_heat_pct=max_heat_pct, portfolio_value=portfolio_value, ) # The max allowed heat in dollars max_heat_dollars = max_heat_pct * portfolio_value # Verify: if adding new position exceeds limit, must be rejected (False) if (current_heat + new_risk_dollars) > max_heat_dollars: assert allowed is False, ( f"Expected rejection: current_heat={current_heat:.2f} + " f"new_risk={new_risk_dollars:.2f} = {current_heat + new_risk_dollars:.2f} " f"> max_heat={max_heat_dollars:.2f}, but got allowed=True" ) else: # If within limit, must be allowed (True) assert allowed is True, ( f"Expected allowance: current_heat={current_heat:.2f} + " f"new_risk={new_risk_dollars:.2f} = {current_heat + new_risk_dollars:.2f} " f"<= max_heat={max_heat_dollars:.2f}, but got allowed=False" ) # --------------------------------------------------------------------------- # Feature: math-core-v3-engine, Property 23: Tier auto-adjustment obeys # downgrade-any, upgrade-all logic # --------------------------------------------------------------------------- # **Validates: Requirements 18.3, 18.4** @settings(max_examples=100) @given( profit_factor=profit_factor_strategy, max_drawdown=drawdown_strategy, calibration_error=calibration_error_strategy, sharpe=sharpe_strategy, n_trades=n_trades_strategy, reserve_pool=reserve_pool_strategy, ) def test_property_23_tier_adjustment_downgrade_any_upgrade_all( profit_factor: float, max_drawdown: float, calibration_error: float, sharpe: float, n_trades: int, reserve_pool: float, ) -> None: """Property 23: Tier auto-adjustment obeys downgrade-any, upgrade-all logic. For any TierMetrics, if ANY single downgrade condition is met (profit_factor < 1.0 OR drawdown > 0.12 OR calibration_error > 0.20 OR sharpe < 0), the result SHALL be "downgrade". An "upgrade" SHALL only occur when ALL upgrade conditions are simultaneously met. """ metrics = TierMetrics( profit_factor_30d=profit_factor, max_drawdown_30d=max_drawdown, calibration_error=calibration_error, realized_sharpe_30d=sharpe, n_trades_30d=n_trades, reserve_pool_pct=reserve_pool, ) result = evaluate_tier_adjustment(metrics) # Check downgrade conditions (any single one triggers downgrade) downgrade_triggered = ( profit_factor < 1.0 or max_drawdown > 0.12 or calibration_error > 0.20 or sharpe < 0 ) # Check upgrade conditions (all must be met simultaneously) upgrade_triggered = ( profit_factor > 1.35 and max_drawdown < 0.05 and calibration_error < 0.12 and reserve_pool > 0.20 and n_trades >= 20 ) if downgrade_triggered: assert result == "downgrade", ( f"Expected 'downgrade' when downgrade condition met: " f"pf={profit_factor}, dd={max_drawdown}, " f"cal_err={calibration_error}, sharpe={sharpe}, " f"but got '{result}'" ) elif upgrade_triggered: assert result == "upgrade", ( f"Expected 'upgrade' when all upgrade conditions met: " f"pf={profit_factor}, dd={max_drawdown}, " f"cal_err={calibration_error}, sharpe={sharpe}, " f"n_trades={n_trades}, reserve={reserve_pool}, " f"but got '{result}'" ) else: assert result == "hold", ( f"Expected 'hold' when neither downgrade nor upgrade: " f"pf={profit_factor}, dd={max_drawdown}, " f"cal_err={calibration_error}, sharpe={sharpe}, " f"n_trades={n_trades}, reserve={reserve_pool}, " f"but got '{result}'" )