318 lines
14 KiB
Python
318 lines
14 KiB
Python
import datetime as dt
|
|
import json
|
|
import logging
|
|
import statistics
|
|
from collections import Counter
|
|
|
|
import httpx
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .config import settings
|
|
from .models import Article, MarketPrice, MarketSpike
|
|
from .textutil import extract_keywords
|
|
|
|
# Two spikes within this many hours of each other are treated as one
|
|
# "incident" — e.g. WTI and Brent crude almost always move together, and a
|
|
# real shock often shows up across several indices within the same hour.
|
|
# Chain-merged: A+B within range and B+C within range merges all three even
|
|
# if A and C individually aren't, same declutter pattern used for the globe.
|
|
INCIDENT_MERGE_HOURS = 1.0
|
|
|
|
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
|
|
) -> tuple[list[int], list[tuple[str, 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)
|
|
|
|
# "Most likely reasons," heuristically: just the words that recur across
|
|
# every headline in the window (stopwords dropped) — no AI/LLM involved,
|
|
# see textutil.py. Computed over the whole window, not just the
|
|
# `limit`-truncated list below, for a more robust signal.
|
|
keywords = extract_keywords([a.title for _, a in scored], top_n=8)
|
|
|
|
return [a.id for _, a in scored[:limit]], keywords
|
|
|
|
|
|
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 _recent_volatility(session: Session, symbol: str, before: dt.datetime) -> float | None:
|
|
"""This instrument's recent typical poll-to-poll move size (mean of
|
|
|% change| between consecutive polls over the lookback window), strictly
|
|
before `before` so the move being evaluated can't inflate its own
|
|
baseline. None if there isn't enough history yet to trust it."""
|
|
since = before - dt.timedelta(days=settings.market_spike_history_days)
|
|
prices = session.execute(
|
|
select(MarketPrice.price)
|
|
.where(MarketPrice.symbol == symbol, MarketPrice.recorded_at >= since, MarketPrice.recorded_at < before)
|
|
.order_by(MarketPrice.recorded_at.asc())
|
|
).scalars().all()
|
|
|
|
changes = [
|
|
abs((p2 - p1) / p1 * 100) for p1, p2 in zip(prices, prices[1:]) if p1
|
|
]
|
|
if len(changes) < settings.market_spike_min_samples:
|
|
return None
|
|
return statistics.mean(changes)
|
|
|
|
|
|
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
|
|
abs_change = abs(pct_change)
|
|
|
|
# Absolute floor: never report a move too small to matter, regardless of
|
|
# this instrument's own volatility (a near-frozen instrument's tiny
|
|
# normal wiggle shouldn't count as "abnormal" just because it's small).
|
|
if abs_change < settings.market_spike_threshold_pct:
|
|
return
|
|
|
|
# Above the floor, also require it to be unusually large *for this
|
|
# instrument* — a historical-norm check, not just an absolute cutoff.
|
|
# Falls back to the floor alone when there's not enough price history
|
|
# yet (new deployment, or an instrument added recently).
|
|
baseline = _recent_volatility(session, symbol, prev.recorded_at)
|
|
if baseline is not None and baseline > 0 and abs_change < baseline * settings.market_spike_volatility_multiplier:
|
|
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, keywords = _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),
|
|
baseline_volatility_pct=round(baseline, 3) if baseline is not None else None,
|
|
window_start=window_start,
|
|
window_end=window_end,
|
|
detected_at=recorded_at,
|
|
article_ids_json=json.dumps(article_ids),
|
|
top_keywords_json=json.dumps(keywords),
|
|
)
|
|
)
|
|
log.info(
|
|
"Spike detected: %s %.2f%% (baseline %s, %d candidate articles)",
|
|
symbol,
|
|
pct_change,
|
|
f"{baseline:.2f}%" if baseline is not None else "n/a",
|
|
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
|
|
|
|
|
|
def merge_spikes_into_incidents(spikes: list[MarketSpike]) -> list[dict]:
|
|
"""Chain-merge spikes within INCIDENT_MERGE_HOURS of each other into
|
|
"incidents" spanning possibly multiple instruments, on the assumption
|
|
that near-simultaneous moves are more likely to share a cause than
|
|
coincidence. Purely a timestamp heuristic — no correlation of the
|
|
actual price movements is attempted."""
|
|
ordered = sorted(spikes, key=lambda s: s.detected_at)
|
|
groups: list[list[MarketSpike]] = []
|
|
threshold = dt.timedelta(hours=INCIDENT_MERGE_HOURS)
|
|
|
|
for spike in ordered:
|
|
target = next(
|
|
(g for g in groups if any(abs(spike.detected_at - member.detected_at) <= threshold for member in g)),
|
|
None,
|
|
)
|
|
if target is not None:
|
|
target.append(spike)
|
|
else:
|
|
groups.append([spike])
|
|
|
|
incidents = []
|
|
for group in groups:
|
|
group.sort(key=lambda s: s.detected_at)
|
|
article_ids: list[int] = []
|
|
seen_ids: set[int] = set()
|
|
keyword_counts: Counter[str] = Counter()
|
|
for s in group:
|
|
for aid in json.loads(s.article_ids_json or "[]"):
|
|
if aid not in seen_ids:
|
|
seen_ids.add(aid)
|
|
article_ids.append(aid)
|
|
for word, count in json.loads(s.top_keywords_json or "[]"):
|
|
keyword_counts[word] += count
|
|
|
|
incidents.append(
|
|
{
|
|
"instruments": [
|
|
{
|
|
"symbol": s.symbol,
|
|
"label": s.label,
|
|
"from_price": s.from_price,
|
|
"to_price": s.to_price,
|
|
"pct_change": s.pct_change,
|
|
"baseline_volatility_pct": s.baseline_volatility_pct,
|
|
"detected_at": s.detected_at,
|
|
}
|
|
for s in group
|
|
],
|
|
"window_start": min(s.window_start for s in group),
|
|
"window_end": max(s.window_end for s in group),
|
|
"detected_at": group[0].detected_at,
|
|
"latest_detected_at": group[-1].detected_at,
|
|
"article_ids": article_ids[:60],
|
|
"top_keywords": keyword_counts.most_common(8),
|
|
}
|
|
)
|
|
|
|
incidents.sort(key=lambda i: i["latest_detected_at"], reverse=True)
|
|
return incidents
|