用 Python 调用免费财经新闻 API
选择可测试的财经新闻接口，用 Python 按股票代码请求新闻，并在生产接入前核对额度、历史、情绪、来源和 JSON 字段。
Python 财经新闻采集器的实现目标
开发者在确定要接入财经新闻后，需要的并不是一次性请求，而是能够应对分页、限流、重复报道、结构变化、部分失败和进程重启的 Python 采集器。本页不以供应商覆盖比较为重点，核心目标是构建一个规模适中但具备生产形态的采集循环，并确保日后更换 API 时不必改写所有下游流程。
只在批次成功落库后保存游标和检查点，确保进程重启时既不漏文章，也不会越过失败任务。
把不同响应转成统一内部文章模型，在进入 NLP、告警或存储前验证必需字段。
结合退避、稳定标识、URL 规范化和数据库唯一约束，确保重复请求不会生成重复结果。
选择能支撑可靠 Python 采集循环的财经新闻 API
优先选择游标分页，而不是脆弱的页码分页
Python 采集器应能在中断后继续运行，同时避免漏掉或重复读取不断变化的结果集。测试接口是否返回不透明的下一页游标、确定的排序规则和明确的结束条件；只有对应批次成功落库后，才保存游标。
先统一文章结构，再交给下游 NLP
把供应商字段映射为类型明确的记录，包括供应商文章 ID、规范 URL、标题、发布时间、来源、语言、实体、股票代码和抓取时间。原始响应可留作排错，但不要让每个模型或 Notebook 都直接依赖供应商特有 JSON。
把重试、去重与检查点作为一个整体设计
对暂时性错误进行退避重试，遵守 Retry-After，并结合供应商 ID 与规范 URL 去重。重试必须幂等：重复运行同一页不能生成重复提醒或重复数据行。
使用带重叠窗口的发布时间水位
同时保存供应商游标和发布时间水位。进程恢复时，不要恰好从最后时间戳开始，而应重新请求一小段重叠窗口并去重，否则延迟到达或被更正的文章可能落在检查点之前。发布时间、供应商收录时间和本地抓取时间应分别保存。
隔离结构与内容异常
不能因为一篇文章格式异常就丢弃整页，也不能静默接受破坏兼容性的字段变化。将异常记录连同供应商、请求 ID、原始响应、验证错误和重试状态放入隔离区；只有确认分页游标仍可靠时才继续采集，并安排后续复核。
把采集与内容增强分开
先获取并提交来源记录，再执行摘要、翻译、情绪、向量化或提醒排序。下游增强可以独立重试和版本化，采集器则始终确定性地记录供应商在什么时间返回了什么内容。
| 检查项 | 要回答的问题 | 证据 |
|---|---|---|
| Python 易用性 | 是否兼容 requests，是否有 SDK、类型与异步支持？ | 最小同步与异步原型 |
| 分页 | 游标是否稳定，顺序、页大小和断点续传如何？ | 两页重放与恢复测试 |
| 限流 | 429 响应、Retry-After、重置头与突发规则？ | 受控负载下的实际响应头 |
| 幂等性 | 重复读取一页是否会产生重复结果？ | 供应商 ID、URL 哈希与数据库约束 |
| 可观察性 | 能否测量延迟、失败、空字段和额度？ | 结构化日志与采集指标 |
| 测试 | 能否离线记录并重放响应？ | 测试样本、结构测试与失败用例 |
免费财经新闻 API 的 Python 请求示例
不要把 API Key 写进源码
从环境变量读取密钥，设置合理的超时时间，不要在日志或 Notebook 中输出密钥。
验证状态码、内容类型和字段
遇到 HTTP 错误时立即停止处理，确认返回内容为 JSON，并拒绝缺少应用必需字段的新闻记录。
明确处理分页和限流
遵循文档中的游标，处理 429 与 Retry-After，按固定 URL 或文章 ID 去重，并保留来源时间戳。
封装一个小型供应商客户端
把基础 URL、鉴权、超时、连接复用、错误映射、游标提取和响应验证集中到同一接口中，返回规范化批次与检查点元数据，不要让原始供应商 JSON 扩散到整个应用。这样更换 API 时修改范围小且容易测试。
测试重启与部分失败
记录代表性响应并在离线环境重放，覆盖两页正常结果、提交前超时、带 Retry-After 的 429、异常 JSON、单条无效文章、重复重放、空结果、更正到达和字段删除。既要断言最终落库记录，也要断言最终检查点。
暴露采集健康状态
测量请求延迟、从发布到收录的延迟、成功率、重试、剩余额度、每页文章数、重复率、隔离记录和检查点年龄。即使请求仍返回 HTTP 200，只要进度过期或重要来源突然消失，也应告警。
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)检查点约束：commit_batch 必须使用数据库唯一约束，并在同一个事务中保存有效文章、隔离记录和 next_cursor。进程在提交前终止时，可以安全重放同一页；提交后终止时，则从已保存游标对应的下一页继续。绝不能先保存游标、后保存文章。
域名和字段名仅用于说明通用请求结构，避免虚构 QVeris 或服务商端点；请按所选服务商的当前文档替换。
QVeris 如何帮助发现财经新闻 API 能力
QVeris 帮助 Agent 发现可调用能力并检查输入输出，但不保证服务商免费、实时、完整或适用于所有用途；这些事实必须以服务商当前文档为准。
- 在 QVeris 工具详情中搜索财经新闻或股票新闻能力。
- 接入工作流前检查鉴权、股票代码参数、返回字段和错误行为。
- 参考 QVeris Python SDK 文档了解受支持的接入方式。
FAQ
有些服务提供免费套餐或试用。选择前应比较配额、延迟、历史、字段、市场覆盖、署名和商用规则。
使用 requests 发送带鉴权的请求，传入股票代码或查询词，检查状态码，验证 JSON，并处理分页与限流。
很多金融专用接口支持股票代码，但代码格式和交易所映射不同，应查阅当前参数文档并用已知代码测试。
有时可以，但免费历史通常限制日期范围、结果数量或访问延迟，不能假设当前新闻接口自动包含完整档案。
部分接口返回情绪、实体或相关性，另一些只返回原始文章。应检查字段定义、置信度、方法和语言支持。
多数需要 API Key，也有少数公开或样例端点。无论是否需要密钥，都应核对限额、来源许可和生产使用条件。
权威参考资料
服务商套餐会变化，应以当前文档为证据；本页提供评估流程，不承诺某个具体免费额度长期有效。