321 lines
11 KiB
Python
321 lines
11 KiB
Python
import httpx
|
|
from typing import List, Dict
|
|
from datetime import datetime, timedelta
|
|
from textblob import TextBlob
|
|
import hashlib
|
|
from app.config import settings
|
|
from app.schemas.schemas import (
|
|
NewsArticle,
|
|
NewsFeedResponse,
|
|
Sentiment,
|
|
EconomicEvent,
|
|
EconomicCalendarResponse,
|
|
)
|
|
|
|
|
|
class NewsService:
|
|
def __init__(self):
|
|
self.alpha_vantage_key = settings.ALPHA_VANTAGE_API_KEY
|
|
self.finnhub_key = settings.FINNHUB_API_KEY
|
|
self.news_api_key = settings.NEWS_API_KEY
|
|
|
|
# Gold-related keywords for relevance scoring
|
|
self.gold_keywords = {
|
|
"high_relevance": [
|
|
"gold", "xau", "precious metals", "bullion", "gold price",
|
|
"gold market", "gold trading", "gold miners", "gold etf"
|
|
],
|
|
"medium_relevance": [
|
|
"federal reserve", "fed", "inflation", "interest rates",
|
|
"dollar", "usd", "monetary policy", "central bank",
|
|
"jerome powell", "treasury", "bonds"
|
|
],
|
|
"context_relevance": [
|
|
"geopolitics", "war", "sanctions", "recession",
|
|
"crisis", "safe haven", "risk off", "uncertainty"
|
|
]
|
|
}
|
|
|
|
# Impact categories
|
|
self.impact_categories = {
|
|
"MONETARY_POLICY": ["federal reserve", "fed", "interest rate", "monetary policy", "central bank"],
|
|
"GEOPOLITICS": ["war", "conflict", "sanctions", "tension", "geopolitical"],
|
|
"ECONOMIC_DATA": ["inflation", "cpi", "gdp", "employment", "jobs", "unemployment"],
|
|
"MARKET_SENTIMENT": ["risk", "sentiment", "volatility", "safe haven"],
|
|
"COMMODITY": ["gold", "precious metals", "bullion", "commodities"],
|
|
}
|
|
|
|
def _calculate_relevance_score(self, text: str) -> float:
|
|
"""Calculate how relevant a news article is to gold trading"""
|
|
text_lower = text.lower()
|
|
score = 0.0
|
|
|
|
# High relevance keywords
|
|
for keyword in self.gold_keywords["high_relevance"]:
|
|
if keyword in text_lower:
|
|
score += 0.4
|
|
|
|
# Medium relevance keywords
|
|
for keyword in self.gold_keywords["medium_relevance"]:
|
|
if keyword in text_lower:
|
|
score += 0.2
|
|
|
|
# Context relevance keywords
|
|
for keyword in self.gold_keywords["context_relevance"]:
|
|
if keyword in text_lower:
|
|
score += 0.1
|
|
|
|
return min(score, 1.0)
|
|
|
|
def _categorize_news(self, text: str) -> str:
|
|
"""Categorize news based on content"""
|
|
text_lower = text.lower()
|
|
|
|
for category, keywords in self.impact_categories.items():
|
|
for keyword in keywords:
|
|
if keyword in text_lower:
|
|
return category
|
|
|
|
return "OTHER"
|
|
|
|
def _analyze_sentiment(self, text: str) -> tuple[Sentiment, float]:
|
|
"""Analyze sentiment using TextBlob"""
|
|
try:
|
|
analysis = TextBlob(text)
|
|
polarity = analysis.sentiment.polarity
|
|
|
|
if polarity > 0.1:
|
|
sentiment = Sentiment.POSITIVE
|
|
elif polarity < -0.1:
|
|
sentiment = Sentiment.NEGATIVE
|
|
else:
|
|
sentiment = Sentiment.NEUTRAL
|
|
|
|
return sentiment, polarity
|
|
except Exception:
|
|
return Sentiment.NEUTRAL, 0.0
|
|
|
|
def _assess_gold_impact(self, sentiment: Sentiment, category: str, relevance: float) -> str:
|
|
"""Assess impact level on gold prices"""
|
|
# High impact categories
|
|
high_impact_cats = ["MONETARY_POLICY", "ECONOMIC_DATA"]
|
|
|
|
if relevance > 0.7:
|
|
if category in high_impact_cats:
|
|
return "HIGH"
|
|
return "MEDIUM"
|
|
elif relevance > 0.4:
|
|
return "MEDIUM"
|
|
else:
|
|
return "LOW"
|
|
|
|
async def fetch_alpha_vantage_news(self, topics: str = "economy_monetary,finance") -> List[NewsArticle]:
|
|
"""Fetch news from Alpha Vantage News Sentiment API"""
|
|
try:
|
|
params = {
|
|
"function": "NEWS_SENTIMENT",
|
|
"topics": topics,
|
|
"limit": 50,
|
|
"apikey": self.alpha_vantage_key,
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
response = await client.get(
|
|
settings.ALPHA_VANTAGE_BASE_URL,
|
|
params=params
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
if "feed" not in data:
|
|
return []
|
|
|
|
articles = []
|
|
for item in data["feed"]:
|
|
title = item.get("title", "")
|
|
summary = item.get("summary", "")
|
|
full_text = f"{title} {summary}"
|
|
|
|
relevance = self._calculate_relevance_score(full_text)
|
|
|
|
# Filter only gold-relevant news
|
|
if relevance < 0.3:
|
|
continue
|
|
|
|
sentiment, score = self._analyze_sentiment(full_text)
|
|
category = self._categorize_news(full_text)
|
|
impact = self._assess_gold_impact(sentiment, category, relevance)
|
|
|
|
# Parse published date
|
|
published_str = item.get("time_published", "")
|
|
try:
|
|
published_at = datetime.strptime(published_str, "%Y%m%dT%H%M%S")
|
|
except:
|
|
published_at = datetime.now()
|
|
|
|
article_id = hashlib.md5(f"{title}{published_str}".encode()).hexdigest()
|
|
|
|
articles.append(
|
|
NewsArticle(
|
|
id=article_id,
|
|
source=item.get("source", "Alpha Vantage"),
|
|
title=title,
|
|
description=summary,
|
|
url=item.get("url", ""),
|
|
published_at=published_at,
|
|
sentiment=sentiment,
|
|
sentiment_score=score,
|
|
impact_on_gold=impact,
|
|
relevance_score=relevance,
|
|
category=category,
|
|
)
|
|
)
|
|
|
|
return articles
|
|
|
|
except Exception as e:
|
|
print(f"Error fetching Alpha Vantage news: {e}")
|
|
return []
|
|
|
|
async def fetch_finnhub_news(self) -> List[NewsArticle]:
|
|
"""Fetch gold-related news from Finnhub"""
|
|
if not self.finnhub_key:
|
|
return []
|
|
|
|
try:
|
|
# Get general market news
|
|
params = {
|
|
"category": "forex",
|
|
"token": self.finnhub_key,
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
response = await client.get(
|
|
f"{settings.FINNHUB_BASE_URL}/news",
|
|
params=params
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
articles = []
|
|
for item in data[:50]: # Limit to 50 articles
|
|
title = item.get("headline", "")
|
|
summary = item.get("summary", "")
|
|
full_text = f"{title} {summary}"
|
|
|
|
relevance = self._calculate_relevance_score(full_text)
|
|
|
|
# Filter only gold-relevant news
|
|
if relevance < 0.3:
|
|
continue
|
|
|
|
sentiment, score = self._analyze_sentiment(full_text)
|
|
category = self._categorize_news(full_text)
|
|
impact = self._assess_gold_impact(sentiment, category, relevance)
|
|
|
|
published_at = datetime.fromtimestamp(item.get("datetime", 0))
|
|
article_id = hashlib.md5(f"{title}{item.get('id', '')}".encode()).hexdigest()
|
|
|
|
articles.append(
|
|
NewsArticle(
|
|
id=article_id,
|
|
source=item.get("source", "Finnhub"),
|
|
title=title,
|
|
description=summary,
|
|
url=item.get("url", ""),
|
|
published_at=published_at,
|
|
sentiment=sentiment,
|
|
sentiment_score=score,
|
|
impact_on_gold=impact,
|
|
relevance_score=relevance,
|
|
category=category,
|
|
)
|
|
)
|
|
|
|
return articles
|
|
|
|
except Exception as e:
|
|
print(f"Error fetching Finnhub news: {e}")
|
|
return []
|
|
|
|
async def get_aggregated_news_feed(self) -> NewsFeedResponse:
|
|
"""Get aggregated news from all sources"""
|
|
# Fetch from multiple sources
|
|
alpha_news = await self.fetch_alpha_vantage_news()
|
|
finnhub_news = await self.fetch_finnhub_news() if self.finnhub_key else []
|
|
|
|
# Combine and deduplicate
|
|
all_articles = alpha_news + finnhub_news
|
|
|
|
# Remove duplicates based on similar titles
|
|
unique_articles = []
|
|
seen_titles = set()
|
|
|
|
for article in all_articles:
|
|
title_key = article.title.lower()[:50] # First 50 chars
|
|
if title_key not in seen_titles:
|
|
seen_titles.add(title_key)
|
|
unique_articles.append(article)
|
|
|
|
# Sort by published date (newest first)
|
|
unique_articles.sort(key=lambda x: x.published_at, reverse=True)
|
|
|
|
# Limit to most recent 50
|
|
unique_articles = unique_articles[:50]
|
|
|
|
# Calculate statistics
|
|
bullish_count = sum(1 for a in unique_articles if a.sentiment == Sentiment.POSITIVE)
|
|
bearish_count = sum(1 for a in unique_articles if a.sentiment == Sentiment.NEGATIVE)
|
|
neutral_count = sum(1 for a in unique_articles if a.sentiment == Sentiment.NEUTRAL)
|
|
|
|
avg_sentiment = (
|
|
sum(a.sentiment_score for a in unique_articles) / len(unique_articles)
|
|
if unique_articles else 0.0
|
|
)
|
|
|
|
# Determine overall sentiment
|
|
if avg_sentiment > 0.1:
|
|
overall_sentiment = Sentiment.POSITIVE
|
|
elif avg_sentiment < -0.1:
|
|
overall_sentiment = Sentiment.NEGATIVE
|
|
else:
|
|
overall_sentiment = Sentiment.NEUTRAL
|
|
|
|
return NewsFeedResponse(
|
|
articles=unique_articles,
|
|
total_count=len(unique_articles),
|
|
bullish_count=bullish_count,
|
|
bearish_count=bearish_count,
|
|
neutral_count=neutral_count,
|
|
overall_sentiment=overall_sentiment,
|
|
avg_sentiment_score=avg_sentiment,
|
|
)
|
|
|
|
async def get_economic_calendar(self) -> EconomicCalendarResponse:
|
|
"""Get upcoming economic events that impact gold"""
|
|
# This would integrate with economic calendar APIs
|
|
# For MVP, return curated list of upcoming events
|
|
|
|
# In production, integrate with:
|
|
# - Forex Factory API
|
|
# - Investing.com Economic Calendar
|
|
# - Alpha Vantage Economic Indicators
|
|
|
|
# For now, return empty with structure
|
|
events = []
|
|
|
|
# Count high-impact upcoming events
|
|
now = datetime.now()
|
|
upcoming_high_impact = sum(
|
|
1 for e in events
|
|
if e.importance == "HIGH" and e.event_date > now
|
|
)
|
|
|
|
return EconomicCalendarResponse(
|
|
events=events,
|
|
upcoming_high_impact=upcoming_high_impact,
|
|
)
|
|
|
|
|
|
news_service = NewsService()
|