94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
from __future__ import annotations
|
|
from datetime import datetime, timedelta
|
|
from newsapi import NewsApiClient
|
|
from ai.claude_client import ClaudeClient
|
|
from ai.prompts import SENTIMENT_SCORE_SYSTEM, SENTIMENT_SCORE_PROMPT
|
|
|
|
|
|
def fetch_news(symbol: str, api_key: str, page_size: int = 10) -> list[dict]:
|
|
try:
|
|
client = NewsApiClient(api_key=api_key)
|
|
from_date = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
|
|
response = client.get_everything(
|
|
q=symbol,
|
|
language="en",
|
|
sort_by="publishedAt",
|
|
page_size=page_size,
|
|
from_param=from_date,
|
|
)
|
|
articles = response.get("articles", [])
|
|
return [
|
|
{
|
|
"title": a.get("title", ""),
|
|
"description": a.get("description", ""),
|
|
"url": a.get("url", ""),
|
|
"source": a.get("source", {}).get("name", ""),
|
|
"published_at": a.get("publishedAt", ""),
|
|
}
|
|
for a in articles
|
|
if a.get("title")
|
|
]
|
|
except Exception as e:
|
|
return []
|
|
|
|
|
|
def analyze_sentiment(symbol: str, articles: list[dict], claude: ClaudeClient) -> tuple[str, dict]:
|
|
if not articles:
|
|
return "No news available for sentiment analysis.", {}
|
|
|
|
headlines = "\n".join(
|
|
f"{i + 1}. {a['title']}" for i, a in enumerate(articles[:15])
|
|
)
|
|
prompt = SENTIMENT_SCORE_PROMPT.format(symbol=symbol, headlines=headlines)
|
|
text, usage = claude.ask(prompt, system=SENTIMENT_SCORE_SYSTEM, max_tokens=800)
|
|
return text, usage
|
|
|
|
|
|
def cache_news(symbol: str, articles: list[dict], sentiment: str = "") -> None:
|
|
from db.database import get_session
|
|
from db.models import NewsCache
|
|
from datetime import datetime
|
|
|
|
with get_session() as session:
|
|
session.query(NewsCache).filter_by(symbol=symbol).delete()
|
|
for a in articles:
|
|
session.add(NewsCache(
|
|
symbol=symbol,
|
|
title=a.get("title", ""),
|
|
description=a.get("description", ""),
|
|
url=a.get("url", ""),
|
|
source=a.get("source", ""),
|
|
published_at=a.get("published_at", ""),
|
|
sentiment=sentiment,
|
|
fetched_at=datetime.utcnow(),
|
|
))
|
|
session.commit()
|
|
|
|
|
|
def get_cached_news(symbol: str, max_age_hours: int = 1) -> list[dict] | None:
|
|
from db.database import get_session
|
|
from db.models import NewsCache
|
|
from datetime import datetime, timedelta
|
|
|
|
cutoff = datetime.utcnow() - timedelta(hours=max_age_hours)
|
|
with get_session() as session:
|
|
rows = (
|
|
session.query(NewsCache)
|
|
.filter(NewsCache.symbol == symbol, NewsCache.fetched_at > cutoff)
|
|
.order_by(NewsCache.published_at.desc())
|
|
.all()
|
|
)
|
|
if not rows:
|
|
return None
|
|
return [
|
|
{
|
|
"title": r.title,
|
|
"description": r.description,
|
|
"url": r.url,
|
|
"source": r.source,
|
|
"published_at": r.published_at,
|
|
"sentiment": r.sentiment,
|
|
}
|
|
for r in rows
|
|
]
|