#!/usr/bin/env python3 # -*- coding: utf-8 -*- """五年规划行业先验 — 完整回测(/reports/policy-sectors solid 版)。 口径(消灭旧版幸存者偏差): · 十三五/十四五/十五五: 官方申万一级指数 sw_daily(801xxx.SI, 2014起, 点位时点) · 十一五/十二五: 中证800十行业指数 index_daily(000928-937.SH, 2006起, 粗粒度扩展样本) · favored = 统一映射字典(规划语言→行业)作用于各期规划原文产业清单 — 预注册, 避免逐期主观挑选 · 事件时间: 规划首年=Y1(自然年), 逐年 favored−非favored 超额; 抢跑段 = 建议公布(10月末)→纲要(次年3月末) · capex: cashflow_vip 购建固定资产(年报, 2015+), sw_member 当前快照聚合(披露) 数据缓存 data/factor_lab/policy_*.parquet(gitignored)。运行: cd /root/cb-allotment && python3 my-app/public/reports/policy-sectors/backtest.py """ import json import os import sys import time import numpy as np import pandas as pd ROOT = "/root/cb-allotment" sys.path.insert(0, f"{ROOT}/scripts") CACHE = f"{ROOT}/data/factor_lab" OUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "backtest_result.json") os.makedirs(CACHE, exist_ok=True) # ---------- 统一映射字典(规划语言 → 申万一级) ---------- LANG2SW = { "新一代信息技术": ["电子", "计算机", "通信"], "高端装备": ["机械设备", "国防军工"], "新材料": ["有色金属", "基础化工"], "生物": ["医药生物"], "新能源": ["电力设备"], "新能源汽车": ["汽车"], "节能环保": ["环保"], "数字创意/数字经济": ["传媒"], "航空航天/低空经济": ["国防军工", "机械设备"], "人工智能": ["计算机"], "集成电路/量子": ["电子"], "6G": ["通信"], "生物制造/脑机接口": ["医药生物"], "氢能/核聚变": ["电力设备"], "工业母机/具身智能": ["机械设备"], } # 各期规划原文产业清单(见报告正文引用与来源) PLAN_LANGS = { "13th": ["新一代信息技术", "高端装备", "新材料", "生物", "新能源汽车", "新能源", "节能环保", "数字创意/数字经济"], "14th": ["新一代信息技术", "高端装备", "新材料", "生物", "新能源汽车", "新能源", "节能环保", "数字创意/数字经济", "航空航天/低空经济"], "15th": ["新能源", "新材料", "航空航天/低空经济", "集成电路/量子", "人工智能", "6G", "生物制造/脑机接口", "氢能/核聚变", "工业母机/具身智能", "新一代信息技术"], } PLAN_YEARS = {"13th": (2016, 2020), "14th": (2021, 2025), "15th": (2026, 2030)} # 建议公布→纲要窗口(抢跑段) PLAN_RUNUP = {"13th": ("20151030", "20160331"), "14th": ("20201030", "20210331"), "15th": ("20251024", "20260331")} def favored_sw(plan): s = set() for lang in PLAN_LANGS[plan]: s.update(LANG2SW[lang]) return sorted(s) # 中证800十行业(十一五/十二五, 粗粒度) CSI = {"000928.SH": "能源", "000929.SH": "材料", "000930.SH": "工业", "000931.SH": "可选消费", "000932.SH": "主要消费", "000933.SH": "医药卫生", "000934.SH": "金融地产", "000935.SH": "信息技术", "000936.SH": "电信业务", "000937.SH": "公用事业"} # 十一五(装备制造/信息产业/生物/新材料→材料): 工业/信息/电信/医药/材料; 十二五(七大战新)同+材料 CSI_FAVORED = {"11th": ["工业", "信息技术", "电信业务", "医药卫生"], "12th": ["工业", "信息技术", "电信业务", "医药卫生", "材料"]} CSI_YEARS = {"11th": (2006, 2010), "12th": (2011, 2015)} def _pro(): from tushare_client import get_pro return get_pro() def _retry(fn, tag, min_rows=50, tries=3, **kw): for i in range(tries): try: df = fn(**kw) if df is not None and len(df) >= min_rows: return df print(f" {tag} 仅 {0 if df is None else len(df)} 行, retry {i+1}", flush=True) except Exception as e: print(f" {tag} ERR({str(e)[:50]}), retry {i+1}", flush=True) time.sleep(2 + 2 * i) return None def fetch_sw_daily() -> pd.DataFrame: cache = f"{CACHE}/policy_sw_daily.parquet" if os.path.exists(cache): df = pd.read_parquet(cache) if str(df.index.max())[:7] >= "2026-06": return df pro = _pro() cls = _retry(pro.index_classify, "index_classify", min_rows=25, level="L1", src="SW2021") codes = dict(zip(cls.index_code, cls.industry_name)) frames = {} for code, name in codes.items(): parts = [] for s, e in [("20140101", "20201231"), ("20210101", "20261231")]: d = _retry(pro.sw_daily, f"sw_daily {name}", min_rows=10, ts_code=code, start_date=s, end_date=e) if d is not None: parts.append(d[["trade_date", "close"]]) time.sleep(0.35) if parts: ser = pd.concat(parts).drop_duplicates("trade_date").set_index("trade_date").close.sort_index() frames[name] = ser print(f" sw {name}: {len(frames.get(name, []))} 天", flush=True) df = pd.DataFrame(frames) df.index = pd.to_datetime(df.index, format="%Y%m%d") df.to_parquet(cache) return df def fetch_csi() -> pd.DataFrame: cache = f"{CACHE}/policy_csi800.parquet" if os.path.exists(cache): df = pd.read_parquet(cache); return df pro = _pro() frames = {} for code, name in CSI.items(): parts = [] for s, e in [("20050101", "20121231"), ("20130101", "20201231"), ("20210101", "20261231")]: d = _retry(pro.index_daily, f"csi {name}", min_rows=10, ts_code=code, start_date=s, end_date=e) if d is not None: parts.append(d[["trade_date", "close"]]) time.sleep(0.35) if parts: frames[name] = pd.concat(parts).drop_duplicates("trade_date").set_index("trade_date").close.sort_index() print(f" csi {name}: {len(frames.get(name, []))} 天", flush=True) df = pd.DataFrame(frames) df.index = pd.to_datetime(df.index, format="%Y%m%d") df.to_parquet(cache) return df def fetch_capex() -> pd.DataFrame: cache = f"{CACHE}/policy_capex.parquet" if os.path.exists(cache): return pd.read_parquet(cache) pro = _pro() frames = [] for year in range(2014, 2026): p = f"{year}1231" d = _retry(pro.cashflow_vip, f"capex {year}", min_rows=1500, period=p, limit=8000, fields="ts_code,end_date,c_pay_acq_const_fiolta") time.sleep(0.5) if d is not None: frames.append(d.drop_duplicates("ts_code")) df = pd.concat(frames, ignore_index=True) df.to_parquet(cache) return df def yearly_ret(prices: pd.DataFrame, year: int) -> pd.Series: """自然年收益(年末/上年末-1)。""" ye = prices[prices.index <= f"{year}-12-31"] y0 = prices[prices.index <= f"{year-1}-12-31"] if not len(ye) or not len(y0): return pd.Series(dtype=float) return ye.iloc[-1] / y0.iloc[-1] - 1 def window_ret(prices: pd.DataFrame, d0: str, d1: str) -> pd.Series: a = prices[prices.index <= pd.Timestamp(d0)] b = prices[prices.index <= pd.Timestamp(d1)] if not len(a) or not len(b): return pd.Series(dtype=float) return b.iloc[-1] / a.iloc[-1] - 1 def event_study(prices, favored, y0, y1, label): all_cols = list(prices.columns) fav = [c for c in favored if c in all_cols] non = [c for c in all_cols if c not in fav] years = [] cum_f, cum_n = 1.0, 1.0 for i, y in enumerate(range(y0, min(y1, 2026) + 1), 1): r = yearly_ret(prices, y) if r.empty or r[fav].isna().all(): continue rf, rn = float(r[fav].mean()), float(r[non].mean()) if y == 2026: # YTD r_ytd = prices.iloc[-1] / prices[prices.index <= "2025-12-31"].iloc[-1] - 1 rf, rn = float(r_ytd[fav].mean()), float(r_ytd[non].mean()) cum_f *= (1 + rf); cum_n *= (1 + rn) years.append({"y": i, "year": y, "fav": round(rf * 100, 1), "non": round(rn * 100, 1), "ex": round((rf - rn) * 100, 1)}) # 命中率: favored 行业年收益 > 全行业中位的比例(全期) hits, tot = 0, 0 per_ind = {} for y in range(y0, min(y1, 2026) + 1): r = yearly_ret(prices, y) if r.empty: continue med = r.median() for c in fav: if pd.notna(r[c]): tot += 1; hits += int(r[c] > med) r_full = window_ret(prices, f"{y0-1}-12-31", f"{min(y1,2026)}-12-31") for c in all_cols: if pd.notna(r_full.get(c)): per_ind[c] = round(float(r_full[c]) * 100, 1) return {"label": label, "favored": fav, "years": years, "cum_fav": round((cum_f - 1) * 100, 1), "cum_non": round((cum_n - 1) * 100, 1), "cum_ex": round((cum_f - cum_n) * 100, 1), "hit_rate": round(hits / tot * 100, 0) if tot else None, "per_industry_full": dict(sorted(per_ind.items(), key=lambda x: -x[1]))} def main(): print("== 拉取/加载数据 ==", flush=True) sw = fetch_sw_daily() csi = fetch_csi() capex_raw = fetch_capex() print(f"sw {sw.shape} {sw.index.min().date()}~{sw.index.max().date()} | csi {csi.shape} {csi.index.min().date()}~", flush=True) plans = [] # 申万口径: 13th/14th/15th for p in ["13th", "14th", "15th"]: y0, y1 = PLAN_YEARS[p] plans.append(event_study(sw, favored_sw(p), y0, y1, p)) # 中证口径: 11th/12th for p in ["11th", "12th"]: y0, y1 = CSI_YEARS[p] plans.append(event_study(csi, CSI_FAVORED[p], y0, y1, p)) # 跨期事件时间平均(Y1..Y5): 用 4 期完整规划(11-14th) evt = {i: [] for i in range(1, 6)} for pl in plans: if pl["label"] == "15th": continue for row in pl["years"]: evt[row["y"]].append(row["ex"]) event_avg = [{"y": i, "avg_ex": round(float(np.mean(v)), 1), "n": len(v), "detail": v} for i, v in evt.items() if v] # 抢跑段(建议→纲要): 13th/14th/15th 用申万 runup = [] for p, (d0, d1) in PLAN_RUNUP.items(): fav = [c for c in favored_sw(p) if c in sw.columns] non = [c for c in sw.columns if c not in fav] r = window_ret(sw, pd.Timestamp(d0).strftime("%Y-%m-%d"), pd.Timestamp(d1).strftime("%Y-%m-%d")) if r.empty or r[fav].isna().all(): continue runup.append({"plan": p, "window": f"{d0}→{d1}", "fav": round(float(r[fav].mean()) * 100, 1), "non": round(float(r[non].mean()) * 100, 1), "ex": round(float(r[fav].mean() - r[non].mean()) * 100, 1)}) # 成长 beta 2×2(申万, 13th/14th): 成长集 = 电子/计算机/通信/传媒/电力设备/国防军工/医药生物 growth = {"电子", "计算机", "通信", "传媒", "电力设备", "国防军工", "医药生物"} beta2x2 = [] for p in ["13th", "14th"]: y0, y1 = PLAN_YEARS[p] fav = set(favored_sw(p)) & set(sw.columns) r = window_ret(sw, f"{y0-1}-12-31", f"{min(y1,2026)}-12-31") cells = {} for name, cols in [("favored∩成长", fav & growth), ("favored∩非成长", fav - growth), ("非favored∩成长", (set(sw.columns) - fav) & growth), ("非favored∩非成长", (set(sw.columns) - fav) - growth)]: cols = [c for c in cols if pd.notna(r.get(c))] cells[name] = {"n": len(cols), "ret": round(float(r[cols].mean()) * 100, 1) if cols else None, "inds": sorted(cols)} beta2x2.append({"plan": p, "cells": cells}) # capex 传导: sw_member 快照聚合(披露), favored vs 非favored 年度 capex 增速 import sqlite3 conn = sqlite3.connect(f"{ROOT}/data/fundamental_cache.db") swm = pd.read_sql("SELECT l1_name,ts_code FROM sw_member", conn) conn.close() s2i = swm.drop_duplicates("ts_code").set_index("ts_code").l1_name cx = capex_raw.copy() cx["ind"] = cx.ts_code.map(s2i) cx["year"] = cx.end_date.str[:4].astype(int) ind_capex = cx.groupby(["year", "ind"]).c_pay_acq_const_fiolta.sum().unstack() capex_g = ind_capex.pct_change() capex_out = [] for p in ["13th", "14th"]: y0, y1 = PLAN_YEARS[p] fav = [c for c in favored_sw(p) if c in capex_g.columns] non = [c for c in capex_g.columns if c not in fav and pd.notna(c)] seg = capex_g.loc[y0:min(y1, 2025)] capex_out.append({"plan": p, "fav_capex_cagr": round(float(((ind_capex.loc[min(y1,2025), fav].sum() / ind_capex.loc[y0-1, fav].sum()) ** (1/(min(y1,2025)-y0+1)) - 1) * 100), 1), "non_capex_cagr": round(float(((ind_capex.loc[min(y1,2025), non].sum() / ind_capex.loc[y0-1, non].sum()) ** (1/(min(y1,2025)-y0+1)) - 1) * 100), 1), "fav_yearly_g": [round(float(seg.loc[y, fav].mean()) * 100, 1) for y in seg.index], "non_yearly_g": [round(float(seg.loc[y, non].mean()) * 100, 1) for y in seg.index], "years": list(map(int, seg.index))}) # capex 加速→次年收益? 行业级: capex增速(t) vs 行业收益(t+1) 相关 swy = {} for y in range(2016, 2026): swy[y] = yearly_ret(sw, y) swy = pd.DataFrame(swy).T # year x industry common = [c for c in swy.columns if c in capex_g.columns] pairs = [] for y in range(2016, 2025): g = capex_g.loc[y, common]; r_next = swy.loc[y + 1, common] m = pd.concat([g, r_next], axis=1).dropna() if len(m) > 10: pairs.append(float(m.corr(method="spearman").iloc[0, 1])) capex_lead = {"spearman_capexg_t_vs_ret_t1_mean": round(float(np.mean(pairs)), 2), "n_years": len(pairs), "yearly": [round(x, 2) for x in pairs]} result = {"as_of": str(sw.index.max().date()), "mapping": {k: v for k, v in LANG2SW.items()}, "plan_langs": PLAN_LANGS, "csi_favored": CSI_FAVORED, "plans": plans, "event_avg": event_avg, "runup": runup, "growth_2x2": beta2x2, "capex": capex_out, "capex_lead": capex_lead} json.dump(result, open(OUT, "w"), ensure_ascii=False, indent=1) print(json.dumps({"event_avg": event_avg, "runup": runup}, ensure_ascii=False, indent=1)) for pl in plans: print(pl["label"], "cum_ex", pl["cum_ex"], "hit", pl["hit_rate"], "years", [(r["year"], r["ex"]) for r in pl["years"]]) print("capex:", capex_out and [(c["plan"], c["fav_capex_cagr"], c["non_capex_cagr"]) for c in capex_out], "| lead:", capex_lead["spearman_capexg_t_vs_ret_t1_mean"]) print("写入", OUT) if __name__ == "__main__": main()