#!/usr/bin/env python3
"""Enrich the stable stock-valuation ledger with valuation-bubble explanations.

The script does not change valuation ranges.  It reads the current ``latest.csv``,
the referenced formal reports, and front-adjusted daily K-lines, then adds a
reproducible explanation layer.  A bubble cause is an inference unless a separate
verified event chain exists; the output deliberately never labels it as proven
fund-flow causality.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import io
import json
import math
import os
import re
import statistics
import sys
from collections import Counter, defaultdict
from dataclasses import dataclass
from datetime import date, timedelta
from pathlib import Path
from typing import Iterable

import pymysql


SCRIPT_VERSION = "1.0"
BUBBLE_COLUMNS = [
    "bubble_status",
    "bubble_premium_pct",
    "bubble_vs_optimistic_pct",
    "bubble_primary_cause",
    "bubble_secondary_cause",
    "bubble_reason",
    "bubble_reason_nature",
    "bubble_confidence",
    "bubble_evidence_basis",
    "bubble_recheck_trigger",
    "return_20d_pct",
    "return_60d_pct",
    "distance_to_ma60_pct",
    "amount_ratio_5d_to_60d",
    "market_sample_count",
]


THEME_RULES: list[tuple[str, tuple[str, ...]]] = [
    (
        "军工、商业航天或低空经济主题溢价",
        (
            "军工",
            "国防",
            "导弹",
            "雷达",
            "航空",
            "航天",
            "无人机",
            "卫星",
            "军品",
            "惯性导航",
            "火箭",
            "低空",
        ),
    ),
    (
        "半导体国产替代、AI算力或先进封装预期",
        (
            "半导体",
            "芯片",
            "晶圆",
            "光刻",
            "刻蚀",
            "封测",
            "先进封装",
            "算力",
            "GPU",
            "服务器",
            "存储",
            "EDA",
            "光模块",
            "PCB",
        ),
    ),
    (
        "机器人、自动化或智能制造成长预期",
        (
            "机器人",
            "减速器",
            "伺服",
            "丝杠",
            "自动化",
            "机器视觉",
            "智能制造",
            "人形",
        ),
    ),
    (
        "创新药、医疗器械或国产替代预期",
        (
            "创新药",
            "医药",
            "生物药",
            "临床",
            "医疗器械",
            "CRO",
            "CXO",
            "疫苗",
            "制药",
            "诊断",
        ),
    ),
    (
        "新能源、储能或电动化成长预期",
        (
            "新能源",
            "锂电",
            "电池",
            "储能",
            "光伏",
            "风电",
            "充电桩",
            "逆变器",
            "固态电池",
            "电解液",
            "正极",
            "负极",
            "隔膜",
        ),
    ),
    (
        "智能汽车、汽车电子或电动化渗透预期",
        (
            "汽车电子",
            "智能驾驶",
            "车载",
            "汽车零部件",
            "新能源汽车",
            "线控",
            "座舱",
        ),
    ),
    (
        "信创、网络安全或软件国产化预期",
        (
            "信创",
            "网络安全",
            "自主计算",
            "国产软件",
            "操作系统",
            "数据库",
            "工业软件",
            "云计算",
            "信息化",
        ),
    ),
    (
        "通信、光通信或卫星互联网成长预期",
        (
            "通信",
            "光通信",
            "光纤",
            "光器件",
            "射频",
            "天线",
            "卫星互联网",
        ),
    ),
    (
        "资源品稀缺性或周期高景气外推",
        (
            "黄金",
            "白银",
            "铜",
            "铝",
            "稀土",
            "钨",
            "锂",
            "钴",
            "镍",
            "矿",
            "有色",
            "资源",
            "化工",
            "化纤",
            "煤炭",
            "油气",
        ),
    ),
    (
        "消费电子新品、AI终端或景气复苏预期",
        (
            "消费电子",
            "智能终端",
            "可穿戴",
            "折叠屏",
            "手机",
            "AR",
            "VR",
        ),
    ),
]


@dataclass(frozen=True)
class MarketMetrics:
    return_20d_pct: float | None
    return_60d_pct: float | None
    distance_to_ma60_pct: float | None
    amount_ratio_5d_to_60d: float | None
    sample_count: int


@dataclass(frozen=True)
class ReportFacts:
    business: str
    driver: str
    company_type: str
    method: str
    first_risk: str
    negative_profit: bool
    negative_fcf: bool


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Add reproducible valuation-bubble explanations to latest.csv/latest.md."
    )
    parser.add_argument("--project-root", type=Path, default=Path.cwd())
    parser.add_argument(
        "--latest-csv",
        type=Path,
        default=Path("ana-data/result/股票估值/估值台账/latest.csv"),
    )
    parser.add_argument(
        "--latest-md",
        type=Path,
        default=Path("ana-data/result/股票估值/估值台账/latest.md"),
    )
    parser.add_argument("--write", action="store_true", help="Atomically replace latest.csv/latest.md.")
    parser.add_argument("--skip-market", action="store_true", help="Do not read front-adjusted K-lines.")
    return parser.parse_args()


def resolve_under(root: Path, path: Path) -> Path:
    root = root.resolve()
    target = path if path.is_absolute() else root / path
    target = target.resolve()
    if target != root and root not in target.parents:
        raise ValueError(f"Path escapes project root: {path}")
    return target


def read_csv(path: Path) -> tuple[list[str], list[dict[str, str]]]:
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        if reader.fieldnames is None:
            raise ValueError(f"CSV has no header: {path}")
        return list(reader.fieldnames), list(reader)


def clean_text(value: str, limit: int = 220) -> str:
    value = re.sub(r"\s+", " ", value).strip(" -；;。")
    return value[:limit]


def section(text: str, heading: str, next_prefix: str) -> str:
    start = text.find(heading)
    if start < 0:
        return ""
    start += len(heading)
    end = text.find(next_prefix, start)
    return text[start:] if end < 0 else text[start:end]


def first_bullet(block: str) -> str:
    for line in block.splitlines():
        stripped = line.strip()
        if stripped.startswith("- "):
            return clean_text(stripped[2:])
    return ""


def all_bullets(block: str, limit: int = 3) -> list[str]:
    values: list[str] = []
    for line in block.splitlines():
        stripped = line.strip()
        if stripped.startswith("- "):
            values.append(clean_text(stripped[2:]))
            if len(values) >= limit:
                break
    return values


def parse_report(path: Path) -> ReportFacts:
    text = path.read_text(encoding="utf-8-sig")
    business_block = section(text, "## 3. 业务与利润来源", "\n## ")
    if not business_block:
        business_block = section(text, "## 4. 业务及利润来源", "\n## ")
    business_bullets = all_bullets(business_block)
    business = clean_text("；".join(business_bullets))
    driver_match = re.search(r"(?:主要驱动是|高增长主要依赖)(.+?)(?:。|；|$)", business)
    driver = clean_text(driver_match.group(1), 120) if driver_match else ""
    type_match = re.search(r"公司类型：(.+)", text)
    company_type = clean_text(type_match.group(1), 100) if type_match else ""
    method_match = re.search(r"主模型：(.+)", text)
    method = clean_text(method_match.group(1), 120) if method_match else ""
    if not method:
        model_block = section(text, "## 9. 模型选择", "\n## ")
        model_lines = [line.strip() for line in model_block.splitlines() if line.strip()]
        if model_lines:
            method = clean_text(model_lines[0], 120)
    risk_block = section(text, "### 12.1 主要风险", "\n### ")
    if not risk_block:
        risk_section = section(text, "## 14. 风险、上调和下调触发器", "\n## ")
        risk_block = section(risk_section, "主要风险：", "\n上调触发器：")
    first_risk = first_bullet(risk_block)
    # A negative deduction-only profit warning does not mean attributable profit
    # is negative.  Use the explicit TTM attributable-profit warning only.
    negative_profit = "QA-NEGATIVE-TTM-PROFIT" in text
    negative_fcf = "QA-NEGATIVE-FCF" in text
    return ReportFacts(
        business=business,
        driver=driver,
        company_type=company_type,
        method=method,
        first_risk=first_risk,
        negative_profit=negative_profit,
        negative_fcf=negative_fcf,
    )


def mysql_connection() -> pymysql.Connection:
    missing = [
        key
        for key in ("MYSQL_HOST", "MYSQL_USER", "MYSQL_PASSWORD")
        if not os.environ.get(key)
    ]
    if missing:
        raise RuntimeError(f"Missing MySQL environment variables: {', '.join(missing)}")
    return pymysql.connect(
        host=os.environ["MYSQL_HOST"],
        port=int(os.environ.get("MYSQL_PORT", "3306")),
        user=os.environ["MYSQL_USER"],
        password=os.environ["MYSQL_PASSWORD"],
        database="trading_xuntou",
        charset="utf8mb4",
        autocommit=True,
        cursorclass=pymysql.cursors.DictCursor,
        read_timeout=60,
        write_timeout=60,
    )


def finite_float(value: object) -> float | None:
    if value in (None, ""):
        return None
    result = float(value)
    return result if math.isfinite(result) else None


def market_metrics(
    tickers: list[str], trade_date: date, skip_market: bool
) -> dict[str, MarketMetrics]:
    if skip_market:
        return {}
    start = trade_date - timedelta(days=150)
    placeholders = ",".join(["%s"] * len(tickers))
    query = f"""
        SELECT symbol, trade_date, close, amount
        FROM cn_stock_kline_1d_front
        WHERE symbol IN ({placeholders})
          AND trade_date BETWEEN %s AND %s
          AND close > 0
          AND source = 'xtquant'
        ORDER BY symbol, trade_date
    """
    grouped: dict[str, list[tuple[date, float, float | None]]] = defaultdict(list)
    with mysql_connection() as connection:
        with connection.cursor() as cursor:
            cursor.execute(query, [*tickers, start, trade_date])
            for row in cursor.fetchall():
                grouped[str(row["symbol"])].append(
                    (
                        row["trade_date"],
                        float(row["close"]),
                        finite_float(row["amount"]),
                    )
                )
    result: dict[str, MarketMetrics] = {}
    for ticker, series in grouped.items():
        if not series or series[-1][0] != trade_date:
            continue
        closes = [item[1] for item in series]
        amounts = [item[2] for item in series]
        current = closes[-1]
        ret20 = current / closes[-21] - 1 if len(closes) >= 21 else None
        ret60 = current / closes[-61] - 1 if len(closes) >= 61 else None
        ma_window = closes[-60:]
        ma60 = statistics.fmean(ma_window) if len(ma_window) >= 40 else None
        distance_ma60 = current / ma60 - 1 if ma60 else None
        valid60 = [value for value in amounts[-60:] if value and value > 0]
        valid5 = [value for value in amounts[-5:] if value and value > 0]
        amount_ratio = None
        if valid60 and valid5:
            amount_ratio = statistics.fmean(valid5) / statistics.fmean(valid60)
        result[ticker] = MarketMetrics(
            return_20d_pct=ret20,
            return_60d_pct=ret60,
            distance_to_ma60_pct=distance_ma60,
            amount_ratio_5d_to_60d=amount_ratio,
            sample_count=len(series),
        )
    return result


def choose_theme(facts: ReportFacts) -> str:
    # Risk paragraphs often contain negated statements such as "no direct military
    # evidence".  They are useful as recheck triggers but must not classify the
    # company's actual business theme.
    haystack = " ".join((facts.business, facts.driver, facts.company_type))
    for label, keywords in THEME_RULES:
        if any(keyword.lower() in haystack.lower() for keyword in keywords):
            return label
    if any(word in haystack for word in ("重组", "并购", "资产注入", "控制权", "整合")):
        return "并购重组、资产整合或控制权期权"
    return "业务增长、订单兑现和利润率改善预期"


def pct(value: float | None) -> str:
    return "" if value is None else f"{value * 100:.2f}"


def short_driver(facts: ReportFacts, theme: str) -> str:
    if facts.driver:
        return facts.driver
    if facts.company_type:
        return f"{facts.company_type}业务兑现"
    if facts.business:
        return facts.business[:100]
    return theme


def enrich_row(
    row: dict[str, str], facts: ReportFacts, market: MarketMetrics | None
) -> dict[str, str]:
    close = float(row["close"])
    base_high = float(row["base_high"])
    optimistic_high = finite_float(row.get("optimistic_high"))
    premium = close / base_high - 1
    vs_optimistic = close / optimistic_high - 1 if optimistic_high else None
    enriched = dict(row)

    if premium <= 0:
        enriched.update(
            {
                "bubble_status": "未识别估值泡沫",
                "bubble_premium_pct": f"{premium * 100:.2f}",
                "bubble_vs_optimistic_pct": pct(vs_optimistic),
                "bubble_primary_cause": "不适用",
                "bubble_secondary_cause": "不适用",
                "bubble_reason": "当前收盘价未高于基准合理区间上沿，按本手册规则不认定估值泡沫。",
                "bubble_reason_nature": "规则判定",
                "bubble_confidence": "高（价格位置）",
                "bubble_evidence_basis": "当前收盘价与正式基准合理区间比较",
                "bubble_recheck_trigger": "价格升破基准合理区间上沿，或公司发生正式复评",
            }
        )
    else:
        theme = choose_theme(facts)
        driver = short_driver(facts, theme)
        primary_method = facts.method.split("，", 1)[0].strip().upper()
        if facts.negative_profit or primary_method.startswith("PB"):
            primary = "盈利修复或扭亏预期提前定价"
            lead = (
                f"当前盈利基线偏弱或常规PE适用性不足，但价格仍高于基准上沿"
                f"{premium * 100:.1f}%，主要在交易盈利修复，并押注{driver}。"
            )
        else:
            primary = theme
            lead = (
                f"价格高于基准合理区间上沿{premium * 100:.1f}%，主要在提前交易"
                f"{theme}，具体押注{driver}。"
            )

        secondary: list[str] = []
        if optimistic_high and close > optimistic_high:
            secondary.append("超出乐观情景的叙事与情绪溢价")
            lead += (
                f" 当前价还高于乐观情景上沿{(close / optimistic_high - 1) * 100:.1f}%，"
                "说明市场计入了超过现有乐观模型的额外预期。"
            )
            status = "泡沫-极端（超过乐观上沿）"
        elif premium > 0.50:
            secondary.append("乐观情景被大幅提前资本化")
            lead += " 当前价虽未超过乐观上沿，但已经大幅提前资本化乐观情景。"
            status = "泡沫-高"
        elif premium > 0.15:
            secondary.append("乐观情景提前定价")
            lead += " 当前价位于基准与乐观上沿之间，市场已经提前支付部分乐观情景。"
            status = "泡沫-中"
        else:
            secondary.append("轻度乐观预期")
            lead += " 溢价幅度较小，更接近估值误差与轻度乐观预期的交界。"
            status = "泡沫-轻"

        momentum = False
        if market:
            if (
                market.return_60d_pct is not None
                and market.distance_to_ma60_pct is not None
                and market.return_60d_pct >= 0.15
                and market.distance_to_ma60_pct >= 0.08
            ):
                momentum = True
                secondary.append("趋势动量与交易拥挤")
                lead += (
                    f" 近60个交易日上涨{market.return_60d_pct * 100:.1f}%，"
                    f"并高于60日均线{market.distance_to_ma60_pct * 100:.1f}%，"
                    "量价趋势可能放大估值溢价。"
                )
            elif (
                market.return_20d_pct is not None
                and market.distance_to_ma60_pct is not None
                and market.return_20d_pct >= 0.10
                and market.distance_to_ma60_pct >= 0.08
            ):
                momentum = True
                secondary.append("短期动量放大")
                lead += (
                    f" 近20个交易日上涨{market.return_20d_pct * 100:.1f}%，"
                    f"高于60日均线{market.distance_to_ma60_pct * 100:.1f}%，"
                    "短期动量可能放大估值溢价。"
                )
            if market.sample_count < 61:
                lead += (
                    f" 前复权行情样本仅{market.sample_count}个交易日，"
                    "不足以形成完整60日涨幅，长期动量不参与原因判断。"
                )
        count = int(row.get("consensus_count") or 0)
        if count == 0:
            lead += " 当前没有可用机构一致预期覆盖，原因判断更依赖估值反推与业务情景。"
        elif count <= 2:
            lead += f" 机构覆盖仅{count}家，预测分歧和样本偏差仍可能较大。"

        evidence = "估值反推+正式报告业务/风险段落"
        if market:
            evidence += f"+前复权20/60日量价（样本{market.sample_count}日）"
        confidence = "中" if facts.business and market else "中低"
        if count == 0 or not facts.business:
            confidence = "中低"
        if not momentum and primary == "业务增长、订单兑现和利润率改善预期":
            confidence = "中低"
        recheck = facts.first_risk or f"{driver}未兑现或估值倍数回落"
        reason_text = clean_text(lead, 620)
        if not reason_text.endswith(("。", "！", "？")):
            reason_text += "。"
        enriched.update(
            {
                "bubble_status": status,
                "bubble_premium_pct": f"{premium * 100:.2f}",
                "bubble_vs_optimistic_pct": pct(vs_optimistic),
                "bubble_primary_cause": primary,
                "bubble_secondary_cause": "；".join(dict.fromkeys(secondary)),
                "bubble_reason": reason_text,
                "bubble_reason_nature": "基于估值反推、业务驱动和量价显影的推断",
                "bubble_confidence": confidence,
                "bubble_evidence_basis": evidence,
                "bubble_recheck_trigger": clean_text(recheck, 180),
            }
        )

    enriched.update(
        {
            "return_20d_pct": pct(market.return_20d_pct) if market else "",
            "return_60d_pct": pct(market.return_60d_pct) if market else "",
            "distance_to_ma60_pct": pct(market.distance_to_ma60_pct) if market else "",
            "amount_ratio_5d_to_60d": (
                "" if not market or market.amount_ratio_5d_to_60d is None else f"{market.amount_ratio_5d_to_60d:.4f}"
            ),
            "market_sample_count": "" if not market else str(market.sample_count),
        }
    )
    return enriched


def canonical_base_csv(fieldnames: list[str], rows: Iterable[dict[str, str]]) -> bytes:
    base_fields = [field for field in fieldnames if field not in BUBBLE_COLUMNS]
    buffer = io.StringIO(newline="")
    writer = csv.DictWriter(
        buffer,
        fieldnames=base_fields,
        extrasaction="ignore",
        lineterminator="\n",
    )
    writer.writeheader()
    writer.writerows(rows)
    return buffer.getvalue().encode("utf-8")


def atomic_write_csv(path: Path, fieldnames: list[str], rows: Iterable[dict[str, str]]) -> bool:
    temp = path.with_name(f".{path.name}.bubble.tmp")
    try:
        with temp.open("w", encoding="utf-8", newline="") as handle:
            writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore", lineterminator="\n")
            writer.writeheader()
            writer.writerows(rows)
        if path.exists() and temp.read_bytes() == path.read_bytes():
            return False
        os.replace(temp, path)
        return True
    finally:
        if temp.exists():
            temp.unlink()


def markdown_escape(value: object) -> str:
    return str(value).replace("|", "\\|").replace("\n", " ")


def render_markdown(rows: list[dict[str, str]], source_hash: str) -> str:
    status_counts = Counter(row["bubble_status"] for row in rows)
    cause_counts = Counter(
        row["bubble_primary_cause"]
        for row in rows
        if row["bubble_status"] != "未识别估值泡沫"
    )
    label_counts = Counter(row["label"] for row in rows)
    trade_dates = sorted({row["trade_date"] for row in rows})
    trade_date_text = trade_dates[0] if len(trade_dates) == 1 else ",".join(trade_dates)
    bubble_count = len(rows) - status_counts.get("未识别估值泡沫", 0)
    lines = [
        "# 股票估值每日台账最新总表",
        "",
        f"- 价格交易日：`{trade_date_text}`",
        f"- 泡沫解释版本：`{trade_date_text}/v{SCRIPT_VERSION}`",
        f"- 记录数：`{len(rows)}`",
        f"- 数值判定覆盖：本表共`{len(rows)}`只；无法形成当日数值判定的证券单列于同目录`latest_gaps.csv`。",
        f"- 估值泡沫：`{bubble_count}`只；定义为当前价高于正式基准合理区间上沿。",
        "- 原因性质：除价格位置外，泡沫原因均是基于反向估值、基础报告和量价显影的最可能解释，不是已证实的资金流因果。",
        f"- 输入快照 SHA-256：`{source_hash}`；泡沫解释脚本：`enrich_valuation_bubbles.py v{SCRIPT_VERSION}`。",
        "- 口径：最近完整交易日收盘价相对最近有效正式估值版本；不构成交易指令。",
        "",
        "## 泡沫分布",
        "",
        "| 泡沫状态 | 数量 |",
        "|---|---:|",
    ]
    for key in (
        "未识别估值泡沫",
        "泡沫-轻",
        "泡沫-中",
        "泡沫-高",
        "泡沫-极端（超过乐观上沿）",
    ):
        lines.append(f"| {key} | {status_counts.get(key, 0)} |")
    lines.extend(
        [
            "",
            "## 主要泡沫原因",
            "",
            "| 主要原因 | 数量 |",
            "|---|---:|",
        ]
    )
    for cause, count in cause_counts.most_common():
        lines.append(f"| {markdown_escape(cause)} | {count} |")
    lines.extend(
        [
            "",
            "## 全量明细",
            "",
            "| 代码 | 公司 | 收盘价 | 基准区间 | 原判定 | 泡沫判定 | 高于基准上沿 | 主要泡沫原因 | 原因说明 | 置信度 | 60日涨幅 | 相对MA60 | 正式报告 |",
            "|---|---|---:|---:|---|---|---:|---|---|---|---:|---:|---|",
        ]
    )
    for row in rows:
        report = f"`{row['report_path']}`"
        base = f"{float(row['base_low']):.4f}—{float(row['base_high']):.4f}"
        values = [
            row["ticker"],
            row["company"],
            f"{float(row['close']):.4f}",
            base,
            row["label"],
            row["bubble_status"],
            f"{float(row['bubble_premium_pct']):.2f}%",
            row["bubble_primary_cause"],
            row["bubble_reason"],
            row["bubble_confidence"],
            "" if not row["return_60d_pct"] else f"{float(row['return_60d_pct']):.2f}%",
            "" if not row["distance_to_ma60_pct"] else f"{float(row['distance_to_ma60_pct']):.2f}%",
            report,
        ]
        lines.append("| " + " | ".join(markdown_escape(value) for value in values) + " |")
    lines.extend(
        [
            "",
            "## 标签复核",
            "",
            "、".join(f"{key}{label_counts[key]}只" for key in ("偏低", "基本合理", "偏贵", "明显偏贵")),
            "",
        ]
    )
    return "\n".join(lines)


def atomic_write_text(path: Path, content: str) -> bool:
    temp = path.with_name(f".{path.name}.bubble.tmp")
    try:
        temp.write_text(content, encoding="utf-8", newline="\n")
        if path.exists() and temp.read_bytes() == path.read_bytes():
            return False
        os.replace(temp, path)
        return True
    finally:
        if temp.exists():
            temp.unlink()


def main() -> int:
    args = parse_args()
    project_root = args.project_root.resolve()
    latest_csv = resolve_under(project_root, args.latest_csv)
    latest_md = resolve_under(project_root, args.latest_md)
    fieldnames, rows = read_csv(latest_csv)
    source_hash = hashlib.sha256(canonical_base_csv(fieldnames, rows)).hexdigest().upper()
    if not rows:
        raise RuntimeError("Current numeric judgement file is empty")
    tickers = [row["ticker"] for row in rows]
    if len(set(tickers)) != len(tickers):
        raise RuntimeError("Duplicate tickers in latest.csv")
    trade_dates = {date.fromisoformat(row["trade_date"]) for row in rows}
    metrics: dict[str, MarketMetrics] = {}
    for trade_date in sorted(trade_dates):
        date_tickers = [row["ticker"] for row in rows if date.fromisoformat(row["trade_date"]) == trade_date]
        metrics.update(market_metrics(date_tickers, trade_date, args.skip_market))
    missing_market = sorted(set(tickers) - set(metrics)) if not args.skip_market else []

    enriched: list[dict[str, str]] = []
    missing_reports: list[str] = []
    missing_business: list[str] = []
    for row in rows:
        report_path = resolve_under(project_root, Path(row["report_path"]))
        if not report_path.is_file():
            missing_reports.append(row["ticker"])
            continue
        facts = parse_report(report_path)
        if not facts.business:
            missing_business.append(row["ticker"])
        enriched.append(enrich_row(row, facts, metrics.get(row["ticker"])))
    if missing_reports:
        raise RuntimeError(f"Missing formal reports: {missing_reports[:10]}")
    if len(enriched) != len(rows):
        raise RuntimeError("Enriched row count changed")

    bubble_rows = [row for row in enriched if row["bubble_status"] != "未识别估值泡沫"]
    expected_bubbles = [row for row in enriched if float(row["close"]) > float(row["base_high"])]
    if {row["ticker"] for row in bubble_rows} != {row["ticker"] for row in expected_bubbles}:
        raise RuntimeError("Bubble rule mismatch")
    label_mismatch = [
        row["ticker"]
        for row in enriched
        if (row["label"] in {"偏贵", "明显偏贵"})
        != (row["bubble_status"] != "未识别估值泡沫")
    ]
    if label_mismatch:
        raise RuntimeError(f"Bubble/valuation label mismatch: {label_mismatch[:10]}")

    output_fields = [field for field in fieldnames if field not in BUBBLE_COLUMNS] + BUBBLE_COLUMNS
    status_counts = Counter(row["bubble_status"] for row in enriched)
    cause_counts = Counter(row["bubble_primary_cause"] for row in bubble_rows)
    summary = {
        "script_version": SCRIPT_VERSION,
        "mode": "WRITE" if args.write else "DRY_RUN",
        "source_csv_sha256": source_hash,
        "row_count": len(enriched),
        "bubble_count": len(bubble_rows),
        "non_bubble_count": len(enriched) - len(bubble_rows),
        "status_counts": dict(sorted(status_counts.items())),
        "primary_cause_counts": dict(cause_counts.most_common()),
        "market_metrics_count": len(metrics),
        "market_under_61_count": sum(metric.sample_count < 61 for metric in metrics.values()),
        "missing_market_count": len(missing_market),
        "missing_market_sample": missing_market[:10],
        "missing_business_count": len(missing_business),
        "missing_business_sample": missing_business[:10],
        "formal_report_count": len(enriched),
        "label_mismatch_count": 0,
    }
    if args.write:
        csv_changed = atomic_write_csv(latest_csv, output_fields, enriched)
        md_changed = atomic_write_text(latest_md, render_markdown(enriched, source_hash))
        summary["files_changed"] = {"latest_csv": csv_changed, "latest_md": md_changed}
        summary["latest_csv_sha256"] = hashlib.sha256(latest_csv.read_bytes()).hexdigest().upper()
        summary["latest_md_sha256"] = hashlib.sha256(latest_md.read_bytes()).hexdigest().upper()
    print(json.dumps(summary, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as exc:
        print(json.dumps({"status": "ERROR", "error": str(exc)}, ensure_ascii=False), file=sys.stderr)
        raise