#!/usr/bin/env python3 """Build auditable continuous panels for the five-year-plan research report.""" from __future__ import annotations import json import math import re from collections import Counter from pathlib import Path import numpy as np import pandas as pd ROOT = Path(__file__).resolve().parents[1] RAW = ROOT / "data" / "research" / "raw" OUT = ROOT / "data" / "research" / "derived" YEARS = list(range(1990, 2026)) def load_json(name: str) -> dict: return json.loads((RAW / name).read_text(encoding="utf-8")) def numeric(value: object) -> float: if value in (None, ""): return math.nan try: return float(value) except (TypeError, ValueError): return math.nan def flatten_queries(payload: dict) -> pd.DataFrame: rows: list[dict] = [] for query_name, query in payload["queries"].items(): request = query["request"] for period in query["response"].get("data", []): for item in period.get("values", []): rows.append( { "query_name": query_name, "period_code": period.get("code"), "period_name": period.get("name"), "indicator_id": item.get("_id"), "catalog_id": item.get("catalogid") or request.get("cid"), "indicator_name": item.get("i_showname", "").strip(), "unit": item.get("du_name", "").strip(), "value": numeric(item.get("value")), "annotation": item.get("i_annotation"), "root_id": request.get("rootId"), } ) return pd.DataFrame(rows) def choose_values( frame: pd.DataFrame, indicator_ids: list[str], frequency: str, ) -> dict[int, dict]: selected = frame[frame["indicator_id"].isin(indicator_ids)].copy() if frequency == "年度": selected = selected[selected["period_code"].str.match(r"^\d{4}YY$", na=False)] else: selected = selected[selected["period_code"].str.match(r"^\d{4}12MM$", na=False)] selected["year"] = selected["period_code"].str[:4].astype(int) selected["priority"] = selected["indicator_id"].map( {indicator_id: index for index, indicator_id in enumerate(indicator_ids)} ) output: dict[int, dict] = {} for year, group in selected.groupby("year"): valid = group[group["value"].notna()].sort_values("priority") if valid.empty: continue chosen = valid.iloc[0] values = valid["value"].astype(float).to_numpy() conflict = bool(len(values) > 1 and np.nanmax(values) - np.nanmin(values) > 0.05) output[int(year)] = { "value": float(chosen["value"]), "indicator_id": chosen["indicator_id"], "catalog_id": chosen["catalog_id"], "indicator_name": chosen["indicator_name"], "unit_returned": chosen["unit"], "annotation": chosen["annotation"], "candidate_count": int(len(valid)), "candidate_conflict": "是" if conflict else "否", } return output def build_four_industry_panel() -> tuple[pd.DataFrame, pd.DataFrame]: annual = flatten_queries(load_json("nbs-four-industry-annual-panel-raw.json")) monthly = flatten_queries(load_json("nbs-four-industry-monthly-panel-raw.json")) definitions = [ ("房地产与基础设施", "房地产开发投资额", "亿元", "年度", ["592bd692b6f24ab4b88d5a3ef36757cb"], "投资"), ("房地产与基础设施", "商品房销售面积", "万平方米", "年度", ["254cd99c243a427188a434885cbf50d1"], "需求"), ("房地产与基础设施", "商品房销售额", "亿元", "年度", ["0379b2ea66e74dcc93c7af2a760a334d"], "需求"), ("房地产与基础设施", "房地产施工面积", "万平方米", "月度累计取12月", ["8d0e68e8f3874f6ba1927517ad6663c7"], "产能/在建"), ("煤炭钢铁与原材料", "原煤产量", "万吨", "年度", ["b6ba346ee94f401eb1e9bf4e4de979ed"], "产量"), ("煤炭钢铁与原材料", "粗钢产量", "万吨", "年度", ["d5581e56f18d42a5be5f8877f2bc384e"], "产量"), ( "煤炭钢铁与原材料", "煤炭开采和洗选业利润总额", "亿元", "月度累计取12月", [ "aaf85a12b98c4a58a24978be84babb51", "84dbc9031feb475098ddd2b11fdfa4f7", "bf61e8faa8ca41dab2115f10ae4dba54", "294182b531754af1a0d20cb006c70934", ], "利润", ), ( "煤炭钢铁与原材料", "黑色金属冶炼和压延加工业利润总额", "亿元", "月度累计取12月", [ "0b2f7cdc2ccc4741ae4b6dd8218f7fe6", "a69c6ea024c74fba966471d0fb26b890", "fc89a806c6ad43d296ab2c7ca17ff281", "aeb813d115894b839c3f4d5093d083a9", ], "利润", ), ("移动互联网与信息产业", "移动电话年末用户", "万户", "年度", ["0ced7bd16cd341e78d7d26705306dc15"], "渗透"), ("移动互联网与信息产业", "互联网宽带接入端口", "万个", "年度", ["82be89f29cfa4e19a5d2ef53917571b9"], "基础设施"), ("移动互联网与信息产业", "电信业务总量", "亿元", "年度", ["c24dbf9fd9d34ccc929036f225b1e4d1"], "产业规模"), ("移动互联网与信息产业", "软件业务收入", "万元", "年度", ["022e11c857ce40d3a827be6ab3263449"], "产业收入"), ("移动互联网与信息产业", "集成电路产量", "万块", "年度", ["95ed3bbbb09c40d6998c9dacaea442f8"], "产量"), ( "新能源汽车与新能源体系", "新能源汽车产量", "万辆", "月度累计取12月", ["5dc8b74d3e6e431aae8064574b02992b", "bf1176ebc395416f895e6b2e67129452"], "产量", ), ( "新能源汽车与新能源体系", "太阳能发电量", "亿千瓦时", "月度累计取12月", ["3989ee06d17b4fe7a60a653a51579b51", "0cf2667f31ed47b894883064cba19e81"], "产出", ), ( "新能源汽车与新能源体系", "风力发电量", "亿千瓦时", "月度累计取12月", ["b16dedabaac543a990bd9fd16f581f47", "fda87554277d45d3b95527e1465b62d9"], "产出", ), ] rows: list[dict] = [] for case, metric, unit, frequency, indicator_ids, stage in definitions: source_frame = annual if frequency == "年度" else monthly values = choose_values( source_frame, indicator_ids, "年度" if frequency == "年度" else "月度", ) valid_years = sorted(values) for year in YEARS: observation = values.get(year, {}) rows.append( { "案例": case, "年份": year, "指标": metric, "产业传导环节": stage, "数值": observation.get("value"), "单位": unit, "原始频率": frequency, "年度化方法": "原始年度值" if frequency == "年度" else "当年12月累计值；12月缺失则保持缺失", "缺失标记": "否" if year in values else "是", "指标ID": observation.get("indicator_id") or "|".join(indicator_ids), "目录ID": observation.get("catalog_id"), "原始指标名称": observation.get("indicator_name"), "重复候选数": observation.get("candidate_count", 0), "候选值冲突": observation.get("candidate_conflict", "否"), "可得起始年": min(valid_years) if valid_years else None, "可得结束年": max(valid_years) if valid_years else None, "口径备注": observation.get("annotation"), "数据来源": "国家统计局数据发布库", "原始文件": ( "nbs-four-industry-annual-panel-raw.json" if frequency == "年度" else "nbs-four-industry-monthly-panel-raw.json" ), "是否计算": "否" if frequency == "年度" else "是：筛选12月累计值，未插值", } ) panel = pd.DataFrame(rows) coverage = ( panel.groupby(["案例", "指标", "单位", "原始频率"], as_index=False) .agg( 可得起始年=("年份", lambda x: panel.loc[x.index][panel.loc[x.index, "缺失标记"] == "否"]["年份"].min()), 可得结束年=("年份", lambda x: panel.loc[x.index][panel.loc[x.index, "缺失标记"] == "否"]["年份"].max()), 有效年度数=("缺失标记", lambda x: int((x == "否").sum())), 全区间年度数=("年份", "size"), 候选冲突年度数=("候选值冲突", lambda x: int((x == "是").sum())), ) ) coverage["覆盖率"] = coverage["有效年度数"] / coverage["全区间年度数"] coverage["可比性结论"] = np.where( coverage["覆盖率"] == 1, "1990-2025连续可得", "仅在可得区间内比较；缺失年份未插值", ) return panel, coverage def main_policy_label(policy_type: str) -> str: first = str(policy_type).split("/")[0] aliases = { "支柱产业": "战略愿景", "约束性任务": "监管限制", "量化硬任务": "监管限制", } return aliases.get(first, first) def cohen_kappa(left: pd.Series, right: pd.Series) -> float: categories = sorted(set(left.dropna()) | set(right.dropna())) total = len(left) observed = float((left == right).sum()) / total expected = sum( float((left == category).sum()) / total * float((right == category).sum()) / total for category in categories ) return (observed - expected) / (1 - expected) if expected < 1 else 1.0 def build_policy_consensus() -> tuple[pd.DataFrame, pd.DataFrame]: first = pd.read_csv(OUT / "five_year_plan_policy_strength_coding.csv") second = pd.read_csv(OUT / "policy_coding_second_review.csv") recheck = pd.read_csv(OUT / "policy_coding_second_recheck.csv") aliases = { ("八五（1991-1995）", "交通运输"): "交通与通信", ("十二五（2011-2015）", "保障性住房"): "城镇化与保障房", ("十四五（2021-2025）", "房地产长效机制"): "房地产与金融风险", ("九五（1996-2000）", "能源交通通信重点建设"): "能源交通通信", ("九五（1996-2000）", "集成电路"): "集成电路与移动通信", ("十五（2001-2005）", "西部开发"): "西部开发与城镇化", ("十五五（2026-2030）", "研发投入"): "研发与数字经济", } second_lookup = { (row["规划周期"], row["政策方向"]): row for _, row in second.iterrows() } recheck_lookup = { (row["规划周期"], row["政策方向"]): row for _, row in recheck.iterrows() } decisions: list[dict] = [] for _, row in first.iterrows(): key = (row["规划周期"], row["政策方向"]) review = recheck_lookup.get(key) review_round = "第二轮引文修订复核" if review is not None else "首轮盲审" if review is None: old_key = (row["规划周期"], aliases.get(key, row["政策方向"])) review = second_lookup.get(old_key) if review is None: raise AssertionError(f"Missing second review for {key}") first_label = main_policy_label(row["政策类型"]) second_label = review["第二编码政策类型主标签"] final_label = first_label if row["政策方向"] == "健康中国": final_label = "社会政策" decisions.append( { "规划周期": row["规划周期"], "政策方向": row["政策方向"], "第一编码主标签": first_label, "第一编码强度": int(row["强度等级"]), "第二编码主标签": second_label, "第二编码强度": int(review["第二编码强度等级(1-4)"]), "引文是否足以复核": review["引文是否足以复核(是/否)"], "复核轮次": review_round, "主标签一致": "是" if first_label == second_label else "否", "强度完全一致": "是" if int(row["强度等级"]) == int(review["第二编码强度等级(1-4)"]) else "否", "裁决主标签": final_label, "裁决强度": int(row["强度等级"]), "裁决理由": ( "按编码手册，以修订后逐字引文呈现的承诺形式为准；" "量化预期性指标不因数值精确自动升为约束性指标。" ), "第二编码理由": review["第二编码理由"], "复核备注": review["复核备注"], "逐字引文": row["逐字引文"], "精确定位": row["精确定位"], "原始链接": row["原始链接"], } ) consensus = pd.DataFrame(decisions) strength_left = consensus["第一编码强度"] strength_right = consensus["第二编码强度"] summary = pd.DataFrame( [ ("编码记录数", len(consensus), "28条高相关目的抽样，不是全文穷尽编码"), ("首轮盲审引文可复核率", float((second["引文是否足以复核(是/否)"] == "是").mean()), "22/28"), ("修订后引文可复核率", float((consensus["引文是否足以复核"] == "是").mean()), "28/28"), ("强度等级完全一致率", float((strength_left == strength_right).mean()), "修订后第一编码与第二编码"), ("政策类型主标签一致率", float((consensus["第一编码主标签"] == consensus["第二编码主标签"]).mean()), "修订后第一编码与第二编码"), ("强度等级Cohen's Kappa", cohen_kappa(strength_left, strength_right), "名义尺度；目的抽样"), ], columns=["指标", "数值", "口径"], ) return consensus, summary def build_a_share_overlay() -> pd.DataFrame: company = pd.read_csv(OUT / "a_share_company_snapshot_ifind.csv.gz") availability = pd.read_csv(OUT / "financial_data_availability_metadata.csv") proxy = { "房地产与基础设施": ["金融地产"], "煤炭钢铁与原材料": ["能源", "原材料"], "移动互联网与信息产业": ["信息技术", "通信服务"], "新能源汽车与新能源体系": ["工业", "原材料", "可选消费", "公用事业"], } rows: list[dict] = [] for case, industries in proxy.items(): for year, group in company[company["stable_industry"].isin(industries)].groupby("year"): profit = group["net_profit_parent"].sum(min_count=1) equity = group["equity_parent"].sum(min_count=1) market_cap = group["total_market_cap"].sum(min_count=1) availability_row = availability[availability["年份"] == year].iloc[0] rows.append( { "案例": case, "年份": int(year), "稳定行业代理": "+".join(industries), "公司数": int(group["code"].nunique()), "营业收入合计": group["operating_revenue"].sum(min_count=1), "归母净利润合计": profit, "归母权益合计": equity, "总市值合计": market_cap, "组合ROE": profit / equity if pd.notna(equity) and equity != 0 else np.nan, "组合PE": market_cap / profit if pd.notna(profit) and profit > 0 else np.nan, "组合PB": market_cap / equity if pd.notna(equity) and equity > 0 else np.nan, "财务报告期末": availability_row["财务报告期末"], "统一保守可得日期代理": availability_row["统一保守可得日期代理"], "同年末市场是否可得": availability_row["市场快照时是否可得"], "代理局限": "稳定一级行业宽代理，不等于细分产业；财务数据不得用于解释同年末市场状态", "数据来源": "同花顺iFinD历史快照；本报告计算", } ) return pd.DataFrame(rows) def build_exposure_candidates() -> tuple[pd.DataFrame, pd.DataFrame]: company = pd.read_csv(OUT / "a_share_company_snapshot_ifind.csv.gz") fields = [ "name", "csrc_industry", "new_csrc_industry", "csi_industry_2021", "stable_industry", ] text = company[fields].fillna("").agg("|".join, axis=1) rules = [ ("房地产开发", r"房地产开发|房地产经营|房地产业", "历史行业分类直接命中", "中"), ("基础设施建设", r"土木工程|建筑业|铁路运输|道路运输|基础设施", "历史行业分类直接或相邻命中", "中"), ("煤炭", r"煤炭开采|煤炭采选", "历史行业分类直接命中", "中"), ("钢铁", r"黑色金属冶炼|钢铁", "历史行业分类直接命中", "中"), ("软件与信息技术服务", r"软件和信息技术服务|计算机应用服务|软件开发", "历史行业分类直接命中", "中"), ("通信服务", r"电信服务|通信服务|电信、广播电视和卫星传输", "历史行业分类直接命中", "中"), ("半导体候选", r"计算机、通信和其他电子设备制造|电子元器件", "宽行业候选，不能证明半导体收入", "低"), ("新能源汽车候选", r"汽车制造|汽车整车|汽车零部件", "汽车宽行业候选，不能证明新能源车收入", "低"), ("光伏风电储能候选", r"电气机械和器材制造|电力、热力生产和供应", "宽行业候选，不能证明新能源收入", "低"), ] rows: list[pd.DataFrame] = [] for target, pattern, reason, confidence in rules: matched = company[text.str.contains(pattern, regex=True, na=False)].copy() if matched.empty: continue matched["细分产业"] = target matched["映射状态"] = "候选映射；主营/分部收入待核验" matched["收入暴露比例"] = np.nan matched["收入暴露比例状态"] = "数据缺失" matched["映射证据"] = reason matched["行业归属可信度"] = confidence matched["收入暴露可信度"] = "未验证" matched["证据日期"] = matched["report_date"] matched["数据来源"] = "同花顺iFinD历史行业分类快照" matched["后续所需证据"] = "年报主营构成或分部收入、业务口径、合并范围与报告期" rows.append( matched[ [ "year", "code", "name", "stable_industry", "csrc_industry", "new_csrc_industry", "csi_industry_2021", "细分产业", "映射状态", "收入暴露比例", "收入暴露比例状态", "映射证据", "行业归属可信度", "收入暴露可信度", "证据日期", "数据来源", "后续所需证据", ] ] ) candidates = pd.concat(rows, ignore_index=True) candidates = candidates.rename(columns={"year": "映射年份", "code": "证券代码", "name": "证券简称"}) coverage = ( candidates.groupby(["映射年份", "细分产业", "行业归属可信度"], as_index=False) .agg(候选公司数=("证券代码", "nunique")) ) coverage["已验证收入暴露公司数"] = 0 coverage["收入暴露覆盖率"] = 0.0 coverage["结论"] = "只建立候选层；没有主营/分部收入时不得汇总细分产业收入、市值或利润" return candidates, coverage def annual_values_from_macro(payload: dict, indicator_id: str) -> dict[int, float]: frame = flatten_queries(payload) return { int(row["period_code"][:4]): float(row["value"]) for _, row in frame[ (frame["indicator_id"] == indicator_id) & frame["period_code"].str.match(r"^\d{4}YY$", na=False) & frame["value"].notna() ].iterrows() } def build_macro_controls(panel: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]: macro = load_json("nbs-macro-controls-annual-raw.json") annual_payload = load_json("nbs-four-industry-annual-panel-raw.json") series = [ ("GDP实际增速代理", "国内生产总值指数(上年=100)", annual_values_from_macro(macro, "93dd15c8a3a3400ea89f8dceec7ab2b3"), "指数", "宏观增长"), ("社会融资规模增量", "社会融资规模增量", annual_values_from_macro(macro, "37968f05a95c499bb601852f565ab1a2"), "亿元", "信用"), ("人民币对美元汇率", "美元=100人民币", annual_values_from_macro(macro, "cd3b99ca016e4e0b94cd4c5647228df6"), "元", "汇率"), ("PPI", "工业生产者出厂价格指数(上年=100)", annual_values_from_macro(annual_payload, "e88e7a5ecd2648f78cecc78a5f3e225d"), "指数", "价格"), ("房地产开发投资额", "房地产开发投资额", annual_values_from_macro(annual_payload, "592bd692b6f24ab4b88d5a3ef36757cb"), "亿元", "地产"), ("商品房销售面积", "商品房销售面积", annual_values_from_macro(annual_payload, "254cd99c243a427188a434885cbf50d1"), "万平方米", "地产"), ] rows: list[dict] = [] for metric, source_name, values, unit, group in series: for year in YEARS: rows.append( { "年份": year, "控制组": group, "指标": metric, "数值": values.get(year), "单位": unit, "缺失标记": "否" if year in values else "是", "原始指标": source_name, "数据来源": "国家统计局数据发布库", "是否计算": "否", } ) controls = pd.DataFrame(rows) gaps = pd.DataFrame( [ ("M2", "月度", "中国人民银行", "已接入FRED分发的IMF序列；有效期1998-12至2019-08，之后仍缺", "部分完成"), ("利率", "月度", "中国人民银行", "已接入FRED分发的3个月同业利率；有效期1997-06至2025-12", "代理序列已接入"), ("全球科技周期", "月度", "全球科技指数数据库", "已接入FRED分发的NASDAQ综合指数；价格指数代理，不是行业总收益指数", "代理序列已接入"), ("全球商品周期", "月度", "国际货币基金组织", "已接入FRED分发的全球大宗商品价格指数；有效期1992-01至2025-12", "代理序列已接入"), ], columns=["控制变量", "目标频率", "优先来源", "当前状态", "证据状态"], ) return controls, gaps def build_ipo_annual_history() -> pd.DataFrame: ipo = pd.read_csv(OUT / "a_share_ipo_company_history_ifind.csv.gz") ipo["上市年份"] = pd.to_datetime(ipo["listed_date"], errors="coerce").dt.year rows = ( ipo[ipo["上市年份"].between(1991, 2025)] .groupby("上市年份", as_index=False) .agg( IPO公司数=("code", "nunique"), IPO募资额=("ipo_proceeds", lambda x: x.sum(min_count=1)), 募资额有效公司数=("ipo_proceeds", "count"), ) ) rows["募资额单位"] = "元" rows["股票宇宙口径"] = "七个年末股票池并集；可能遗漏快照间上市后又退出的证券" rows["完整性"] = "不称为完整全市场融资史" rows["数据来源"] = "同花顺iFinD HTTP API历史记录；本报告汇总" return rows def build_gap_register() -> pd.DataFrame: return pd.DataFrame( [ ( "连续年度A股行业市值、利润、ROE和估值", "1991-2025逐年全A历史成分与退市证券", "当前只有1995/2000/2005/2010/2015/2020/2025七个快照", "IFIND_REFRESH_TOKEN或Wind/CSMAR公司级历史快照", "权限阻塞", ), ( "完整再融资年度史", "逐公司逐事件增发、配股、可转债和其他再融资", "已实现可选公司年度增发/配股区间汇总；无令牌未执行，且不能还原逐笔事件", "IFIND_REFRESH_TOKEN执行年度模式；逐笔事件仍需iFinD事件字段或Wind/CSMAR融资事件库", "权限/字段阻塞", ), ( "细分产业主营收入暴露", "公司-年份-业务分部收入与占比", "已接入东方财富结构化主营构成并生成2025全A公司池历史年度面板", "扩展到已退市公司并逐页复核年报原文；iFinD可作交叉验证", "第三方结构化数据已完成；退市与原文复核待补", ), ], columns=["数据模块", "目标口径", "当前可用数据", "解除阻塞所需", "状态"], ) def main() -> None: OUT.mkdir(parents=True, exist_ok=True) panel, coverage = build_four_industry_panel() panel.to_csv(OUT / "four_industry_continuous_panel.csv", index=False, encoding="utf-8-sig") coverage.to_csv(OUT / "four_industry_panel_coverage.csv", index=False, encoding="utf-8-sig") consensus, consensus_summary = build_policy_consensus() consensus.to_csv(OUT / "policy_coding_consensus.csv", index=False, encoding="utf-8-sig") consensus_summary.to_csv(OUT / "policy_coding_reliability_summary.csv", index=False, encoding="utf-8-sig") overlay = build_a_share_overlay() overlay.to_csv(OUT / "four_industry_a_share_snapshot_overlay.csv", index=False, encoding="utf-8-sig") exposure, exposure_coverage = build_exposure_candidates() exposure.to_csv(OUT / "company_subindustry_exposure_candidates.csv", index=False, encoding="utf-8-sig") exposure_coverage.to_csv(OUT / "company_subindustry_exposure_coverage.csv", index=False, encoding="utf-8-sig") controls, control_gaps = build_macro_controls(panel) controls.to_csv(OUT / "macro_controls_annual.csv", index=False, encoding="utf-8-sig") control_gaps.to_csv(OUT / "macro_control_gap_register.csv", index=False, encoding="utf-8-sig") ipo = build_ipo_annual_history() ipo.to_csv(OUT / "a_share_ipo_annual_history.csv", index=False, encoding="utf-8-sig") build_gap_register().to_csv(OUT / "research_data_gap_register.csv", index=False, encoding="utf-8-sig") assert len(panel) == len(YEARS) * 16 assert panel["缺失标记"].isin(["是", "否"]).all() assert not panel[panel["缺失标记"] == "是"]["数值"].notna().any() assert consensus.shape[0] == 28 assert consensus["引文是否足以复核"].eq("是").all() assert exposure["收入暴露比例"].isna().all() assert exposure["收入暴露可信度"].eq("未验证").all() assert controls["年份"].between(1990, 2025).all() print("continuous-panel assertions: OK") if __name__ == "__main__": main()