手记

如何使用财务报表API做基本面分析

很多人使用行情 API 时,只关注实时价格、K 线、盘口和成交明细。但真正做中长期选股、财务质量分析、估值判断、股息组合监控时,价格数据是不够的。价格只是反映了市场如何给这家公司定价,但财务报表能真实揭露这家公司到底赚不赚钱、现金流好不好、负债压力大不大、分红是否可持续。

这一期我们将基于财务报表接口,通过代码快速帮我们分析一家公司的基本面。

一、财报接口数据总览

下面是Infoway API的财务报表接口能查询到的数据:

分析方向 接口路径 适合做什么
利润表 /common/basic/financial/income_statement 收入、成本、利润、净利润增速分析
收入明细 /common/basic/financial/revenue 业务结构、地区收入结构、产品线变化
现金流量表 /common/basic/financial/cash_flow 经营现金流、自由现金流、利润含金量
资产负债表 /common/basic/financial/balance_sheet 总资产、负债、权益、现金、债务分析
估值与统计指标 /common/basic/financial/statistics PE、PB、ROE、市值、净利率等指标
股息指标 /common/basic/financial/dividend DPS、股息率、派息率、分红趋势
历史派息记录 /common/basic/financial/dividend_payout 除权日、登记日、派息日、派息金额
业绩预测 /common/basic/financial/earnings EPS 和营收实际值与预期值对比

二、封装一个财务报表客户端

先安装依赖:

pip install requests pandas matplotlib python-dotenv

把 API Key 放到环境变量里,不要直接写死在代码中。

创建 .env 文件:

INFOWAY_API_KEY=你的_api_key

最后写一个通用请求函数:

import os
import time
import requests
import pandas as pd
import matplotlib.pyplot as plt
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("INFOWAY_API_KEY")
BASE_URL = "https://data.infoway.io"

if not API_KEY:
    raise RuntimeError("请先在 .env 文件中设置 INFOWAY_API_KEY")


class InfowayFinancialClient:
    def __init__(self, api_key: str, base_url: str = BASE_URL, timeout: int = 20):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.headers = {
            "User-Agent": "Mozilla/5.0",
            "Accept": "application/json",
            "apiKey": api_key,
        }

    def get(self, path: str, params: dict | None = None) -> dict:
        url = f"{self.base_url}{path}"
        resp = requests.get(
            url,
            headers=self.headers,
            params=params,
            timeout=self.timeout,
        )

        if resp.status_code != 200:
            raise RuntimeError(
                f"HTTP 请求失败: {resp.status_code}, body={resp.text[:500]}"
            )

        data = resp.json()

        if data.get("ret") != 200:
            raise RuntimeError(f"API 返回异常: {data}")

        return data

    def financial_statement(
        self,
        endpoint: str,
        symbol: str,
        stock_type: str,
        period_type: str | None = None,
    ) -> pd.DataFrame:
        """
        通用财报接口。

        endpoint 示例:
        - income_statement
        - revenue
        - cash_flow
        - balance_sheet
        - statistics
        - dividend
        - earnings
        """
        params = {
            "symbol": symbol,
            "type": stock_type,
        }

        if period_type:
            params["period_type"] = period_type

        raw = self.get(f"/common/basic/financial/{endpoint}", params=params)
        rows = raw.get("data", [])

        return pd.DataFrame(rows)

    def dividend_payout(self, symbol: str, stock_type: str) -> pd.DataFrame:
        """
        历史派息记录接口。
        官方说明该接口没有 period_type 参数。
        """
        params = {
            "symbol": symbol,
            "type": stock_type,
        }

        raw = self.get("/common/basic/financial/dividend_payout", params=params)
        rows = raw.get("data", [])

        df = pd.DataFrame(rows)

        for col in ["exDate", "recordDate", "paymentDate"]:
            if col in df.columns:
                # 官方示例是 Unix 时间戳。这里兼容秒级和毫秒级。
                df[col] = pd.to_numeric(df[col], errors="coerce")
                unit = "ms" if df[col].dropna().gt(10**12).any() else "s"
                df[col] = pd.to_datetime(df[col], unit=unit, errors="coerce")

        return df


client = InfowayFinancialClient(API_KEY)

三、基本面分析

1. 净利润增速追踪器

如果我们想了解公司最近几个季度的盈利动能是在增强,还是在减弱,可以用利润表接口拉取季度数据,筛选 net_income,然后计算环比和同比。

def extract_item(df: pd.DataFrame, item_id: str) -> pd.DataFrame:
    """
    从长表中提取某个财务科目。
    """
    if df.empty:
        return pd.DataFrame()

    result = df[df["itemId"] == item_id].copy()

    if result.empty:
        return result

    result["periodDate"] = pd.to_datetime(result["periodDate"])
    result["itemValue"] = pd.to_numeric(result["itemValue"], errors="coerce")
    result = result.sort_values("periodDate")

    return result


income = client.financial_statement(
    endpoint="income_statement",
    symbol="AAPL.US",
    stock_type="STOCK_US",
    period_type="fq",
)

net_income = extract_item(income, "net_income")

net_income["qoq_growth"] = net_income["itemValue"].pct_change()
net_income["yoy_growth"] = net_income["itemValue"].pct_change(4)

print(
    net_income[
        ["symbol", "periodDate", "itemValue", "qoq_growth", "yoy_growth"]
    ].tail(8)
)

将结果可视化:

plt.figure(figsize=(10, 5))
plt.plot(net_income["periodDate"], net_income["itemValue"] / 1e9, marker="o")
plt.title("Net Income Trend")
plt.xlabel("Period")
plt.ylabel("Net Income")
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

进一步可以画同比增速:

plt.figure(figsize=(10, 5))
plt.bar(net_income["periodDate"], net_income["yoy_growth"] * 100)
plt.title("Net Income YoY Growth")
plt.xlabel("Period")
plt.ylabel("YoY Growth (%)")
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

这个分析很适合做成一个盈利动能仪表盘。如果净利润连续几个季度同比改善,说明利润趋势可能正在修复;如果收入增长但净利润下滑,就要进一步看毛利率、费用率或一次性损益。

2:利润含金量分析

利润表上的净利润不等于真金白银。一个公司可能净利润很好看,但经营现金流很差,这时就要警惕应收账款、存货、资本化处理或利润质量问题。

我们可以用:

经营现金流 / 净利润
自由现金流 / 净利润
自由现金流率 = 自由现金流 / 营业收入

代码如下:


cash_flow = client.financial_statement(
    endpoint="cash_flow",
    symbol="AAPL.US",
    stock_type="STOCK_US",
    period_type="fq",
)

income = client.financial_statement(
    endpoint="income_statement",
    symbol="AAPL.US",
    stock_type="STOCK_US",
    period_type="fq",
)

net_income = extract_item(income, "net_income")
revenue = extract_item(income, "total_revenue")
free_cash_flow = extract_item(cash_flow, "free_cash_flow")
operating_cf = extract_item(cash_flow, "cf_oper")

quality = (
    net_income[["periodDate", "itemValue"]]
    .rename(columns={"itemValue": "net_income"})
    .merge(
        revenue[["periodDate", "itemValue"]].rename(columns={"itemValue": "revenue"}),
        on="periodDate",
        how="left",
    )
    .merge(
        free_cash_flow[["periodDate", "itemValue"]].rename(
            columns={"itemValue": "free_cash_flow"}
        ),
        on="periodDate",
        how="left",
    )
    .merge(
        operating_cf[["periodDate", "itemValue"]].rename(
            columns={"itemValue": "operating_cash_flow"}
        ),
        on="periodDate",
        how="left",
    )
)

quality["ocf_to_net_income"] = quality["operating_cash_flow"] / quality["net_income"]
quality["fcf_to_net_income"] = quality["free_cash_flow"] / quality["net_income"]
quality["fcf_margin"] = quality["free_cash_flow"] / quality["revenue"]

print(quality.tail(8))

同样将结果可视化:

plt.figure(figsize=(10, 5))
plt.plot(
    quality["periodDate"],
    quality["ocf_to_net_income"],
    marker="o",
    label="Operating CF / Net Income",
)
plt.plot(
    quality["periodDate"],
    quality["fcf_to_net_income"],
    marker="o",
    label="FCF / Net Income",
)
plt.axhline(1.0, line)
plt.title("Profit Quality")
plt.xlabel("Period")
plt.ylabel("Ratio")
plt.legend()
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

一般来说,长期经营现金流明显低于净利润,要谨慎;如果自由现金流长期为负,但公司又持续扩张,也要进一步区分是成长性投入,还是经营质量恶化。

3:资产负债表健康度分析

资产负债表可以回答:公司是不是靠高负债撑起来的?现金是否充足?权益是否稳定增长?

我们可以计算:

资产负债率 = 总负债 / 总资产
净债务率 = 净债务 / 股东权益
现金债务覆盖率 = 现金及等价物 / 总债务

代码如下:

balance = client.financial_statement(
    endpoint="balance_sheet",
    symbol="00700.HK",
    stock_type="STOCK_HK",
    period_type="fq",
)

total_assets = extract_item(balance, "total_assets")
total_liabilities = extract_item(balance, "total_liabilities")
total_equity = extract_item(balance, "total_equity")
cash = extract_item(balance, "cash_and_equiv")
total_debt = extract_item(balance, "total_debt")
net_debt = extract_item(balance, "net_debt")

bs = (
    total_assets[["periodDate", "itemValue"]]
    .rename(columns={"itemValue": "total_assets"})
    .merge(
        total_liabilities[["periodDate", "itemValue"]].rename(
            columns={"itemValue": "total_liabilities"}
        ),
        on="periodDate",
        how="left",
    )
    .merge(
        total_equity[["periodDate", "itemValue"]].rename(
            columns={"itemValue": "total_equity"}
        ),
        on="periodDate",
        how="left",
    )
    .merge(
        cash[["periodDate", "itemValue"]].rename(
            columns={"itemValue": "cash_and_equiv"}
        ),
        on="periodDate",
        how="left",
    )
    .merge(
        total_debt[["periodDate", "itemValue"]].rename(
            columns={"itemValue": "total_debt"}
        ),
        on="periodDate",
        how="left",
    )
    .merge(
        net_debt[["periodDate", "itemValue"]].rename(
            columns={"itemValue": "net_debt"}
        ),
        on="periodDate",
        how="left",
    )
)

bs["debt_asset_ratio"] = bs["total_liabilities"] / bs["total_assets"]
bs["net_debt_to_equity"] = bs["net_debt"] / bs["total_equity"]
bs["cash_to_debt"] = bs["cash_and_equiv"] / bs["total_debt"]

print(bs.tail(8))

可视化:

plt.figure(figsize=(10, 5))
plt.plot(bs["periodDate"], bs["debt_asset_ratio"] * 100, marker="o")
plt.title("Debt Asset Ratio")
plt.xlabel("Period")
plt.ylabel("Debt / Assets (%)")
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

这个分析适合用来筛查“高杠杆扩张”风险。比如收入和利润增长很快,但资产负债率同步快速上升,且经营现金流没有改善,就可能是比较危险的增长。

4:收入结构可视化

收入明细接口非常适合分析公司业务结构。比如苹果可以拆成 iPhone、Mac、iPad、Services、Wearables;也可以按地区拆成 Americas、Europe、Greater China 等。

revenue_detail = client.financial_statement(
    endpoint="revenue",
    symbol="AAPL.US",
    stock_type="STOCK_US",
    period_type="fy",
)

print(revenue_detail.head())
print(revenue_detail["itemGroup"].dropna().unique())

筛选业务收入:

business_rev = revenue_detail[
    revenue_detail["itemGroup"] == "Revenue by business"
].copy()

business_rev["periodDate"] = business_rev["periodDate"].astype(str)
business_rev["itemValue"] = pd.to_numeric(business_rev["itemValue"], errors="coerce")

pivot = business_rev.pivot_table(
    index="periodDate",
    columns="itemName",
    values="itemValue",
    aggfunc="sum",
).sort_index()

print(pivot.tail())

画堆叠面积图:

plt.figure(figsize=(12, 6))
plt.stackplot(
    pivot.index,
    [pivot[col] / 1e9 for col in pivot.columns],
    labels=pivot.columns,
)
plt.title("Revenue by Business")
plt.xlabel("Fiscal Year")
plt.ylabel("Revenue")
plt.legend(loc="upper left")
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

这个分析比单纯看总收入有意思得多。因为很多公司的估值重估并不是来自总收入增长,而是来自收入结构变化。比如硬件公司服务化、周期股产品结构升级、传统公司高毛利业务占比提升,都可能改变市场给它的估值方式。

5:价值股筛选器

估值与统计指标接口可以用来批量筛选股票。比如我们想找:

  • PE < 15
  • PB < 2
  • ROE > 15%
  • 净利率 > 10%

先写一个函数,把 statistics 中的关键指标取出来:

def safe_float(x):
    if x is None:
        return None
    if isinstance(x, str) and x.strip() in {"", "-"}:
        return None
    try:
        return float(x)
    except Exception:
        return None


def latest_statistics(symbol: str, stock_type: str) -> dict:
    df = client.financial_statement(
        endpoint="statistics",
        symbol=symbol,
        stock_type=stock_type,
        period_type="fq",
    )

    if df.empty:
        return {"symbol": symbol}

    # 优先用 currentValue,没有就用 itemValue
    latest_date = df["periodDate"].iloc[0]
    latest = df[df["periodDate"] == latest_date].copy()

    result = {
        "symbol": symbol,
        "periodDate": latest_date,
    }

    for _, row in latest.iterrows():
        item_id = row.get("itemId")
        value = row.get("currentValue")
        if value is None or value == "":
            value = row.get("itemValue")
        result[item_id] = safe_float(value)

    return result

批量筛选:

symbols = [
    ("AAPL.US", "STOCK_US"),
    ("MSFT.US", "STOCK_US"),
    ("TSLA.US", "STOCK_US"),
    ("00700.HK", "STOCK_HK"),
    ("000001.SZ", "STOCK_CN"),
]

rows = []

for symbol, stock_type in symbols:
    try:
        rows.append(latest_statistics(symbol, stock_type))
        time.sleep(0.2)
    except Exception as e:
        print(f"{symbol} 查询失败: {e}")

factor_df = pd.DataFrame(rows)

print(factor_df.head())

不同接口中的 itemId 命名可能会随市场和数据源略有差异。你可以先打印所有字段:

print(factor_df.columns.tolist())
0人推荐
随时随地看视频
慕课网APP