#!/usr/bin/env python3 """Build auditable derived tables used by the five-year-plan research report.""" from __future__ import annotations import gzip import re from pathlib import Path import pandas as pd ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / "data" / "research" / "derived" def read_company_snapshot() -> pd.DataFrame: with gzip.open(OUT / "a_share_company_snapshot_ifind.csv.gz", "rt", encoding="utf-8-sig") as handle: return pd.read_csv(handle) def safe_cagr(start: float, end: float, years: int) -> float | None: if start <= 0 or end <= 0: return None return (end / start) ** (1 / years) - 1 def build_mapping_sensitivity(market: pd.DataFrame, metadata: pd.DataFrame) -> pd.DataFrame: output = market.copy() mapped_total = ( output[output["industry"] != "未映射"] .groupby("year")["total_market_cap"] .sum() ) output["mapped_only_total_market_cap_share"] = output.apply( lambda row: row["total_market_cap"] / mapped_total[row["year"]] if row["industry"] != "未映射" and mapped_total[row["year"]] else None, axis=1, ) output["share_sensitivity_pp"] = ( output["mapped_only_total_market_cap_share"] - output["total_market_cap_share"] ) * 100 coverage = metadata[ [ "year", "industry_mapping_rate", "mapped_total_market_cap_rate", "mapped_circulating_market_cap_rate", "mapped_free_float_market_cap_rate", ] ] return output.merge(coverage, on="year", how="left") def grouped_snapshot( market: pd.DataFrame, profit: pd.DataFrame, year: int, industries: tuple[str, ...], ) -> tuple[float, float]: market_value = market[ (market["year"] == year) & market["industry"].isin(industries) ]["total_market_cap_share"].sum() profit_value = profit[ (profit["year"] == year) & profit["industry"].isin(industries) ]["net_profit_share"].sum() return float(market_value), float(profit_value) def build_planning_direction_table( company: pd.DataFrame, market: pd.DataFrame, profit: pd.DataFrame, ) -> pd.DataFrame: rows: list[dict[str, object]] = [] definitions = ( ( "电子/信息/数字", 1995, 2025, ("信息技术", "通信服务"), "稳定十行业中的信息技术+通信服务", "八五以来持续强化", "较强关联", ), ( "能源/原材料", 1995, 2025, ("能源", "原材料"), "稳定十行业中的能源+原材料", "早期扩产，后期约束与安全并重", "周期性强关联", ), ) for direction, start, end, industries, rule, evidence, judgment in definitions: start_market, start_profit = grouped_snapshot(market, profit, start, industries) end_market, end_profit = grouped_snapshot(market, profit, end, industries) rows.append( { "规划方向": direction, "起始年份": start, "结束年份": end, "识别规则": rule, "文本证据": evidence, "起始市值权重": start_market, "结束市值权重": end_market, "市值权重变化百分点": (end_market - start_market) * 100, "起始利润占比": start_profit, "结束利润占比": end_profit, "利润占比变化百分点": (end_profit - start_profit) * 100, "初步判断": judgment, "数据来源": "同花顺iFinD HTTP API；本报告计算", } ) text = company[ ["csrc_industry", "new_csrc_industry", "csi_industry_2021"] ].fillna("").astype(str).agg("|".join, axis=1) real_estate = company[text.str.contains("房地产", regex=False)].copy() for year in (2000, 2015, 2025): total_market = company.loc[company["year"] == year, "total_market_cap"].sum() total_profit = company.loc[company["year"] == year, "net_profit_parent"].sum() members = real_estate[real_estate["year"] == year] rows.append( { "规划方向": "房地产/住房", "起始年份": year, "结束年份": year, "识别规则": "任一历史行业字段包含“房地产”；不以金融地产大类代替", "文本证据": "九五后制度推动，十一五后调控增强", "起始市值权重": members["total_market_cap"].sum() / total_market, "结束市值权重": members["total_market_cap"].sum() / total_market, "市值权重变化百分点": None, "起始利润占比": members["net_profit_parent"].sum() / total_profit, "结束利润占比": members["net_profit_parent"].sum() / total_profit, "利润占比变化百分点": None, "初步判断": "结构变化已验证，因果需克制", "数据来源": "同花顺iFinD HTTP API；本报告计算", } ) rows.append( { "规划方向": "新能源汽车/光伏/储能", "起始年份": None, "结束年份": None, "识别规则": "稳定一级行业不能可靠拆分细分产业", "文本证据": "十二五后连续强化", "起始市值权重": None, "结束市值权重": None, "市值权重变化百分点": None, "起始利润占比": None, "结束利润占比": None, "利润占比变化百分点": None, "初步判断": "数据不足；不得以工业大类冒充细分产业", "数据来源": "同花顺iFinD HTTP API；本报告计算", } ) return pd.DataFrame(rows) def build_cycle_fundamentals( fundamentals: pd.DataFrame, returns: pd.DataFrame, ) -> pd.DataFrame: periods = { "十二五(2011-2015)": (2010, 2015), "十三五(2016-2020)": (2015, 2020), "十四五(2021-2025)": (2020, 2025), } output: list[dict[str, object]] = [] for period, (start_year, end_year) in periods.items(): for industry in sorted(set(fundamentals["industry"]) - {"未映射"}): start = fundamentals[ (fundamentals["year"] == start_year) & (fundamentals["industry"] == industry) ].iloc[0] end = fundamentals[ (fundamentals["year"] == end_year) & (fundamentals["industry"] == industry) ].iloc[0] market_industry = "电信业务" if industry == "通信服务" else industry market = returns[ (returns["规划周期"] == period) & (returns["行业"] == market_industry) ] relative = float(market.iloc[0]["相对沪深300"]) if not market.empty else None output.append( { "规划周期": period, "行业": industry, "起始年份": start_year, "结束年份": end_year, "营业收入复合增速": safe_cagr( float(start["operating_revenue"]), float(end["operating_revenue"]), end_year - start_year, ), "归母净利润变化": float(end["net_profit_parent"] - start["net_profit_parent"]), "起始行业ROE": start["aggregate_roe"], "结束行业ROE": end["aggregate_roe"], "起始PE中位数": start["median_company_pe_ttm_positive"], "结束PE中位数": end["median_company_pe_ttm_positive"], "结束PE七期快照分位": end["median_company_pe_ttm_positive_snapshot_percentile"], "起始PB中位数": start["median_company_pb_mrq_positive"], "结束PB中位数": end["median_company_pb_mrq_positive"], "结束PB七期快照分位": end["median_company_pb_mrq_positive_snapshot_percentile"], "起始年换手率中位数": start["median_company_annual_turnover"], "结束年换手率中位数": end["median_company_annual_turnover"], "累计相对沪深300": relative, "数据来源": "同花顺iFinD HTTP API、腾讯证券行情；本报告计算", } ) return pd.DataFrame(output) def build_listing_composition_decomposition(company: pd.DataFrame) -> pd.DataFrame: """Build an exact revenue bridge instead of treating sample expansion as growth.""" periods = ( (1995, 2000), (2000, 2005), (2005, 2010), (2010, 2015), (2015, 2020), (2020, 2025), ) ipo = pd.read_csv( OUT / "a_share_ipo_company_history_ifind.csv.gz", parse_dates=["listed_date"], ) listed_date = ipo.drop_duplicates("code").set_index("code")["listed_date"] industries = sorted(set(company["stable_industry"]) - {"未映射"}) rows: list[dict[str, object]] = [] for start_year, end_year in periods: start_all = company[company["year"] == start_year].copy() end_all = company[company["year"] == end_year].copy() start_by_code = start_all.drop_duplicates("code").set_index("code") end_by_code = end_all.drop_duplicates("code").set_index("code") start_codes = set(start_by_code.index) end_codes = set(end_by_code.index) start_market_date = pd.Timestamp(start_all["market_date"].iloc[0]) end_market_date = pd.Timestamp(end_all["market_date"].iloc[0]) for industry in industries: start_industry_codes = set( start_by_code[start_by_code["stable_industry"] == industry].index ) end_industry_codes = set( end_by_code[end_by_code["stable_industry"] == industry].index ) same_industry_codes = start_industry_codes & end_industry_codes same_start = start_by_code.loc[ list(same_industry_codes), "operating_revenue" ] same_end = end_by_code.loc[ list(same_industry_codes), "operating_revenue" ] comparable_codes = same_start.dropna().index.intersection( same_end.dropna().index ) coverage_gain_codes = same_start[same_start.isna()].index.intersection( same_end.dropna().index ) coverage_loss_codes = same_start.dropna().index.intersection( same_end[same_end.isna()].index ) migration_in_codes = { code for code in end_industry_codes & start_codes if start_by_code.at[code, "stable_industry"] not in (industry, "未映射") } mapping_in_codes = { code for code in end_industry_codes & start_codes if start_by_code.at[code, "stable_industry"] == "未映射" } migration_out_codes = { code for code in start_industry_codes & end_codes if end_by_code.at[code, "stable_industry"] not in (industry, "未映射") } mapping_out_codes = { code for code in start_industry_codes & end_codes if end_by_code.at[code, "stable_industry"] == "未映射" } new_universe_codes = end_industry_codes - start_codes ipo_new_codes = { code for code in new_universe_codes if code in listed_date.index and pd.notna(listed_date.at[code]) and start_market_date < listed_date.at[code] <= end_market_date } newly_observed_codes = new_universe_codes - ipo_new_codes exit_universe_codes = start_industry_codes - end_codes def revenue_sum( snapshot: pd.DataFrame, codes: set[str] | pd.Index, ) -> float: if len(codes) == 0: return 0.0 value = snapshot.loc[ snapshot.index.intersection(list(codes)), "operating_revenue", ].sum(min_count=1) return ( float(value) if pd.notna(value) else 0.0 ) start_revenue = revenue_sum(start_by_code, start_industry_codes) end_revenue = revenue_sum(end_by_code, end_industry_codes) comparable_start_revenue = revenue_sum(start_by_code, comparable_codes) comparable_end_revenue = revenue_sum(end_by_code, comparable_codes) continuing_change = ( comparable_end_revenue - comparable_start_revenue ) coverage_gain = revenue_sum(end_by_code, coverage_gain_codes) coverage_loss = revenue_sum(start_by_code, coverage_loss_codes) migration_in = revenue_sum(end_by_code, migration_in_codes) migration_out = revenue_sum(start_by_code, migration_out_codes) mapping_in = revenue_sum(end_by_code, mapping_in_codes) mapping_out = revenue_sum(start_by_code, mapping_out_codes) ipo_new = revenue_sum(end_by_code, ipo_new_codes) newly_observed = revenue_sum(end_by_code, newly_observed_codes) exit_universe = revenue_sum(start_by_code, exit_universe_codes) explained_change = ( continuing_change + coverage_gain - coverage_loss + migration_in - migration_out + mapping_in - mapping_out + ipo_new + newly_observed - exit_universe ) actual_change = end_revenue - start_revenue bridge_residual = actual_change - explained_change rows.append( { "起始年份": start_year, "结束年份": end_year, "稳定行业": industry, "起始公司数": len(start_industry_codes), "结束公司数": len(end_industry_codes), "同一行业持续公司数": len(same_industry_codes), "收入可比持续公司数": len(comparable_codes), "起始A股公司收入合计": start_revenue, "结束A股公司收入合计": end_revenue, "收入合计变化额": actual_change, "A股公司收入合计CAGR": safe_cagr( float(start_revenue), float(end_revenue), end_year - start_year ), "持续公司起始收入": comparable_start_revenue, "持续公司结束收入": comparable_end_revenue, "持续公司收入CAGR": safe_cagr( float(comparable_start_revenue), float(comparable_end_revenue), end_year - start_year, ), "持续公司收入变化额": continuing_change, "同一行业收入覆盖新增": coverage_gain, "同一行业收入覆盖减少": coverage_loss, "跨稳定行业迁入收入": migration_in, "跨稳定行业迁出收入": migration_out, "未映射转入收入": mapping_in, "转为未映射收入": mapping_out, "区间IPO且期末在样本收入": ipo_new, "非IPO新进入观察宇宙收入": newly_observed, "退出期末观察宇宙收入": exit_universe, "桥接解释变化额": explained_change, "桥接残差": bridge_residual, "区间IPO期末收入占比": ipo_new / end_revenue if end_revenue else None, "非IPO新观察收入占比": newly_observed / end_revenue if end_revenue else None, "全产业真实增长": "未测量；需国家统计局、行业部门或协会的全行业口径", "口径说明": ( "收入变化=持续公司变化+覆盖新增-覆盖减少+跨行业迁入-迁出" "+未映射转入-转出+区间IPO+非IPO新观察-退出观察宇宙；" "退出观察宇宙未等同退市" ), "数据来源": "同花顺iFinD HTTP API；本报告计算", } ) return pd.DataFrame(rows) def build_listing_composition_summary( decomposition: pd.DataFrame, ) -> pd.DataFrame: component_columns = [ "起始公司数", "结束公司数", "收入可比持续公司数", "起始A股公司收入合计", "结束A股公司收入合计", "收入合计变化额", "持续公司起始收入", "持续公司结束收入", "持续公司收入变化额", "同一行业收入覆盖新增", "同一行业收入覆盖减少", "跨稳定行业迁入收入", "跨稳定行业迁出收入", "未映射转入收入", "转为未映射收入", "区间IPO且期末在样本收入", "非IPO新进入观察宇宙收入", "退出期末观察宇宙收入", "桥接解释变化额", "桥接残差", ] summary = ( decomposition.groupby(["起始年份", "结束年份"], as_index=False)[ component_columns ] .sum() ) summary["A股公司收入合计CAGR"] = summary.apply( lambda row: safe_cagr( float(row["起始A股公司收入合计"]), float(row["结束A股公司收入合计"]), int(row["结束年份"] - row["起始年份"]), ), axis=1, ) summary["持续公司收入CAGR"] = summary.apply( lambda row: safe_cagr( float(row["持续公司起始收入"]), float(row["持续公司结束收入"]), int(row["结束年份"] - row["起始年份"]), ), axis=1, ) for column, output_column in ( ("区间IPO且期末在样本收入", "区间IPO期末收入占比"), ("非IPO新进入观察宇宙收入", "非IPO新观察收入占比"), ): summary[output_column] = ( summary[column] / summary["结束A股公司收入合计"] ) summary["结论边界"] = ( "仅拆解A股快照收入变化；不等于全产业增长，亦不能把新进入观察宇宙全部解释为上市扩容" ) summary["数据来源"] = "同花顺iFinD HTTP API；本报告计算" return summary def build_financial_data_availability(company: pd.DataFrame) -> pd.DataFrame: """Record when annual financial data could conservatively enter an information set.""" rows: list[dict[str, object]] = [] for year, group in company.groupby("year"): market_date = pd.Timestamp(group["market_date"].iloc[0]) report_date = pd.Timestamp(group["report_date"].iloc[0]) conservative_available_date = pd.Timestamp( year=int(year) + 1, month=4, day=30 ) rows.append( { "年份": int(year), "市场快照日期": market_date.date().isoformat(), "财务报告期末": report_date.date().isoformat(), "实际公告日期": None, "实际公告日期状态": "现有iFinD原始响应未取该字段", "统一保守可得日期代理": conservative_available_date.date().isoformat(), "代理性质": "研究口径代理，不是逐公司实际公告日，也不宣称为各历史时期统一法定截止日", "市场快照时是否可得": ( "否：不得用该年度年报解释同年末市场状态" if market_date < conservative_available_date else "是" ), "收入有效公司数": int(group["operating_revenue"].notna().sum()), "利润有效公司数": int(group["net_profit_parent"].notna().sum()), "权益有效公司数": int(group["equity_parent"].notna().sum()), "数据来源": "同花顺iFinD HTTP API原始快照；本报告可得性标注", } ) return pd.DataFrame(rows) def equal_weight_proxy_relative_return( monthly: pd.DataFrame, industries: tuple[str, ...], start: str = "2020-12-31", end: str = "2025-12-31", ) -> float: frame = monthly.copy() frame["date"] = pd.to_datetime(frame["date"]) frame = frame[(frame["date"] >= start) & (frame["date"] <= end)].set_index("date") market_industries = [ "电信业务" if industry == "通信服务" else industry for industry in industries ] returns = frame[market_industries + ["沪深300"]].pct_change().dropna() basket = returns[market_industries].mean(axis=1) basket_wealth = (1 + basket).prod() benchmark_wealth = (1 + returns["沪深300"]).prod() return float(basket_wealth / benchmark_wealth - 1) def build_case_cards( company: pd.DataFrame, monthly: pd.DataFrame, ) -> pd.DataFrame: cases = ( ( "房地产与基础设施", ("金融地产",), "国务院《关于进一步深化城镇住房制度改革加快住房建设的通知》", "1998-07-03", "原文链接数据不足", "信用、土地财政、人口、库存", "政策和制度强驱动，A股承接受信用周期影响", "较强关联", ), ( "煤炭、有色、钢铁和化工", ("能源", "原材料"), "十三五供给侧结构性改革及相关去产能政策", "2016起", "https://www.gov.cn/zhengce/content/2016-02/05/content_5039686.htm", "全球商品价格、美元、地产投资", "供给约束比单纯扩产更可能改善利润", "较强关联", ), ( "移动互联网、传媒和计算机", ("信息技术", "通信服务"), "国务院《关于加快培育和发展战略性新兴产业的决定》", "2010-10-18", "https://www.gov.cn/zwgk/2010-10/18/content_1724848.htm", "3G/4G、智能手机、创业板、并购与流动性", "产业扩张与A股估值扩张并存", "较强关联", ), ( "半导体和国产替代", ("信息技术",), "国务院《中国制造2025》", "2015-05-08", "https://www.gov.cn/zhengce/content/2015-05/19/content_9784.htm", "全球半导体周期、外部限制、资本开支", "长期方向较强，分环节盈利差异大", "较强关联", ), ( "新能源汽车、光伏、风电和储能", ("工业", "原材料", "可选消费", "公用事业"), "新能源汽车产业发展规划（2021-2035年）", "2020-10-20", "https://www.gov.cn/zhengce/content/2020-11/02/content_5556716.htm", "全球需求、价格战、并网消纳、贸易壁垒", "政策与产业扩张已发生，A股结构承接分化", "已验证结构变化+较强关联", ), ( "创新药、医疗服务和医疗器械", ("医药卫生",), "十三五规划“健康中国”及药品审评审批改革", "2015起", "https://www.gov.cn/zhengce/content/2015-08/18/content_10101.htm", "临床成功率、医保支付、集采、融资环境", "扶持创新与降低支付成本同时存在", "较强关联", ), ) output: list[dict[str, object]] = [] for ( case, proxies, policy, policy_date, policy_link, counterfactual, conclusion, grade, ) in cases: snapshot = company[ (company["year"] == 2025) & company["stable_industry"].isin(proxies) ].copy() market_cap = snapshot["total_market_cap"].sum() profit = snapshot["net_profit_parent"].sum() equity = snapshot["equity_parent"].sum() output.append( { "案例": case, "稳定行业代理": "+".join(proxies), "关键政策": policy, "政策日期": policy_date, "政策原始链接": policy_link, "2025代理公司数": snapshot["code"].nunique(), "2025组合ROE": profit / equity if equity else None, "2025组合PE": market_cap / profit if profit > 0 else None, "2025组合PB": market_cap / equity if equity > 0 else None, "十四五月度等权行业代理组合相对沪深300": equal_weight_proxy_relative_return( monthly, proxies ), "主要反事实": counterfactual, "结论": conclusion, "证据等级": grade, "组合计算": "ROE=组合归母净利润/组合归母权益；PE=组合总市值/组合归母净利润；PB=组合总市值/组合归母权益；市场结果为月度再平衡等权行业代理组合", "代理局限": "稳定一级行业代理不能代替细分产业；组合包含不属于案例产业的公司，只用于A股结构背景", "数据来源": "中央政策原文、同花顺iFinD HTTP API、腾讯证券行情；本报告计算", } ) return pd.DataFrame(output) def build_policy_strength_coding() -> pd.DataFrame: sources = { "八五（1991-1995）": "https://www.ndrc.gov.cn/fggz/fzzlgh/gjfzgh/200709/P020191029595681819982.pdf", "九五（1996-2000）": "https://www.ndrc.gov.cn/fggz/fzzlgh/gjfzgh/200709/P020250225661408673913.pdf", "十五（2001-2005）": "https://www.ndrc.gov.cn/fggz/fzzlgh/gjfzgh/200709/P020191029595691974319.pdf", "十一五（2006-2010）": "https://www.gov.cn/gongbao/content/2006/content_268766.htm", "十二五（2011-2015）": "https://www.ndrc.gov.cn/fggz/fzzlgh/gjfzgh/201109/P020191029595702423333.pdf", "十三五（2016-2020）": "https://www.ndrc.gov.cn/fggz/fzzlgh/gjfzgh/201603/P020191104614882474091.pdf", "十四五（2021-2025）": "https://www.ndrc.gov.cn/fggz/fzzlgh/gjfzgh/202103/P020210323405614585384.pdf", "十五五（2026-2030）": "https://www.ndrc.gov.cn/fggz/fzzlgh/gjfzgh/202603/U020260317369114704096.pdf", } records = ( ("八五（1991-1995）", "能源与原材料", "产业扶持", 2, "能源工业及原材料工业分节", "开发与节约并重", "产业方向", "节能降耗"), ("八五（1991-1995）", "交通运输", "产业扶持", 2, "交通运输和邮电通信业分节", "建设综合运输体系、增加铁路运力", "产业方向", ""), ("八五（1991-1995）", "电子与高技术", "战略愿景/产业扶持", 2, "电子工业及科技相关分节", "推动电子、微电子、计算机和通信", "技术改造、科研投入", ""), ("九五（1996-2000）", "机械电子石化汽车建筑", "支柱产业/产业扶持", 2, "振兴支柱产业相关章节", "明确支柱产业定位", "国企投资、技术改造、信贷", "调整高耗低效产品"), ("九五（1996-2000）", "能源交通通信重点建设", "产业扶持", 2, "第三章 投资政策", "能源、交通和通信列为重点建设方向", "投资政策", "限制新开工项目总量"), ("九五（1996-2000）", "集成电路", "重大工程/产业扶持", 3, "电子工业分节", "优惠政策与集成电路专项工程并行", "优惠政策、专项工程", ""), ("十五（2001-2005）", "信息化", "重大工程/产业扶持", 3, "第六章 加速发展信息产业 大力推进信息化", "信息基础设施、软件和网络应用", "政府工程、企业投资、资本市场制度建设", ""), ("十五（2001-2005）", "西部开发", "重大工程/产业扶持", 3, "第八章 实施西部大开发战略", "西部基础设施、生态环境和投资环境建设", "财政转移支付、建设资金、税收、土地和人才政策", ""), ("十五（2001-2005）", "过剩和落后产能", "监管限制", 3, "结构调整和技术改造章节", "淘汰落后、压缩过剩", "淘汰和压缩", "限制低水平重复建设"), ("十一五（2006-2010）", "节能减排", "约束性指标/监管限制", 4, "资源节约、环境保护及指标说明", "单位GDP能耗等约束性指标", "环保准入、能耗标准、责任考核", "淘汰小火电和落后钢铁水泥"), ("十一五（2006-2010）", "自主创新与高端技术", "重大工程/产业扶持", 3, "第十章 加快发展高技术产业", "核心电子器件、高端芯片、基础软件和生物", "重大专项、科研投入、政府采购", ""), ("十一五（2006-2010）", "铁路电网与能源通道", "重大工程", 3, "基础设施相关章节", "铁路、电网、西气东输和城市基础设施", "政府投资、政策性金融、国企投资", ""), ("十二五（2011-2015）", "七大战略性新兴产业", "产业扶持", 2, "第十章 培育发展战略性新兴产业", "七大产业明确列示", "产业方向", ""), ("十二五（2011-2015）", "节能环保与能源约束", "量化硬任务/监管限制", 4, "第三章 主要目标", "能源消耗和二氧化碳排放量化下降目标", "目标责任、准入、淘汰落后", "限制高耗能和过剩产能"), ("十二五（2011-2015）", "保障性住房", "社会政策/重大工程", 3, "保障性安居工程相关章节", "加大保障性安居工程建设力度", "财政、信贷、土地和政府投资", "房地产调控并行"), ("十三五（2016-2020）", "创新与先进制造", "重大工程/产业扶持", 3, "第二篇、第五篇、第二十三章", "集成电路、人工智能、机器人和智能制造", "重大工程、产业基金、科研、采购", ""), ("十三五（2016-2020）", "钢铁煤炭去产能", "约束性任务/监管限制", 3, "供给侧结构性改革相关章节", "压减过剩产能、处置僵尸企业", "专项奖补、兼并重组、债务重组、破产清算", "明确去产能"), ("十三五（2016-2020）", "绿色发展", "约束性指标/监管限制", 4, "绿色发展及规划指标章节", "能源、排放和环境质量约束", "考核、准入、排放标准", ""), ("十三五（2016-2020）", "健康中国", "战略愿景/社会政策", 2, "健康中国相关章节", "医疗服务、药品和健康产业", "医保、采购、审评审批", "支付约束与产业扶持并存"), ("十四五（2021-2025）", "科技自立自强", "重大工程/产业扶持", 3, "第二篇 坚持创新驱动发展", "集成电路、人工智能、量子、生物和航空航天", "国家实验室、重大项目、产业金融、采购", ""), ("十四五（2021-2025）", "数字中国", "战略愿景/制度建设", 2, "第四篇及数字化相关章节", "数字化发展战略", "制度建设", ""), ("十四五（2021-2025）", "双碳与绿色转型", "约束性指标/监管限制", 4, "第十一篇 推动绿色发展", "碳强度、能源结构和绿色转型", "能耗与排放约束、绿色金融、重大工程", "高耗能高排放约束"), ("十四五（2021-2025）", "房地产长效机制", "制度建设/监管限制", 3, "住房制度相关章节", "建立房地产长效机制", "住房制度、土地与金融协同", ""), ("十五五（2026-2030）", "现代化产业体系", "重大工程/产业扶持", 3, "第二篇、第四章", "先进制造、数字化和产业基础再造", "产业基础再造工程、技术装备攻关、中长期和信用贷款", "推动落后低效产能有序退出"), ("十五五（2026-2030）", "新兴产业与未来产业", "产业扶持/战略愿景", 3, "第五章 培育壮大新兴产业和未来产业", "AI+、机器人、生物制造、低空、量子、6G等", "重大工程、产业金融、场景和标准", ""), ("十五五（2026-2030）", "研发投入", "预期性指标", 2, "主要指标表", "全社会研发经费投入年均增长7%以上", "财政科技投入、税收激励和长期资助", ""), ("十五五（2026-2030）", "绿色低碳", "约束性指标/监管限制", 4, "主要指标和绿色发展章节", "单位GDP二氧化碳排放累计下降17%", "碳约束、能源政策、重大工程", "高碳和低效产能退出"), ("十五五（2026-2030）", "产能治理", "监管限制/制度建设", 3, "产业治理和规划实施章节", "产能监测预警、低效退出、治理无序竞争", "准入、监测、标准、退出机制", "约束无序扩产"), ) columns = [ "规划周期", "政策方向", "政策类型", "强度等级", "原文章节", "政策表述摘要", "可能传导工具", "限制或约束", ] frame = pd.DataFrame(records, columns=columns) quote_anchors = [ ("坚持开发与节约并重的方针", 20), ("交通运输的建设要着眼于", 20), ("大力推广电子技术在各行各业的应用", 20), ("机械、电子、石油化工、汽车", 5), ("能源、交通、通信和支柱产业", 10), ("对集成电路的发展实行优惠政策", 20), ("加速推进信息化", 5), ("西部大开发的政策措施", 10), ("淘汰落后设 备", 5), ("单位国内生产总值能源消耗降低20%左右", 1), ("根据数字化、网络化、智能化总体趋势", 1), ("建设西电东送三大输电通道", 1), ("大力发展节能环保、新一代信息技术", 20), ("单位国内生产总值能源消耗降低 16%", 1), ("强化各级政府责任，加大保障性安居工程建设力度", 70), ("深入实施《中国制造 2025》", 30), ("加快钢铁、 煤炭等行业过剩产能退出", 30), ("单位 GDP 能源消耗降低", 10), ("深化医药卫生体制改革", 100), ("具有前瞻性、战略性的国家重大科技项目", 5), ("迎接数字时代，激活数据要素潜能", 30), ("单位国内生产总值能源消耗和二氧化碳排放分别", 5), ("房地产市场平稳健康发展", 20), ("实施产业基础再造工程", 5), ("建立未来产业投入 增长和风险分担机制", 5), ("全社会研发经费投入年均增长 7%以上", 5), ("单位国内生产总值二氧化碳排放降低 17%", 5), ("健全产能监测预警机制", 5), ] source_keys = { "八五（1991-1995）": "08", "九五（1996-2000）": "09", "十五（2001-2005）": "10", "十一五（2006-2010）": "11", "十二五（2011-2015）": "12", "十三五（2016-2020）": "13", "十四五（2021-2025）": "14", "十五五（2026-2030）": "15", } def locate_quote( plan: str, anchor: str, minimum_page: int, ) -> tuple[str | None, str]: source_key = source_keys[plan] path = ( ROOT / "data" / "research" / "raw" / "policy" / "text" / f"{source_key}-five-year-plan.txt" ) if not path.exists(): return None, "原文文本待提取" pages = path.read_text(encoding="utf-8", errors="ignore").split("\f") normalized_anchor = r"\s*".join( re.escape(part) for part in re.split(r"\s+", anchor.strip()) ) for page_number, page in enumerate(pages, start=1): if page_number < minimum_page: continue normalized_page = re.sub(r"\s+", " ", page).strip() match = re.search(normalized_anchor, normalized_page) if not match: continue start_candidates = [ normalized_page.rfind(mark, 0, match.start()) for mark in ("。", "！", "？") ] start = max(start_candidates) + 1 end_candidates = [ position for mark in ("。", "！", "？") if (position := normalized_page.find(mark, match.end())) >= 0 ] end = min(end_candidates) + 1 if end_candidates else match.end() + 160 if end - start > 500: start = match.start() end = min(len(normalized_page), match.end() + 260) quote = normalized_page[start:end].strip() locator = ( "网页正文；章节见原文章节字段" if source_key == "11" else f"PDF第{page_number}页" ) return quote, locator return None, "原文定位失败" quotes: list[str | None] = [] locators: list[str] = [] for row, (anchor, minimum_page) in zip(records, quote_anchors): quote, locator = locate_quote(row[0], anchor, minimum_page) quotes.append(quote) locators.append(locator) quote_overrides = { ("十五（2001-2005）", "西部开发"): ( "国家实行重点支持西部大开发的政策措施 增加对西部地区的财政转移支付和建设资金投入 " "并在对外开放 税收 土地 资源 人才等方面采取优惠政策", "PDF第13页；第八章第一节", ), } for index, row in enumerate(records): override = quote_overrides.get((row[0], row[1])) if override: quotes[index], locators[index] = override frame["逐字引文"] = quotes frame["精确定位"] = locators frame["编码单元"] = "完整句；若连续句共同构成同一工程或工具组合则合并" frame["第一编码者"] = "研究主稿编码" frame["第一编码"] = frame["强度等级"] frame["等级定义"] = frame["强度等级"].map( { 1: "战略愿景", 2: "明确方向或单类支持", 3: "重大工程、多工具支持或明确监管", 4: "约束性指标或可分解落实的国家级硬任务", } ) frame["编码理由"] = frame.apply( lambda row: f"{row['等级定义']}；依据逐字引文中的目标、工程、工具或约束形式", axis=1, ) frame["编码性质"] = "人工分类编码；不是关键词频次，也不是政策效果评分" frame["原始链接"] = frame["规划周期"].map(sources) frame["证据等级"] = frame["逐字引文"].apply( lambda value: "政策原文已定位；强度等级为第一研究者编码" if pd.notna(value) else "数据不足：逐字引文未定位" ) frame["编码版本"] = "1.0" return frame def build_deep_industry_timelines() -> pd.DataFrame: rows = ( ("房地产与基础设施", "1996-2000", "九五（1996-2000）", "规划与制度导入", "支柱产业、建筑和城镇住房方向；1998年住房制度改革", "规划原文；国发〔1998〕23号原文链接数据不足", "政策事实已交叉确认"), ("房地产与基础设施", "2001-2005", "十五（2001-2005）", "投资与城市化", "城镇化、西部开发和基础设施扩大地产链需求", "十五规划第八章、第九章", "政策事实+合理推断"), ("房地产与基础设施", "2006-2015", "十一五至十二五", "信用扩张与结构调控", "保障房、城市基础设施、住房结构调控并行；A股地产市值权重2015年达到可识别样本阶段高点", "规划原文；planning_direction_market_profit_comparison.csv", "政策事实+iFinD结构事实"), ("房地产与基础设施", "2016-2020", "十三五（2016-2020）", "去库存与集中度变化", "去库存、棚改和融资环境共同影响行业结构", "十三五规划；需补全行业销售与开工官方序列", "合理推断"), ("房地产与基础设施", "2021-2025", "十四五（2021-2025）", "风险化解", "房地产可识别公司市值占比降至1.3%，利润占比降至-4.0%", "planning_direction_market_profit_comparison.csv", "iFinD结构事实"), ("房地产与基础设施", "2026-2030", "十五五（2026-2030）", "新模式与城市更新", "关注城市更新、保障性住房和房地产新模式的财政与运营结构", "十五五规划原文", "政策事实；产业结果待观察"), ("煤炭钢铁与原材料", "1991-2005", "八五至十五", "扩产与工业化", "能源、钢铁、有色和化工承担瓶颈扩张与重化工业需求", "八五、九五、十五规划原文", "政策事实+合理推断"), ("煤炭钢铁与原材料", "2006-2010", "十一五（2006-2010）", "高景气与约束并行", "商品周期和投资需求扩张，同时节能环保和淘汰落后约束加强", "十一五规划原文；全球商品与信用为反事实", "较强关联"), ("煤炭钢铁与原材料", "2011-2015", "十二五（2011-2015）", "过剩与利润下行", "A股原材料和能源上市公司收入增长放缓，ROE明显下降", "listing_composition_revenue_decomposition.csv；a_share_industry_fundamentals_valuation.csv", "iFinD上市公司事实"), ("煤炭钢铁与原材料", "2016-2020", "十三五（2016-2020）", "去产能与利润修复", "钢铁煤炭去产能由产业扶持转为供给约束，改变产能和利润分配", "国发〔2016〕7号及十三五规划", "政策事实+较强关联"), ("煤炭钢铁与原材料", "2021-2025", "十四五（2021-2025）", "能源安全与再定价", "能源相对沪深300约+99.6个百分点，期末组合ROE改善；市场结果同时受商品价格影响", "csi_sector_plan_period_returns.csv；a_share_industry_fundamentals_valuation.csv", "市场与上市公司事实"), ("移动互联网与信息产业", "2001-2005", "十五（2001-2005）", "信息化基础", "信息基础设施、软件、互联网和电子政务进入国家规划", "十五规划第六章", "政策事实"), ("移动互联网与信息产业", "2006-2010", "十一五（2006-2010）", "核心技术与网络建设", "核心电子器件、高端芯片、基础软件和通信网络继续投入", "十一五规划第十章", "政策事实"), ("移动互联网与信息产业", "2010-2012", "十二五准备期", "专项政策落地", "战略性新兴产业决定把新一代信息技术纳入多工具支持", "国发〔2010〕32号", "政策事实"), ("移动互联网与信息产业", "2013-2015", "十二五（2011-2015）", "渗透与A股扩容", "移动互联网渗透、创业板和并购共同扩大信息技术市值与估值", "iFinD结构快照；中证行业收益；并购制度为反事实", "较强关联"), ("移动互联网与信息产业", "2016-2020", "十三五（2016-2020）", "数字化深化", "大数据、人工智能、云计算和产业数字化扩展政策范围", "十三五规划原文", "政策事实"), ("移动互联网与信息产业", "2021-2025", "十四五（2021-2025）", "数字基础设施与分化", "通信服务和信息技术合计市值权重继续提升，但ROE与估值分化", "iFinD结构与基本面快照", "上市公司结构事实"), ("新能源汽车与新能源体系", "2010-2015", "十二五（2011-2015）", "政策导入与资本开支", "新能源汽车、新能源和节能环保进入七大战略性新兴产业", "国发〔2010〕32号；十二五规划第十章", "政策事实"), ("新能源汽车与新能源体系", "2016-2020", "十三五（2016-2020）", "规模化与国产化", "新能源汽车、智能制造、能源互联网持续获得规划和产业政策支持", "十三五规划原文", "政策事实；细分产量待官方序列"), ("新能源汽车与新能源体系", "2020-2021", "十四五准备期", "目标与制度强化", "新能源汽车产业规划和碳达峰行动方案形成需求、基础设施和约束框架", "国办发〔2020〕39号；国发〔2021〕23号", "政策事实"), ("新能源汽车与新能源体系", "2021-2022", "十四五（2021-2025）", "扩产与市场结构提升", "工业、材料、可选消费和公用事业代理组合扩大，但不能代替细分产业", "case_evidence_cards.csv；细分产业数据不足", "A股代理事实"), ("新能源汽车与新能源体系", "2022-2025", "十四五（2021-2025）", "产能释放与盈利分化", "价格下降、利用率和利润分化成为检验供给纪律的关键变量", "报告案例归纳；需补细分价格、产能和利用率序列", "合理推断"), ("新能源汽车与新能源体系", "2026-2030", "十五五（2026-2030）", "产业治理与出清观察", "政策同时支持新型能源体系并强调产能预警、低效退出和无序竞争治理", "十五五规划原文", "政策事实；产业结果待观察"), ) return pd.DataFrame( rows, columns=[ "案例", "时期", "规划周期", "产业阶段", "关键变化", "证据来源", "证据等级", ], ) def main() -> None: company = read_company_snapshot() market = pd.read_csv(OUT / "a_share_industry_market_cap_structure.csv") profit = pd.read_csv(OUT / "a_share_industry_profit_structure.csv") metadata = pd.read_csv(OUT / "ifind_market_structure_metadata.csv") fundamentals = pd.read_csv(OUT / "a_share_industry_fundamentals_valuation.csv") returns = pd.read_csv(OUT / "csi_sector_plan_period_returns.csv") monthly = pd.read_csv(OUT / "csi_sector_monthly_closes_2009_2025.csv") sensitivity = build_mapping_sensitivity(market, metadata) sensitivity.to_csv( OUT / "industry_mapping_coverage_sensitivity.csv", index=False, encoding="utf-8-sig", ) planning = build_planning_direction_table(company, market, profit) planning.to_csv( OUT / "planning_direction_market_profit_comparison.csv", index=False, encoding="utf-8-sig", ) cycle = build_cycle_fundamentals(fundamentals, returns) cycle.to_csv( OUT / "plan_cycle_fundamental_valuation_comparison.csv", index=False, encoding="utf-8-sig", ) decomposition = build_listing_composition_decomposition(company) decomposition.to_csv( OUT / "listing_composition_revenue_decomposition.csv", index=False, encoding="utf-8-sig", ) decomposition_summary = build_listing_composition_summary(decomposition) decomposition_summary.to_csv( OUT / "listing_composition_revenue_bridge_summary.csv", index=False, encoding="utf-8-sig", ) availability = build_financial_data_availability(company) availability.to_csv( OUT / "financial_data_availability_metadata.csv", index=False, encoding="utf-8-sig", ) cases = build_case_cards(company, monthly) cases.to_csv( OUT / "case_evidence_cards.csv", index=False, encoding="utf-8-sig", ) policy_strength = build_policy_strength_coding() policy_strength.to_csv( OUT / "five_year_plan_policy_strength_coding.csv", index=False, encoding="utf-8-sig", ) policy_strength[ [ "规划周期", "政策方向", "原文章节", "逐字引文", "精确定位", "原始链接", ] ].to_csv( OUT / "policy_coding_blind_review_input.csv", index=False, encoding="utf-8-sig", ) policy_strength[ policy_strength["政策方向"].isin( [ "机械电子石化汽车建筑", "能源交通通信重点建设", "集成电路", "西部开发", "节能环保与能源约束", "研发投入", ] ) ][ [ "规划周期", "政策方向", "原文章节", "逐字引文", "精确定位", "原始链接", ] ].to_csv( OUT / "policy_coding_recheck_input.csv", index=False, encoding="utf-8-sig", ) timelines = build_deep_industry_timelines() timelines.to_csv( OUT / "deep_industry_timelines.csv", index=False, encoding="utf-8-sig", ) fundamentals[ (fundamentals["year"] == 2025) & (fundamentals["industry"] != "未映射") ].to_csv( OUT / "fifteenth_plan_2025_market_baseline.csv", index=False, encoding="utf-8-sig", ) assert len(sensitivity) == 7 * 11 assert planning.loc[ planning["规划方向"] == "电子/信息/数字", "市值权重变化百分点" ].notna().all() assert len(cycle) == 3 * 10 assert len(decomposition) == 6 * 10 assert decomposition["收入可比持续公司数"].le( decomposition["同一行业持续公司数"] ).all() assert decomposition["桥接残差"].abs().max() < 1 assert decomposition_summary["桥接残差"].abs().max() < 1 assert decomposition["全产业真实增长"].str.startswith("未测量").all() assert availability["实际公告日期"].isna().all() assert availability["市场快照时是否可得"].str.startswith("否").all() assert len(cases) == 6 assert cases["政策原始链接"].notna().all() assert cases[["2025组合ROE", "2025组合PE", "2025组合PB"]].notna().all().all() assert (cases["2025组合PE"] > 0).all() assert (cases["2025组合PB"] > 0).all() assert len(policy_strength) == 28 assert policy_strength["强度等级"].between(1, 4).all() assert policy_strength["原始链接"].notna().all() assert policy_strength["逐字引文"].notna().all() assert ~policy_strength["精确定位"].eq("原文定位失败").any() assert timelines["案例"].nunique() == 4 assert timelines.groupby("案例").size().ge(5).all() print("evidence-table assertions: OK") if __name__ == "__main__": main()