"""Property-based tests for the Portfolio Rebalancer. Feature: autonomous-trading-engine Property 17: Portfolio rebalancing generates correct sell orders. """ from __future__ import annotations from hypothesis import assume, given, settings from hypothesis import strategies as st from services.trading.models import OpenPosition, RiskTierConfig from services.trading.rebalancer import PortfolioRebalancer # --------------------------------------------------------------------------- # Hypothesis strategies # --------------------------------------------------------------------------- SECTORS = ["Technology", "Healthcare", "Energy", "Financials", "Consumer"] def _risk_tier_config_strategy() -> st.SearchStrategy[RiskTierConfig]: """Generate random RiskTierConfig objects with valid parameter ranges.""" return st.builds( RiskTierConfig, name=st.sampled_from(["conservative", "moderate", "aggressive"]), min_confidence=st.floats(min_value=0.10, max_value=0.95, allow_nan=False, allow_infinity=False), max_position_pct=st.floats(min_value=0.05, max_value=0.30, allow_nan=False, allow_infinity=False), stop_loss_atr_multiplier=st.floats(min_value=0.5, max_value=5.0, allow_nan=False, allow_infinity=False), reward_risk_ratio=st.floats(min_value=0.5, max_value=5.0, allow_nan=False, allow_infinity=False), max_sector_pct=st.floats(min_value=0.10, max_value=0.50, allow_nan=False, allow_infinity=False), max_portfolio_heat=st.floats(min_value=0.05, max_value=0.50, allow_nan=False, allow_infinity=False), ) def _open_position_strategy( sector: st.SearchStrategy[str] | None = None, market_value: st.SearchStrategy[float] | None = None, signal_confidence: st.SearchStrategy[float] | None = None, ) -> st.SearchStrategy[OpenPosition]: """Generate random OpenPosition objects.""" return st.builds( OpenPosition, ticker=st.from_regex(r"[A-Z]{3,5}", fullmatch=True), quantity=st.integers(min_value=1, max_value=200), entry_price=st.floats(min_value=5.0, max_value=500.0, allow_nan=False, allow_infinity=False), current_price=st.floats(min_value=5.0, max_value=500.0, allow_nan=False, allow_infinity=False), unrealized_pnl=st.floats(min_value=-1000.0, max_value=1000.0, allow_nan=False, allow_infinity=False), market_value=market_value if market_value is not None else st.floats( min_value=50.0, max_value=5000.0, allow_nan=False, allow_infinity=False, ), sector=sector if sector is not None else st.sampled_from(SECTORS), stop_loss_price=st.floats(min_value=1.0, max_value=400.0, allow_nan=False, allow_infinity=False), take_profit_price=st.floats(min_value=10.0, max_value=1000.0, allow_nan=False, allow_infinity=False), signal_confidence=signal_confidence if signal_confidence is not None else st.floats( min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False, ), is_micro_trade=st.just(False), ) # --------------------------------------------------------------------------- # Property 17: Portfolio rebalancing generates correct sell orders # **Validates: Requirements 8.2, 8.3** # --------------------------------------------------------------------------- class TestProperty17PortfolioRebalancing: """Property 17: Portfolio rebalancing generates correct sell orders. **Validates: Requirements 8.2, 8.3** """ rebalancer = PortfolioRebalancer() @settings(max_examples=100) @given( risk_tier=_risk_tier_config_strategy(), active_pool=st.floats(min_value=1000.0, max_value=50000.0, allow_nan=False, allow_infinity=False), excess_factor=st.floats(min_value=1.1, max_value=3.0, allow_nan=False, allow_infinity=False), current_price=st.floats(min_value=5.0, max_value=200.0, allow_nan=False, allow_infinity=False), ) def test_sell_order_generated_for_over_concentrated_position( self, risk_tier: RiskTierConfig, active_pool: float, excess_factor: float, current_price: float, ) -> None: """A sell order is generated when a single stock exceeds max_position_pct.""" max_dollars = risk_tier.max_position_pct * active_pool over_value = max_dollars * excess_factor quantity = max(1, int(over_value / current_price)) actual_market_value = quantity * current_price # Only test when the position actually exceeds the limit assume(actual_market_value > max_dollars) # Ensure we have enough shares to sell at least 1 assume(int((actual_market_value - max_dollars) / current_price) >= 1) pos = OpenPosition( ticker="OVER", quantity=quantity, entry_price=current_price, current_price=current_price, unrealized_pnl=0.0, market_value=actual_market_value, sector="Technology", stop_loss_price=current_price * 0.9, take_profit_price=current_price * 1.2, signal_confidence=0.7, ) orders = self.rebalancer.evaluate( positions=[pos], risk_tier=risk_tier, active_pool=active_pool, ) assert len(orders) >= 1 over_order = next(o for o in orders if o.ticker == "OVER") assert over_order.action == "sell" assert over_order.quantity >= 1 assert over_order.tag == "rebalance" # After selling, the remaining value should be within the limit remaining_value = actual_market_value - (over_order.quantity * current_price) assert remaining_value <= max_dollars + current_price, ( f"Remaining value {remaining_value} still exceeds limit {max_dollars}" ) @settings(max_examples=100) @given( risk_tier=_risk_tier_config_strategy(), active_pool=st.floats(min_value=1000.0, max_value=50000.0, allow_nan=False, allow_infinity=False), within_factor=st.floats(min_value=0.1, max_value=0.99, allow_nan=False, allow_infinity=False), ) def test_no_sell_order_when_within_limits( self, risk_tier: RiskTierConfig, active_pool: float, within_factor: float, ) -> None: """No sell orders when all positions are within limits.""" max_dollars = risk_tier.max_position_pct * active_pool position_value = max_dollars * within_factor current_price = 10.0 quantity = max(1, int(position_value / current_price)) pos = OpenPosition( ticker="OK", quantity=quantity, entry_price=current_price, current_price=current_price, unrealized_pnl=0.0, market_value=quantity * current_price, sector="Technology", stop_loss_price=9.0, take_profit_price=12.0, signal_confidence=0.7, ) # Ensure the position is actually within limits assume(pos.market_value <= max_dollars) # Also ensure sector is within limits max_sector = risk_tier.max_sector_pct * active_pool assume(pos.market_value <= max_sector) orders = self.rebalancer.evaluate( positions=[pos], risk_tier=risk_tier, active_pool=active_pool, ) assert len(orders) == 0 @settings(max_examples=100) @given( active_pool=st.floats(min_value=5000.0, max_value=50000.0, allow_nan=False, allow_infinity=False), conf_low=st.floats(min_value=0.1, max_value=0.4, allow_nan=False, allow_infinity=False), conf_high=st.floats(min_value=0.6, max_value=0.9, allow_nan=False, allow_infinity=False), ) def test_lowest_confidence_sold_first_for_sector_rebalancing( self, active_pool: float, conf_low: float, conf_high: float, ) -> None: """Lowest-confidence positions are targeted first for sector rebalancing.""" assume(conf_low < conf_high) risk_tier = RiskTierConfig( name="moderate", min_confidence=0.5, max_position_pct=0.50, # High so single-stock check doesn't trigger stop_loss_atr_multiplier=2.0, reward_risk_ratio=1.5, max_sector_pct=0.20, # Low sector limit to trigger rebalancing max_portfolio_heat=0.50, ) max_sector_dollars = risk_tier.max_sector_pct * active_pool # Each position is 60% of the sector limit, so two together exceed it per_position_value = max_sector_dollars * 0.6 current_price = 50.0 quantity = max(1, int(per_position_value / current_price)) actual_value = quantity * current_price # Ensure two positions together exceed the sector limit assume(actual_value * 2 > max_sector_dollars) # Ensure each position alone is within the single-stock limit assume(actual_value <= risk_tier.max_position_pct * active_pool) pos_low = OpenPosition( ticker="LOW", quantity=quantity, entry_price=current_price, current_price=current_price, unrealized_pnl=0.0, market_value=actual_value, sector="Technology", stop_loss_price=45.0, take_profit_price=60.0, signal_confidence=conf_low, ) pos_high = OpenPosition( ticker="HIGH", quantity=quantity, entry_price=current_price, current_price=current_price, unrealized_pnl=0.0, market_value=actual_value, sector="Technology", stop_loss_price=45.0, take_profit_price=60.0, signal_confidence=conf_high, ) orders = self.rebalancer.evaluate( positions=[pos_low, pos_high], risk_tier=risk_tier, active_pool=active_pool, ) # Should have at least one order assert len(orders) >= 1 # The first order should target the lowest-confidence position tickers_ordered = [o.ticker for o in orders] if "LOW" in tickers_ordered and "HIGH" in tickers_ordered: # If both are being sold, LOW should have more shares sold low_order = next(o for o in orders if o.ticker == "LOW") high_order = next(o for o in orders if o.ticker == "HIGH") assert low_order.quantity >= high_order.quantity elif len(tickers_ordered) == 1: # If only one is being sold, it should be the low-confidence one assert "LOW" in tickers_ordered @settings(max_examples=100) @given( num_positions=st.integers(min_value=11, max_value=15), max_positions=st.integers(min_value=5, max_value=10), ) def test_excess_positions_sold_lowest_confidence_first( self, num_positions: int, max_positions: int, ) -> None: """When exceeding max positions, lowest-confidence positions are sold first.""" assume(num_positions > max_positions) active_pool = 100000.0 # Large pool so no single-stock/sector triggers risk_tier = RiskTierConfig( name="moderate", min_confidence=0.5, max_position_pct=0.50, stop_loss_atr_multiplier=2.0, reward_risk_ratio=1.5, max_sector_pct=0.90, max_portfolio_heat=0.50, ) positions = [] for i in range(num_positions): conf = (i + 1) / (num_positions + 1) # Increasing confidence pos = OpenPosition( ticker=f"T{i:02d}", quantity=10, entry_price=50.0, current_price=50.0, unrealized_pnl=0.0, market_value=500.0, sector=SECTORS[i % len(SECTORS)], stop_loss_price=45.0, take_profit_price=60.0, signal_confidence=conf, ) positions.append(pos) orders = self.rebalancer.evaluate( positions=positions, risk_tier=risk_tier, active_pool=active_pool, max_positions=max_positions, ) assert len(orders) >= num_positions - max_positions # Verify that the sold positions have lower confidence than the kept ones sold_tickers = {o.ticker for o in orders} sold_confs = [p.signal_confidence for p in positions if p.ticker in sold_tickers] kept_confs = [p.signal_confidence for p in positions if p.ticker not in sold_tickers] if sold_confs and kept_confs: assert max(sold_confs) <= max(kept_confs), ( f"Sold max conf {max(sold_confs)} > kept max conf {max(kept_confs)}" ) @settings(max_examples=100) @given( active_pool=st.floats(min_value=1000.0, max_value=50000.0, allow_nan=False, allow_infinity=False), ) def test_empty_portfolio_returns_no_orders(self, active_pool: float) -> None: """Empty portfolio produces no rebalancing orders.""" risk_tier = RiskTierConfig( name="moderate", min_confidence=0.5, max_position_pct=0.10, stop_loss_atr_multiplier=2.0, reward_risk_ratio=1.5, max_sector_pct=0.30, max_portfolio_heat=0.20, ) orders = self.rebalancer.evaluate( positions=[], risk_tier=risk_tier, active_pool=active_pool, ) assert orders == [] @settings(max_examples=100) @given( risk_tier=_risk_tier_config_strategy(), ) def test_all_orders_are_sell_with_rebalance_tag( self, risk_tier: RiskTierConfig, ) -> None: """All rebalancing orders have action='sell' and tag='rebalance'.""" active_pool = 5000.0 max_dollars = risk_tier.max_position_pct * active_pool # Create an over-concentrated position over_value = max_dollars * 2.0 current_price = 50.0 quantity = max(1, int(over_value / current_price)) pos = OpenPosition( ticker="BIG", quantity=quantity, entry_price=current_price, current_price=current_price, unrealized_pnl=0.0, market_value=quantity * current_price, sector="Technology", stop_loss_price=45.0, take_profit_price=60.0, signal_confidence=0.5, ) orders = self.rebalancer.evaluate( positions=[pos], risk_tier=risk_tier, active_pool=active_pool, ) for order in orders: assert order.action == "sell" assert order.tag == "rebalance" assert order.quantity >= 1 assert len(order.reason) > 0