面向 AI Agent 的 财务报表 API

import hashlib
import json
import os
from datetime import datetime, timezone

import requests

def fetch_statement_bundle(ticker: str, period: str = "annual", limit: int = 8):
    endpoint = "https://api.financialdatasets.ai/financials"
    params = {"ticker": ticker, "period": period, "limit": limit}
    response = requests.get(
        endpoint,
        params=params,
        headers={"X-API-KEY": os.environ["FINANCIAL_DATASETS_API_KEY"]},
        timeout=30,
    )
    response.raise_for_status()
    raw = response.json()
    stable = json.dumps(raw, sort_keys=True, separators=(",", ":"))

    return {
        "provider": "financial_datasets",
        "endpoint": endpoint,
        "request": params,
        "retrievedAt": datetime.now(timezone.utc).isoformat(),
        "rawSha256": hashlib.sha256(stable.encode()).hexdigest(),
        "raw": raw,
        "statements": raw.get("financials", raw),
    }

bundle = fetch_statement_bundle("AAPL", period="annual")
print(bundle["rawSha256"], bundle["statements"].keys())

def close_enough(left, right, abs_tol=1_000, rel_tol=0.002):
    scale = max(abs(left), abs(right), 1)
    return abs(left - right) <= max(abs_tol, rel_tol * scale)

def validate_period(bundle):
    errors, warnings = [], []
    bs, cf = bundle["balance"], bundle["cashFlow"]

    if not close_enough(
        bs["totalAssets"],
        bs["totalLiabilities"] + bs["totalEquity"],
    ):
        errors.append("balance_sheet_equation_failed")

    if all(cf.get(k) is not None for k in
           ["beginningCash", "changeInCash", "endingCash"]):
        if not close_enough(
            cf["beginningCash"] + cf["changeInCash"],
            cf["endingCash"],
        ):
            warnings.append("cash_rollforward_failed")

    if len({bundle[x]["currency"] for x in
            ["income", "balance", "cashFlow"]}) != 1:
        errors.append("currency_mismatch")

    if len({bundle[x]["periodEnd"] for x in
            ["income", "balance", "cashFlow"]}) != 1:
        errors.append("period_end_mismatch")

    return {"valid": not errors, "errors": errors, "warnings": warnings}