Financial News API Free Python: A Practical Guide用 Python 调用免费财经新闻 API

import os
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from urllib.parse import urldefrag

import requests


@dataclass(frozen=True)
class Article:
    provider_id: str
    canonical_url: str
    title: str
    published_at: datetime
    source: str
    language: str
    retrieved_at: datetime

    @property
    def dedupe_key(self) -> tuple[str, str]:
        return self.provider_id, self.canonical_url


class FinancialNewsClient:
    def __init__(self, base_url: str, api_key: str) -> None:
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {api_key}"})

    def fetch_page(self, cursor: str | None) -> tuple[list[Article], list[dict], str | None]:
        params = {"symbols": "AAPL", "language": "en", "limit": 100}
        if cursor:
            params["cursor"] = cursor

        for attempt in range(4):
            response = self.session.get(self.base_url, params=params, timeout=(5, 20))
            if response.status_code == 429:
                retry_after = response.headers.get("Retry-After", "1")
                delay = int(retry_after) if retry_after.isdigit() else 1
                time.sleep(min(delay, 30))
                continue
            if 500 <= response.status_code < 600 and attempt < 3:
                time.sleep(2 ** attempt)
                continue

            response.raise_for_status()
            if "application/json" not in response.headers.get("Content-Type", ""):
                raise ValueError("Expected a JSON response")

            payload = response.json()
            raw_items = payload.get("results")
            if not isinstance(raw_items, list):
                raise ValueError("Expected results to be a list")

            retrieved_at = datetime.now(timezone.utc)
            valid, quarantine = [], []
            for raw in raw_items:
                try:
                    required = {"id", "title", "published_at", "source_url", "source"}
                    if not required.issubset(raw):
                        raise ValueError("Missing a required article field")
                    published = datetime.fromisoformat(raw["published_at"].replace("Z", "+00:00"))
                    if published.tzinfo is None:
                        raise ValueError("published_at must include a timezone")
                    valid.append(Article(
                        provider_id=str(raw["id"]),
                        canonical_url=urldefrag(raw["source_url"])[0],
                        title=raw["title"].strip(),
                        published_at=published.astimezone(timezone.utc),
                        source=raw["source"].strip(),
                        language=raw.get("language", "und"),
                        retrieved_at=retrieved_at,
                    ))
                except (KeyError, TypeError, ValueError) as error:
                    quarantine.append({"raw": raw, "error": str(error)})

            return valid, quarantine, payload.get("next_cursor")

        raise RuntimeError("The provider remained unavailable after bounded retries")


def collect(client, start_cursor, commit_batch):
    cursor = start_cursor
    while True:
        articles, quarantine, next_cursor = client.fetch_page(cursor)
        unique = {article.dedupe_key: article for article in articles}

        # This function must commit articles, quarantine records, and the
        # next cursor in one database transaction. Advance no checkpoint first.
        commit_batch(list(unique.values()), quarantine, next_cursor)

        if not next_cursor:
            break
        cursor = next_cursor


client = FinancialNewsClient(
    "https://provider.example/v1/financial-news",
    os.environ["FINANCIAL_NEWS_API_KEY"],
)
# Wire in application-owned transactional functions before running:
# collect(client, start_cursor=load_checkpoint(), commit_batch=commit_batch)