import datetime as dt import json import logging import httpx from sqlalchemy import select from sqlalchemy.orm import Session from .config import settings from .models import Article, MarketPrice, MarketSpike log = logging.getLogger("newsatlas.markets") # A spread of major indices as rough proxies for national/regional economies, # plus crude oil benchmarks. INSTRUMENTS = [ ("^GSPC", "S&P 500 (US)", "index", "USD"), ("^DJI", "Dow Jones Industrial Average (US)", "index", "USD"), ("^IXIC", "Nasdaq Composite (US)", "index", "USD"), ("^FTSE", "FTSE 100 (UK)", "index", "GBP"), ("^GDAXI", "DAX (Germany)", "index", "EUR"), ("^FCHI", "CAC 40 (France)", "index", "EUR"), ("^N225", "Nikkei 225 (Japan)", "index", "JPY"), ("^HSI", "Hang Seng (Hong Kong/China)", "index", "HKD"), ("000001.SS", "Shanghai Composite (China)", "index", "CNY"), ("^BSESN", "BSE Sensex (India)", "index", "INR"), # Yahoo's feed for this appears frozen since mid-2022 (Western data # providers largely cut off live Russian market data after sanctions); # treat it as historical, not live. ("IMOEX.ME", "MOEX Russia Index", "index", "RUB"), ("^BVSP", "Bovespa (Brazil)", "index", "BRL"), ("CL=F", "WTI Crude Oil", "commodity", "USD"), ("BZ=F", "Brent Crude Oil", "commodity", "USD"), ] # Yahoo's unauthenticated chart endpoint — no crumb/cookie dance required, # unlike the quote/quoteSummary endpoints that the yfinance library wraps # (which have proven flaky here: empty bodies once its session's crumb goes # stale). This is the same data source, just called directly. CHART_URL = "https://query1.finance.yahoo.com/v8/finance/chart/{symbol}" _HEADERS = {"User-Agent": "Mozilla/5.0 (NewsAtlas market poller)"} # Rough relevance hints used only to *rank* candidate-cause articles for a # detected spike — not a claim of causation, just "these are more likely to # be about this instrument than the average headline in the window". _SPIKE_HINTS: dict[str, dict] = { "CL=F": {"keywords": ["oil", "crude", "opec", "barrel", "pipeline", "refinery", "hormuz"], "country": None}, "BZ=F": {"keywords": ["oil", "crude", "opec", "barrel", "pipeline", "refinery", "hormuz"], "country": None}, "^GSPC": {"keywords": ["fed", "federal reserve", "tariff", "inflation", "rate"], "country": "United States"}, "^DJI": {"keywords": ["fed", "federal reserve", "tariff", "inflation", "rate"], "country": "United States"}, "^IXIC": {"keywords": ["fed", "federal reserve", "tariff", "tech", "inflation", "rate"], "country": "United States"}, "^FTSE": {"keywords": ["boe", "bank of england", "tariff", "inflation", "rate"], "country": "United Kingdom"}, "^GDAXI": {"keywords": ["ecb", "european central bank", "tariff", "inflation", "energy"], "country": "Germany"}, "^FCHI": {"keywords": ["ecb", "european central bank", "tariff", "inflation", "energy"], "country": "France"}, "^N225": {"keywords": ["boj", "bank of japan", "yen", "tariff", "inflation"], "country": "Japan"}, "^HSI": {"keywords": ["china", "yuan", "tariff", "property", "beijing"], "country": "China"}, "000001.SS": {"keywords": ["china", "yuan", "tariff", "property", "beijing", "pboc"], "country": "China"}, "^BSESN": {"keywords": ["rbi", "rupee", "tariff", "inflation"], "country": "India"}, "IMOEX.ME": {"keywords": ["sanctions", "ruble", "central bank", "war"], "country": "Russia"}, "^BVSP": {"keywords": ["real", "central bank", "tariff", "inflation", "election"], "country": "Brazil"}, } _GENERIC_SHOCK_KEYWORDS = [ "war", "attack", "strike", "sanctions", "tariff", "default", "crisis", "embargo", "shutdown", "election", "earthquake", "coup", "ceasefire", ] def _fetch_one(client: httpx.Client, symbol: str) -> tuple[float, float | None] | None: resp = client.get( CHART_URL.format(symbol=symbol), params={"range": "5d", "interval": "1d"}, headers=_HEADERS, timeout=15, ) resp.raise_for_status() result = resp.json()["chart"]["result"][0] meta = result["meta"] price = meta.get("regularMarketPrice") if price is None: return None prev_close = meta.get("previousClose") or meta.get("chartPreviousClose") if prev_close is None: closes = [c for c in result["indicators"]["quote"][0]["close"] if c is not None] if len(closes) >= 2: prev_close = closes[-2] change_pct = None if prev_close: change_pct = round((price - prev_close) / prev_close * 100, 3) return float(price), change_pct def _score_article(article: Article, keywords: list[str], country: str | None) -> int: text = f"{article.title} {article.summary}".lower() score = sum(1 for kw in keywords if kw in text) score += sum(1 for kw in _GENERIC_SHOCK_KEYWORDS if kw in text) if country and (article.country == country or country.lower() in text): score += 3 return score def _find_candidate_articles(session: Session, symbol: str, window_start: dt.datetime, window_end: dt.datetime, limit: int = 60) -> list[int]: hints = _SPIKE_HINTS.get(symbol, {"keywords": [], "country": None}) rows = session.execute( select(Article) .where(Article.published_at >= window_start, Article.published_at <= window_end) .order_by(Article.published_at.desc()) .limit(300) # cap the scan; this is a heuristic ranker, not a full-text search index ).scalars().all() # Every article published in the window is a candidate — timing alone is # "possibly related" for the purposes of this log, not just keyword # matches. Score still decides ordering (keyword/country hits surface # first), it just no longer excludes zero-score articles. scored = [(_score_article(a, hints["keywords"], hints["country"]), a) for a in rows] scored.sort(key=lambda pair: (pair[0], pair[1].published_at), reverse=True) return [a.id for _, a in scored[:limit]] def _floor_hour(t: dt.datetime) -> dt.datetime: return t.replace(minute=0, second=0, microsecond=0) def _ceil_hour(t: dt.datetime) -> dt.datetime: floored = _floor_hour(t) return floored if floored == t else floored + dt.timedelta(hours=1) def _detect_and_record_spike( session: Session, symbol: str, label: str, prev: MarketPrice | None, price: float, recorded_at: dt.datetime ) -> None: if prev is None or not prev.price: return pct_change = (price - prev.price) / prev.price * 100 if abs(pct_change) < settings.market_spike_threshold_pct: return # Snapped to whole clock hours so the log reads as clean ranges (e.g. # "14:00-16:00"), covering at least the full hour the spike landed in. window_start = _floor_hour(prev.recorded_at - dt.timedelta(hours=settings.market_spike_lookback_hours)) window_end = _ceil_hour(recorded_at) article_ids = _find_candidate_articles(session, symbol, window_start, window_end) session.add( MarketSpike( symbol=symbol, label=label, from_price=prev.price, to_price=price, pct_change=round(pct_change, 3), window_start=window_start, window_end=window_end, detected_at=recorded_at, article_ids_json=json.dumps(article_ids), ) ) log.info("Spike detected: %s %.2f%% (%d candidate articles)", symbol, pct_change, len(article_ids)) def poll_markets(session: Session) -> int: added = 0 with httpx.Client() as client: for symbol, label, category, currency in INSTRUMENTS: try: result = _fetch_one(client, symbol) if result is None: continue price, change_pct = result recorded_at = dt.datetime.utcnow() prev = session.execute( select(MarketPrice) .where(MarketPrice.symbol == symbol) .order_by(MarketPrice.recorded_at.desc()) .limit(1) ).scalar_one_or_none() session.add( MarketPrice( symbol=symbol, label=label, category=category, price=price, currency=currency, change_pct=change_pct, recorded_at=recorded_at, ) ) _detect_and_record_spike(session, symbol, label, prev, price, recorded_at) added += 1 except Exception: log.exception("Failed to fetch %s", symbol) session.commit() return added