blob: 2caa2667446b0d19b955e6a1d8e2526f2b3c4034 (
plain)
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
"""Tests for market sentiment module."""
import pytest
from unittest.mock import AsyncMock, MagicMock
from shared.sentiment import SentimentData, SentimentProvider
# --- SentimentData tests ---
def test_sentiment_should_buy_on_fear():
s = SentimentData(fear_greed_value=15) # Extreme fear
assert s.should_buy is True
def test_sentiment_should_not_buy_on_greed():
s = SentimentData(fear_greed_value=75) # Greed
assert s.should_buy is False
def test_sentiment_should_block_extreme_greed():
s = SentimentData(fear_greed_value=85)
assert s.should_block is True
def test_sentiment_should_block_very_negative_news():
s = SentimentData(news_sentiment=-0.6)
assert s.should_block is True
def test_sentiment_no_block_on_neutral():
s = SentimentData(fear_greed_value=50, news_sentiment=0.0)
assert s.should_block is False
def test_sentiment_should_buy_default_no_data():
s = SentimentData()
assert s.should_buy is True
assert s.should_block is False
def test_sentiment_positive_news_allows_buy():
s = SentimentData(fear_greed_value=50, news_sentiment=0.3)
assert s.should_buy is True
def test_sentiment_outflow_bullish():
s = SentimentData(exchange_netflow=-100.0) # Outflow = bullish
assert s.should_buy is True
def test_sentiment_inflow_bearish():
s = SentimentData(fear_greed_value=50, exchange_netflow=100.0) # Inflow = bearish
assert s.should_buy is False
# --- SentimentProvider tests ---
@pytest.mark.asyncio
async def test_provider_fetch_fear_greed():
provider = SentimentProvider()
mock_response = AsyncMock()
mock_response.status = 200
mock_response.json = AsyncMock(
return_value={"data": [{"value": "25", "value_classification": "Extreme Fear"}]}
)
mock_response.__aenter__ = AsyncMock(return_value=mock_response)
mock_response.__aexit__ = AsyncMock(return_value=False)
mock_session = MagicMock()
mock_session.closed = False
mock_session.get = MagicMock(return_value=mock_response)
mock_session.close = AsyncMock()
provider._session = mock_session
value, label = await provider.fetch_fear_greed()
assert value == 25
assert label == "Extreme Fear"
await provider.close()
@pytest.mark.asyncio
async def test_provider_fetch_fear_greed_failure():
provider = SentimentProvider()
mock_response = AsyncMock()
mock_response.status = 500
mock_response.__aenter__ = AsyncMock(return_value=mock_response)
mock_response.__aexit__ = AsyncMock(return_value=False)
mock_session = MagicMock()
mock_session.closed = False
mock_session.get = MagicMock(return_value=mock_response)
mock_session.close = AsyncMock()
provider._session = mock_session
value, label = await provider.fetch_fear_greed()
assert value is None
await provider.close()
@pytest.mark.asyncio
async def test_provider_news_disabled_without_key():
provider = SentimentProvider(cryptopanic_api_key="")
score, count = await provider.fetch_news_sentiment()
assert score is None
assert count == 0
@pytest.mark.asyncio
async def test_provider_netflow_disabled_without_key():
provider = SentimentProvider(cryptoquant_api_key="")
result = await provider.fetch_exchange_netflow()
assert result is None
@pytest.mark.asyncio
async def test_provider_get_sentiment_aggregates():
provider = SentimentProvider()
mock_response = AsyncMock()
mock_response.status = 200
mock_response.json = AsyncMock(
return_value={"data": [{"value": "20", "value_classification": "Extreme Fear"}]}
)
mock_response.__aenter__ = AsyncMock(return_value=mock_response)
mock_response.__aexit__ = AsyncMock(return_value=False)
mock_session = MagicMock()
mock_session.closed = False
mock_session.get = MagicMock(return_value=mock_response)
mock_session.close = AsyncMock()
provider._session = mock_session
sentiment = await provider.get_sentiment("SOL")
assert sentiment.fear_greed_value == 20
assert sentiment.fear_greed_label == "Extreme Fear"
assert provider.cached is sentiment
await provider.close()
|
