Celes Renata
c85c0068a2
- Replace all datetime.utcnow() with datetime.now(tz=timezone.utc) across 8 files - Fix 12 failing tests to match current implementation behavior - Fix pytest_plugins in non-top-level conftest (moved to root conftest.py) - Auto-fix 189 lint issues (import sorting, unused imports) - Add CI/CD pipeline infrastructure (ARC, ArgoCD, Kargo manifests) - Add values-beta.yaml and values-paper.yaml for staged deployments - Update GitHub Actions workflow to use self-hosted-gremlin runners - Add integration-test job to CI pipeline Result: 1596 passed, 0 failed, 0 warnings
923 lines
37 KiB
Python
923 lines
37 KiB
Python
"""Property-based tests for the Position Sizer.
|
||
|
||
Feature: autonomous-trading-engine
|
||
|
||
Tests properties 1–5, 7, 19, and 24 from the design specification,
|
||
covering position sizing formula, correlation adjustment, sector exposure,
|
||
diversification bonus, Active Pool computation, earnings proximity,
|
||
portfolio heat, and Active Pool minimum enforcement.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from hypothesis import assume, given, settings
|
||
from hypothesis import strategies as st
|
||
|
||
from services.trading.models import (
|
||
OpenPosition,
|
||
PortfolioState,
|
||
RiskTierConfig,
|
||
)
|
||
from services.trading.position_sizer import PositionSizer
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Hypothesis strategies
|
||
# ---------------------------------------------------------------------------
|
||
|
||
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.02, max_value=0.50, 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.05, max_value=0.60, 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,
|
||
) -> st.SearchStrategy[OpenPosition]:
|
||
"""Generate random OpenPosition objects."""
|
||
sector_st = sector if sector is not None else st.sampled_from(
|
||
["Technology", "Healthcare", "Energy", "Financials", "Consumer"]
|
||
)
|
||
return st.builds(
|
||
OpenPosition,
|
||
ticker=st.from_regex(r"[A-Z]{3,5}", fullmatch=True),
|
||
quantity=st.integers(min_value=1, max_value=100),
|
||
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=st.floats(min_value=10.0, max_value=5000.0, allow_nan=False, allow_infinity=False),
|
||
sector=sector_st,
|
||
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=st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
|
||
is_micro_trade=st.just(False),
|
||
)
|
||
|
||
|
||
def _portfolio_state_strategy(
|
||
positions: st.SearchStrategy[list] | None = None,
|
||
sector_exposure: st.SearchStrategy[dict] | None = None,
|
||
active_pool: st.SearchStrategy[float] | None = None,
|
||
portfolio_heat: st.SearchStrategy[float] | None = None,
|
||
) -> st.SearchStrategy[PortfolioState]:
|
||
"""Generate random PortfolioState objects."""
|
||
return st.builds(
|
||
PortfolioState,
|
||
positions=positions if positions is not None else st.just([]),
|
||
total_value=st.floats(min_value=100.0, max_value=20000.0, allow_nan=False, allow_infinity=False),
|
||
cash=st.floats(min_value=0.0, max_value=10000.0, allow_nan=False, allow_infinity=False),
|
||
active_pool=active_pool if active_pool is not None else st.floats(
|
||
min_value=100.0, max_value=10000.0, allow_nan=False, allow_infinity=False
|
||
),
|
||
reserve_pool=st.floats(min_value=0.0, max_value=5000.0, allow_nan=False, allow_infinity=False),
|
||
sector_exposure=sector_exposure if sector_exposure is not None else st.just({}),
|
||
portfolio_heat=portfolio_heat if portfolio_heat is not None else st.just(0.0),
|
||
open_position_count=st.integers(min_value=0, max_value=10),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 1: Position sizing formula and invariants
|
||
# **Validates: Requirements 2.1, 2.2, 2.3, 2.4, 2.7**
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestProperty1PositionSizingFormula:
|
||
"""Property 1: Position sizing formula and invariants.
|
||
|
||
**Validates: Requirements 2.1, 2.2, 2.3, 2.4, 2.7**
|
||
"""
|
||
|
||
sizer = PositionSizer()
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
confidence=st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
|
||
active_pool=st.floats(min_value=100.0, max_value=10000.0, allow_nan=False, allow_infinity=False),
|
||
price=st.floats(min_value=1.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||
risk_tier=_risk_tier_config_strategy(),
|
||
absolute_cap=st.floats(min_value=10.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_zero_allocation_below_min_confidence(
|
||
self, confidence: float, active_pool: float, price: float,
|
||
risk_tier: RiskTierConfig, absolute_cap: float,
|
||
) -> None:
|
||
"""Confidence below min_confidence yields zero allocation."""
|
||
assume(confidence < risk_tier.min_confidence)
|
||
|
||
portfolio = PortfolioState(active_pool=active_pool)
|
||
result = self.sizer.compute(
|
||
confidence=confidence,
|
||
ticker="TEST",
|
||
sector="Technology",
|
||
current_price=price,
|
||
active_pool=active_pool,
|
||
risk_tier=risk_tier,
|
||
portfolio_state=portfolio,
|
||
correlation_matrix={},
|
||
earnings_calendar={},
|
||
absolute_position_cap=absolute_cap,
|
||
)
|
||
assert result.rejected is True
|
||
assert result.dollar_amount == 0.0
|
||
assert result.share_quantity == 0
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
active_pool=st.floats(min_value=100.0, max_value=10000.0, allow_nan=False, allow_infinity=False),
|
||
price=st.floats(min_value=1.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||
risk_tier=_risk_tier_config_strategy(),
|
||
absolute_cap=st.floats(min_value=10.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_allocation_never_exceeds_max_position_pct_or_cap(
|
||
self, active_pool: float, price: float,
|
||
risk_tier: RiskTierConfig, absolute_cap: float,
|
||
) -> None:
|
||
"""Allocation never exceeds max_position_pct * active_pool or absolute cap."""
|
||
# Use confidence well above threshold to get a non-rejected result
|
||
confidence = min(risk_tier.min_confidence + 0.3, 1.0)
|
||
|
||
portfolio = PortfolioState(active_pool=active_pool)
|
||
result = self.sizer.compute(
|
||
confidence=confidence,
|
||
ticker="TEST",
|
||
sector="Technology",
|
||
current_price=price,
|
||
active_pool=active_pool,
|
||
risk_tier=risk_tier,
|
||
portfolio_state=portfolio,
|
||
correlation_matrix={},
|
||
earnings_calendar={},
|
||
absolute_position_cap=absolute_cap,
|
||
)
|
||
if not result.rejected:
|
||
max_allowed = risk_tier.max_position_pct * active_pool
|
||
# Dollar amount (based on whole shares) should not exceed the cap or max_position_pct
|
||
assert result.dollar_amount <= max_allowed + 0.01, (
|
||
f"dollar_amount {result.dollar_amount} > max_allowed {max_allowed}"
|
||
)
|
||
assert result.dollar_amount <= absolute_cap + 0.01, (
|
||
f"dollar_amount {result.dollar_amount} > absolute_cap {absolute_cap}"
|
||
)
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
active_pool=st.floats(min_value=200.0, max_value=10000.0, allow_nan=False, allow_infinity=False),
|
||
price=st.floats(min_value=1.0, max_value=50.0, allow_nan=False, allow_infinity=False),
|
||
risk_tier=_risk_tier_config_strategy(),
|
||
absolute_cap=st.floats(min_value=50.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_share_quantity_rounded_down(
|
||
self, active_pool: float, price: float,
|
||
risk_tier: RiskTierConfig, absolute_cap: float,
|
||
) -> None:
|
||
"""Share quantity is always rounded down to whole shares (math.floor)."""
|
||
confidence = min(risk_tier.min_confidence + 0.3, 1.0)
|
||
|
||
portfolio = PortfolioState(active_pool=active_pool)
|
||
result = self.sizer.compute(
|
||
confidence=confidence,
|
||
ticker="TEST",
|
||
sector="Technology",
|
||
current_price=price,
|
||
active_pool=active_pool,
|
||
risk_tier=risk_tier,
|
||
portfolio_state=portfolio,
|
||
correlation_matrix={},
|
||
earnings_calendar={},
|
||
absolute_position_cap=absolute_cap,
|
||
)
|
||
if not result.rejected:
|
||
assert result.share_quantity == int(result.share_quantity)
|
||
assert result.share_quantity >= 1
|
||
# Verify it's floor: share_quantity * price should be <= the pre-rounding dollar amount
|
||
assert result.dollar_amount == result.share_quantity * price
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
active_pool=st.floats(min_value=100.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||
risk_tier=_risk_tier_config_strategy(),
|
||
)
|
||
def test_rejection_when_zero_shares(
|
||
self, active_pool: float, risk_tier: RiskTierConfig,
|
||
) -> None:
|
||
"""Trade rejected when rounded share quantity is zero (price too high)."""
|
||
confidence = min(risk_tier.min_confidence + 0.1, 1.0)
|
||
# Use a very high price so dollar_amount / price < 1
|
||
price = active_pool * 10.0
|
||
|
||
portfolio = PortfolioState(active_pool=active_pool)
|
||
result = self.sizer.compute(
|
||
confidence=confidence,
|
||
ticker="TEST",
|
||
sector="Technology",
|
||
current_price=price,
|
||
active_pool=active_pool,
|
||
risk_tier=risk_tier,
|
||
portfolio_state=portfolio,
|
||
correlation_matrix={},
|
||
earnings_calendar={},
|
||
absolute_position_cap=50.0,
|
||
)
|
||
assert result.rejected is True
|
||
assert result.share_quantity == 0
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 2: Correlation-based allocation adjustment
|
||
# **Validates: Requirements 2.5, 9.2, 9.3**
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestProperty2CorrelationAdjustment:
|
||
"""Property 2: Correlation-based allocation adjustment.
|
||
|
||
**Validates: Requirements 2.5, 9.2, 9.3**
|
||
"""
|
||
|
||
sizer = PositionSizer()
|
||
|
||
def _make_portfolio_with_position(
|
||
self, ticker: str = "EXIST", sector: str = "Technology",
|
||
market_value: float = 500.0, active_pool: float = 5000.0,
|
||
) -> PortfolioState:
|
||
"""Create a portfolio with one existing position."""
|
||
pos = OpenPosition(
|
||
ticker=ticker, quantity=10, entry_price=50.0,
|
||
current_price=50.0, unrealized_pnl=0.0,
|
||
market_value=market_value, sector=sector,
|
||
stop_loss_price=45.0, take_profit_price=60.0,
|
||
signal_confidence=0.7,
|
||
)
|
||
return PortfolioState(
|
||
positions=[pos],
|
||
total_value=active_pool + 500.0,
|
||
cash=active_pool - market_value,
|
||
active_pool=active_pool,
|
||
reserve_pool=500.0,
|
||
sector_exposure={sector: market_value},
|
||
portfolio_heat=0.0,
|
||
open_position_count=1,
|
||
)
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
corr=st.floats(min_value=0.51, max_value=0.79, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_allocation_reduced_when_correlation_above_half(self, corr: float) -> None:
|
||
"""Allocation reduced when weighted avg correlation > 0.5."""
|
||
portfolio = self._make_portfolio_with_position(active_pool=5000.0)
|
||
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.50, max_portfolio_heat=0.50,
|
||
)
|
||
corr_matrix = {("NEW", "EXIST"): corr}
|
||
|
||
result_with_corr = self.sizer.compute(
|
||
confidence=0.8, ticker="NEW", sector="Healthcare",
|
||
current_price=10.0, active_pool=5000.0,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix=corr_matrix, earnings_calendar={},
|
||
absolute_position_cap=500.0,
|
||
)
|
||
|
||
result_no_corr = self.sizer.compute(
|
||
confidence=0.8, ticker="NEW", sector="Healthcare",
|
||
current_price=10.0, active_pool=5000.0,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix={("NEW", "EXIST"): 0.0}, earnings_calendar={},
|
||
absolute_position_cap=500.0,
|
||
)
|
||
|
||
if not result_with_corr.rejected and not result_no_corr.rejected:
|
||
assert result_with_corr.dollar_amount <= result_no_corr.dollar_amount
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
corr=st.floats(min_value=0.81, max_value=1.0, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_trade_rejected_when_correlation_above_0_8(self, corr: float) -> None:
|
||
"""Trade rejected when weighted avg correlation > 0.8."""
|
||
portfolio = self._make_portfolio_with_position(active_pool=5000.0)
|
||
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.50, max_portfolio_heat=0.50,
|
||
)
|
||
corr_matrix = {("NEW", "EXIST"): corr}
|
||
|
||
result = self.sizer.compute(
|
||
confidence=0.8, ticker="NEW", sector="Healthcare",
|
||
current_price=10.0, active_pool=5000.0,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix=corr_matrix, earnings_calendar={},
|
||
absolute_position_cap=500.0,
|
||
)
|
||
assert result.rejected is True
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
corr=st.floats(min_value=-1.0, max_value=0.5, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_allocation_unchanged_when_correlation_at_or_below_half(self, corr: float) -> None:
|
||
"""Allocation unchanged when weighted avg correlation <= 0.5."""
|
||
portfolio = self._make_portfolio_with_position(active_pool=5000.0)
|
||
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.50, max_portfolio_heat=0.50,
|
||
)
|
||
corr_matrix_with = {("NEW", "EXIST"): corr}
|
||
corr_matrix_zero = {("NEW", "EXIST"): 0.0}
|
||
|
||
result_with = self.sizer.compute(
|
||
confidence=0.8, ticker="NEW", sector="Healthcare",
|
||
current_price=10.0, active_pool=5000.0,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix=corr_matrix_with, earnings_calendar={},
|
||
absolute_position_cap=500.0,
|
||
)
|
||
result_zero = self.sizer.compute(
|
||
confidence=0.8, ticker="NEW", sector="Healthcare",
|
||
current_price=10.0, active_pool=5000.0,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix=corr_matrix_zero, earnings_calendar={},
|
||
absolute_position_cap=500.0,
|
||
)
|
||
|
||
# Both should produce the same dollar amount (correlation <= 0.5 has no effect)
|
||
if not result_with.rejected and not result_zero.rejected:
|
||
assert result_with.dollar_amount == result_zero.dollar_amount
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
corr_low=st.floats(min_value=0.51, max_value=0.79, allow_nan=False, allow_infinity=False),
|
||
corr_high=st.floats(min_value=0.51, max_value=0.79, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_monotonic_non_increase_with_higher_correlation(
|
||
self, corr_low: float, corr_high: float,
|
||
) -> None:
|
||
"""Higher correlation produces lower or equal allocation (monotonic non-increase)."""
|
||
assume(corr_low < corr_high)
|
||
|
||
portfolio = self._make_portfolio_with_position(active_pool=5000.0)
|
||
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.50, max_portfolio_heat=0.50,
|
||
)
|
||
|
||
result_low = self.sizer.compute(
|
||
confidence=0.8, ticker="NEW", sector="Healthcare",
|
||
current_price=10.0, active_pool=5000.0,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix={("NEW", "EXIST"): corr_low},
|
||
earnings_calendar={}, absolute_position_cap=500.0,
|
||
)
|
||
result_high = self.sizer.compute(
|
||
confidence=0.8, ticker="NEW", sector="Healthcare",
|
||
current_price=10.0, active_pool=5000.0,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix={("NEW", "EXIST"): corr_high},
|
||
earnings_calendar={}, absolute_position_cap=500.0,
|
||
)
|
||
|
||
# Higher correlation → lower or equal allocation
|
||
low_amount = result_low.dollar_amount if not result_low.rejected else 0.0
|
||
high_amount = result_high.dollar_amount if not result_high.rejected else 0.0
|
||
assert high_amount <= low_amount + 0.01
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 3: Sector exposure computation and enforcement
|
||
# **Validates: Requirements 2.6, 9.4**
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestProperty3SectorExposure:
|
||
"""Property 3: Sector exposure computation and enforcement.
|
||
|
||
**Validates: Requirements 2.6, 9.4**
|
||
"""
|
||
|
||
sizer = PositionSizer()
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
max_sector_pct=st.floats(min_value=0.10, max_value=0.50, allow_nan=False, allow_infinity=False),
|
||
active_pool=st.floats(min_value=1000.0, max_value=10000.0, allow_nan=False, allow_infinity=False),
|
||
exposure_frac=st.floats(min_value=0.51, max_value=0.99, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_allocation_reduced_when_exceeding_sector_limit(
|
||
self, max_sector_pct: float, active_pool: float, exposure_frac: float,
|
||
) -> None:
|
||
"""Allocation reduced when adding position would exceed max_sector_pct."""
|
||
max_sector_dollars = max_sector_pct * active_pool
|
||
# Derive existing_exposure directly from the sector limit so it's always valid
|
||
existing_exposure = max_sector_dollars * exposure_frac
|
||
|
||
risk_tier = RiskTierConfig(
|
||
name="moderate", min_confidence=0.5, max_position_pct=0.15,
|
||
stop_loss_atr_multiplier=2.0, reward_risk_ratio=1.5,
|
||
max_sector_pct=max_sector_pct, max_portfolio_heat=0.50,
|
||
)
|
||
|
||
pos = OpenPosition(
|
||
ticker="EXIST", quantity=10, entry_price=50.0,
|
||
current_price=50.0, unrealized_pnl=0.0,
|
||
market_value=existing_exposure, sector="Technology",
|
||
stop_loss_price=45.0, take_profit_price=60.0,
|
||
signal_confidence=0.7,
|
||
)
|
||
portfolio = PortfolioState(
|
||
positions=[pos],
|
||
total_value=active_pool + 500.0,
|
||
cash=active_pool - existing_exposure,
|
||
active_pool=active_pool,
|
||
reserve_pool=500.0,
|
||
sector_exposure={"Technology": existing_exposure},
|
||
portfolio_heat=0.0,
|
||
open_position_count=1,
|
||
)
|
||
|
||
result = self.sizer.compute(
|
||
confidence=0.9, ticker="NEW", sector="Technology",
|
||
current_price=5.0, active_pool=active_pool,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix={}, earnings_calendar={},
|
||
absolute_position_cap=max_sector_dollars,
|
||
)
|
||
|
||
if not result.rejected:
|
||
# The resulting position + existing should not exceed the sector limit
|
||
assert existing_exposure + result.dollar_amount <= max_sector_dollars + 0.01
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 4: Diversification bonus for under-represented sectors
|
||
# **Validates: Requirements 9.5**
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestProperty4DiversificationBonus:
|
||
"""Property 4: Diversification bonus for under-represented sectors.
|
||
|
||
**Validates: Requirements 9.5**
|
||
"""
|
||
|
||
sizer = PositionSizer()
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
num_existing_sectors=st.integers(min_value=0, max_value=2),
|
||
)
|
||
def test_bonus_applied_when_fewer_than_3_sectors_and_new_sector(
|
||
self, num_existing_sectors: int,
|
||
) -> None:
|
||
"""1.2x bonus applied when portfolio has < 3 sectors and trade is in new sector."""
|
||
existing_sectors = ["Technology", "Healthcare", "Energy"][:num_existing_sectors]
|
||
sector_exposure = {s: 200.0 for s in existing_sectors}
|
||
positions = [
|
||
OpenPosition(
|
||
ticker=f"T{i}", quantity=5, entry_price=40.0,
|
||
current_price=40.0, unrealized_pnl=0.0,
|
||
market_value=200.0, sector=s,
|
||
stop_loss_price=35.0, take_profit_price=50.0,
|
||
signal_confidence=0.7,
|
||
)
|
||
for i, s in enumerate(existing_sectors)
|
||
]
|
||
|
||
active_pool = 5000.0
|
||
risk_tier = RiskTierConfig(
|
||
name="moderate", min_confidence=0.5, max_position_pct=0.15,
|
||
stop_loss_atr_multiplier=2.0, reward_risk_ratio=1.5,
|
||
max_sector_pct=0.50, max_portfolio_heat=0.50,
|
||
)
|
||
|
||
portfolio = PortfolioState(
|
||
positions=positions,
|
||
total_value=active_pool + 500.0,
|
||
cash=active_pool - sum(200.0 for _ in existing_sectors),
|
||
active_pool=active_pool,
|
||
reserve_pool=500.0,
|
||
sector_exposure=sector_exposure,
|
||
portfolio_heat=0.0,
|
||
open_position_count=len(positions),
|
||
)
|
||
|
||
# Trade in a new sector not in existing sectors
|
||
new_sector = "Financials"
|
||
|
||
result_new_sector = self.sizer.compute(
|
||
confidence=0.7, ticker="NEW", sector=new_sector,
|
||
current_price=5.0, active_pool=active_pool,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix={}, earnings_calendar={},
|
||
absolute_position_cap=500.0,
|
||
)
|
||
|
||
# Trade in an existing sector (no bonus)
|
||
if existing_sectors:
|
||
result_existing_sector = self.sizer.compute(
|
||
confidence=0.7, ticker="NEW2", sector=existing_sectors[0],
|
||
current_price=5.0, active_pool=active_pool,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix={}, earnings_calendar={},
|
||
absolute_position_cap=500.0,
|
||
)
|
||
|
||
if not result_new_sector.rejected and not result_existing_sector.rejected:
|
||
# New sector should get a bonus (higher allocation)
|
||
assert result_new_sector.dollar_amount >= result_existing_sector.dollar_amount
|
||
else:
|
||
# With 0 existing sectors, new sector should still get the bonus
|
||
if not result_new_sector.rejected:
|
||
assert result_new_sector.dollar_amount > 0
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
active_pool=st.floats(min_value=2000.0, max_value=10000.0, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_no_bonus_when_3_or_more_sectors(self, active_pool: float) -> None:
|
||
"""No bonus when portfolio has >= 3 sectors."""
|
||
existing_sectors = ["Technology", "Healthcare", "Energy"]
|
||
sector_exposure = {s: 200.0 for s in existing_sectors}
|
||
positions = [
|
||
OpenPosition(
|
||
ticker=f"T{i}", quantity=5, entry_price=40.0,
|
||
current_price=40.0, unrealized_pnl=0.0,
|
||
market_value=200.0, sector=s,
|
||
stop_loss_price=35.0, take_profit_price=50.0,
|
||
signal_confidence=0.7,
|
||
)
|
||
for i, s in enumerate(existing_sectors)
|
||
]
|
||
|
||
risk_tier = RiskTierConfig(
|
||
name="moderate", min_confidence=0.5, max_position_pct=0.15,
|
||
stop_loss_atr_multiplier=2.0, reward_risk_ratio=1.5,
|
||
max_sector_pct=0.50, max_portfolio_heat=0.50,
|
||
)
|
||
|
||
portfolio = PortfolioState(
|
||
positions=positions,
|
||
total_value=active_pool + 500.0,
|
||
cash=active_pool - 600.0,
|
||
active_pool=active_pool,
|
||
reserve_pool=500.0,
|
||
sector_exposure=sector_exposure,
|
||
portfolio_heat=0.0,
|
||
open_position_count=3,
|
||
)
|
||
|
||
# Trade in a new sector — should NOT get bonus since we already have 3 sectors
|
||
result_new = self.sizer.compute(
|
||
confidence=0.7, ticker="NEW", sector="Financials",
|
||
current_price=5.0, active_pool=active_pool,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix={}, earnings_calendar={},
|
||
absolute_position_cap=500.0,
|
||
)
|
||
|
||
# Trade in an existing sector
|
||
result_existing = self.sizer.compute(
|
||
confidence=0.7, ticker="NEW2", sector="Technology",
|
||
current_price=5.0, active_pool=active_pool,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix={}, earnings_calendar={},
|
||
absolute_position_cap=500.0,
|
||
)
|
||
|
||
if not result_new.rejected and not result_existing.rejected:
|
||
# With >= 3 sectors, no bonus — allocations should be equal
|
||
assert result_new.dollar_amount == result_existing.dollar_amount
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 5: Active Pool computation invariant
|
||
# **Validates: Requirements 3.3**
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestProperty5ActivePoolComputation:
|
||
"""Property 5: Active Pool computation invariant.
|
||
|
||
**Validates: Requirements 3.3**
|
||
"""
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
total_portfolio_value=st.floats(min_value=100.0, max_value=50000.0, allow_nan=False, allow_infinity=False),
|
||
reserve_pool_pct=st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_active_pool_equals_total_minus_reserve(
|
||
self, total_portfolio_value: float, reserve_pool_pct: float,
|
||
) -> None:
|
||
"""Active Pool = total_portfolio_value - reserve_pool_balance."""
|
||
reserve_pool_balance = total_portfolio_value * reserve_pool_pct
|
||
active_pool = total_portfolio_value - reserve_pool_balance
|
||
|
||
assert active_pool >= -0.01, (
|
||
f"Active pool should be non-negative: {active_pool}"
|
||
)
|
||
assert abs(active_pool - (total_portfolio_value - reserve_pool_balance)) < 1e-9
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 19: Earnings proximity adjustments
|
||
# **Validates: Requirements 10.2, 10.3**
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestProperty19EarningsProximity:
|
||
"""Property 19: Earnings proximity adjustments.
|
||
|
||
**Validates: Requirements 10.2, 10.3**
|
||
"""
|
||
|
||
sizer = PositionSizer()
|
||
|
||
def _base_args(self, active_pool: float = 5000.0) -> dict:
|
||
"""Common arguments for position sizer calls."""
|
||
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.50, max_portfolio_heat=0.50,
|
||
)
|
||
portfolio = PortfolioState(active_pool=active_pool)
|
||
return dict(
|
||
confidence=0.8,
|
||
ticker="EARN",
|
||
sector="Technology",
|
||
current_price=10.0,
|
||
active_pool=active_pool,
|
||
risk_tier=risk_tier,
|
||
portfolio_state=portfolio,
|
||
correlation_matrix={},
|
||
absolute_position_cap=500.0,
|
||
)
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
days_until=st.floats(min_value=1.01, max_value=3.0, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_50_pct_reduction_within_3_trading_days(self, days_until: float) -> None:
|
||
"""50% reduction when earnings within 3 trading days (but > 1 day)."""
|
||
now = datetime.now(tz=timezone.utc)
|
||
earnings_date = now + timedelta(days=days_until)
|
||
|
||
args = self._base_args()
|
||
result_with_earnings = self.sizer.compute(
|
||
**args, earnings_calendar={"EARN": earnings_date},
|
||
)
|
||
result_no_earnings = self.sizer.compute(
|
||
**args, earnings_calendar={},
|
||
)
|
||
|
||
if not result_with_earnings.rejected and not result_no_earnings.rejected:
|
||
# With earnings proximity, allocation should be ~50% of normal
|
||
ratio = result_with_earnings.dollar_amount / result_no_earnings.dollar_amount
|
||
# Allow some tolerance due to share rounding
|
||
assert ratio <= 0.6, (
|
||
f"Expected ~50% reduction, got ratio={ratio:.4f}"
|
||
)
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
days_until=st.floats(min_value=0.01, max_value=1.0, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_rejection_within_1_trading_day(self, days_until: float) -> None:
|
||
"""Trade rejected when earnings within 1 trading day."""
|
||
now = datetime.now(tz=timezone.utc)
|
||
earnings_date = now + timedelta(days=days_until)
|
||
|
||
args = self._base_args()
|
||
result = self.sizer.compute(
|
||
**args, earnings_calendar={"EARN": earnings_date},
|
||
)
|
||
assert result.rejected is True
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
days_until=st.floats(min_value=4.0, max_value=30.0, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_normal_sizing_outside_earnings_window(self, days_until: float) -> None:
|
||
"""Normal sizing when earnings are outside the 3-day window."""
|
||
now = datetime.now(tz=timezone.utc)
|
||
earnings_date = now + timedelta(days=days_until)
|
||
|
||
args = self._base_args()
|
||
result_with_earnings = self.sizer.compute(
|
||
**args, earnings_calendar={"EARN": earnings_date},
|
||
)
|
||
result_no_earnings = self.sizer.compute(
|
||
**args, earnings_calendar={},
|
||
)
|
||
|
||
if not result_with_earnings.rejected and not result_no_earnings.rejected:
|
||
assert result_with_earnings.dollar_amount == result_no_earnings.dollar_amount
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 24: Portfolio heat computation and threshold enforcement
|
||
# **Validates: Requirements 13.1, 13.2**
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestProperty24PortfolioHeat:
|
||
"""Property 24: Portfolio heat computation and threshold enforcement.
|
||
|
||
**Validates: Requirements 13.1, 13.2**
|
||
"""
|
||
|
||
sizer = PositionSizer()
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
entry_prices=st.lists(
|
||
st.floats(min_value=10.0, max_value=200.0, allow_nan=False, allow_infinity=False),
|
||
min_size=1, max_size=5,
|
||
),
|
||
stop_loss_pcts=st.lists(
|
||
st.floats(min_value=0.02, max_value=0.20, allow_nan=False, allow_infinity=False),
|
||
min_size=1, max_size=5,
|
||
),
|
||
quantities=st.lists(
|
||
st.integers(min_value=1, max_value=50),
|
||
min_size=1, max_size=5,
|
||
),
|
||
)
|
||
def test_heat_computation_formula(
|
||
self, entry_prices: list[float], stop_loss_pcts: list[float],
|
||
quantities: list[int],
|
||
) -> None:
|
||
"""Heat = sum of position_value * (entry_price - stop_loss_price) / entry_price."""
|
||
n = min(len(entry_prices), len(stop_loss_pcts), len(quantities))
|
||
assume(n >= 1)
|
||
|
||
expected_heat = 0.0
|
||
for i in range(n):
|
||
entry = entry_prices[i]
|
||
stop_pct = stop_loss_pcts[i]
|
||
qty = quantities[i]
|
||
stop_loss = entry * (1.0 - stop_pct)
|
||
position_value = qty * entry
|
||
heat_contribution = position_value * (entry - stop_loss) / entry
|
||
expected_heat += heat_contribution
|
||
|
||
# Verify the formula: heat_contribution = position_value * stop_pct
|
||
recomputed = 0.0
|
||
for i in range(n):
|
||
entry = entry_prices[i]
|
||
stop_pct = stop_loss_pcts[i]
|
||
qty = quantities[i]
|
||
position_value = qty * entry
|
||
recomputed += position_value * stop_pct
|
||
|
||
assert abs(expected_heat - recomputed) < 1e-6
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
active_pool=st.floats(min_value=2000.0, max_value=10000.0, allow_nan=False, allow_infinity=False),
|
||
max_heat_pct=st.floats(min_value=0.05, max_value=0.30, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_new_entries_rejected_when_heat_exceeds_max(
|
||
self, active_pool: float, max_heat_pct: float,
|
||
) -> None:
|
||
"""New entries rejected when portfolio heat exceeds max_portfolio_heat."""
|
||
# Set portfolio heat to exceed the max
|
||
max_heat_dollars = max_heat_pct * active_pool
|
||
current_heat = max_heat_dollars * 1.1 # 10% over the limit
|
||
|
||
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.50, max_portfolio_heat=max_heat_pct,
|
||
)
|
||
|
||
portfolio = PortfolioState(
|
||
positions=[],
|
||
total_value=active_pool + 500.0,
|
||
cash=active_pool,
|
||
active_pool=active_pool,
|
||
reserve_pool=500.0,
|
||
sector_exposure={},
|
||
portfolio_heat=current_heat,
|
||
open_position_count=0,
|
||
)
|
||
|
||
result = self.sizer.compute(
|
||
confidence=0.8, ticker="NEW", sector="Technology",
|
||
current_price=10.0, active_pool=active_pool,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix={}, earnings_calendar={},
|
||
absolute_position_cap=500.0,
|
||
)
|
||
assert result.rejected is True
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 7: Active Pool minimum halts new entries but allows exits
|
||
# **Validates: Requirements 3.5**
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestProperty7ActivePoolMinimum:
|
||
"""Property 7: Active Pool minimum halts new entries but allows exits.
|
||
|
||
**Validates: Requirements 3.5**
|
||
"""
|
||
|
||
sizer = PositionSizer()
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
active_pool=st.floats(min_value=1.0, max_value=99.99, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_buy_orders_rejected_when_active_pool_below_minimum(
|
||
self, active_pool: float,
|
||
) -> None:
|
||
"""Buy orders rejected when Active Pool < minimum ($100 default)."""
|
||
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,
|
||
)
|
||
|
||
portfolio = PortfolioState(
|
||
positions=[],
|
||
total_value=active_pool + 50.0,
|
||
cash=active_pool,
|
||
active_pool=active_pool,
|
||
reserve_pool=50.0,
|
||
sector_exposure={},
|
||
portfolio_heat=0.0,
|
||
open_position_count=0,
|
||
)
|
||
|
||
result = self.sizer.compute(
|
||
confidence=0.8, ticker="TEST", sector="Technology",
|
||
current_price=10.0, active_pool=active_pool,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix={}, earnings_calendar={},
|
||
absolute_position_cap=50.0,
|
||
active_pool_minimum=100.0,
|
||
)
|
||
assert result.rejected is True
|
||
assert "below minimum" in result.rejection_reason.lower() or "active pool" in result.rejection_reason.lower()
|
||
|
||
@settings(max_examples=100)
|
||
@given(
|
||
active_pool=st.floats(min_value=100.0, max_value=10000.0, allow_nan=False, allow_infinity=False),
|
||
)
|
||
def test_buy_orders_allowed_when_active_pool_above_minimum(
|
||
self, active_pool: float,
|
||
) -> None:
|
||
"""Buy orders allowed when Active Pool >= minimum."""
|
||
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.50, max_portfolio_heat=0.50,
|
||
)
|
||
|
||
portfolio = PortfolioState(
|
||
positions=[],
|
||
total_value=active_pool + 500.0,
|
||
cash=active_pool,
|
||
active_pool=active_pool,
|
||
reserve_pool=500.0,
|
||
sector_exposure={},
|
||
portfolio_heat=0.0,
|
||
open_position_count=0,
|
||
)
|
||
|
||
result = self.sizer.compute(
|
||
confidence=0.8, ticker="TEST", sector="Technology",
|
||
current_price=5.0, active_pool=active_pool,
|
||
risk_tier=risk_tier, portfolio_state=portfolio,
|
||
correlation_matrix={}, earnings_calendar={},
|
||
absolute_position_cap=500.0,
|
||
active_pool_minimum=100.0,
|
||
)
|
||
# Should not be rejected due to active pool minimum
|
||
# (may still be rejected for other reasons like heat, but not for active pool)
|
||
if result.rejected:
|
||
assert "active pool" not in result.rejection_reason.lower() or "below minimum" not in result.rejection_reason.lower()
|