Celes Renata
284 lines
9.7 KiB
Python
284 lines
9.7 KiB
Python
"""Tests for deterministic recommendation eligibility logic."""
|
|
from typing import Any
|
|
|
|
from services.recommendation.eligibility import (
|
|
DEFAULT_ELIGIBILITY_CONFIG,
|
|
EligibilityConfig,
|
|
RejectionReason,
|
|
evaluate_eligibility,
|
|
)
|
|
from services.shared.schemas import (
|
|
ActionType,
|
|
RecommendationMode,
|
|
TrendDirection,
|
|
TrendSummary,
|
|
TrendWindow,
|
|
)
|
|
|
|
|
|
def _make_summary(**overrides: Any) -> TrendSummary:
|
|
"""Build a TrendSummary with sensible defaults for testing."""
|
|
defaults = dict(
|
|
entity_type="company",
|
|
entity_id="AAPL",
|
|
window=TrendWindow.SEVEN_DAY,
|
|
trend_direction=TrendDirection.BULLISH,
|
|
trend_strength=0.5,
|
|
confidence=0.6,
|
|
top_supporting_evidence=["doc1", "doc2", "doc3"],
|
|
top_opposing_evidence=[],
|
|
dominant_catalysts=["earnings"],
|
|
material_risks=["regulatory scrutiny"],
|
|
contradiction_score=0.1,
|
|
)
|
|
defaults.update(overrides)
|
|
return TrendSummary(**defaults)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gate checks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_eligible_strong_bullish():
|
|
"""A strong bullish trend with good confidence passes all gates."""
|
|
summary = _make_summary(
|
|
trend_strength=0.5, confidence=0.6, contradiction_score=0.1,
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.eligible is True
|
|
assert result.rejection_reasons == []
|
|
assert result.action == ActionType.BUY
|
|
|
|
|
|
def test_rejected_low_confidence():
|
|
"""Below min_confidence → rejected."""
|
|
summary = _make_summary(confidence=0.2)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.eligible is False
|
|
assert RejectionReason.LOW_CONFIDENCE in result.rejection_reasons
|
|
|
|
|
|
def test_rejected_low_strength():
|
|
"""Below min_trend_strength → rejected."""
|
|
summary = _make_summary(trend_strength=0.05)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.eligible is False
|
|
assert RejectionReason.LOW_TREND_STRENGTH in result.rejection_reasons
|
|
|
|
|
|
def test_rejected_high_contradiction():
|
|
"""Above max_contradiction_score → rejected."""
|
|
summary = _make_summary(contradiction_score=0.7)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.eligible is False
|
|
assert RejectionReason.HIGH_CONTRADICTION in result.rejection_reasons
|
|
|
|
|
|
def test_rejected_insufficient_evidence():
|
|
"""Too few evidence documents → rejected."""
|
|
summary = _make_summary(
|
|
top_supporting_evidence=["doc1"],
|
|
top_opposing_evidence=[],
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.eligible is False
|
|
assert RejectionReason.INSUFFICIENT_EVIDENCE in result.rejection_reasons
|
|
|
|
|
|
def test_rejected_neutral_direction():
|
|
"""Neutral trend direction → rejected."""
|
|
summary = _make_summary(trend_direction=TrendDirection.NEUTRAL)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.eligible is False
|
|
assert RejectionReason.NEUTRAL_DIRECTION in result.rejection_reasons
|
|
|
|
|
|
def test_rejected_forces_informational_mode():
|
|
"""Any rejection forces mode to informational (Req 7.4)."""
|
|
summary = _make_summary(confidence=0.2)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.eligible is False
|
|
assert result.mode == RecommendationMode.INFORMATIONAL
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action mapping
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_action_buy_strong_bullish():
|
|
summary = _make_summary(
|
|
trend_direction=TrendDirection.BULLISH, trend_strength=0.4,
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.action == ActionType.BUY
|
|
|
|
|
|
def test_action_sell_strong_bearish():
|
|
summary = _make_summary(
|
|
trend_direction=TrendDirection.BEARISH, trend_strength=0.4,
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.action == ActionType.SELL
|
|
|
|
|
|
def test_action_hold_weak_bullish_decent_confidence():
|
|
"""Weak bullish with decent confidence → HOLD."""
|
|
summary = _make_summary(
|
|
trend_direction=TrendDirection.BULLISH,
|
|
trend_strength=0.15,
|
|
confidence=0.55,
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.action == ActionType.HOLD
|
|
|
|
|
|
def test_action_watch_weak_bullish_low_confidence():
|
|
"""Weak bullish with low confidence → WATCH."""
|
|
summary = _make_summary(
|
|
trend_direction=TrendDirection.BULLISH,
|
|
trend_strength=0.15,
|
|
confidence=0.40,
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.action == ActionType.WATCH
|
|
|
|
|
|
def test_action_watch_mixed():
|
|
summary = _make_summary(trend_direction=TrendDirection.MIXED)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.action == ActionType.WATCH
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mode escalation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_mode_informational_for_hold():
|
|
"""HOLD actions are always informational."""
|
|
summary = _make_summary(
|
|
trend_direction=TrendDirection.BULLISH,
|
|
trend_strength=0.15,
|
|
confidence=0.55,
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.action == ActionType.HOLD
|
|
assert result.mode == RecommendationMode.INFORMATIONAL
|
|
|
|
|
|
def test_mode_paper_eligible():
|
|
"""BUY with confidence >= paper threshold → paper_eligible."""
|
|
summary = _make_summary(
|
|
trend_strength=0.4, confidence=0.55, contradiction_score=0.1,
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.action == ActionType.BUY
|
|
assert result.mode == RecommendationMode.PAPER_ELIGIBLE
|
|
|
|
|
|
def test_mode_live_eligible():
|
|
"""BUY with high confidence, low contradiction, enough evidence → live_eligible."""
|
|
summary = _make_summary(
|
|
trend_strength=0.5,
|
|
confidence=0.75,
|
|
contradiction_score=0.1,
|
|
top_supporting_evidence=["d1", "d2", "d3", "d4"],
|
|
top_opposing_evidence=["d5"],
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.action == ActionType.BUY
|
|
assert result.mode == RecommendationMode.LIVE_ELIGIBLE
|
|
|
|
|
|
def test_mode_not_live_high_contradiction():
|
|
"""High contradiction blocks live even with high confidence."""
|
|
summary = _make_summary(
|
|
trend_strength=0.5,
|
|
confidence=0.75,
|
|
contradiction_score=0.4,
|
|
top_supporting_evidence=["d1", "d2", "d3", "d4", "d5"],
|
|
top_opposing_evidence=[],
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.mode != RecommendationMode.LIVE_ELIGIBLE
|
|
|
|
|
|
def test_mode_informational_low_confidence_buy():
|
|
"""BUY with confidence below paper threshold → informational."""
|
|
summary = _make_summary(
|
|
trend_strength=0.4, confidence=0.40,
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.action == ActionType.BUY
|
|
assert result.mode == RecommendationMode.INFORMATIONAL
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Position sizing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_position_sizing_scales_with_confidence():
|
|
"""Higher confidence → larger portfolio allocation."""
|
|
low = _make_summary(confidence=0.40, trend_strength=0.4)
|
|
high = _make_summary(confidence=0.80, trend_strength=0.4)
|
|
r_low = evaluate_eligibility(low)
|
|
r_high = evaluate_eligibility(high)
|
|
assert r_high.position_sizing.portfolio_pct > r_low.position_sizing.portfolio_pct
|
|
|
|
|
|
def test_position_sizing_penalised_by_contradiction():
|
|
"""Higher contradiction → smaller portfolio allocation."""
|
|
clean = _make_summary(contradiction_score=0.05, trend_strength=0.4)
|
|
messy = _make_summary(contradiction_score=0.50, trend_strength=0.4)
|
|
r_clean = evaluate_eligibility(clean)
|
|
r_messy = evaluate_eligibility(messy)
|
|
assert r_clean.position_sizing.portfolio_pct > r_messy.position_sizing.portfolio_pct
|
|
|
|
|
|
def test_position_sizing_within_bounds():
|
|
"""Sizing should always stay within configured bounds."""
|
|
cfg = DEFAULT_ELIGIBILITY_CONFIG
|
|
for conf in [0.35, 0.5, 0.7, 0.9]:
|
|
for contra in [0.0, 0.3, 0.55]:
|
|
summary = _make_summary(confidence=conf, contradiction_score=contra, trend_strength=0.4)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.position_sizing.portfolio_pct >= cfg.base_portfolio_pct * 0.5
|
|
assert result.position_sizing.portfolio_pct <= cfg.max_portfolio_pct
|
|
assert result.position_sizing.max_loss_pct >= cfg.base_max_loss_pct * 0.5
|
|
assert result.position_sizing.max_loss_pct <= cfg.max_max_loss_pct
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Time horizon and invalidation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_time_horizon_mapped():
|
|
summary = _make_summary(window=TrendWindow.SEVEN_DAY)
|
|
result = evaluate_eligibility(summary)
|
|
assert result.time_horizon == "swing_1d_10d"
|
|
|
|
|
|
def test_invalidation_conditions_present():
|
|
summary = _make_summary()
|
|
result = evaluate_eligibility(summary)
|
|
assert len(result.invalidation_conditions) > 0
|
|
assert any("AAPL" in c for c in result.invalidation_conditions)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Custom config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_custom_config_stricter_gates():
|
|
"""A stricter config rejects what the default would accept."""
|
|
strict = EligibilityConfig(min_confidence=0.80)
|
|
summary = _make_summary(confidence=0.60)
|
|
result = evaluate_eligibility(summary, config=strict)
|
|
assert result.eligible is False
|
|
assert RejectionReason.LOW_CONFIDENCE in result.rejection_reasons
|