1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
|
"""Tests for sentiment aggregator."""
import pytest
from datetime import datetime, timezone, timedelta
from shared.sentiment import SentimentAggregator
@pytest.fixture
def aggregator():
return SentimentAggregator()
def test_freshness_decay_recent():
a = SentimentAggregator()
now = datetime.now(timezone.utc)
assert a._freshness_decay(now, now) == 1.0
def test_freshness_decay_3_hours():
a = SentimentAggregator()
now = datetime.now(timezone.utc)
assert a._freshness_decay(now - timedelta(hours=3), now) == 0.7
def test_freshness_decay_12_hours():
a = SentimentAggregator()
now = datetime.now(timezone.utc)
assert a._freshness_decay(now - timedelta(hours=12), now) == 0.3
def test_freshness_decay_old():
a = SentimentAggregator()
now = datetime.now(timezone.utc)
assert a._freshness_decay(now - timedelta(days=2), now) == 0.0
def test_compute_composite():
a = SentimentAggregator()
composite = a._compute_composite(news_score=0.5, social_score=0.3, policy_score=0.8, filing_score=0.2)
expected = 0.5 * 0.3 + 0.3 * 0.2 + 0.8 * 0.3 + 0.2 * 0.2
assert abs(composite - expected) < 0.001
def test_aggregate_news_by_symbol(aggregator):
now = datetime.now(timezone.utc)
news_items = [
{"symbols": ["AAPL"], "sentiment": 0.8, "category": "earnings", "published_at": now},
{"symbols": ["AAPL"], "sentiment": 0.3, "category": "macro", "published_at": now - timedelta(hours=2)},
{"symbols": ["MSFT"], "sentiment": -0.5, "category": "policy", "published_at": now},
]
scores = aggregator.aggregate(news_items, now)
assert "AAPL" in scores
assert "MSFT" in scores
assert scores["AAPL"].news_count == 2
assert scores["AAPL"].news_score > 0
assert scores["MSFT"].policy_score < 0
def test_aggregate_empty(aggregator):
now = datetime.now(timezone.utc)
assert aggregator.aggregate([], now) == {}
def test_determine_regime():
a = SentimentAggregator()
assert a.determine_regime(15, None) == "risk_off"
assert a.determine_regime(15, 35.0) == "risk_off"
assert a.determine_regime(50, 35.0) == "risk_off"
assert a.determine_regime(70, 15.0) == "risk_on"
assert a.determine_regime(50, 20.0) == "neutral"
|
