#!/usr/bin/env python3 """Reproduce the bank-core-holdings case study used by the public blog. The simulation deliberately uses only local, pre-existing price and dividend evidence. It is a historical illustration rather than a trading strategy or investment recommendation. Cash dividends stay in cash; they are not used to buy additional shares. All monetary results are pre-personal-income-tax. """ from __future__ import annotations import csv import html import json import math import sqlite3 from collections import defaultdict from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Iterable import pandas as pd ROOT = Path(__file__).resolve().parents[2] PRICE_DB = ROOT / "data/sqlite/stock_daily_v2.sqlite3" SEED_DIVIDENDS = ROOT / "data/history_2015_all_v1/dividends_seed_v1/raw" EXTENDED_DIVIDENDS = ROOT / "cache/annual_stability_dividends_v1" OUT = ROOT / "research/experiments/a_share_bank_core_holdings_v1" WEB_IMAGES = ROOT / "web/static/images" START = "2015-01-05" END = "2026-09-04" PER_BANK_INITIAL = 50_000.0 LOT_BUDGET = 25_000.0 COMMISSION_RATE = 0.0003 MIN_COMMISSION = 5.0 LOOKBACK = 250 FIRST_DRAWDOWN = 0.20 SECOND_DRAWDOWN = 0.35 BANKS = ( ("sh601398", "工商银行", "国有大行"), ("sh601939", "建设银行", "国有大行"), ("sh601288", "农业银行", "国有大行"), ("sh601988", "中国银行", "国有大行"), ("sh601328", "交通银行", "国有大行"), ("sh600036", "招商银行", "股份行"), ("sh601166", "兴业银行", "股份行"), ("sz000001", "平安银行", "股份行"), ("sz002142", "宁波银行", "城商行"), ("sh601009", "南京银行", "城商行"), ) @dataclass class Position: code: str name: str category: str shares: int = 0 spent: float = 0.0 cash_dividends: float = 0.0 additions_used: int = 0 @dataclass class Simulation: label: str initial_per_bank: float scheduled_additions: bool positions: dict[str, Position] = field(default_factory=dict) cash: float = 0.0 external_contributions: float = 0.0 total_commission: float = 0.0 orders: list[dict] = field(default_factory=list) dividend_ledger: list[dict] = field(default_factory=list) nav_rows: list[dict] = field(default_factory=list) def to_float(value: object) -> float: if value in (None, "", "None"): return 0.0 return float(value) def load_prices() -> pd.DataFrame: placeholders = ",".join("?" for _ in BANKS) query = f""" SELECT code, date, open, high, low, close, preclose FROM daily_bars WHERE code IN ({placeholders}) AND date BETWEEN ? AND ? ORDER BY date, code """ with sqlite3.connect(PRICE_DB) as connection: frame = pd.read_sql_query(query, connection, params=[*(code for code, *_ in BANKS), START, END]) for column in ("open", "high", "low", "close", "preclose"): frame[column] = pd.to_numeric(frame[column]) expected_days = frame.groupby("code")["date"].nunique() if expected_days.nunique() != 1: raise ValueError(f"Uneven price coverage: {expected_days.to_dict()}") return frame def local_dividend_files(code: str) -> Iterable[Path]: """Yield cached source files covering 2015 onward, without networking.""" yield from sorted(SEED_DIVIDENDS.glob(f"{code}_*.csv")) yield from sorted(EXTENDED_DIVIDENDS.glob(f"{code}_*.csv")) def load_dividends() -> tuple[dict[str, dict[str, dict]], list[dict]]: events: dict[str, dict[str, dict]] = defaultdict(dict) source_rows: list[dict] = [] for code, name, _ in BANKS: for path in local_dividend_files(code): with path.open(encoding="utf-8-sig", newline="") as handle: for row in csv.DictReader(handle): event_date = row.get("dividOperateDate") or "" if not START <= event_date <= END: continue cash = to_float(row.get("dividCashPsBeforeTax")) stock = to_float(row.get("dividStocksPs")) reserve = to_float(row.get("dividReserveToStockPs")) if not event_date or (cash == 0 and stock == 0 and reserve == 0): continue payload = { "code": code, "name": name, "date": event_date, "cash_per_share": cash, "stock_per_share": stock, "reserve_per_share": reserve, "source_file": str(path.relative_to(ROOT)), } # The same distribution can appear in both a preliminary and a # final record. Its economic fields and ex-date identify it. key = f"{cash:.12f}|{stock:.12f}|{reserve:.12f}" if key not in events[code].get(event_date, {}): events[code].setdefault(event_date, {})[key] = payload source_rows.append(payload) flattened: dict[str, dict[str, dict]] = defaultdict(dict) duplicate_audit = [] for code, by_date in events.items(): for event_date, unique in by_date.items(): if len(unique) != 1: raise ValueError(f"Multiple different dividend events on {code} {event_date}: {unique}") flattened[code][event_date] = next(iter(unique.values())) source_count = sum(1 for row in source_rows if row["code"] == code and row["date"] == event_date) if source_count > 1: duplicate_audit.append({"code": code, "date": event_date, "raw_rows": source_count, "deduplicated_to": 1}) return flattened, duplicate_audit def purchase(position: Position, simulation: Simulation, day: str, price: float, budget: float, reason: str) -> None: lots = int(budget // (price * 100)) while lots: shares = lots * 100 gross = shares * price commission = max(MIN_COMMISSION, gross * COMMISSION_RATE) if gross + commission <= budget + 1e-9: break lots -= 1 if not lots: raise ValueError(f"Budget {budget} cannot buy one lot of {position.code} at {price}") shares = lots * 100 gross = shares * price commission = max(MIN_COMMISSION, gross * COMMISSION_RATE) total = gross + commission if simulation.cash + 1e-9 < total: raise ValueError("Insufficient portfolio cash") simulation.cash -= total position.shares += shares position.spent += total simulation.total_commission += commission simulation.orders.append({ "date": day, "code": position.code, "name": position.name, "reason": reason, "open_price": round(price, 4), "shares": shares, "gross_buy": round(gross, 2), "commission": round(commission, 2), "cash_after": round(simulation.cash, 2), }) def apply_dividend(position: Position, simulation: Simulation, event: dict) -> None: old_shares = position.shares cash = old_shares * event["cash_per_share"] bonus_shares = round(old_shares * (event["stock_per_share"] + event["reserve_per_share"])) position.shares += bonus_shares position.cash_dividends += cash simulation.cash += cash simulation.dividend_ledger.append({ "date": event["date"], "code": position.code, "name": position.name, "shares_before": old_shares, "cash_per_share_pre_tax": event["cash_per_share"], "cash_dividend_pre_tax": round(cash, 2), "bonus_shares": bonus_shares, "source_file": event["source_file"], }) def total_return_indices(prices: pd.DataFrame, dividends: dict[str, dict[str, dict]]) -> pd.DataFrame: """Construct a pre-tax, ex-date-close dividend-reinvestment index per stock.""" rows = [] for code, _, _ in BANKS: frame = prices[prices.code == code].sort_values("date").reset_index(drop=True) index = 1.0 for i, row in frame.iterrows(): if i: event = dividends.get(code, {}).get(row.date) cash = event["cash_per_share"] if event else 0.0 stock = (event["stock_per_share"] + event["reserve_per_share"]) if event else 0.0 previous_close = float(frame.loc[i - 1, "close"]) if previous_close <= 0: raise ValueError(f"Invalid close for {code} {row.date}") index *= ((1 + stock) * float(row.close) + cash) / previous_close rows.append({"date": row.date, "code": code, "total_return_index": index}) result = pd.DataFrame(rows) result["rolling_high"] = result.groupby("code")["total_return_index"].transform( lambda series: series.rolling(LOOKBACK, min_periods=LOOKBACK).max() ) result["drawdown"] = result["total_return_index"] / result["rolling_high"] - 1 return result def simulate(label: str, initial_per_bank: float, scheduled_additions: bool, prices: pd.DataFrame, dividends: dict[str, dict[str, dict]], indices: pd.DataFrame) -> Simulation: simulation = Simulation(label=label, initial_per_bank=initial_per_bank, scheduled_additions=scheduled_additions) simulation.positions = {code: Position(code, name, category) for code, name, category in BANKS} quotes = {date: frame.set_index("code") for date, frame in prices.groupby("date")} index_rows = {(row.date, row.code): row for row in indices.itertuples(index=False)} dates = sorted(quotes) # execution day, stock code, trigger description, trigger day pending: list[tuple[str, str, str, str]] = [] previous_nav = None twr_index = 1.0 for date_index, day in enumerate(dates): external_flow = 0.0 if day == START: contribution = initial_per_bank * len(BANKS) simulation.cash += contribution simulation.external_contributions += contribution external_flow += contribution for code, _, _ in BANKS: purchase(simulation.positions[code], simulation, day, float(quotes[day].loc[code, "open"]), initial_per_bank, "初始建仓") for _, code, reason, trigger_day in [p for p in pending if p[0] == day]: simulation.cash += LOT_BUDGET simulation.external_contributions += LOT_BUDGET external_flow += LOT_BUDGET position = simulation.positions[code] purchase(position, simulation, day, float(quotes[day].loc[code, "open"]), LOT_BUDGET, f"{reason}（信号日 {trigger_day}）") position.additions_used += 1 pending = [p for p in pending if p[0] != day] for code, _, _ in BANKS: event = dividends.get(code, {}).get(day) if event: apply_dividend(simulation.positions[code], simulation, event) if scheduled_additions and date_index + 1 < len(dates): next_day = dates[date_index + 1] scheduled_codes = {code for _, code, _, _ in pending} for code, _, _ in BANKS: position = simulation.positions[code] row = index_rows[(day, code)] if pd.isna(row.drawdown) or code in scheduled_codes: continue threshold = FIRST_DRAWDOWN if position.additions_used == 0 else SECOND_DRAWDOWN if position.additions_used < 2 and row.drawdown <= -threshold: pending.append((next_day, code, f"总回报回撤 {threshold:.0%}", day)) market_value = sum(position.shares * float(quotes[day].loc[code, "close"]) for code, position in simulation.positions.items()) nav = simulation.cash + market_value if previous_nav is not None: denominator = previous_nav + external_flow if denominator <= 0: raise ValueError("Invalid return denominator") twr_index *= nav / denominator previous_nav = nav simulation.nav_rows.append({ "date": day, "nav": round(nav, 2), "market_value": round(market_value, 2), "cash": round(simulation.cash, 2), "external_flow": round(external_flow, 2), "twr_index": twr_index, }) return simulation def xirr(cashflows: list[tuple[str, float]]) -> float | None: days = [(datetime.fromisoformat(day).date() - datetime.fromisoformat(cashflows[0][0]).date()).days for day, _ in cashflows] amounts = [amount for _, amount in cashflows] def value(rate: float) -> float: return sum(amount / (1 + rate) ** (days[index] / 365.25) for index, amount in enumerate(amounts)) low, high = -0.9999, 10.0 if value(low) * value(high) > 0: return None for _ in range(200): middle = (low + high) / 2 result = value(middle) if abs(result) < 1e-8: return middle if value(low) * result <= 0: high = middle else: low = middle return (low + high) / 2 def summarize(simulation: Simulation, prices: pd.DataFrame) -> dict: nav = pd.DataFrame(simulation.nav_rows) terminal = nav.iloc[-1] highwater = nav.twr_index.cummax() max_drawdown = (nav.twr_index / highwater - 1).min() days = (pd.Timestamp(END) - pd.Timestamp(START)).days annualized_twr = float(nav.twr_index.iloc[-1]) ** (365.25 / days) - 1 cashflows = [(START, -simulation.initial_per_bank * len(BANKS))] for order in simulation.orders: if order["reason"] != "初始建仓": cashflows.append((order["date"], -LOT_BUDGET)) cashflows.append((END, float(terminal.nav))) xirr_value = xirr(cashflows) closing = prices[prices.date == END].set_index("code") contributions = simulation.external_contributions per_bank = [] for code, position in simulation.positions.items(): per_bank.append({ "code": code, "name": position.name, "category": position.category, "shares_at_end": position.shares, "market_value_at_end": round(position.shares * float(closing.loc[code, "close"]), 2), "cash_dividends_pre_tax": round(position.cash_dividends, 2), "purchase_cost_with_commission": round(position.spent, 2), "additions_used": position.additions_used, }) return { "label": simulation.label, "start": START, "end": END, "external_contributions": round(contributions, 2), "ending_nav": round(float(terminal.nav), 2), "ending_market_value": round(float(terminal.market_value), 2), "ending_cash": round(float(terminal.cash), 2), "cash_dividends_pre_tax": round(sum(p.cash_dividends for p in simulation.positions.values()), 2), "commissions": round(simulation.total_commission, 2), "absolute_return_vs_contributions": round(float(terminal.nav) / contributions - 1, 8), "time_weighted_annualized_return": round(annualized_twr, 8), "money_weighted_annualized_return": None if xirr_value is None else round(xirr_value, 8), "max_drawdown_time_weighted": round(float(max_drawdown), 8), "orders": len(simulation.orders), "dividend_events": len(simulation.dividend_ledger), "per_bank": per_bank, } def svg_document(title: str, desc: str, body: str, width: int, height: int) -> str: """Return a self-contained, browser-rendered SVG with explicit metadata.""" return f''' {html.escape(title)} {html.escape(desc)} {body} ''' def chart_text(x: float, y: float, value: str, size: int = 16, color: str = "#2d4a40", anchor: str = "start", weight: int = 400) -> str: return (f'{html.escape(value)}') def monthly_rows(rows: list[dict]) -> pd.DataFrame: """Keep the last observation per calendar month: readable charts, unchanged data source.""" frame = pd.DataFrame(rows).copy() frame["month"] = frame["date"].str[:7] return frame.groupby("month", as_index=False).tail(1).reset_index(drop=True) def path_for(frame: pd.DataFrame, value_column: str, x_fn, y_fn) -> str: return " ".join( ("M" if index == 0 else "L") + f"{x_fn(row):.1f},{y_fn(row[value_column]):.1f}" for index, (_, row) in enumerate(frame.iterrows()) ) def nav_and_drawdown_chart(simulations: dict[str, Simulation]) -> str: """Chart cash-dividend-inclusive time-weighted wealth and the realised drawdown.""" width, height = 1200, 760 left, right = 104, 1126 top, mid_bottom, lower_top, bottom = 144, 402, 510, 666 labels = { "initial_only": ("初始 50 万、不加仓", "#537d6a"), "rule_additions": ("50 万起步、规则加仓", "#145f6b"), "lump_sum": ("2015 年初一次 100 万", "#b67536"), } # Full daily paths retain the inception date and intramonth extrema. frames = {key: pd.DataFrame(value.nav_rows) for key, value in simulations.items()} all_dates = pd.concat([frame["date"] for frame in frames.values()]) start, end = pd.Timestamp(all_dates.min()), pd.Timestamp(all_dates.max()) span = max((end - start).days, 1) x_fn = lambda date: left + (pd.Timestamp(date) - start).days / span * (right - left) all_twr = pd.concat([frame["twr_index"] for frame in frames.values()]) ymax = max(2.6, math.ceil(float(all_twr.max()) * 10) / 10) y_top = lambda value: mid_bottom - (value / ymax) * (mid_bottom - top) rule_full = pd.DataFrame(simulations["rule_additions"].nav_rows).copy() rule_full["highwater"] = rule_full["twr_index"].cummax() rule_full["drawdown"] = rule_full["twr_index"] / rule_full["highwater"] - 1 min_dd = float(rule_full["drawdown"].min()) trough = rule_full.loc[rule_full["drawdown"].idxmin()] rule = rule_full y_bottom = lambda value: lower_top + (value / min_dd) * (bottom - lower_top) parts = [ chart_text(left, 58, "总回报的路径，比终点数字更重要", 30, "#173e34", weight=700), chart_text(left, 86, "十只固定样本；含税前现金分红；时间加权口径剔除外部追加资金的影响；2015-01-05 = 1.00", 16, "#587067"), chart_text(left, 124, "时间加权财富指数（倍）", 15, "#35584b", weight=700), f'', chart_text(left, 490, "规则加仓情景的回撤（相对历史高点）", 15, "#35584b", weight=700), f'', ] for tick in (0, 1, 2): y = y_top(tick) parts.append(f'') parts.append(chart_text(left - 13, y + 5, f"{tick:.1f}", 13, "#668077", anchor="end")) for year in range(start.year, end.year + 1): x = x_fn(f"{year}-01-01") if left <= x <= right: parts.append(f'') parts.append(chart_text(x, bottom + 28, str(year), 13, "#668077", anchor="middle")) for value in (0, -0.1, -0.2, -0.3): y = y_bottom(value) parts.append(f'') parts.append(chart_text(left - 13, y + 5, f"{value:.0%}", 13, "#668077", anchor="end")) for key, frame in frames.items(): label, color = labels[key] parts.append(f'') legend_x = {"initial_only": 150, "rule_additions": 485, "lump_sum": 820}[key] parts.append(f'') parts.append(chart_text(legend_x, 447, label, 15, "#35584b")) dd_path = path_for(rule, "drawdown", lambda row: x_fn(row["date"]), y_bottom) parts.append(f'') parts.append(f'') tx, ty = x_fn(trough["date"]), y_bottom(float(trough["drawdown"])) parts.extend([ f'', chart_text(tx + 13, ty - 11, f"{trough['date']}：{trough['drawdown']:.1%}", 14, "#7c3d25", weight=700), chart_text(left, 714, "数据：本仓库本地日线与分红缓存。图中指数把现金分红留在账户内；它不是账户金额，也不是未来收益预测。", 14, "#587067"), ]) return svg_document("十只 A 股银行股情景的总回报路径与最大回撤", "上图比较三种资金路径的时间加权财富指数；下图显示规则化加仓情景在 2015 年 8 月出现约 36.9% 最大回撤。", "\n ".join(parts), width, height) def stock_dispersion_chart(indices: pd.DataFrame) -> str: """Show how widely ten banks' cash-dividend-inclusive returns diverged.""" width, height = 1200, 760 left, plot_left, right, top, bottom = 54, 365, 1122, 158, 655 name_map = {code: name for code, name, _ in BANKS} category_map = {code: category for code, _, category in BANKS} colors = {"国有大行": "#537d6a", "股份行": "#2f7181", "城商行": "#b67536"} endpoint = indices.sort_values("date").groupby("code", as_index=False).tail(1).copy() endpoint["name"] = endpoint["code"].map(name_map) endpoint["category"] = endpoint["code"].map(category_map) endpoint = endpoint.sort_values("total_return_index").reset_index(drop=True) upper = 5.0 x = lambda value: plot_left + value / upper * (right - plot_left) parts = [ chart_text(left, 58, "同属“银行”，并不意味着回报相近", 30, "#173e34", weight=700), chart_text(left, 86, "各股税前分红再投资指数（含送转）；2015-01-05 = 1.00；与组合保留现金的口径不同。", 16, "#587067"), chart_text(plot_left, 130, "期末总回报倍数（截至 2026-09-04）", 15, "#35584b", weight=700), f'', ] for tick in range(0, 6): px = x(tick) parts.append(f'') parts.append(chart_text(px, bottom + 28, f"{tick:.0f}×", 13, "#668077", anchor="middle")) baseline = x(1) parts.append(f'') parts.append(chart_text(baseline, top - 10, "起点 1.00×", 13, "#587067", anchor="middle")) row_height = (bottom - top) / len(endpoint) for index, row in endpoint.iterrows(): y = top + row_height * index + 12 color = colors[row["category"]] parts.append(chart_text(plot_left - 18, y + 18, str(row["name"]), 17, "#28483d", anchor="end", weight=700)) parts.append(chart_text(plot_left - 18, y + 37, str(row["category"]), 12, "#668077", anchor="end")) parts.append(f'') parts.append(chart_text(x(float(row["total_return_index"])) + 12, y + 20, f"{row['total_return_index']:.2f}×", 16, "#28483d", weight=700)) parts.extend([ f'', chart_text(left + 21, 704, "国有大行", 14, "#587067"), f'', chart_text(left + 146, 704, "股份行", 14, "#587067"), f'', chart_text(left + 271, 704, "城商行", 14, "#587067"), chart_text(left, 738, "数据：本仓库本地日线与分红缓存；指数按除权日收盘再投资，不扣交易费用或个人红利税。", 14, "#587067"), ]) return svg_document("十只 A 股银行股的总回报离散度", "2015 年初至 2026 年 9 月初，十只固定样本的总回报倍数从 1.44 倍到 4.52 倍，说明行业标签不能替代单家银行研究。", "\n ".join(parts), width, height) def dividend_history_chart(simulation: Simulation) -> tuple[str, list[dict]]: """Turn the reproducible cash-dividend ledger into a sourceable annual chart.""" width, height = 1200, 700 left, right, top, bottom = 96, 1124, 155, 548 ledger = pd.DataFrame(simulation.dividend_ledger) ledger["year"] = ledger["date"].str[:4].astype(int) annual = ledger.groupby("year", as_index=False)["cash_dividend_pre_tax"].sum() annual["cash_dividend_pre_tax"] = annual["cash_dividend_pre_tax"].round(2) upper = math.ceil(float(annual["cash_dividend_pre_tax"].max()) / 10_000) * 10_000 y = lambda value: bottom - value / upper * (bottom - top) band = (right - left) / len(annual) parts = [ chart_text(left, 58, "分红是现金流，但也只是一条回报来源", 30, "#173e34", weight=700), chart_text(left, 86, "规则加仓情景的税前现金分红：不再投资，直接保留在账户现金中；2026 年仅统计至 9 月 4 日。", 16, "#587067"), chart_text(left, 132, "税前现金分红（元）", 15, "#35584b", weight=700), f'', ] for tick in range(0, upper + 1, 20_000): py = y(tick) parts.append(f'') parts.append(chart_text(left - 13, py + 5, f"{tick/10_000:.0f} 万", 13, "#668077", anchor="end")) for index, row in annual.iterrows(): bar_x = left + band * index + band * .19 bar_width = band * .62 bar_y = y(float(row["cash_dividend_pre_tax"])) parts.append(f'') parts.append(chart_text(bar_x + bar_width / 2, bottom + 28, str(int(row["year"])), 13, "#668077", anchor="middle")) if index in (0, len(annual) - 1, int(annual["cash_dividend_pre_tax"].idxmax())): parts.append(chart_text(bar_x + bar_width / 2, bar_y - 10, f"{row['cash_dividend_pre_tax']/10_000:.1f} 万", 14, "#28483d", anchor="middle", weight=700)) parts.extend([ chart_text(left, 597, "累计税前现金分红：57.2 万元；期末账户现金 58.9 万元，其中还含未使用预算、交易余款等现金。", 16, "#35584b", weight=700), chart_text(left, 642, "数据：本仓库分红事件台账。税前金额；不计个人投资者持有期差异产生的红利税；不可与未来分红承诺等同。", 14, "#587067"), ]) return svg_document("规则加仓情景的年度税前现金分红", "十只固定样本在规则化加仓情景中每年的税前现金分红。2026 年只覆盖至 9 月 4 日；分红未再投资。", "\n ".join(parts), width, height), annual.to_dict("records") def state_ownership_chart() -> tuple[str, list[dict]]: """Render dated ownership facts without pretending they form one live cap table.""" width, height = 1200, 850 left_x, right_x, panel_y, panel_w, panel_h = 52, 626, 130, 522, 610 central = [ ("1998", "财政部特别国债", "2,700 亿元补充四家国有独资商业银行资本。", "资本补充，不是二级市场买入。"), ("2003", "中央汇金成立", "代表国家对重点金融企业行使出资人权利。", "出资人角色与日常经营分开。"), ("2019", "工行：财政部 → 社保", "财政部划转 123.32 亿股 A 股至社保基金。", "国有资本持有主体发生迁移。"), ("2023", "中央汇金增持四大行", "公开增持工、农、中、建四家银行 A 股。", "影响预期，但不是普通股收益担保。"), ("2025", "财政部认购补充 CET1", "建行向财政部发行 115.894 亿股，募资 1,050 亿元。", "是定向资本补充的公司行为。"), ] snapshots = [ ("工行｜2025", "汇金 34.79% · 财政部 31.14% · 社保基金 A+H 5.35%", "三个国家背景主要股东均超过 5%。"), ("农行｜2025", "汇金 40.14% · 财政部 35.29% · 社保基金 6.72%", "两家大股东均超过 15%。"), ("兴业｜2025", "福建省财政厅 + 福建金投：20.57%", "合并第一大股东，并派驻董事。"), ("宁波｜官方股权介绍", "宁波国资一致行动合计约 20%", "宁波开发投资 18.74% + 宁兴资产 1.29%。"), ("南京 / 平安｜2015", "紫金投资 12.41% / 平安集团及平安寿险约 58%", "地方国资与保险集团控制不是同一类持股。"), ] parts = [ chart_text(52, 58, "政府资本进入银行：必须按动作、持有人与报告期拆开看", 29, "#173e34", weight=700), chart_text(52, 86, "以下均为可追溯的日期化事实；左右两栏的报告期不同，不能合并成一张“实时持股比例表”。", 16, "#587067"), f'', f'', chart_text(left_x + 22, panel_y + 39, "中央财政与国家资本：可核验的持股 / 资本动作", 17, "#245044", weight=700), chart_text(right_x + 22, panel_y + 39, "样本银行：日期化的股权快照", 17, "#245044", weight=700), ] for index, (year, heading, detail, implication) in enumerate(central): y = panel_y + 64 + index * 106 if index: parts.append(f'') parts.extend([ f'', chart_text(left_x + 53.5, y + 21, year, 14, "#166b59", anchor="middle", weight=700), chart_text(left_x + 100, y + 20, heading, 17, "#294a3f", weight=700), chart_text(left_x + 100, y + 47, detail, 14, "#587067"), chart_text(left_x + 100, y + 72, implication, 14, "#587067"), ]) for index, (heading, holding, implication) in enumerate(snapshots): y = panel_y + 64 + index * 106 if index: parts.append(f'') parts.extend([ chart_text(right_x + 22, y + 20, heading, 17, "#294a3f", weight=700), chart_text(right_x + 22, y + 47, holding, 14, "#166b59", weight=700), chart_text(right_x + 22, y + 72, implication, 14, "#587067"), ]) parts.extend([ chart_text(52, 786, "读图原则：财政部、中央汇金、社保基金、地方国资和保险集团的持股权利与政策目标不同；存款保护、系统稳定与普通股收益也不同。", 15, "#35584b", weight=700), chart_text(52, 821, "来源：各公司 2015 / 2025 年报、公司股东评估报告、财政部及中央汇金公开披露；逐项链接见正文。", 14, "#587067"), ]) facts = [ {"period": year, "type": heading, "fact": detail, "interpretation": implication} for year, heading, detail, implication in central ] + [ {"period": heading, "type": "ownership_snapshot", "fact": holding, "interpretation": implication} for heading, holding, implication in snapshots ] return svg_document("中国银行样本中的国家与地方资本持股路径", "左栏按时间显示财政部与中央汇金的资本和持股动作，右栏显示十只样本中部分银行的日期化股权快照。报告期不同，不能当作同日持股表。", "\n ".join(parts), width, height), facts def render_charts(simulations: dict[str, Simulation], indices: pd.DataFrame) -> None: """Create site assets and a compact machine-readable summary from local evidence.""" WEB_IMAGES.mkdir(parents=True, exist_ok=True) (WEB_IMAGES / "a-share-bank-core-holdings-nav-drawdown.svg").write_text( nav_and_drawdown_chart(simulations), encoding="utf-8" ) (WEB_IMAGES / "a-share-bank-core-holdings-stock-dispersion.svg").write_text( stock_dispersion_chart(indices), encoding="utf-8" ) dividend_svg, annual_dividends = dividend_history_chart(simulations["rule_additions"]) (WEB_IMAGES / "a-share-bank-core-holdings-dividend-history.svg").write_text(dividend_svg, encoding="utf-8") ownership_svg, ownership_facts = state_ownership_chart() (WEB_IMAGES / "a-share-bank-core-holdings-state-ownership.svg").write_text(ownership_svg, encoding="utf-8") endpoint = indices.sort_values("date").groupby("code", as_index=False).tail(1) name_map = {code: name for code, name, _ in BANKS} category_map = {code: category for code, _, category in BANKS} endpoint = endpoint.assign(name=endpoint.code.map(name_map), category=endpoint.code.map(category_map)) endpoint = endpoint.sort_values("total_return_index") chart_data = { "as_of": END, "portfolio_path_sampling": "month-end observations; SVG paths use every daily observation", "stock_index_method": "pre-tax dividend reinvestment at ex-date close; excludes dealing costs", "portfolio_path": { key: monthly_rows(simulation.nav_rows)[["date", "twr_index"]].to_dict("records") for key, simulation in simulations.items() }, "stock_total_return_endpoints": endpoint[["code", "name", "category", "total_return_index"]].to_dict("records"), "rule_additions_annual_cash_dividends_pre_tax": annual_dividends, "dated_state_and_local_ownership_facts": ownership_facts, "caveat": "Local cached prices and dividends only; portfolios retain pre-tax cash dividends, whereas stock indices assume ex-date-close reinvestment. Untriggered contribution budgets stay outside portfolio NAV.", } with (OUT / "chart_data.json").open("w", encoding="utf-8") as handle: json.dump(chart_data, handle, ensure_ascii=False, indent=2) def main() -> None: OUT.mkdir(parents=True, exist_ok=True) prices = load_prices() dividends, duplicate_audit = load_dividends() indices = total_return_indices(prices, dividends) simulations = [ simulate("初始50万，持有不加仓", PER_BANK_INITIAL, False, prices, dividends, indices), simulate("初始50万，规则化加仓最多50万", PER_BANK_INITIAL, True, prices, dividends, indices), simulate("2015年初一次性投入100万", 100_000.0, False, prices, dividends, indices), ] simulations_by_chart_key = { "initial_only": simulations[0], "rule_additions": simulations[1], "lump_sum": simulations[2], } summaries = [summarize(simulation, prices) for simulation in simulations] for simulation in simulations: label = {"初始50万，持有不加仓": "initial_only", "初始50万，规则化加仓最多50万": "rule_additions", "2015年初一次性投入100万": "lump_sum"}[simulation.label] pd.DataFrame(simulation.orders).to_csv(OUT / f"{label}_orders.csv", index=False) pd.DataFrame(simulation.dividend_ledger).to_csv(OUT / f"{label}_dividends.csv", index=False) pd.DataFrame(simulation.nav_rows).to_csv(OUT / f"{label}_daily_nav.csv", index=False) pd.DataFrame(summarize(simulation, prices)["per_bank"]).to_csv(OUT / f"{label}_per_bank.csv", index=False) indices.to_csv(OUT / "bank_total_return_indices.csv", index=False) with (OUT / "summary.json").open("w", encoding="utf-8") as handle: json.dump({ "method": { "price_source": str(PRICE_DB.relative_to(ROOT)), "dividend_sources": [str(SEED_DIVIDENDS.relative_to(ROOT)), str(EXTENDED_DIVIDENDS.relative_to(ROOT))], "network_requests": 0, "start": START, "end": END, "commission_rate": COMMISSION_RATE, "minimum_commission": MIN_COMMISSION, "cash_dividends": "pre_personal_income_tax; retained as cash; not reinvested", "addition_rule": "250-trading-day total-return drawdown of 20% then 35%; next-day open; two 25,000 RMB lots per bank", "known_limitations": [ "The simulation does not apply discretionary fundamental vetoes or sell rules.", "It reports pre-personal-income-tax cash dividends; investor-specific tax lots are not modelled.", "It is an illustration using a fixed, survivorship-aware 2015 investable universe, not an investable index.", ], }, "raw_dividend_duplicate_audit": duplicate_audit, "summaries": summaries, }, handle, ensure_ascii=False, indent=2) pd.DataFrame([{key: value for key, value in summary.items() if key != "per_bank"} for summary in summaries]).to_csv( OUT / "comparison.csv", index=False ) render_charts(simulations_by_chart_key, indices) print(json.dumps({"output": str(OUT), "summaries": summaries}, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()