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.
112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
"""
|
|
Finnhub /company-news client (free tier).
|
|
|
|
Rate limit: 60 calls/min on the free plan. We enforce 1 second between calls
|
|
via an asyncio.Semaphore + sleep, which is the simplest correct shape since
|
|
the entire ingest pipeline is single-process.
|
|
|
|
Endpoint shape: /company-news?symbol=AAPL&from=2026-04-01&to=2026-04-25
|
|
Free-tier history: ~12 months. Cumulative archive must be self-built by
|
|
running `scripts/news_backfill.py` once at install time, then daily.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import date
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class FinnhubClient:
|
|
def __init__(
|
|
self,
|
|
api_key: str | None = None,
|
|
base_url: str | None = None,
|
|
request_interval_sec: float = 1.05,
|
|
):
|
|
self.api_key = api_key or settings.FINNHUB_API_KEY
|
|
self.base_url = (base_url or settings.FINNHUB_BASE_URL).rstrip("/")
|
|
self._client: httpx.AsyncClient | None = None
|
|
# Single-slot semaphore + sleep enforces 60-calls/min ceiling
|
|
self._gate = asyncio.Semaphore(1)
|
|
self._interval = request_interval_sec
|
|
|
|
def is_configured(self) -> bool:
|
|
return bool(self.api_key)
|
|
|
|
async def _get_client(self) -> httpx.AsyncClient:
|
|
if self._client is None or self._client.is_closed:
|
|
self._client = httpx.AsyncClient(base_url=self.base_url, timeout=30.0)
|
|
return self._client
|
|
|
|
async def close(self) -> None:
|
|
if self._client and not self._client.is_closed:
|
|
await self._client.aclose()
|
|
|
|
async def _request(self, path: str, params: dict[str, Any], retries: int = 3) -> Any:
|
|
client = await self._get_client()
|
|
params = {**params, "token": self.api_key}
|
|
|
|
async with self._gate:
|
|
last_exc: Exception | None = None
|
|
for attempt in range(retries):
|
|
try:
|
|
resp = await client.get(path, params=params)
|
|
if resp.status_code == 429:
|
|
wait = 2 ** attempt
|
|
logger.warning(f"Finnhub 429 — sleeping {wait}s (attempt {attempt + 1})")
|
|
await asyncio.sleep(wait)
|
|
continue
|
|
if resp.status_code >= 500:
|
|
wait = 2 ** attempt
|
|
logger.warning(f"Finnhub {resp.status_code} — retry in {wait}s")
|
|
await asyncio.sleep(wait)
|
|
continue
|
|
resp.raise_for_status()
|
|
await asyncio.sleep(self._interval)
|
|
return resp.json()
|
|
except httpx.HTTPStatusError as exc:
|
|
last_exc = exc
|
|
if attempt < retries - 1 and exc.response.status_code in (429, 500, 502, 503, 504):
|
|
await asyncio.sleep(2 ** attempt)
|
|
continue
|
|
raise
|
|
except (httpx.ConnectError, httpx.ReadTimeout) as exc:
|
|
last_exc = exc
|
|
if attempt < retries - 1:
|
|
await asyncio.sleep(2 ** attempt)
|
|
continue
|
|
raise
|
|
raise last_exc # type: ignore[misc]
|
|
|
|
async def fetch_company_news(
|
|
self,
|
|
symbol: str,
|
|
from_date: date,
|
|
to_date: date,
|
|
) -> list[dict[str, Any]]:
|
|
"""Fetch company-news for [from_date, to_date] inclusive.
|
|
|
|
One call covers the full range, so split by month at the caller side
|
|
for backfill granularity.
|
|
"""
|
|
payload = await self._request(
|
|
"/company-news",
|
|
params={
|
|
"symbol": symbol.strip().upper(),
|
|
"from": from_date.isoformat(),
|
|
"to": to_date.isoformat(),
|
|
},
|
|
)
|
|
if not isinstance(payload, list):
|
|
logger.warning(f"Finnhub /company-news returned non-list for {symbol}")
|
|
return []
|
|
return payload
|