財務報表

import tqx_data
result = tqx_data.get_financial_statement(
    market="hk",
    symbol="0700.HK",
    start_quarter="2020q1",
    end_quarter="2024q4",
    date="20241014",
    is_latest=False,
    fields=[]
)
print(result)

symbol fy_period date ... is_write_off_real_estate is_zakat quarter
0 0700.HK FY2020Q1 20200513 ... None None 2020q1
1 0700.HK FY2020Q2 20200826 ... None None 2020q2
... ... ... ... ... ... ...
14 0700.HK FY2023Q3 20231115 ... None None 2023q3

import tqx_data

result = tqx_data.get_financial_statement(
    market="hk",
    symbol="0700.HK",
    start_quarter="2020q1",
    end_quarter="2024q4",
    date="20241014",
    is_latest=True,
    interim_type="cumulative",
    fields=["symbol", "fy_period", "date", "bs_asset_accruals"],
)
print(result)

symbol fy_period date bs_asset_accruals
0 0700.HK FY2024Q2 20240827 1.497963e+12

import re
import numpy as np
import pandas as pd
import tqx_data

as_of_date = "20250101"
rows = tqx_data.get_financial_statement(
    market="nb",
    symbol=["NVDA.NB", "AVGO.NB"],
    start_quarter="2023q1",
    end_quarter="2024q4",
    date=as_of_date,
    is_latest=False,
    interim_type="cumulative",
    fields=[
        "symbol", "fy_period", "date",
        "is_reported_revenue", "is_revenue_business_total",
        "is_reported_gross_profit", "is_gross_profit",
        "cfs_net_cf_operating", "cfs_reported_cf_operating",
    ],
)

# 同一财期可能存在多次披露，只保留截止日前公告日期最新的版本。
rows["date_key"] = pd.to_datetime(rows["date"], errors="coerce")
rows = rows.dropna(subset=["symbol", "fy_period", "date_key"])
rows = rows.sort_values("date_key").drop_duplicates(
    ["symbol", "fy_period"], keep="last"
)

def parse_period(value):
    match = re.fullmatch(r"FY(\d{4})(.+)", str(value).upper())
    return (int(match.group(1)), match.group(2)) if match else None

def previous_year_period(value):
    parsed = parse_period(value)
    return f"FY{parsed[0] - 1}{parsed[1]}" if parsed else None

def period_sort_key(value):
    parsed = parse_period(value)
    if not parsed:
        return (0, 0, "")
    year, suffix = parsed
    rank = {"Q1": 1, "H1": 2, "Q2": 2, "Q3": 3,
            "H2": 4, "Q4": 4, "FY": 4}.get(suffix, 0)
    return (year, rank, suffix)

def first_finite(row, candidates):
    for field in candidates:
        value = pd.to_numeric(row.get(field), errors="coerce")
        if pd.notna(value) and np.isfinite(value):
            return float(value)
    return None

# 按财期而不是公告日期选择当前可见的最新财期，再匹配上年同财期。
period_keys = rows["fy_period"].map(period_sort_key)
rows[["period_year", "period_rank", "period_suffix"]] = pd.DataFrame(
    period_keys.tolist(), index=rows.index
)
latest = rows.sort_values(
    ["symbol", "period_year", "period_rank", "period_suffix", "date_key"]
).groupby("symbol", as_index=False).tail(1)
history = rows.set_index(["symbol", "fy_period"])
screening = []
for _, current in latest.iterrows():
    prior_key = (current["symbol"], previous_year_period(current["fy_period"]))
    if prior_key[1] is None or prior_key not in history.index:
        continue
    prior = history.loc[prior_key]
    current_revenue = first_finite(
        current, ["is_reported_revenue", "is_revenue_business_total"]
    )
    prior_revenue = first_finite(
        prior, ["is_reported_revenue", "is_revenue_business_total"]
    )
    gross_profit = first_finite(
        current, ["is_reported_gross_profit", "is_gross_profit"]
    )
    operating_cf = first_finite(
        current, ["cfs_net_cf_operating", "cfs_reported_cf_operating"]
    )
    if None in (current_revenue, prior_revenue, gross_profit, operating_cf):
        continue
    if current_revenue <= 0 or prior_revenue <= 0:
        continue
    revenue_growth = current_revenue / prior_revenue - 1.0
    gross_margin = gross_profit / current_revenue
    screening.append({
        "symbol": current["symbol"],
        "fy_period": current["fy_period"],
        "revenue_growth": revenue_growth,
        "gross_margin": gross_margin,
        "operating_cf": operating_cf,
        "eligible": revenue_growth > 0.20 and gross_margin > 0.20 and operating_cf > 0,
    })

screening = pd.DataFrame(screening)

import pandas as pd
import tqx_data

backtest_end = "20251231"
all_rows = tqx_data.get_financial_statement(
    market="nb",
    symbol=nasdaq_100_symbols,
    # 公司财年可能领先公告自然年；同时要覆盖同比所需的上年同财期。
    start_quarter="2023q1",
    end_quarter="2026q4",
    date=backtest_end,
    is_latest=False,
    interim_type="cumulative",
    fields=required_financial_fields,
)
all_rows["date_key"] = pd.to_datetime(all_rows["date"], errors="coerce")
all_rows = all_rows.dropna(subset=["symbol", "fy_period", "date_key"])

def visible_financials(as_of_date):
    cutoff = pd.to_datetime(as_of_date, format="%Y%m%d")

    # 必须先按调仓日过滤，再处理同一财期的重述版本。
    visible = all_rows[all_rows["date_key"] <= cutoff].copy()
    visible = visible.sort_values("date_key").drop_duplicates(
        ["symbol", "fy_period"], keep="last"
    )
    assert visible.empty or visible["date_key"].max() <= cutoff

    keys = visible["fy_period"].map(period_sort_key)
    visible[["period_year", "period_rank", "period_suffix"]] = pd.DataFrame(
        keys.tolist(), index=visible.index
    )
    latest = visible.sort_values(
        ["symbol", "period_year", "period_rank", "period_suffix", "date_key"]
    ).groupby("symbol", as_index=False).tail(1)
    return visible, latest

panel_rows = []
for rebalance_date in rebalance_dates:  # 例如每 5 个交易日
    history, latest = visible_financials(rebalance_date)
    history = history.set_index(["symbol", "fy_period"])
    for _, current in latest.iterrows():
        prior_period = previous_year_period(current["fy_period"])
        prior_key = (current["symbol"], prior_period)
        if prior_period is None or prior_key not in history.index:
            continue
        prior = history.loc[prior_key]
        # 按 2.4.3 的兼容字段和有限值规则计算同比、毛利率、经营现金流。
        metrics = calculate_financial_metrics(current, prior)
        if metrics is None:
            continue
        panel_rows.append({
            "date": rebalance_date,
            "symbol": current["symbol"],
            "fy_period": current["fy_period"],
            "source_date": current["date"],
            **metrics,
        })

financial_panel = pd.DataFrame(panel_rows)