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
670 lines
24 KiB
Python
670 lines
24 KiB
Python
"""Property-based tests for the TradingEngine core decision loop.
|
|
|
|
Feature: autonomous-trading-engine
|
|
|
|
Tests properties 27, 28, 16, and 18 from the design specification,
|
|
covering recommendation deduplication, decision record completeness,
|
|
multiple declining positions halt, and maximum open positions enforcement.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from hypothesis import assume, given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from services.shared.config import TradingConfig
|
|
from services.trading.correlation import CorrelationMatrix
|
|
from services.trading.engine import TradingEngine
|
|
from services.trading.models import (
|
|
CircuitBreakerState,
|
|
OpenPosition,
|
|
PortfolioState,
|
|
RiskTierConfig,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
ET = ZoneInfo("America/New_York")
|
|
|
|
# A valid trading window datetime: Wednesday 10:00 AM ET
|
|
VALID_TRADING_DT = datetime(2025, 1, 8, 10, 0, 0, tzinfo=ET)
|
|
|
|
|
|
def _make_engine() -> TradingEngine:
|
|
"""Create a TradingEngine with default TradingConfig and no pool/redis."""
|
|
return TradingEngine(pool=None, redis=None, config=TradingConfig())
|
|
|
|
|
|
def _moderate_risk_tier() -> RiskTierConfig:
|
|
"""Return a moderate risk tier with reasonable defaults for testing."""
|
|
return RiskTierConfig(
|
|
name="moderate",
|
|
min_confidence=0.55,
|
|
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,
|
|
)
|
|
|
|
|
|
def _inactive_cb() -> CircuitBreakerState:
|
|
"""Return an inactive circuit breaker state."""
|
|
return CircuitBreakerState(active=False)
|
|
|
|
|
|
def _empty_correlation_matrix() -> CorrelationMatrix:
|
|
"""Return an empty correlation matrix."""
|
|
return CorrelationMatrix()
|
|
|
|
|
|
def _base_portfolio(
|
|
active_pool: float = 5000.0,
|
|
positions: list | None = None,
|
|
portfolio_heat: float = 0.0,
|
|
open_position_count: int = 0,
|
|
) -> PortfolioState:
|
|
"""Return a PortfolioState with sensible defaults for testing."""
|
|
return PortfolioState(
|
|
positions=positions or [],
|
|
total_value=active_pool + 500.0,
|
|
cash=active_pool,
|
|
active_pool=active_pool,
|
|
reserve_pool=500.0,
|
|
sector_exposure={},
|
|
portfolio_heat=portfolio_heat,
|
|
open_position_count=open_position_count,
|
|
)
|
|
|
|
|
|
def _make_recommendation(
|
|
rec_id: str = "rec-001",
|
|
ticker: str = "AAPL",
|
|
confidence: float = 0.80,
|
|
sector: str = "Technology",
|
|
current_price: float = 10.0,
|
|
action: str = "buy",
|
|
) -> dict:
|
|
"""Build a recommendation dict suitable for evaluate_recommendation."""
|
|
return {
|
|
"recommendation_id": rec_id,
|
|
"ticker": ticker,
|
|
"confidence": confidence,
|
|
"sector": sector,
|
|
"current_price": current_price,
|
|
"action": action,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Hypothesis strategies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _recommendation_id_strategy() -> st.SearchStrategy[str]:
|
|
"""Generate random recommendation IDs."""
|
|
return st.text(
|
|
alphabet=st.sampled_from("abcdefghijklmnopqrstuvwxyz0123456789-"),
|
|
min_size=5,
|
|
max_size=20,
|
|
).filter(lambda s: len(s.strip()) > 0)
|
|
|
|
|
|
def _open_position_strategy(
|
|
unrealized_pnl: 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=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=unrealized_pnl if unrealized_pnl is not None else 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=st.sampled_from(["Technology", "Healthcare", "Energy", "Financials", "Consumer"]),
|
|
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),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 27: Recommendation deduplication (idempotence)
|
|
# **Validates: Requirements 1.5**
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProperty27RecommendationDeduplication:
|
|
"""Property 27: Recommendation deduplication (idempotence).
|
|
|
|
**Validates: Requirements 1.5**
|
|
"""
|
|
|
|
@settings(max_examples=100)
|
|
@given(rec_id=_recommendation_id_strategy())
|
|
def test_duplicate_recommendation_produces_skip(self, rec_id: str) -> None:
|
|
"""Processing the same recommendation twice produces a skip on the second call."""
|
|
engine = _make_engine()
|
|
risk_tier = _moderate_risk_tier()
|
|
portfolio = _base_portfolio(active_pool=5000.0)
|
|
cb_state = _inactive_cb()
|
|
corr_matrix = _empty_correlation_matrix()
|
|
|
|
rec = _make_recommendation(
|
|
rec_id=rec_id,
|
|
ticker="AAPL",
|
|
confidence=0.80,
|
|
sector="Technology",
|
|
current_price=10.0,
|
|
)
|
|
|
|
# First processing — should produce an "act" decision
|
|
decision1 = engine.evaluate_recommendation(
|
|
rec=rec,
|
|
portfolio_state=portfolio,
|
|
risk_tier=risk_tier,
|
|
circuit_breaker_state=cb_state,
|
|
correlation_matrix=corr_matrix,
|
|
earnings_calendar={},
|
|
now=VALID_TRADING_DT,
|
|
)
|
|
assert decision1.decision == "act", (
|
|
f"First processing should produce 'act', got '{decision1.decision}' "
|
|
f"with skip_reason={decision1.skip_reason}"
|
|
)
|
|
|
|
# Second processing — should produce a "skip" with duplicate reason
|
|
decision2 = engine.evaluate_recommendation(
|
|
rec=rec,
|
|
portfolio_state=portfolio,
|
|
risk_tier=risk_tier,
|
|
circuit_breaker_state=cb_state,
|
|
correlation_matrix=corr_matrix,
|
|
earnings_calendar={},
|
|
now=VALID_TRADING_DT,
|
|
)
|
|
assert decision2.decision == "skip"
|
|
assert decision2.skip_reason == "duplicate_recommendation"
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
rec_id_a=_recommendation_id_strategy(),
|
|
rec_id_b=_recommendation_id_strategy(),
|
|
)
|
|
def test_different_recommendations_not_deduplicated(
|
|
self, rec_id_a: str, rec_id_b: str,
|
|
) -> None:
|
|
"""Different recommendation IDs are processed independently."""
|
|
assume(rec_id_a != rec_id_b)
|
|
|
|
engine = _make_engine()
|
|
risk_tier = _moderate_risk_tier()
|
|
portfolio = _base_portfolio(active_pool=5000.0)
|
|
cb_state = _inactive_cb()
|
|
corr_matrix = _empty_correlation_matrix()
|
|
|
|
rec_a = _make_recommendation(rec_id=rec_id_a, ticker="AAPL", current_price=10.0)
|
|
rec_b = _make_recommendation(rec_id=rec_id_b, ticker="MSFT", current_price=10.0)
|
|
|
|
decision_a = engine.evaluate_recommendation(
|
|
rec=rec_a,
|
|
portfolio_state=portfolio,
|
|
risk_tier=risk_tier,
|
|
circuit_breaker_state=cb_state,
|
|
correlation_matrix=corr_matrix,
|
|
earnings_calendar={},
|
|
now=VALID_TRADING_DT,
|
|
)
|
|
|
|
decision_b = engine.evaluate_recommendation(
|
|
rec=rec_b,
|
|
portfolio_state=portfolio,
|
|
risk_tier=risk_tier,
|
|
circuit_breaker_state=cb_state,
|
|
correlation_matrix=corr_matrix,
|
|
earnings_calendar={},
|
|
now=VALID_TRADING_DT,
|
|
)
|
|
|
|
# Both should be "act" (not deduplicated against each other)
|
|
assert decision_a.decision == "act"
|
|
assert decision_b.decision == "act"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 28: Trading decision record completeness and traceability
|
|
# **Validates: Requirements 1.4, 17.1, 17.2**
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProperty28DecisionRecordCompleteness:
|
|
"""Property 28: Trading decision record completeness and traceability.
|
|
|
|
**Validates: Requirements 1.4, 17.1, 17.2**
|
|
"""
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
confidence=st.floats(min_value=0.60, max_value=0.99, allow_nan=False, allow_infinity=False),
|
|
ticker=st.from_regex(r"[A-Z]{3,5}", fullmatch=True),
|
|
price=st.floats(min_value=1.0, max_value=50.0, allow_nan=False, allow_infinity=False),
|
|
)
|
|
def test_act_decision_has_all_required_fields(
|
|
self, confidence: float, ticker: str, price: float,
|
|
) -> None:
|
|
"""Act decisions have all required fields including position size."""
|
|
engine = _make_engine()
|
|
risk_tier = _moderate_risk_tier()
|
|
portfolio = _base_portfolio(active_pool=5000.0)
|
|
cb_state = _inactive_cb()
|
|
corr_matrix = _empty_correlation_matrix()
|
|
|
|
rec = _make_recommendation(
|
|
rec_id=f"rec-{ticker}",
|
|
ticker=ticker,
|
|
confidence=confidence,
|
|
current_price=price,
|
|
)
|
|
|
|
decision = engine.evaluate_recommendation(
|
|
rec=rec,
|
|
portfolio_state=portfolio,
|
|
risk_tier=risk_tier,
|
|
circuit_breaker_state=cb_state,
|
|
correlation_matrix=corr_matrix,
|
|
earnings_calendar={},
|
|
now=VALID_TRADING_DT,
|
|
)
|
|
|
|
# All required fields must be present and non-None
|
|
assert decision.id is not None and len(decision.id) > 0
|
|
assert decision.recommendation_id == f"rec-{ticker}"
|
|
assert decision.decision in ("act", "skip")
|
|
assert decision.ticker == ticker
|
|
assert decision.risk_tier_at_decision == "moderate"
|
|
assert decision.circuit_breaker_status in ("active", "inactive")
|
|
assert decision.decision_trace is not None
|
|
assert isinstance(decision.decision_trace, dict)
|
|
assert decision.created_at is not None
|
|
|
|
if decision.decision == "act":
|
|
assert decision.computed_position_size is not None
|
|
assert decision.computed_position_size > 0
|
|
assert decision.computed_share_quantity is not None
|
|
assert decision.computed_share_quantity > 0
|
|
assert decision.skip_reason is None
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
confidence=st.floats(min_value=0.01, max_value=0.50, allow_nan=False, allow_infinity=False),
|
|
ticker=st.from_regex(r"[A-Z]{3,5}", fullmatch=True),
|
|
)
|
|
def test_skip_decision_has_skip_reason(
|
|
self, confidence: float, ticker: str,
|
|
) -> None:
|
|
"""Skip decisions have skip_reason set."""
|
|
engine = _make_engine()
|
|
risk_tier = _moderate_risk_tier()
|
|
portfolio = _base_portfolio(active_pool=5000.0)
|
|
cb_state = _inactive_cb()
|
|
corr_matrix = _empty_correlation_matrix()
|
|
|
|
# Use low confidence to trigger a skip
|
|
assume(confidence < risk_tier.min_confidence)
|
|
|
|
rec = _make_recommendation(
|
|
rec_id=f"rec-{ticker}",
|
|
ticker=ticker,
|
|
confidence=confidence,
|
|
current_price=10.0,
|
|
)
|
|
|
|
decision = engine.evaluate_recommendation(
|
|
rec=rec,
|
|
portfolio_state=portfolio,
|
|
risk_tier=risk_tier,
|
|
circuit_breaker_state=cb_state,
|
|
correlation_matrix=corr_matrix,
|
|
earnings_calendar={},
|
|
now=VALID_TRADING_DT,
|
|
)
|
|
|
|
assert decision.decision == "skip"
|
|
assert decision.skip_reason is not None
|
|
assert len(decision.skip_reason) > 0
|
|
|
|
# Required fields still present on skip decisions
|
|
assert decision.id is not None
|
|
assert decision.recommendation_id is not None
|
|
assert decision.ticker == ticker
|
|
assert decision.risk_tier_at_decision == "moderate"
|
|
assert decision.circuit_breaker_status == "inactive"
|
|
assert decision.decision_trace is not None
|
|
assert decision.created_at is not None
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
confidence=st.floats(min_value=0.60, max_value=0.99, allow_nan=False, allow_infinity=False),
|
|
ticker=st.from_regex(r"[A-Z]{3,5}", fullmatch=True),
|
|
price=st.floats(min_value=1.0, max_value=50.0, allow_nan=False, allow_infinity=False),
|
|
)
|
|
def test_decision_trace_contains_reasoning(
|
|
self, confidence: float, ticker: str, price: float,
|
|
) -> None:
|
|
"""Decision trace contains a reasoning list for traceability."""
|
|
engine = _make_engine()
|
|
risk_tier = _moderate_risk_tier()
|
|
portfolio = _base_portfolio(active_pool=5000.0)
|
|
cb_state = _inactive_cb()
|
|
corr_matrix = _empty_correlation_matrix()
|
|
|
|
rec = _make_recommendation(
|
|
rec_id=f"rec-trace-{ticker}",
|
|
ticker=ticker,
|
|
confidence=confidence,
|
|
current_price=price,
|
|
)
|
|
|
|
decision = engine.evaluate_recommendation(
|
|
rec=rec,
|
|
portfolio_state=portfolio,
|
|
risk_tier=risk_tier,
|
|
circuit_breaker_state=cb_state,
|
|
correlation_matrix=corr_matrix,
|
|
earnings_calendar={},
|
|
now=VALID_TRADING_DT,
|
|
)
|
|
|
|
assert "reasoning" in decision.decision_trace
|
|
assert isinstance(decision.decision_trace["reasoning"], list)
|
|
assert len(decision.decision_trace["reasoning"]) > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 16: Multiple declining positions halts new entries
|
|
# **Validates: Requirements 7.5**
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProperty16DecliningPositionsHalt:
|
|
"""Property 16: Multiple declining positions halts new entries.
|
|
|
|
**Validates: Requirements 7.5**
|
|
"""
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
total_positions=st.integers(min_value=4, max_value=20),
|
|
declining_fraction=st.floats(min_value=0.55, max_value=1.0, allow_nan=False, allow_infinity=False),
|
|
)
|
|
def test_entries_halted_when_majority_declining(
|
|
self, total_positions: int, declining_fraction: float,
|
|
) -> None:
|
|
"""New entries halted when > 50% of positions have > 2% negative unrealized P&L."""
|
|
declining_count = int(total_positions * declining_fraction)
|
|
assume(declining_count > total_positions * 0.50) # strictly > 50%
|
|
non_declining_count = total_positions - declining_count
|
|
|
|
positions: list[OpenPosition] = []
|
|
|
|
# Declining positions: unrealized_pnl is negative and > 2% of entry value
|
|
for i in range(declining_count):
|
|
entry_price = 100.0
|
|
quantity = 10
|
|
entry_value = entry_price * quantity
|
|
# Loss > 2% of entry value
|
|
loss = entry_value * 0.03 # 3% loss
|
|
positions.append(OpenPosition(
|
|
ticker=f"DEC{i}",
|
|
quantity=quantity,
|
|
entry_price=entry_price,
|
|
current_price=entry_price - (loss / quantity),
|
|
unrealized_pnl=-loss,
|
|
market_value=entry_value - loss,
|
|
sector="Technology",
|
|
stop_loss_price=90.0,
|
|
take_profit_price=120.0,
|
|
signal_confidence=0.6,
|
|
))
|
|
|
|
# Non-declining positions: small positive or flat P&L
|
|
for i in range(non_declining_count):
|
|
entry_price = 100.0
|
|
quantity = 10
|
|
positions.append(OpenPosition(
|
|
ticker=f"OK{i}",
|
|
quantity=quantity,
|
|
entry_price=entry_price,
|
|
current_price=entry_price + 1.0,
|
|
unrealized_pnl=10.0,
|
|
market_value=1010.0,
|
|
sector="Healthcare",
|
|
stop_loss_price=90.0,
|
|
take_profit_price=120.0,
|
|
signal_confidence=0.7,
|
|
))
|
|
|
|
engine = _make_engine()
|
|
|
|
# Verify the check_declining_positions method returns True
|
|
assert engine.check_declining_positions(positions) is True
|
|
|
|
# Verify the full engine skips new entries
|
|
risk_tier = _moderate_risk_tier()
|
|
portfolio = _base_portfolio(
|
|
active_pool=5000.0,
|
|
positions=positions,
|
|
open_position_count=total_positions,
|
|
)
|
|
cb_state = _inactive_cb()
|
|
corr_matrix = _empty_correlation_matrix()
|
|
|
|
rec = _make_recommendation(
|
|
rec_id="rec-decline-test",
|
|
ticker="NEWSTOCK",
|
|
confidence=0.80,
|
|
current_price=10.0,
|
|
)
|
|
|
|
decision = engine.evaluate_recommendation(
|
|
rec=rec,
|
|
portfolio_state=portfolio,
|
|
risk_tier=risk_tier,
|
|
circuit_breaker_state=cb_state,
|
|
correlation_matrix=corr_matrix,
|
|
earnings_calendar={},
|
|
now=VALID_TRADING_DT,
|
|
)
|
|
|
|
assert decision.decision == "skip"
|
|
assert decision.skip_reason == "multiple_declining_positions"
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
total_positions=st.integers(min_value=4, max_value=20),
|
|
declining_fraction=st.floats(min_value=0.0, max_value=0.45, allow_nan=False, allow_infinity=False),
|
|
)
|
|
def test_entries_allowed_when_minority_declining(
|
|
self, total_positions: int, declining_fraction: float,
|
|
) -> None:
|
|
"""Entries allowed when <= 50% of positions are declining."""
|
|
declining_count = int(total_positions * declining_fraction)
|
|
assume(declining_count <= total_positions * 0.50) # at or below 50%
|
|
non_declining_count = total_positions - declining_count
|
|
|
|
positions: list[OpenPosition] = []
|
|
|
|
# Declining positions
|
|
for i in range(declining_count):
|
|
entry_price = 100.0
|
|
quantity = 10
|
|
entry_value = entry_price * quantity
|
|
loss = entry_value * 0.03
|
|
positions.append(OpenPosition(
|
|
ticker=f"DEC{i}",
|
|
quantity=quantity,
|
|
entry_price=entry_price,
|
|
current_price=entry_price - (loss / quantity),
|
|
unrealized_pnl=-loss,
|
|
market_value=entry_value - loss,
|
|
sector="Technology",
|
|
stop_loss_price=90.0,
|
|
take_profit_price=120.0,
|
|
signal_confidence=0.6,
|
|
))
|
|
|
|
# Non-declining positions
|
|
for i in range(non_declining_count):
|
|
entry_price = 100.0
|
|
quantity = 10
|
|
positions.append(OpenPosition(
|
|
ticker=f"OK{i}",
|
|
quantity=quantity,
|
|
entry_price=entry_price,
|
|
current_price=entry_price + 1.0,
|
|
unrealized_pnl=10.0,
|
|
market_value=1010.0,
|
|
sector="Healthcare",
|
|
stop_loss_price=90.0,
|
|
take_profit_price=120.0,
|
|
signal_confidence=0.7,
|
|
))
|
|
|
|
engine = _make_engine()
|
|
|
|
# Verify the check returns False (entries allowed)
|
|
assert engine.check_declining_positions(positions) is False
|
|
|
|
@settings(max_examples=100)
|
|
@given(data=st.data())
|
|
def test_empty_positions_allows_entries(self, data: st.DataObject) -> None:
|
|
"""Empty position list always allows new entries."""
|
|
engine = _make_engine()
|
|
assert engine.check_declining_positions([]) is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 18: Maximum open positions enforcement
|
|
# **Validates: Requirements 8.4**
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProperty18MaxOpenPositions:
|
|
"""Property 18: Maximum open positions enforcement.
|
|
|
|
**Validates: Requirements 8.4**
|
|
"""
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
max_positions=st.integers(min_value=1, max_value=20),
|
|
)
|
|
def test_entries_rejected_at_max_positions(self, max_positions: int) -> None:
|
|
"""New entries rejected when open_position_count >= max_open_positions."""
|
|
engine = _make_engine()
|
|
|
|
# Verify the check_max_positions method returns True at the limit
|
|
assert engine.check_max_positions(max_positions, max_positions) is True
|
|
|
|
# Also verify above the limit
|
|
assert engine.check_max_positions(max_positions + 1, max_positions) is True
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
max_positions=st.integers(min_value=2, max_value=20),
|
|
open_count_offset=st.integers(min_value=1, max_value=10),
|
|
)
|
|
def test_entries_allowed_below_max_positions(
|
|
self, max_positions: int, open_count_offset: int,
|
|
) -> None:
|
|
"""Entries allowed when open_position_count < max_open_positions."""
|
|
open_count = max(0, max_positions - open_count_offset)
|
|
assume(open_count < max_positions)
|
|
|
|
engine = _make_engine()
|
|
assert engine.check_max_positions(open_count, max_positions) is False
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
max_positions=st.integers(min_value=1, max_value=15),
|
|
)
|
|
def test_full_engine_rejects_at_max_positions(self, max_positions: int) -> None:
|
|
"""Full engine evaluation rejects new entries at max positions."""
|
|
engine = _make_engine()
|
|
# Override the config's max_open_positions via hasattr fallback
|
|
# The engine uses: self.config.max_open_positions if hasattr(...) else 10
|
|
# TradingConfig doesn't have max_open_positions, so it defaults to 10.
|
|
# We test with the default (10) by setting open_position_count = 10.
|
|
risk_tier = _moderate_risk_tier()
|
|
portfolio = _base_portfolio(
|
|
active_pool=5000.0,
|
|
open_position_count=10, # default max is 10
|
|
)
|
|
cb_state = _inactive_cb()
|
|
corr_matrix = _empty_correlation_matrix()
|
|
|
|
rec = _make_recommendation(
|
|
rec_id=f"rec-max-{max_positions}",
|
|
ticker="MAXTEST",
|
|
confidence=0.80,
|
|
current_price=10.0,
|
|
)
|
|
|
|
decision = engine.evaluate_recommendation(
|
|
rec=rec,
|
|
portfolio_state=portfolio,
|
|
risk_tier=risk_tier,
|
|
circuit_breaker_state=cb_state,
|
|
correlation_matrix=corr_matrix,
|
|
earnings_calendar={},
|
|
now=VALID_TRADING_DT,
|
|
)
|
|
|
|
assert decision.decision == "skip"
|
|
assert decision.skip_reason == "max_positions_reached"
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
open_count=st.integers(min_value=0, max_value=8),
|
|
)
|
|
def test_full_engine_allows_below_max_positions(self, open_count: int) -> None:
|
|
"""Full engine evaluation allows entries below max positions (default 10)."""
|
|
assume(open_count < 10) # default max is 10
|
|
|
|
engine = _make_engine()
|
|
risk_tier = _moderate_risk_tier()
|
|
portfolio = _base_portfolio(
|
|
active_pool=5000.0,
|
|
open_position_count=open_count,
|
|
)
|
|
cb_state = _inactive_cb()
|
|
corr_matrix = _empty_correlation_matrix()
|
|
|
|
rec = _make_recommendation(
|
|
rec_id=f"rec-below-{open_count}",
|
|
ticker="BELOWMAX",
|
|
confidence=0.80,
|
|
current_price=10.0,
|
|
)
|
|
|
|
decision = engine.evaluate_recommendation(
|
|
rec=rec,
|
|
portfolio_state=portfolio,
|
|
risk_tier=risk_tier,
|
|
circuit_breaker_state=cb_state,
|
|
correlation_matrix=corr_matrix,
|
|
earnings_calendar={},
|
|
now=VALID_TRADING_DT,
|
|
)
|
|
|
|
# Should not be rejected for max positions
|
|
if decision.decision == "skip":
|
|
assert decision.skip_reason != "max_positions_reached"
|