Celes Renata
284 lines
9.4 KiB
Python
284 lines
9.4 KiB
Python
"""Tests for recommendation worker — generating recommendations from trend data.
|
|
|
|
Tests the pure logic functions (no DB required). Async DB functions
|
|
are covered by integration tests.
|
|
"""
|
|
from datetime import datetime, timezone
|
|
|
|
from services.recommendation.eligibility import evaluate_eligibility
|
|
from services.recommendation.worker import (
|
|
_extract_risk_classification,
|
|
build_recommendation,
|
|
build_thesis,
|
|
classify_risk,
|
|
)
|
|
from services.shared.schemas import (
|
|
ActionType,
|
|
RecommendationMode,
|
|
TrendDirection,
|
|
TrendSummary,
|
|
TrendWindow,
|
|
)
|
|
|
|
NOW = datetime(2026, 4, 11, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _make_summary(
|
|
ticker: str = "AAPL",
|
|
direction: TrendDirection = TrendDirection.BULLISH,
|
|
strength: float = 0.5,
|
|
confidence: float = 0.65,
|
|
contradiction: float = 0.1,
|
|
supporting: list[str] | None = None,
|
|
opposing: list[str] | None = None,
|
|
catalysts: list[str] | None = None,
|
|
risks: list[str] | None = None,
|
|
window: TrendWindow = TrendWindow.SEVEN_DAY,
|
|
) -> TrendSummary:
|
|
return TrendSummary(
|
|
entity_type="company",
|
|
entity_id=ticker,
|
|
window=window,
|
|
trend_direction=direction,
|
|
trend_strength=strength,
|
|
confidence=confidence,
|
|
top_supporting_evidence=supporting or ["doc1", "doc2", "doc3"],
|
|
top_opposing_evidence=opposing or [],
|
|
dominant_catalysts=catalysts or ["earnings"],
|
|
material_risks=risks or ["regulatory scrutiny"],
|
|
contradiction_score=contradiction,
|
|
generated_at=NOW,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# build_thesis
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_thesis_contains_ticker_and_direction():
|
|
summary = _make_summary()
|
|
result = evaluate_eligibility(summary)
|
|
thesis = build_thesis(summary, result)
|
|
assert "AAPL" in thesis
|
|
assert "bullish" in thesis
|
|
|
|
|
|
def test_thesis_includes_catalysts():
|
|
summary = _make_summary(catalysts=["product", "m_and_a"])
|
|
result = evaluate_eligibility(summary)
|
|
thesis = build_thesis(summary, result)
|
|
assert "product" in thesis
|
|
|
|
|
|
def test_thesis_includes_contradiction_note():
|
|
summary = _make_summary(contradiction=0.3)
|
|
result = evaluate_eligibility(summary)
|
|
thesis = build_thesis(summary, result)
|
|
assert "disagreement" in thesis
|
|
|
|
|
|
def test_thesis_includes_risks():
|
|
summary = _make_summary(risks=["supply chain disruption"])
|
|
result = evaluate_eligibility(summary)
|
|
thesis = build_thesis(summary, result)
|
|
assert "supply chain disruption" in thesis
|
|
|
|
|
|
def test_thesis_includes_evidence_counts():
|
|
summary = _make_summary(
|
|
supporting=["d1", "d2"],
|
|
opposing=["d3"],
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
thesis = build_thesis(summary, result)
|
|
assert "2 supporting" in thesis
|
|
assert "1 opposing" in thesis
|
|
|
|
|
|
def test_thesis_includes_action():
|
|
summary = _make_summary()
|
|
result = evaluate_eligibility(summary)
|
|
thesis = build_thesis(summary, result)
|
|
assert "BUY" in thesis
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# classify_risk
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_risk_low_for_strong_signal():
|
|
summary = _make_summary(
|
|
confidence=0.8,
|
|
contradiction=0.05,
|
|
supporting=["d1", "d2", "d3", "d4", "d5"],
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert classify_risk(summary, result) == "low"
|
|
|
|
|
|
def test_risk_high_for_weak_signal():
|
|
summary = _make_summary(
|
|
confidence=0.36,
|
|
contradiction=0.55,
|
|
supporting=["d1"],
|
|
opposing=[],
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
risk = classify_risk(summary, result)
|
|
assert risk in ("high", "very_high")
|
|
|
|
|
|
def test_risk_moderate_for_mixed():
|
|
summary = _make_summary(
|
|
confidence=0.5,
|
|
contradiction=0.2,
|
|
supporting=["d1", "d2"],
|
|
opposing=["d3"],
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
assert classify_risk(summary, result) == "moderate"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# build_recommendation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_recommendation_basic():
|
|
summary = _make_summary()
|
|
result = evaluate_eligibility(summary)
|
|
rec = build_recommendation(summary, result, reference_time=NOW)
|
|
|
|
assert rec.ticker == "AAPL"
|
|
assert rec.action == ActionType.BUY
|
|
assert rec.confidence == summary.confidence
|
|
assert rec.time_horizon == "swing_1d_10d"
|
|
assert rec.generated_at == NOW
|
|
assert len(rec.evidence_refs) == 3 # 3 supporting + 0 opposing
|
|
assert rec.model_metadata.provider == "deterministic"
|
|
|
|
|
|
def test_build_recommendation_includes_risk_in_thesis():
|
|
summary = _make_summary()
|
|
result = evaluate_eligibility(summary)
|
|
rec = build_recommendation(summary, result)
|
|
assert rec.thesis.startswith("[risk:")
|
|
|
|
|
|
def test_build_recommendation_with_llm_thesis():
|
|
"""When llm_thesis is provided, it replaces the deterministic body."""
|
|
summary = _make_summary()
|
|
result = evaluate_eligibility(summary)
|
|
llm_text = "Apple exhibits a bullish posture driven by strong earnings."
|
|
rec = build_recommendation(summary, result, llm_thesis=llm_text)
|
|
assert llm_text in rec.thesis
|
|
assert rec.thesis.startswith("[risk:")
|
|
assert rec.model_metadata.provider == "ollama"
|
|
assert rec.model_metadata.model_name == "thesis-rewrite"
|
|
|
|
|
|
def test_build_recommendation_without_llm_thesis_uses_deterministic():
|
|
"""When llm_thesis is None, the deterministic thesis is used."""
|
|
summary = _make_summary()
|
|
result = evaluate_eligibility(summary)
|
|
rec = build_recommendation(summary, result)
|
|
assert rec.model_metadata.provider == "deterministic"
|
|
assert rec.model_metadata.model_name == "eligibility-v1"
|
|
|
|
|
|
def test_build_recommendation_combines_evidence():
|
|
summary = _make_summary(
|
|
supporting=["s1", "s2"],
|
|
opposing=["o1"],
|
|
)
|
|
result = evaluate_eligibility(summary)
|
|
rec = build_recommendation(summary, result)
|
|
assert rec.evidence_refs == ["s1", "s2", "o1"]
|
|
|
|
|
|
def test_build_recommendation_position_sizing():
|
|
summary = _make_summary(confidence=0.7)
|
|
result = evaluate_eligibility(summary)
|
|
rec = build_recommendation(summary, result)
|
|
assert rec.position_sizing.portfolio_pct == result.position_sizing.portfolio_pct
|
|
assert rec.position_sizing.max_loss_pct == result.position_sizing.max_loss_pct
|
|
|
|
|
|
def test_build_recommendation_invalidation_conditions():
|
|
summary = _make_summary()
|
|
result = evaluate_eligibility(summary)
|
|
rec = build_recommendation(summary, result)
|
|
assert len(rec.invalidation_conditions) > 0
|
|
|
|
|
|
def test_build_recommendation_ineligible_is_informational():
|
|
"""When eligibility fails, mode should be informational (Req 7.4)."""
|
|
summary = _make_summary(confidence=0.2)
|
|
result = evaluate_eligibility(summary)
|
|
rec = build_recommendation(summary, result)
|
|
assert rec.mode == RecommendationMode.INFORMATIONAL
|
|
|
|
|
|
def test_build_recommendation_sell_action():
|
|
summary = _make_summary(direction=TrendDirection.BEARISH, strength=0.5)
|
|
result = evaluate_eligibility(summary)
|
|
rec = build_recommendation(summary, result)
|
|
assert rec.action == ActionType.SELL
|
|
assert "SELL" in rec.thesis
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extract_risk_classification
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_extract_risk_classification_from_thesis():
|
|
assert _extract_risk_classification("[risk:low] Some thesis text") == "low"
|
|
assert _extract_risk_classification("[risk:very_high] Bad signal") == "very_high"
|
|
|
|
|
|
def test_extract_risk_classification_missing_prefix():
|
|
assert _extract_risk_classification("No risk prefix here") == "moderate"
|
|
|
|
|
|
def test_extract_risk_classification_empty():
|
|
assert _extract_risk_classification("") == "moderate"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# build_recommendation stores full model metadata
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_recommendation_model_metadata_deterministic():
|
|
summary = _make_summary()
|
|
result = evaluate_eligibility(summary)
|
|
rec = build_recommendation(summary, result, reference_time=NOW)
|
|
assert rec.model_metadata.provider == "deterministic"
|
|
assert rec.model_metadata.model_name == "eligibility-v1"
|
|
assert rec.model_metadata.schema_version == "1.0.0"
|
|
|
|
|
|
def test_build_recommendation_model_metadata_llm():
|
|
summary = _make_summary()
|
|
result = evaluate_eligibility(summary)
|
|
rec = build_recommendation(
|
|
summary, result, reference_time=NOW,
|
|
llm_thesis="Rewritten thesis text.",
|
|
)
|
|
assert rec.model_metadata.provider == "ollama"
|
|
assert rec.model_metadata.model_name == "thesis-rewrite"
|
|
assert rec.model_metadata.prompt_version != ""
|
|
|
|
|
|
def test_build_recommendation_risk_classification_in_thesis():
|
|
"""The risk classification should be embedded in the thesis prefix."""
|
|
summary = _make_summary(confidence=0.8, contradiction=0.05,
|
|
supporting=["d1", "d2", "d3", "d4", "d5"])
|
|
result = evaluate_eligibility(summary)
|
|
rec = build_recommendation(summary, result, reference_time=NOW)
|
|
risk = _extract_risk_classification(rec.thesis)
|
|
assert risk == classify_risk(summary, result)
|