Celes Renata
351 lines
13 KiB
Python
351 lines
13 KiB
Python
"""Tests for sector and market rollup aggregation.
|
|
|
|
Tests the pure rollup logic (no DB required).
|
|
|
|
Requirements: 6.3, 6.4, 6.5
|
|
"""
|
|
from datetime import datetime, timezone
|
|
|
|
from services.aggregation.rollups import (
|
|
CompanyTrendRow,
|
|
rollup_trends,
|
|
_build_rollup_disagreement,
|
|
_derive_rollup_direction,
|
|
)
|
|
from services.shared.schemas import TrendDirection, TrendWindow
|
|
|
|
NOW = datetime(2026, 4, 11, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _make_trend(
|
|
ticker: str = "AAPL",
|
|
sector: str = "Technology",
|
|
window: str = "7d",
|
|
direction: str = "bullish",
|
|
strength: float = 0.6,
|
|
confidence: float = 0.8,
|
|
contradiction: float = 0.1,
|
|
catalysts: list[str] | None = None,
|
|
risks: list[str] | None = None,
|
|
supporting: list[str] | None = None,
|
|
opposing: list[str] | None = None,
|
|
) -> CompanyTrendRow:
|
|
return CompanyTrendRow(
|
|
entity_id=ticker,
|
|
sector=sector,
|
|
window=window,
|
|
trend_direction=direction,
|
|
trend_strength=strength,
|
|
confidence=confidence,
|
|
contradiction_score=contradiction,
|
|
dominant_catalysts=catalysts or ["earnings"],
|
|
material_risks=risks or [],
|
|
top_supporting_evidence=supporting or ["doc-1"],
|
|
top_opposing_evidence=opposing or [],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# rollup_trends
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_rollup_empty():
|
|
summary = rollup_trends([], "sector", "Technology", "7d", NOW)
|
|
assert summary.entity_type == "sector"
|
|
assert summary.entity_id == "Technology"
|
|
assert summary.trend_direction == TrendDirection.NEUTRAL
|
|
assert summary.trend_strength == 0.0
|
|
assert summary.confidence == 0.0
|
|
|
|
|
|
def test_rollup_single_bullish():
|
|
trends = [_make_trend("AAPL", direction="bullish", strength=0.7, confidence=0.9)]
|
|
summary = rollup_trends(trends, "sector", "Technology", "7d", NOW)
|
|
assert summary.trend_direction == TrendDirection.BULLISH
|
|
assert summary.trend_strength > 0
|
|
assert summary.confidence > 0
|
|
assert summary.window == TrendWindow.SEVEN_DAY
|
|
|
|
|
|
def test_rollup_mixed_directions():
|
|
trends = [
|
|
_make_trend("AAPL", direction="bullish", strength=0.6, confidence=0.8),
|
|
_make_trend("MSFT", direction="bearish", strength=0.6, confidence=0.8),
|
|
]
|
|
summary = rollup_trends(trends, "sector", "Technology", "7d", NOW)
|
|
# Equal and opposite → neutral or mixed
|
|
assert summary.trend_direction in (TrendDirection.NEUTRAL, TrendDirection.MIXED)
|
|
|
|
|
|
def test_rollup_confidence_weighted():
|
|
"""Higher-confidence company should dominate the rollup direction."""
|
|
trends = [
|
|
_make_trend("AAPL", direction="bullish", strength=0.8, confidence=0.95),
|
|
_make_trend("MSFT", direction="bearish", strength=0.3, confidence=0.2),
|
|
]
|
|
summary = rollup_trends(trends, "sector", "Technology", "7d", NOW)
|
|
assert summary.trend_direction == TrendDirection.BULLISH
|
|
|
|
|
|
def test_rollup_catalysts_aggregated():
|
|
trends = [
|
|
_make_trend("AAPL", catalysts=["earnings", "product"], confidence=0.8),
|
|
_make_trend("MSFT", catalysts=["product", "macro"], confidence=0.6),
|
|
]
|
|
summary = rollup_trends(trends, "sector", "Technology", "7d", NOW)
|
|
# "product" appears in both → should be top catalyst
|
|
assert "product" in summary.dominant_catalysts
|
|
|
|
|
|
def test_rollup_risks_deduplicated():
|
|
trends = [
|
|
_make_trend("AAPL", risks=["regulatory risk", "supply chain"], confidence=0.8),
|
|
_make_trend("MSFT", risks=["Regulatory Risk", "tariffs"], confidence=0.6),
|
|
]
|
|
summary = rollup_trends(trends, "sector", "Technology", "7d", NOW)
|
|
risk_lower = [r.lower() for r in summary.material_risks]
|
|
assert risk_lower.count("regulatory risk") == 1
|
|
|
|
|
|
def test_rollup_evidence_collected():
|
|
trends = [
|
|
_make_trend("AAPL", supporting=["doc-1", "doc-2"], opposing=["doc-3"]),
|
|
_make_trend("MSFT", supporting=["doc-4"], opposing=["doc-5"]),
|
|
]
|
|
summary = rollup_trends(trends, "market", "all", "7d", NOW)
|
|
assert "doc-1" in summary.top_supporting_evidence
|
|
assert "doc-4" in summary.top_supporting_evidence
|
|
assert "doc-3" in summary.top_opposing_evidence
|
|
|
|
|
|
def test_rollup_market_entity_type():
|
|
trends = [_make_trend("AAPL"), _make_trend("JPM", sector="Financials")]
|
|
summary = rollup_trends(trends, "market", "all", "7d", NOW)
|
|
assert summary.entity_type == "market"
|
|
assert summary.entity_id == "all"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _derive_rollup_direction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_derive_direction_bullish():
|
|
assert _derive_rollup_direction(0.5, 0.0) == TrendDirection.BULLISH
|
|
|
|
|
|
def test_derive_direction_bearish():
|
|
assert _derive_rollup_direction(-0.5, 0.0) == TrendDirection.BEARISH
|
|
|
|
|
|
def test_derive_direction_neutral():
|
|
assert _derive_rollup_direction(0.05, 0.0) == TrendDirection.NEUTRAL
|
|
|
|
|
|
def test_derive_direction_mixed_high_contradiction():
|
|
assert _derive_rollup_direction(0.1, 0.2) == TrendDirection.MIXED
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _build_rollup_disagreement
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_disagreement_no_conflict():
|
|
trends = [
|
|
_make_trend("AAPL", direction="bullish"),
|
|
_make_trend("MSFT", direction="bullish"),
|
|
]
|
|
details = _build_rollup_disagreement(trends, "Technology")
|
|
assert details == []
|
|
|
|
|
|
def test_disagreement_with_conflict():
|
|
trends = [
|
|
_make_trend("AAPL", direction="bullish", confidence=0.8),
|
|
_make_trend("MSFT", direction="bearish", confidence=0.7),
|
|
]
|
|
details = _build_rollup_disagreement(trends, "Technology")
|
|
assert len(details) == 1
|
|
assert details[0].dimension == "company_direction"
|
|
assert "AAPL" in details[0].positive_doc_ids
|
|
assert "MSFT" in details[0].negative_doc_ids
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Macro rollup integration (Requirements 6.1, 6.2, 6.3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from services.aggregation.rollups import (
|
|
SectorMacroImpact,
|
|
compute_sector_macro_concentration,
|
|
SECTOR_CONCENTRATION_THRESHOLD,
|
|
)
|
|
|
|
|
|
def _make_sector_macro(
|
|
sector: str = "Technology",
|
|
total_impact: float = 1.0,
|
|
avg_impact: float = 0.5,
|
|
company_count: int = 2,
|
|
net_direction: float = -1.0,
|
|
event_ids: list[str] | None = None,
|
|
) -> SectorMacroImpact:
|
|
return SectorMacroImpact(
|
|
sector=sector,
|
|
total_impact=total_impact,
|
|
avg_impact=avg_impact,
|
|
company_count=company_count,
|
|
net_direction=net_direction,
|
|
event_ids=event_ids or ["evt-1"],
|
|
)
|
|
|
|
|
|
def test_rollup_no_macro_unchanged():
|
|
"""Without macro data, rollup output is identical to original behavior."""
|
|
trends = [_make_trend("AAPL", direction="bullish", strength=0.7, confidence=0.9)]
|
|
without_macro = rollup_trends(trends, "sector", "Technology", "7d", NOW)
|
|
with_none = rollup_trends(trends, "sector", "Technology", "7d", NOW, macro_impacts=None)
|
|
with_empty = rollup_trends(trends, "sector", "Technology", "7d", NOW, macro_impacts={})
|
|
assert without_macro.trend_strength == with_none.trend_strength
|
|
assert without_macro.trend_strength == with_empty.trend_strength
|
|
assert without_macro.confidence == with_none.confidence
|
|
assert without_macro.confidence == with_empty.confidence
|
|
|
|
|
|
def test_sector_rollup_with_macro_adjusts_strength():
|
|
"""Sector rollup with macro data should adjust strength."""
|
|
trends = [
|
|
_make_trend("AAPL", sector="Technology", direction="bullish", strength=0.5, confidence=0.8),
|
|
_make_trend("MSFT", sector="Technology", direction="bullish", strength=0.4, confidence=0.7),
|
|
]
|
|
macro = {"Technology": _make_sector_macro("Technology", total_impact=2.0, avg_impact=0.6, company_count=2)}
|
|
|
|
without = rollup_trends(trends, "sector", "Technology", "7d", NOW)
|
|
with_macro = rollup_trends(trends, "sector", "Technology", "7d", NOW, macro_impacts=macro)
|
|
|
|
# Macro should increase strength
|
|
assert with_macro.trend_strength >= without.trend_strength
|
|
|
|
|
|
def test_sector_rollup_macro_no_match_unchanged():
|
|
"""Sector rollup with macro data for a different sector is unchanged."""
|
|
trends = [_make_trend("AAPL", sector="Technology", direction="bullish", strength=0.5, confidence=0.8)]
|
|
macro = {"Financials": _make_sector_macro("Financials")}
|
|
|
|
without = rollup_trends(trends, "sector", "Technology", "7d", NOW)
|
|
with_macro = rollup_trends(trends, "sector", "Technology", "7d", NOW, macro_impacts=macro)
|
|
|
|
assert without.trend_strength == with_macro.trend_strength
|
|
assert without.confidence == with_macro.confidence
|
|
|
|
|
|
def test_market_rollup_with_macro_adjusts():
|
|
"""Market rollup with macro data should adjust strength and confidence."""
|
|
trends = [
|
|
_make_trend("AAPL", sector="Technology", direction="bullish", strength=0.5, confidence=0.8),
|
|
_make_trend("JPM", sector="Financials", direction="bearish", strength=0.4, confidence=0.7),
|
|
]
|
|
macro = {
|
|
"Technology": _make_sector_macro("Technology", total_impact=1.5, avg_impact=0.5, company_count=1),
|
|
"Financials": _make_sector_macro("Financials", total_impact=0.5, avg_impact=0.3, company_count=1),
|
|
}
|
|
|
|
without = rollup_trends(trends, "market", "all", "7d", NOW)
|
|
with_macro = rollup_trends(trends, "market", "all", "7d", NOW, macro_impacts=macro)
|
|
|
|
# With macro data, at least one of strength or confidence should differ
|
|
differs = (
|
|
with_macro.trend_strength != without.trend_strength
|
|
or with_macro.confidence != without.confidence
|
|
)
|
|
assert differs
|
|
|
|
|
|
def test_market_rollup_disproportionate_sector_surfaced():
|
|
"""When one sector has >60% of macro impact, it appears in risks or catalysts."""
|
|
trends = [
|
|
_make_trend("AAPL", sector="Technology", direction="bullish", strength=0.5, confidence=0.8),
|
|
_make_trend("JPM", sector="Financials", direction="bullish", strength=0.4, confidence=0.7),
|
|
]
|
|
# Technology has 90% of total macro impact
|
|
macro = {
|
|
"Technology": _make_sector_macro("Technology", total_impact=9.0, avg_impact=0.9, company_count=1, net_direction=-1.0),
|
|
"Financials": _make_sector_macro("Financials", total_impact=1.0, avg_impact=0.1, company_count=1, net_direction=0.5),
|
|
}
|
|
|
|
summary = rollup_trends(trends, "market", "all", "7d", NOW, macro_impacts=macro)
|
|
|
|
# Technology should appear in material_risks (negative direction) or dominant_catalysts
|
|
all_labels = summary.material_risks + summary.dominant_catalysts
|
|
tech_found = any("Technology" in label for label in all_labels)
|
|
assert tech_found, f"Expected Technology in risks/catalysts, got: {all_labels}"
|
|
|
|
|
|
def test_market_rollup_no_disproportionate_sector():
|
|
"""When no sector has >60% of macro impact, no macro labels are surfaced."""
|
|
trends = [
|
|
_make_trend("AAPL", sector="Technology", direction="bullish", strength=0.5, confidence=0.8),
|
|
_make_trend("JPM", sector="Financials", direction="bullish", strength=0.4, confidence=0.7),
|
|
]
|
|
# Even split: 50/50
|
|
macro = {
|
|
"Technology": _make_sector_macro("Technology", total_impact=5.0, avg_impact=0.5, company_count=1),
|
|
"Financials": _make_sector_macro("Financials", total_impact=5.0, avg_impact=0.5, company_count=1),
|
|
}
|
|
|
|
summary = rollup_trends(trends, "market", "all", "7d", NOW, macro_impacts=macro)
|
|
|
|
all_labels = summary.material_risks + summary.dominant_catalysts
|
|
macro_labels = [l for l in all_labels if l.startswith("Macro:")]
|
|
assert len(macro_labels) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# compute_sector_macro_concentration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_concentration_empty():
|
|
assert compute_sector_macro_concentration({}) == []
|
|
|
|
|
|
def test_concentration_single_sector():
|
|
impacts = {"Technology": _make_sector_macro("Technology", total_impact=5.0)}
|
|
result = compute_sector_macro_concentration(impacts)
|
|
assert len(result) == 1
|
|
assert result[0] == ("Technology", 1.0)
|
|
|
|
|
|
def test_concentration_multiple_sectors():
|
|
impacts = {
|
|
"Technology": _make_sector_macro("Technology", total_impact=7.0),
|
|
"Financials": _make_sector_macro("Financials", total_impact=3.0),
|
|
}
|
|
result = compute_sector_macro_concentration(impacts)
|
|
assert result[0][0] == "Technology"
|
|
assert abs(result[0][1] - 0.7) < 0.01
|
|
assert result[1][0] == "Financials"
|
|
assert abs(result[1][1] - 0.3) < 0.01
|
|
|
|
|
|
def test_concentration_threshold_boundary():
|
|
"""Exactly at 60% should not be considered disproportionate (>60% required)."""
|
|
impacts = {
|
|
"Technology": _make_sector_macro("Technology", total_impact=6.0),
|
|
"Financials": _make_sector_macro("Financials", total_impact=4.0),
|
|
}
|
|
result = compute_sector_macro_concentration(impacts)
|
|
# 60% is exactly at threshold, not above it
|
|
assert result[0][1] <= SECTOR_CONCENTRATION_THRESHOLD
|
|
|
|
|
|
def test_concentration_above_threshold():
|
|
impacts = {
|
|
"Technology": _make_sector_macro("Technology", total_impact=7.0),
|
|
"Financials": _make_sector_macro("Financials", total_impact=3.0),
|
|
}
|
|
result = compute_sector_macro_concentration(impacts)
|
|
assert result[0][1] > SECTOR_CONCENTRATION_THRESHOLD
|