FMP 财务报表 API： 代码示例、三表对齐与数据校验

curl --fail --silent --show-error \
  --get "https://financialmodelingprep.com/stable/income-statement" \
  --data-urlencode "symbol=AAPL" \
  --data-urlencode "period=annual" \
  --data-urlencode "limit=5" \
  --header "apikey: $FMP_API_KEY"

import os
from concurrent.futures import ThreadPoolExecutor
import requests

BASE = "https://financialmodelingprep.com/stable"
ENDPOINTS = {
    "income": "income-statement",
    "balance": "balance-sheet-statement",
    "cashflow": "cash-flow-statement",
}

def fetch(name: str, symbol: str, period="annual", limit=5):
    response = requests.get(
        f"{BASE}/{ENDPOINTS[name]}",
        params={"symbol": symbol.upper(), "period": period, "limit": limit},
        headers={"apikey": os.environ["FMP_API_KEY"]},
        timeout=10,
    )
    response.raise_for_status()
    rows = response.json()
    if not isinstance(rows, list) or not rows:
        raise ValueError(f"No {name} rows for {symbol}")
    return rows

def fiscal_key(row: dict):
    year = row.get("fiscalYear") or row.get("calendarYear")
    return row.get("symbol"), str(year), row.get("period")

def get_three_statements(symbol: str):
    with ThreadPoolExecutor(max_workers=3) as pool:
        futures = {name: pool.submit(fetch, name, symbol) for name in ENDPOINTS}
        data = {name: future.result() for name, future in futures.items()}

    indexed = {
        name: {fiscal_key(row): row for row in rows}
        for name, rows in data.items()
    }
    common = set.intersection(*(set(rows) for rows in indexed.values()))
    if not common:
        raise ValueError("The three statements share no fiscal periods")

    return [
        {"key": key, **{name: indexed[name][key] for name in ENDPOINTS}}
        for key in sorted(common, reverse=True)
    ]

const paths = {
  income: "income-statement",
  balance: "balance-sheet-statement",
  cashflow: "cash-flow-statement",
} as const;

async function getStatements(symbol: string, period = "annual") {
  const key = process.env.FMP_API_KEY;
  if (!key) throw new Error("FMP_API_KEY is missing");

  const entries = await Promise.all(
    Object.entries(paths).map(async ([name, path]) => {
      const url = new URL(\`https://financialmodelingprep.com/stable/\${path}\`);
      url.searchParams.set("symbol", symbol.toUpperCase());
      url.searchParams.set("period", period);
      url.searchParams.set("limit", "5");

      const response = await fetch(url, {
        headers: { apikey: key },
        signal: AbortSignal.timeout(10_000),
      });
      if (!response.ok) throw new Error(\`\${name}: HTTP \${response.status}\`);
      const rows: unknown = await response.json();
      if (!Array.isArray(rows) || rows.length === 0) {
        throw new Error(\`\${name}: empty statement response\`);
      }
      return [name, rows] as const;
    }),
  );
  return Object.fromEntries(entries);
}

def close_enough(left, right, rel_tol=0.001, abs_tol=1.0):
    if left is None or right is None:
        return None  # unknown, not pass
    return abs(left - right) <= max(abs_tol, rel_tol * max(abs(left), abs(right)))

def validate_balance_sheet(row):
    assets = row.get("totalAssets")
    liabilities = row.get("totalLiabilities")
    equity = row.get("totalStockholdersEquity")
    return {
        "balanceEquation": close_enough(assets, liabilities + equity)
        if None not in (assets, liabilities, equity) else None,
        "currencyPresent": bool(row.get("reportedCurrency")),
        "periodPresent": bool(row.get("period")),
    }