You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
150 lines
5.1 KiB
Python
150 lines
5.1 KiB
Python
"""
|
|
Yahoo Finance RSS feed adapter - collect headline events and match to tickers.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Dict, List, Optional
|
|
|
|
import aiohttp
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
|
|
from app.core.http_client import get_http_session
|
|
|
|
from app.models.overlay_raw_event import OverlayHeadlineEvent
|
|
from app.services.overlay.entity_resolver import EntityResolver
|
|
from app.core.overlay_config import YAHOO_RSS_FEEDS, TOP_50_SYMBOLS
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class YahooRSSAdapter:
|
|
"""Collect Yahoo Finance RSS headlines and match to tickers."""
|
|
|
|
def __init__(self):
|
|
self.resolver = EntityResolver()
|
|
|
|
async def fetch_feed(self, url: str) -> List[Dict]:
|
|
"""Fetch and parse a single RSS feed URL."""
|
|
try:
|
|
import feedparser
|
|
except ImportError:
|
|
logger.error("feedparser not installed; Yahoo RSS adapter inactive")
|
|
return []
|
|
|
|
try:
|
|
session = await get_http_session()
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0 StockOracle/1.0",
|
|
"Accept": "application/rss+xml, application/xml, text/xml",
|
|
}
|
|
async with session.get(
|
|
url, headers=headers, timeout=aiohttp.ClientTimeout(total=30)
|
|
) as resp:
|
|
if resp.status != 200:
|
|
logger.warning(f"Yahoo RSS: non-200 from {url}: {resp.status}")
|
|
return []
|
|
text = await resp.text()
|
|
|
|
feed = feedparser.parse(text)
|
|
entries = []
|
|
for entry in feed.entries:
|
|
guid = getattr(entry, "id", None) or getattr(entry, "link", None)
|
|
title = getattr(entry, "title", "")
|
|
publisher = getattr(entry, "publisher", None)
|
|
if not publisher:
|
|
src = getattr(entry, "source", {})
|
|
publisher = src.get("title") if isinstance(src, dict) else None
|
|
|
|
published = getattr(entry, "published_parsed", None)
|
|
if published:
|
|
try:
|
|
pub_dt = datetime(*published[:6], tzinfo=timezone.utc)
|
|
except Exception:
|
|
pub_dt = datetime.now(timezone.utc)
|
|
else:
|
|
pub_dt = datetime.now(timezone.utc)
|
|
|
|
entries.append({
|
|
"guid": guid or title,
|
|
"title": title,
|
|
"publisher": publisher,
|
|
"published_at": pub_dt,
|
|
})
|
|
return entries
|
|
|
|
except Exception as e:
|
|
logger.error(f"Yahoo RSS fetch error for {url}: {e}")
|
|
return []
|
|
|
|
async def collect(self, db: AsyncSession, symbols: Optional[List[str]] = None) -> int:
|
|
"""
|
|
Collect RSS headlines from general feeds and per-symbol feeds.
|
|
|
|
Persists new events (deduped by article_guid), skips duplicates.
|
|
Returns number of new records inserted.
|
|
"""
|
|
await self.resolver.load_aliases(db)
|
|
|
|
all_entries: List[Dict] = []
|
|
|
|
# General feeds
|
|
tasks = [self.fetch_feed(url) for url in YAHOO_RSS_FEEDS]
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
for r in results:
|
|
if isinstance(r, list):
|
|
all_entries.extend(r)
|
|
|
|
# Per-symbol feeds for watch-list
|
|
watch_symbols = symbols or TOP_50_SYMBOLS
|
|
sym_tasks = [
|
|
self.fetch_feed(f"https://finance.yahoo.com/rss/headline?s={sym}")
|
|
for sym in watch_symbols
|
|
]
|
|
sym_results = await asyncio.gather(*sym_tasks, return_exceptions=True)
|
|
for r in sym_results:
|
|
if isinstance(r, list):
|
|
all_entries.extend(r)
|
|
|
|
# Deduplicate by guid within this batch
|
|
seen_guids: set = set()
|
|
unique_entries = []
|
|
for entry in all_entries:
|
|
g = entry.get("guid")
|
|
if g and g not in seen_guids:
|
|
seen_guids.add(g)
|
|
unique_entries.append(entry)
|
|
|
|
inserted = 0
|
|
for entry in unique_entries:
|
|
if not entry.get("guid") or not entry.get("title"):
|
|
continue
|
|
|
|
# Check DB duplicate
|
|
existing = await db.execute(
|
|
select(OverlayHeadlineEvent.id).where(
|
|
OverlayHeadlineEvent.article_guid == entry["guid"]
|
|
)
|
|
)
|
|
if existing.first():
|
|
continue
|
|
|
|
matched = self.resolver.resolve_from_title(entry["title"])
|
|
event = OverlayHeadlineEvent(
|
|
article_guid=entry["guid"],
|
|
title=entry["title"],
|
|
publisher=entry.get("publisher"),
|
|
published_at=entry["published_at"],
|
|
matched_symbols=matched,
|
|
)
|
|
db.add(event)
|
|
inserted += 1
|
|
|
|
if inserted:
|
|
await db.commit()
|
|
logger.info(f"Yahoo RSS: inserted {inserted} new headline events")
|
|
|
|
return inserted
|