choijaewook
adad553a65
Add ThreadedWebsocketManager-based BinanceWSClient for real-time price streaming, NewsClient for CryptoPanic/NewsAPI fetching with coin filtering, and SocialClient for Reddit post retrieval with keyword filtering and simple keyword-based sentiment scoring. Includes unit tests for news and social clients (4/4 passing). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
32 lines
1.2 KiB
Python
32 lines
1.2 KiB
Python
import pytest
|
|
from data.news_client import NewsClient
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
return NewsClient(cryptopanic_key="test", newsapi_key="test")
|
|
|
|
def test_parse_cryptopanic_response(client):
|
|
raw = {
|
|
"results": [
|
|
{"title": "Bitcoin hits new high", "published_at": "2026-03-20T10:00:00Z",
|
|
"currencies": [{"code": "BTC"}], "kind": "news",
|
|
"votes": {"positive": 10, "negative": 2}},
|
|
{"title": "ETH upgrade coming", "published_at": "2026-03-20T09:00:00Z",
|
|
"currencies": [{"code": "ETH"}], "kind": "news",
|
|
"votes": {"positive": 5, "negative": 1}},
|
|
]
|
|
}
|
|
articles = client.parse_cryptopanic(raw)
|
|
assert len(articles) == 2
|
|
assert articles[0]["coin"] == "BTC"
|
|
assert articles[0]["title"] == "Bitcoin hits new high"
|
|
|
|
def test_get_news_for_coin_filters(client):
|
|
articles = [
|
|
{"coin": "BTC", "title": "BTC news", "sentiment_votes": {"positive": 5, "negative": 1}},
|
|
{"coin": "ETH", "title": "ETH news", "sentiment_votes": {"positive": 3, "negative": 3}},
|
|
]
|
|
btc_news = client.filter_by_coin(articles, "BTC")
|
|
assert len(btc_news) == 1
|
|
assert btc_news[0]["coin"] == "BTC"
|