4. 金融数据获取与清洗:财务报表数据

任务说明

本案例从 SEC EDGAR 的 companyfacts 接口获取一家公司的 XBRL 财务数据,提取三个关键科目:营业收入(Revenues)、净利润(NetIncomeLoss)、总资产(Assets),并处理季度与年度对齐问题。

核心难点在于:SEC 返回的 units.USD 列表中,季度值与年度值混在一起,且部分季度值缺失(Q4 常需用全年减前三季度推算)。本案例用 form 与 fp 字段区分报告类型,按 start/end 日期跨度判定季度或年度,并输出对齐后的时间序列。

环境与依赖

pip install requests pandas

SEC 要求请求头携带 User-Agent(含邮箱),否则返回 403。示例使用 CIK 0000320193(Apple)。

核心实现

import requests
import pandas as pd
from datetime import datetime

HEADERS = {"User-Agent": "MLCourse research@example.com"}
CIK = "0000320193"
URL = f"https://data.sec.gov/api/xbrl/companyfacts/CIK{CIK}.json"

TAGS = {
    "Revenues": "revenue",
    "NetIncomeLoss": "net_income",
    "Assets": "assets",
}

def fetch_facts():
    r = requests.get(URL, headers=HEADERS, timeout=30)
    r.raise_for_status()
    return r.json()["facts"]["us-gaap"]

def extract(facts, tag):
    """提取单个科目的年度/季度记录,返回 DataFrame。"""
    if tag not in facts:
        return pd.DataFrame()
    rows = []
    for item in facts[tag]["units"]["USD"]:
        start = item.get("start")
        end = item["end"]
        # 无 start 的是时点值(如 Assets),按年度末对齐
        if start is None:
            rows.append({"end": end, "start": end, "val": item["val"],
                         "form": item["form"], "fp": item["fp"], "period": "instant"})
            continue
        days = (datetime.fromisoformat(end) - datetime.fromisoformat(start)).days
        period = "annual" if days > 300 else "quarter"
        rows.append({"end": end, "start": start, "val": item["val"],
                     "form": item["form"], "fp": item["fp"], "period": period})
    df = pd.DataFrame(rows)
    df["end"] = pd.to_datetime(df["end"])
    df["start"] = pd.to_datetime(df["start"])
    # 同一 end 保留最新提交(form 优先级 10-K > 10-Q)
    df["prio"] = df["form"].map({"10-K": 0, "10-Q": 1}).fillna(2)
    df = df.sort_values(["end", "prio"]).drop_duplicates("end", keep="first")
    return df.drop(columns="prio").sort_values("end").reset_index(drop=True)

def align_quarterly(df):
    """季度对齐:若某季度缺失,用年度值减同年前三季度推算 Q4。"""
    if df.empty:
        return df
    df = df.copy()
    df["year"] = df["end"].dt.year
    df["q"] = df["end"].dt.quarter
    annual = df[df["period"] == "annual"].set_index("year")["val"]
    quarters = df[df["period"] == "quarter"]
    filled = []
    for year, grp in quarters.groupby("year"):
        have = set(grp["q"])
        if year in annual.index and 4 not in have and {1, 2, 3} <= have:
            q4_val = annual[year] - grp[grp["q"].isin([1, 2, 3])]["val"].sum()
            filled.append({"end": pd.Timestamp(f"{year}-12-31"), "start": pd.Timestamp(f"{year}-10-01"),
                           "val": q4_val, "form": "derived", "fp": "Q4",
                           "period": "quarter", "year": year, "q": 4})
    if filled:
        df = pd.concat([df, pd.DataFrame(filled)], ignore_index=True)
    return df.sort_values("end").reset_index(drop=True)

if __name__ == "__main__":
    facts = fetch_facts()
    result = {}
    for tag, name in TAGS.items():
        df = extract(facts, tag)
        if name != "assets":
            df = align_quarterly(df)
        result[name] = df
        print(f"\n=== {name} ({tag}) ===")
        print(df[["end", "val", "period", "form"]].tail(6).to_string(index=False))

运行与验证

直接运行脚本:python sec_financials.py。脚本会打印每个科目的最近 6 条记录。

判定通过的条件:

  • 输出包含 revenue、net_income、assets 三组数据,无异常报错。
  • period 列正确区分 annual 与 quarter(跨度 >300 天为年度)。
  • 若某年 Q4 缺失且前三季度齐全,会出现 form=derived 的推算记录,其 val 等于该年年度值减前三季度之和。
  • assets 为时点值,period=instant,不参与季度对齐。

关键点

SEC 的 companyfacts 中同一 end 日期可能有多条记录(不同 form 或 frame),代码用 10-K 优先于 10-Q 去重。季度对齐采用"年度减前三季度"推算 Q4,这是处理 XBRL 数据缺失的常见做法。若需其他公司,替换 CIK 即可;若科目名不同(如 RevenueFromContractWithCustomerExcludingAssessedTax),在 TAGS 中调整。