#!/usr/bin/env python3 """Build reproducible market tables and charts for the five-year-plan report.""" from __future__ import annotations import json from pathlib import Path import pandas as pd ROOT = Path(__file__).resolve().parents[1] RAW_DIR = ROOT / "data" / "research" / "raw" OUT_DIR = ROOT / "data" / "research" / "derived" FIG_DIR = ROOT / "docs" / "research" / "figures" INDEX_NAMES = { "000300": "沪深300", "000928": "能源", "000929": "原材料", "000930": "工业", "000931": "可选消费", "000932": "主要消费", "000933": "医药卫生", "000934": "金融地产", "000935": "信息技术", "000936": "电信业务", "000937": "公用事业", } PLAN_PERIODS = { "十二五(2011-2015)": ("2010-12-31", "2015-12-31"), "十三五(2016-2020)": ("2015-12-31", "2020-12-31"), "十四五(2021-2025)": ("2020-12-31", "2025-12-31"), } APPROVAL_DATES = { "十二五": "2011-03-14", "十三五": "2016-03-16", "十四五": "2021-03-11", } POLICY_EVENTS = ( { "event": "十二五建议", "type": "规划建议", "date": "2010-10-27", "source": "https://www.gov.cn/jrzg/2010-10/27/content_1731694.htm", "exposed": (), }, { "event": "十二五纲要", "type": "规划纲要", "date": "2011-03-14", "source": "https://www.ndrc.gov.cn/fggz/fzzlgh/gjfzgh/201109/P020191029595702423333.pdf", "exposed": (), }, { "event": "十三五建议", "type": "规划建议", "date": "2015-11-03", "source": "https://www.gov.cn/xinwen/2015-11/03/content_5004093.htm", "exposed": (), }, { "event": "十三五纲要", "type": "规划纲要", "date": "2016-03-16", "source": "https://www.ndrc.gov.cn/fggz/fzzlgh/gjfzgh/201603/P020191104614882474091.pdf", "exposed": (), }, { "event": "十四五建议", "type": "规划建议", "date": "2020-11-03", "source": "https://www.gov.cn/zhengce/2020-11/03/content_5556991.htm", "exposed": (), }, { "event": "十四五纲要", "type": "规划纲要", "date": "2021-03-11", "source": "https://www.ndrc.gov.cn/fggz/fzzlgh/gjfzgh/202103/P020210323405614585384.pdf", "exposed": (), }, { "event": "战略性新兴产业决定", "type": "专项政策", "date": "2010-10-18", "source": "https://www.gov.cn/zwgk/2010-10/18/content_1724848.htm", "exposed": ("工业", "医药卫生", "信息技术", "电信业务", "公用事业"), }, { "event": "中国制造2025", "type": "专项政策", "date": "2015-05-08", "source": "https://www.gov.cn/zhengce/content/2015-05/19/content_9784.htm", "exposed": ("工业", "原材料", "信息技术"), }, { "event": "新能源汽车产业规划", "type": "专项政策", "date": "2020-10-20", "source": "https://www.gov.cn/zhengce/content/2020-11/02/content_5556716.htm", "exposed": ("工业", "原材料", "可选消费", "公用事业"), }, { "event": "碳达峰行动方案", "type": "专项政策", "date": "2021-10-24", "source": "https://www.gov.cn/zhengce/content/2021-10/26/content_5644984.htm", "exposed": ("能源", "原材料", "工业", "公用事业"), }, { "event": "十四五数字经济规划", "type": "专项政策", "date": "2022-01-12", "source": "https://www.gov.cn/zhengce/content/2022-01/12/content_5667817.htm", "exposed": ("工业", "信息技术", "电信业务"), }, ) def load_monthly(code: str) -> pd.Series: payload = json.loads((RAW_DIR / f"tencent_{code}_monthly.json").read_text()) rows = payload["data"][f"sh{code}"]["month"] frame = pd.DataFrame(rows, columns=["date", "open", "close", "high", "low", "volume"]) frame["date"] = pd.to_datetime(frame["date"]) frame["close"] = pd.to_numeric(frame["close"]) frame = frame[frame["date"] <= pd.Timestamp("2025-12-31")] return frame.set_index("date")["close"].sort_index().rename(INDEX_NAMES[code]) def previous_close(series: pd.Series, target: str) -> float: eligible = series.loc[: pd.Timestamp(target)] if eligible.empty: raise ValueError(f"No observation on or before {target}") return float(eligible.iloc[-1]) def forward_close(series: pd.Series, target: pd.Timestamp) -> float: eligible = series.loc[target:] if eligible.empty: raise ValueError(f"No observation on or after {target.date()}") return float(eligible.iloc[0]) def total_return(series: pd.Series, start: str, end: str) -> float: return previous_close(series, end) / previous_close(series, start) - 1 def window_prices(series: pd.Series, start: str, end: pd.Timestamp) -> pd.Series: start_timestamp = pd.Timestamp(start) start_value = previous_close(series, start) observations = series.loc[(series.index > start_timestamp) & (series.index <= end)] start_series = pd.Series([start_value], index=[start_timestamp], name=series.name) return pd.concat([start_series, observations]).sort_index() def max_drawdown(wealth: pd.Series) -> float: drawdown = wealth / wealth.cummax() - 1 return float(drawdown.min()) def relative_window_return( sector: pd.Series, benchmark: pd.Series, start: pd.Timestamp, end: pd.Timestamp, ) -> tuple[float, float, float, float]: sector_window = window_prices(sector, str(start.date()), end) benchmark_window = window_prices(benchmark, str(start.date()), end) common_index = sector_window.index.union(benchmark_window.index).sort_values() sector_wealth = sector_window.reindex(common_index).ffill() / sector_window.iloc[0] benchmark_wealth = benchmark_window.reindex(common_index).ffill() / benchmark_window.iloc[0] sector_return = float(sector_wealth.iloc[-1] - 1) benchmark_return = float(benchmark_wealth.iloc[-1] - 1) return ( sector_return, benchmark_return, max_drawdown(sector_wealth), max_drawdown(sector_wealth / benchmark_wealth), ) def heat_color(value: float, limit: float = 60.0) -> str: normalized = max(-1.0, min(1.0, value / limit)) if normalized >= 0: red = int(248 - 112 * normalized) green = int(250 - 45 * normalized) blue = int(248 - 115 * normalized) else: strength = -normalized red = int(248 - 28 * strength) green = int(250 - 132 * strength) blue = int(248 - 118 * strength) return f"rgb({red},{green},{blue})" def heatmap_html(frame: pd.DataFrame, title: str, section_id: str, decimals: int = 0) -> str: values = frame.astype(float) * 100 header = "".join(f"{column}" for column in values.columns) rows = [] for index, row in values.iterrows(): cells = "".join( f'{value:.{decimals}f}' for value in row ) rows.append(f"{index}{cells}") return ( f'
' f"

{title}

" '
弱相对沪深300收益，百分点强
' f"{header}{''.join(rows)}
行业
" '

来源：腾讯证券公开行情接口；计算：本报告。价格指数月末收盘口径，不含股息。

' "
" ) def main() -> None: OUT_DIR.mkdir(parents=True, exist_ok=True) FIG_DIR.mkdir(parents=True, exist_ok=True) series = {code: load_monthly(code) for code in INDEX_NAMES} prices = pd.concat(series.values(), axis=1) prices.to_csv(OUT_DIR / "csi_sector_monthly_closes_2009_2025.csv", encoding="utf-8-sig") year_end = prices.resample("YE").last() annual_returns = year_end.pct_change().loc["2010":"2025"] annual_returns.to_csv(OUT_DIR / "csi_sector_annual_returns_2010_2025.csv", encoding="utf-8-sig") sector_cols = [name for name in INDEX_NAMES.values() if name != "沪深300"] annual_relative = annual_returns[sector_cols].sub(annual_returns["沪深300"], axis=0) annual_relative.to_csv( OUT_DIR / "csi_sector_annual_excess_vs_csi300_2010_2025.csv", encoding="utf-8-sig" ) period_rows = [] for period, (start, end) in PLAN_PERIODS.items(): benchmark = total_return(series["000300"], start, end) for code, name in INDEX_NAMES.items(): if code == "000300": continue absolute = total_return(series[code], start, end) period_rows.append( { "规划周期": period, "行业": name, "区间收益率": absolute, "沪深300收益率": benchmark, "相对沪深300": absolute - benchmark, } ) period_returns = pd.DataFrame(period_rows) period_returns.to_csv(OUT_DIR / "csi_sector_plan_period_returns.csv", index=False, encoding="utf-8-sig") event_rows = [] for plan, date_text in APPROVAL_DATES.items(): event_date = pd.Timestamp(date_text) benchmark_start = previous_close(series["000300"], date_text) for months in (12, 24, 36): target = event_date + pd.DateOffset(months=months) benchmark_return = forward_close(series["000300"], target) / benchmark_start - 1 for code, name in INDEX_NAMES.items(): if code == "000300": continue sector_window = window_prices(series[code], date_text, target) benchmark_window = window_prices(series["000300"], date_text, target) common_index = sector_window.index.union(benchmark_window.index).sort_values() sector_wealth = sector_window.reindex(common_index).ffill() / sector_window.iloc[0] benchmark_wealth = benchmark_window.reindex(common_index).ffill() / benchmark_window.iloc[0] relative_wealth = sector_wealth / benchmark_wealth sector_return = forward_close(series[code], target) / previous_close(series[code], date_text) - 1 event_rows.append( { "规划": plan, "通过日期": date_text, "观察窗口(月)": months, "行业": name, "行业收益率": sector_return, "沪深300收益率": benchmark_return, "相对沪深300": sector_return - benchmark_return, "行业最大回撤": max_drawdown(sector_wealth), "相对基准最大回撤": max_drawdown(relative_wealth), } ) event_study = pd.DataFrame(event_rows) event_study.to_csv(OUT_DIR / "plan_approval_event_study.csv", index=False, encoding="utf-8-sig") event_summary_rows = [] summary_groups = [("全部周期", months, group) for months, group in event_study.groupby("观察窗口(月)")] summary_groups += [ (plan, months, group) for (plan, months), group in event_study.groupby(["规划", "观察窗口(月)"]) ] for plan, months, group in summary_groups: excess = group["相对沪深300"] event_summary_rows.append( { "规划": plan, "观察窗口(月)": months, "样本数": len(group), "相对收益为正占比": (excess > 0).mean(), "相对收益均值": excess.mean(), "相对收益中位数": excess.median(), "相对收益25分位": excess.quantile(0.25), "相对收益75分位": excess.quantile(0.75), "相对收益最小值": excess.min(), "相对收益最大值": excess.max(), "行业最大回撤中位数": group["行业最大回撤"].median(), "行业最差最大回撤": group["行业最大回撤"].min(), "相对基准最大回撤中位数": group["相对基准最大回撤"].median(), "相对基准最差最大回撤": group["相对基准最大回撤"].min(), } ) event_summary = pd.DataFrame(event_summary_rows) event_summary.to_csv( OUT_DIR / "plan_approval_event_summary.csv", index=False, encoding="utf-8-sig" ) policy_event_rows = [] for event in POLICY_EVENTS: event_date = pd.Timestamp(event["date"]) windows = ((-12, event_date - pd.DateOffset(months=12), event_date),) windows += tuple( (months, event_date, event_date + pd.DateOffset(months=months)) for months in (12, 24, 36) ) for months, start, end in windows: for code, name in INDEX_NAMES.items(): if code == "000300": continue sector_return, benchmark_return, sector_drawdown, relative_drawdown = ( relative_window_return(series[code], series["000300"], start, end) ) exposed = name in event["exposed"] if event["exposed"] else None policy_event_rows.append( { "事件": event["event"], "事件类型": event["type"], "事件日期": event["date"], "观察窗口(月)": months, "窗口含义": "发布前12个月" if months < 0 else f"发布后{months}个月", "行业": name, "政策暴露标记": exposed, "行业收益率": sector_return, "沪深300收益率": benchmark_return, "相对沪深300": sector_return - benchmark_return, "行业最大回撤": sector_drawdown, "相对基准最大回撤": relative_drawdown, "原始链接": event["source"], } ) policy_event_study = pd.DataFrame(policy_event_rows) policy_event_study.to_csv( OUT_DIR / "policy_multi_event_study.csv", index=False, encoding="utf-8-sig" ) policy_summary_rows = [] for keys, group in policy_event_study.groupby(["事件类型", "观察窗口(月)"], dropna=False): event_type, months = keys excess = group["相对沪深300"] policy_summary_rows.append( { "事件类型": event_type, "政策暴露": "全部行业", "观察窗口(月)": months, "事件数": group["事件"].nunique(), "样本数": len(group), "相对收益为正占比": (excess > 0).mean(), "相对收益中位数": excess.median(), "相对收益25分位": excess.quantile(0.25), "相对收益75分位": excess.quantile(0.75), "行业最大回撤中位数": group["行业最大回撤"].median(), } ) special = policy_event_study[policy_event_study["事件类型"] == "专项政策"].copy() for keys, group in special.groupby(["政策暴露标记", "观察窗口(月)"], dropna=False): exposed, months = keys excess = group["相对沪深300"] policy_summary_rows.append( { "事件类型": "专项政策", "政策暴露": "政策直接相关行业" if exposed else "其他行业", "观察窗口(月)": months, "事件数": group["事件"].nunique(), "样本数": len(group), "相对收益为正占比": (excess > 0).mean(), "相对收益中位数": excess.median(), "相对收益25分位": excess.quantile(0.25), "相对收益75分位": excess.quantile(0.75), "行业最大回撤中位数": group["行业最大回撤"].median(), } ) policy_event_summary = pd.DataFrame(policy_summary_rows) policy_event_summary.to_csv( OUT_DIR / "policy_multi_event_summary.csv", index=False, encoding="utf-8-sig" ) ranking_rows = [] for year, row in annual_relative.iterrows(): ordered = row.sort_values(ascending=False) ranking_rows.append( { "年份": year.year, "相对收益第一": ordered.index[0], "第一名相对沪深300": ordered.iloc[0], "相对收益第二": ordered.index[1], "第二名相对沪深300": ordered.iloc[1], "相对收益倒数第一": ordered.index[-1], "倒数第一相对沪深300": ordered.iloc[-1], } ) pd.DataFrame(ranking_rows).to_csv( OUT_DIR / "csi_sector_annual_leaders_laggards.csv", index=False, encoding="utf-8-sig" ) period_pivot = period_returns.pivot(index="行业", columns="规划周期", values="相对沪深300") period_pivot = period_pivot.reindex(columns=list(PLAN_PERIODS)) event_pivot = event_study.pivot_table( index="行业", columns=["规划", "观察窗口(月)"], values="相对沪深300" ) event_pivot = event_pivot.reindex( columns=pd.MultiIndex.from_product( [["十二五", "十三五", "十四五"], [12, 24, 36]], names=["规划", "观察窗口(月)"] ) ) event_pivot.columns = [f"{plan}-{months}个月" for plan, months in event_pivot.columns] multi_event_pivot = policy_event_study.pivot_table( index="行业", columns=["事件", "观察窗口(月)"], values="相对沪深300", ) multi_event_pivot.columns = [ f"{event}-{'前12个月' if months < 0 else f'后{months}个月'}" for event, months in multi_event_pivot.columns ] annual_heatmap = annual_relative.T.copy() annual_heatmap.columns = [str(column.year) for column in annual_heatmap.columns] html = """""" html += heatmap_html( annual_heatmap, "中证十行业年度相对沪深300收益热力图（2010-2025）", "annual-heatmap", ) html += heatmap_html( period_pivot, "五年规划周期内中证行业相对收益", "period-heatmap", decimals=1, ) html += heatmap_html( event_pivot, "规划纲要通过后12/24/36个月行业相对收益", "event-heatmap", ) html += heatmap_html( multi_event_pivot, "建议、纲要与专项政策发布前后行业相对收益", "multi-event-heatmap", ) html += "" (FIG_DIR / "market_heatmaps.html").write_text(html, encoding="utf-8") metadata = pd.DataFrame( [ { "字段": "数据来源", "说明": "腾讯证券公开行情接口；指数名称与代码应再用中证指数公司资料复核", }, {"字段": "数据区间", "说明": "2009-07至2025-12；年度比较为2010-2025"}, {"字段": "指标口径", "说明": "月末收盘点位价格收益，不含股息；行业指数代码000928-000937"}, { "字段": "计算方法", "说明": "年度/周期收盘点位涨跌幅；相对收益=行业收益-沪深300收益；最大回撤按月末净值计算", }, { "字段": "事件研究样本", "说明": "纲要研究为三轮规划×十行业；多事件研究含3份建议、3份纲要、5项专项政策及前12/后12/24/36个月窗口；均非独立政策实验", }, { "字段": "可比性限制", "说明": "不能覆盖八五至十一五前期；指数样本与行业分类会调整；不等同申万行业全收益", }, ] ) metadata.to_csv(OUT_DIR / "market_data_metadata.csv", index=False, encoding="utf-8-sig") assert len(event_study) == len(APPROVAL_DATES) * 3 * (len(INDEX_NAMES) - 1) assert event_study["相对沪深300"].notna().all() assert event_study["行业最大回撤"].between(-1, 0).all() assert event_study["相对基准最大回撤"].between(-1, 0).all() assert set(event_summary["样本数"]) == {10, 30} assert len(policy_event_study) == len(POLICY_EVENTS) * 4 * (len(INDEX_NAMES) - 1) assert set(policy_event_study["观察窗口(月)"]) == {-12, 12, 24, 36} assert policy_event_study["相对沪深300"].notna().all() print("numeric assertions: OK") if __name__ == "__main__": main()