blob: 42e8f8834c5d1fa9b82391fd6101609ad995587e (
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
|
"""CNN Fear & Greed Index collector."""
import logging
from dataclasses import dataclass
import aiohttp
from news_collector.collectors.base import BaseCollector
logger = logging.getLogger(__name__)
FEAR_GREED_URL = "https://production.dataviz.cnn.io/index/fearandgreed/graphdata"
@dataclass
class FearGreedResult:
fear_greed: int
fear_greed_label: str
class FearGreedCollector(BaseCollector):
name = "fear_greed"
poll_interval = 3600 # 1 hour
async def is_available(self) -> bool:
return True
async def _fetch_index(self) -> dict | None:
headers = {"User-Agent": "Mozilla/5.0"}
try:
async with aiohttp.ClientSession() as session:
async with session.get(
FEAR_GREED_URL, headers=headers, timeout=aiohttp.ClientTimeout(total=10)
) as resp:
if resp.status != 200:
return None
return await resp.json()
except Exception:
return None
def _classify(self, score: int) -> str:
if score <= 20:
return "Extreme Fear"
if score <= 40:
return "Fear"
if score <= 60:
return "Neutral"
if score <= 80:
return "Greed"
return "Extreme Greed"
async def collect(self) -> FearGreedResult | None:
data = await self._fetch_index()
if data is None:
return None
try:
fg = data["fear_and_greed"]
score = int(fg["score"])
label = fg.get("rating", self._classify(score))
return FearGreedResult(fear_greed=score, fear_greed_label=label)
except (KeyError, ValueError, TypeError):
return None
|
