#!/usr/bin/env python3
"""Generate valuations for uncovered semiconductor/robot/defense/new-energy names.

The builder reuses V2 provider evidence and the frozen V1 calculation core.
Research universes are read from their governed artifacts, current prices come
only from the local front-adjusted daily table, and final outputs are confined
to the valuation analyst's formal directories.
"""

from __future__ import annotations

import csv
import hashlib
import importlib.util
import json
import os
import re
import statistics
import sys
from collections import Counter, defaultdict
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any

import pymysql


ROOT = Path(__file__).resolve().parents[2]
for package_root in (ROOT / "dev/ana-dev", ROOT / "dev/project-dev"):
    sys.path.insert(0, str(package_root))

BASE_PATH = ROOT / "ai-valuation-analyst/tools/generate_semiconductor_batch_20260804.py"
SPEC = importlib.util.spec_from_file_location("industry_gap_base", BASE_PATH)
if SPEC is None or SPEC.loader is None:
    raise RuntimeError(f"cannot import {BASE_PATH}")
base = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = base
SPEC.loader.exec_module(base)

AS_OF = "2026-08-13"
STAMP = "20260813"
BATCH_ID = "BATCH-STOCK-VALUATION-INDUSTRY-RESEARCH-GAPS-20260814-001"
RESULT_ROOT = ROOT / "ana-data/result/股票估值"
SUMMARY_DIR = RESULT_ROOT / "行业调研标的估值覆盖"
CASE_ROOT = ROOT / "ana-data/cases/股票估值" / BATCH_ID
V2_ROOT = ROOT / "ai-valuation-analyst/tmp/valuation_industry_research_gaps_20260814"
CACHE = ROOT / "ai-valuation-analyst/tmp/v2_cache/blobs/sha256"
REGISTRY = ROOT / "dev/ana-dev/stock_valuation_pipeline/source_registry.json"

# Rebind the reusable parser/calculation helpers to this batch.  Formulae stay
# in the frozen V1 core; only the task date and evidence locations change.
base.AS_OF = AS_OF
base.CACHE = CACHE
base.RESULT_ROOT = RESULT_ROOT
base.CASE_ROOT = CASE_ROOT
base.REGISTRY = REGISTRY

SHANGHAI_PREFIX = ("6",)
BEIJING_PREFIX = ("8", "9")


@dataclass(frozen=True)
class Target:
    ticker: str
    research_name: str
    industries: tuple[str, ...]
    research_layers: tuple[str, ...]


def mysql_connection(database: str) -> pymysql.Connection:
    return pymysql.connect(
        host=os.environ.get("MYSQL_HOST", "127.0.0.1"),
        port=int(os.environ.get("MYSQL_PORT", "3306")),
        user=os.environ.get("MYSQL_USER", "root"),
        password=os.environ["MYSQL_PASSWORD"],
        database=database,
        charset="utf8mb4",
        autocommit=True,
        cursorclass=pymysql.cursors.DictCursor,
    )


def write_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def safe_filename(text: str) -> str:
    return re.sub(r'[<>:"/\\|?*]', "", text).strip() or "company"


def premium_text(price: float, upper: float) -> str:
    return f"{(price / upper - 1) * 100:.2f}%" if upper > 0 else "不可计算（区间上沿为0）"


def canonical_bse(ticker: str) -> str:
    return "920179.BJ" if ticker == "835179.BJ" else ticker


def universe_rows() -> tuple[dict[str, Target], dict[str, list[dict[str, str]]]]:
    membership: dict[str, dict[str, Any]] = {}

    def add(ticker: str, name: str, industry: str, layer: str) -> None:
        ticker = canonical_bse(ticker)
        item = membership.setdefault(ticker, {"names": [], "industries": set(), "layers": set()})
        item["names"].append(name)
        item["industries"].add(industry)
        item["layers"].add(layer)

    semi_path = ROOT / "ana-data/cases/半导体案例/ANA-SEMI-20260722-001/outputs/核心文档/国内企业信息表.csv"
    market_map = {"SSE": "SH", "SSE STAR": "SH", "SZSE": "SZ", "SZSE CHINEXT": "SZ", "BSE": "BJ"}
    for row in csv.DictReader(semi_path.open(encoding="utf-8-sig", newline="")):
        code = row["ticker"].strip()
        suffix = market_map.get(row["listed_market"].strip())
        if suffix and re.fullmatch(r"\d{6}", code):
            add(f"{code}.{suffix}", row["canonical_name"].strip(), "半导体", row["profile_research_state"].strip())

    robot_path = ROOT / "ana-data/cases/机器人案例/ANA-ROBOT-INDUSTRY-001/evidence/next_robot_036_market_database_company_universe_authority_20260726.csv"
    for row in csv.DictReader(robot_path.open(encoding="utf-8-sig", newline="")):
        add(row["symbol"].strip(), row["canonical_target_name"].strip(), "机器人", row["universe_layer"].strip())

    newenergy_path = ROOT / "ana-data/cases/新能源案例/核心文档/全量公司横向总表.md"
    text = newenergy_path.read_text(encoding="utf-8")
    for code, name, tier in re.findall(r"\| (\d{6}) \| \[([^\]]+)\]\([^\)]*\) \| [^|]+ \| [^|]+ \| `([^`]+)`", text):
        suffix = "SH" if code.startswith(SHANGHAI_PREFIX) else ("BJ" if code.startswith(BEIJING_PREFIX) else "SZ")
        add(f"{code}.{suffix}", name.strip(), "新能源", tier.strip())

    defense_path = RESULT_ROOT / "20260806_batch_defense_research_and_added_valuation/军工调研企业估值全量清单_20260806.csv"
    defense_rows = list(csv.DictReader(defense_path.open(encoding="utf-8-sig", newline="")))
    for row in defense_rows:
        add(row["ticker"].strip(), row["company"].strip(), "军工", row["scope_type"].strip())

    targets = {
        ticker: Target(
            ticker=ticker,
            research_name=next((name for name in item["names"] if "/" not in name and "-" not in name), item["names"][0]),
            industries=tuple(sorted(item["industries"])),
            research_layers=tuple(sorted(item["layers"])),
        )
        for ticker, item in membership.items()
    }
    sources = {
        "半导体": [{"path": str(semi_path.relative_to(ROOT)).replace("\\", "/"), "scope": "179只国内A/B股研究证券；北交所旧代码835179映射920179"}],
        "机器人": [{"path": str(robot_path.relative_to(ROOT)).replace("\\", "/"), "scope": "100只本地行情研究证券；含正式、核心、相邻、OEM、HOLD与EXCLUDE层，按用户‘调研涉及’口径全纳入"}],
        "新能源": [{"path": str(newenergy_path.relative_to(ROOT)).replace("\\", "/"), "scope": "40家由已接受BATCH-001/002支撑的当前公司池"}],
        "军工": [{"path": str(defense_path.relative_to(ROOT)).replace("\\", "/"), "scope": "33批528只已接受调研母池"}],
    }
    return targets, sources


def existing_security() -> dict[str, str]:
    with mysql_connection("stock_valuation") as connection, connection.cursor() as cursor:
        cursor.execute("SELECT ticker, company FROM security")
        result = {row["ticker"]: row["company"] for row in cursor.fetchall()}
    # Two defense names have explicit, already completed special conclusions
    # and intentionally never enter the numeric daily ledger.
    defense_path = RESULT_ROOT / "20260806_batch_defense_research_and_added_valuation/军工调研企业估值全量清单_20260806.csv"
    for row in csv.DictReader(defense_path.open(encoding="utf-8-sig", newline="")):
        if row["status"] in {"NON_EVALUABLE", "DELISTED_EXCLUDED"}:
            result[row["ticker"]] = row["company"]
    return result


def failed_dir(ticker: str) -> Path:
    matches = sorted(V2_ROOT.glob(f"v2_{ticker.replace('.', '_')}.failed-*"), key=lambda path: path.stat().st_mtime, reverse=True)
    if not matches:
        raise FileNotFoundError(f"V2 package missing for {ticker}")
    return matches[0]


def load_provider(ticker: str) -> tuple[dict[str, Any], dict[str, Any], Path]:
    directory = failed_dir(ticker)
    return base.read_json(directory / "provider_results.json"), base.read_json(directory / "failure.json"), directory


def latest_kline(ticker: str) -> dict[str, Any]:
    start = date.fromisoformat(AS_OF) - timedelta(days=180)
    with mysql_connection("trading_xuntou") as connection, connection.cursor() as cursor:
        cursor.execute(
            "SELECT trade_date,close,amount,source,source_batch_id,updated_at "
            "FROM cn_stock_kline_1d_front WHERE symbol=%s AND trade_date BETWEEN %s AND %s "
            "AND close>0 AND source='xtquant' ORDER BY trade_date",
            (ticker, start, AS_OF),
        )
        rows = cursor.fetchall()
    if not rows or str(rows[-1]["trade_date"]) != AS_OF:
        raise RuntimeError(f"missing {AS_OF} front-adjusted close for {ticker}")
    closes = [float(row["close"]) for row in rows]
    amounts = [float(row["amount"] or 0) for row in rows]
    ma60 = statistics.fmean(closes[-60:]) if len(closes) >= 60 else None
    return {
        "price": closes[-1], "date": AS_OF, "amount": amounts[-1], "source": rows[-1]["source"],
        "source_batch_id": rows[-1]["source_batch_id"], "updated_at": str(rows[-1]["updated_at"]),
        "return_20d": closes[-1] / closes[-21] - 1 if len(closes) >= 21 else None,
        "return_60d": closes[-1] / closes[-61] - 1 if len(closes) >= 61 else None,
        "ma60": ma60, "distance_to_ma60": closes[-1] / ma60 - 1 if ma60 else None,
        "amount_ratio_5d_to_60d": statistics.fmean(amounts[-5:]) / statistics.fmean(amounts[-60:]) if len(amounts) >= 60 and statistics.fmean(amounts[-60:]) else None,
        "sample_count": len(rows),
    }


def static_total_shares(ticker: str) -> float | None:
    """Return xtquant TotalVolume as a unit sanity check, not a date proof."""
    with mysql_connection("trading_xuntou") as connection, connection.cursor() as cursor:
        cursor.execute("SELECT raw_json FROM cn_stock_instrument_static WHERE BINARY symbol=%s", (ticker,))
        row = cursor.fetchone()
    if not row or not row.get("raw_json"):
        return None
    try:
        value = float(json.loads(row["raw_json"]).get("TotalVolume") or 0)
    except (TypeError, ValueError, json.JSONDecodeError):
        return None
    return value if value > 0 else None


def quote_payload(provider: dict[str, Any]) -> tuple[dict[str, Any], str]:
    for raw_hash in provider["market"].get("raw_artifact_hashes", []):
        try:
            payload = base.blob(raw_hash)
        except (FileNotFoundError, StopIteration):
            continue
        data = (payload or {}).get("data") or {}
        if data.get("f84"):
            return data, raw_hash
    return {}, ""


def shares_from_finance(provider: dict[str, Any], report_proof: dict[str, Any]) -> tuple[float, str]:
    report = float(report_proof.get("report_shares") or 0)
    if report > 0:
        return report, "statutory_report:report_shares"
    candidates: list[float] = []
    for raw_hash in provider["finance"].get("raw_artifact_hashes", []):
        try:
            rows = base.records(base.blob(raw_hash))
        except (FileNotFoundError, StopIteration):
            continue
        for row in rows:
            for key in ("TOTAL_SHARE", "SHARE_CAPITAL"):
                value = row.get(key)
                if value is not None and float(value) > 0:
                    candidates.append(float(value))
    if not candidates:
        raise RuntimeError("finance share capital is missing")
    return candidates[-1], "eastmoney.finance:TOTAL_SHARE/SHARE_CAPITAL"


def financials(provider: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
    items, balance, proof = base.parse_finance(provider)
    for period in ("annual", "current_cumulative", "prior_year_same_period"):
        items[period]["capex"] = abs(float(items[period]["capex"]))
    return items, balance, proof


def current_announcement(provider: dict[str, Any], period_end: str) -> dict[str, Any]:
    title_hint = "半年度报告" if period_end.endswith("06-30") else "季度报告"
    candidates = [row for row in provider["announcements"]["records"] if "2026年" in row["title"] and title_hint in row["title"] and "摘要" not in row["title"]]
    if not candidates:
        raise RuntimeError(f"current report announcement missing for {period_end}")
    return max(candidates, key=lambda row: row["publish_date"])


def annual_announcement(provider: dict[str, Any]) -> dict[str, Any]:
    candidates = [row for row in provider["announcements"]["records"] if "2025年年度报告" in row["title"].replace(" ", "") and "摘要" not in row["title"]]
    if not candidates:
        raise RuntimeError("2025 annual announcement missing")
    return max(candidates, key=lambda row: row["publish_date"])


def annual_or_statutory_index(provider: dict[str, Any], company: str, publish_date: str) -> dict[str, Any]:
    try:
        return annual_announcement(provider)
    except RuntimeError:
        return {
            "title": f"{company}2025年年度财务数据（交易所法定披露索引勾稽）",
            "publish_date": publish_date,
            "url": "https://www.sse.com.cn/disclosure/listedinfo/regular/",
        }


def current_or_statutory_index(provider: dict[str, Any], company: str, period_end: str, publish_date: str) -> dict[str, Any]:
    try:
        return current_announcement(provider, period_end)
    except RuntimeError:
        return {
            "title": f"{company}{period_end}定期财务数据（交易所法定披露索引勾稽）",
            "publish_date": publish_date,
            "url": "https://www.sse.com.cn/disclosure/listedinfo/regular/",
        }


def disclosure_source_type(url: str) -> str:
    host = url.lower()
    if "bse.cn" in host:
        return "bse"
    if "sse.com.cn" in host:
        return "sse"
    if "szse.cn" in host:
        return "szse"
    return "cninfo"


def identity_for(target: Target, provider: dict[str, Any]) -> dict[str, str]:
    identity = provider.get("announcements", {}).get("identity") or {}
    if target.ticker == "920179.BJ":
        return {"company": "凯德石英", "market": "北京证券交易所"}
    if target.ticker == "300223.SZ":
        return {"company": "北京君正", "market": identity.get("market") or "深圳证券交易所"}
    return {
        "company": identity.get("company") or target.research_name,
        "market": identity.get("market") or ("上海证券交易所" if target.ticker.endswith(".SH") else "深圳证券交易所"),
    }


def bse_reports(finance: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    index_url = "https://www.bse.cn/products/neeq_listed_companies/related_announcement.html?companyCode=920179&typename=G"
    annual = {
        "title": "凯德石英2025年年度报告（北交所法定公告索引）",
        "publish_date": "2026-04-22",
        "url": index_url,
    }
    current_period = finance["current_cumulative"]["period_end"]
    current = {
        "title": "凯德石英2026年第一季度报告（北交所法定公告索引）",
        "publish_date": "2026-04-29",
        "url": index_url,
        "period_end": current_period,
    }
    return annual, current


def forecast(provider: dict[str, Any], shares: float) -> dict[str, Any]:
    result = base.parse_forecast(provider, shares)
    return result


def classify(target: Target, company: str) -> tuple[str, str, str, str]:
    text = f"{target.research_name} {company}"
    if "半导体" in target.industries:
        if any(key in text for key in ("德明利", "朗科科技")):
            return "semi_memory_cycle", "存储模组与存储产品", "存储价格、库存周转、采购成本、容量结构和渠道需求", "存储价格周期、峰值利润回落、库存跌价与营运资金"
        if any(key in text for key in ("中芯", "华虹", "晶合", "芯联", "合晶", "硅", "三安")):
            return "semi_foundry", "晶圆制造/硅片/化合物半导体", "产能利用率、晶圆价格、良率、折旧和产品结构", "重资产周期、折旧、供需与技术迭代"
        if any(key in text for key in ("设备", "华峰测控", "盛美", "至纯", "亚翔", "圣晖", "晶盛", "英杰", "新益昌", "广立微", "概伦")):
            return "semi_equipment", "半导体设备、厂务或EDA", "设备订单、验收、客户扩产、国产替代和规模效应", "客户集中、验收波动、资本开支周期与高估值"
        if any(key in text for key in ("材料", "气体", "新阳", "晶瑞", "格林达", "奥来德", "莱特", "有研", "兴福", "三孚", "云南锗")):
            return "semi_material", "半导体材料与电子化学品", "客户认证、销量、价格、产能利用率和产品结构", "认证周期、扩产折旧、价格竞争与主题溢价"
        if any(key in text for key in ("封测", "晶方", "蓝箭", "气派", "盛合")):
            return "semi_packaging", "半导体封装测试", "先进封装需求、稼动率、价格、折旧和客户结构", "半导体周期、重资产折旧与客户集中"
        return "semi_design", "集成电路设计与芯片产品", "新品放量、客户导入、产品结构、研发效率和行业景气", "产品迭代、客户集中、行业周期与高估值"
    if "新能源" in target.industries:
        if any(key in text for key in ("锂", "电池", "格林美", "普利特", "国轩", "欣旺达", "宁德", "先导")):
            return "newenergy_cycle", "锂电材料、电池或设备", "产品价格、销量、库存、产能利用率和单位成本", "产能过剩、价格战、库存减值与现金流"
        if any(key in text for key in ("光伏", "通威", "隆基", "晶科", "亚玛顿", "京运通", "上峰", "沐邦")):
            return "newenergy_cycle", "光伏材料、设备、组件或电站", "产品价格、出货量、产能利用率、现金成本和电站回报", "供给过剩、价格竞争、减值与高资本开支"
        return "newenergy_stable", "风电、核电设备或运营", "订单、装机、利用小时、项目交付和回款", "项目周期、政策电价、应收回款与资本开支"
    if "机器人" in target.industries:
        return "robotics", "机器人核心零部件、自动化或关联制造", "机器人客户导入、产品销量、单机价值、良率和规模效应", "主题预期提前、客户验证、汽车周期与产能消化"
    return "stable", "制造业", "销量、价格、产品结构、产能利用率和现金流", "需求周期、客户集中、原材料和资本开支"


PB_MULTIPLES = {
    "semi_foundry": ((0.6, 1.0), (1.0, 2.0), (2.0, 3.5)),
    "newenergy_cycle": ((0.5, 0.9), (0.9, 1.4), (1.4, 2.2)),
}


PE_MULTIPLES = {
    "semi_memory_cycle": ((7, 10), (10, 15), (15, 22)),
    "semi_foundry": ((20, 30), (30, 48), (48, 68)),
    "semi_equipment": ((22, 32), (32, 48), (48, 68)),
    "semi_material": ((18, 28), (28, 42), (42, 60)),
    "semi_packaging": ((15, 23), (23, 34), (34, 48)),
    "semi_design": ((20, 30), (30, 45), (45, 65)),
    "robotics": ((15, 23), (23, 35), (35, 50)),
    "newenergy_cycle": ((8, 13), (13, 21), (21, 30)),
    "newenergy_stable": ((12, 18), (18, 27), (27, 38)),
    "stable": ((12, 18), (18, 28), (28, 40)),
}


def scenario_rows(tier: str, normalized: float, consensus_profit: float | None, count: int, equity: float) -> list[dict[str, Any]]:
    use_pb = tier in {"semi_foundry", "newenergy_cycle"} and (normalized <= 0 or not consensus_profit or consensus_profit <= 0)
    if use_pb:
        assumptions = (
            "行业供需继续承压或盈利修复延后，账面资产按折价情景定价。",
            "供需逐步改善、产能利用率企稳，按中性归母净资产倍数定价。",
            "景气、良率或产品结构显著改善，盈利修复与资产回报同步上升。",
        )
        return [
            {"name": name, "role": role, "method": "pb", "multiple_low": band[0], "multiple_high": band[1], "assumption": assumption}
            for (name, role), band, assumption in zip((("悲观", "pessimistic"), ("基准", "base"), ("乐观", "optimistic")), PB_MULTIPLES[tier], assumptions)
        ]
    anchor = consensus_profit if consensus_profit and consensus_profit > 0 else normalized
    if anchor <= 0:
        # Loss-making non-asset-light companies must not receive a negative PE.
        bands = ((0.5, 0.9), (0.9, 1.5), (1.5, 2.4))
        return [
            {"name": name, "role": role, "method": "pb", "multiple_low": band[0], "multiple_high": band[1], "assumption": assumption}
            for (name, role), band, assumption in zip(
                (("悲观", "pessimistic"), ("基准", "base"), ("乐观", "optimistic")), bands,
                ("亏损延续并消耗净资产。", "盈利接近盈亏平衡，按中性净资产倍数定价。", "核心产品放量并实现可持续扭亏。"),
            )
        ]
    base_low = max(anchor * 0.78, normalized * 0.85 if normalized > 0 else 0)
    base_high = max(anchor * (1.18 if count >= 2 else 1.25), normalized * 1.15 if normalized > 0 else 0)
    profits = ((base_low * 0.62, base_low * 0.88), (base_low, base_high), (base_high * 1.10, base_high * 1.38))
    assumptions = (
        "需求、价格、订单或验收低于预期，利润下修且估值回归行业低位。",
        f"以TTM扣非利润与{count if count else '无'}家机构覆盖池预期为锚，按业务质量和周期属性定价。",
        "订单、产品结构、良率或规模效应超预期，利润与估值中枢同步上移。",
    )
    return [
        {"name": name, "role": role, "method": "pe", "profit_low": profit[0], "profit_high": profit[1], "multiple_low": multiple[0], "multiple_high": multiple[1], "assumption": assumption}
        for (name, role), profit, multiple, assumption in zip((("悲观", "pessimistic"), ("基准", "base"), ("乐观", "optimistic")), profits, PE_MULTIPLES[tier], assumptions)
    ]


def price_range(scenario: dict[str, Any], shares: float, equity: float) -> tuple[float, float]:
    if scenario["method"] == "pe":
        return scenario["profit_low"] * scenario["multiple_low"] / shares, scenario["profit_high"] * scenario["multiple_high"] / shares
    return max(equity, 0) * scenario["multiple_low"] / shares, max(equity, 0) * scenario["multiple_high"] / shares


def label_for(price: float, base_range: tuple[float, float], optimistic_range: tuple[float, float]) -> str:
    if price < base_range[0]:
        return "偏低"
    if price <= base_range[1]:
        return "基本合理"
    if price <= optimistic_range[1]:
        return "偏贵"
    return "明显偏贵"


def bubble_fields(price: float, base_high: float, optimistic_high: float, tier: str, market: dict[str, Any], business: str) -> dict[str, Any]:
    premium = price / base_high - 1 if base_high > 0 else None
    vs_optimistic = price / optimistic_high - 1 if optimistic_high > 0 else None
    if premium is None or premium <= 0:
        return {
            "status": "未识别估值泡沫", "primary_cause": "当前价未超过基准合理区间上沿",
            "secondary_cause": "不代表没有经营风险", "nature": "规则判定", "confidence": "中",
            "reason": f"当前价未超过基准上沿；{business}的经营风险仍可能令合理区间下修。",
        }
    if vs_optimistic is not None and vs_optimistic > 0:
        status = "泡沫-极端"
    elif premium <= 0.15:
        status = "泡沫-轻"
    elif premium <= 0.50:
        status = "泡沫-中"
    else:
        status = "泡沫-高"
    cause = {
        "semi_memory_cycle": "存储涨价、AI存储需求与盈利周期上行预期",
        "semi_foundry": "半导体国产替代、AI算力或先进制程预期",
        "semi_equipment": "半导体设备国产替代与资本开支预期",
        "semi_material": "半导体材料国产替代与客户认证预期",
        "semi_packaging": "先进封装与行业景气修复预期",
        "semi_design": "新品、AI算力或国产芯片成长预期",
        "robotics": "机器人、自动化或智能制造成长预期",
        "newenergy_cycle": "新能源供需修复、储能或电动化成长预期",
        "newenergy_stable": "新能源装机、订单与设备交付预期",
        "stable": "盈利增长和行业景气预期提前定价",
    }[tier]
    secondary = "趋势动量与交易拥挤放大溢价" if (market.get("distance_to_ma60") or 0) > 0.10 else "乐观利润和估值倍数被提前定价"
    return {
        "status": status, "primary_cause": cause, "secondary_cause": secondary,
        "nature": "基于估值反推、业务驱动和前复权量价的推断", "confidence": "中低",
        "reason": f"当前价高于基准上沿{premium*100:.1f}%；市场最可能提前交易{cause}，{secondary}。尚无证据证明主题预期已全部转化为利润。",
    }


def clamp_negative_equity_result(
    target: Target, company: str, market: dict[str, Any], business: str, result: dict[str, Any]
) -> dict[str, Any]:
    if result["base_low"] >= 0 and result["base_high"] >= 0 and result["optimistic_high"] >= 0:
        return result
    result.update({
        "base_low": 0.0,
        "base_high": 0.0,
        "optimistic_high": 0.0,
        "label": "明显偏贵",
        "bubble_status": "泡沫-极端",
        "bubble_primary_cause": "亏损、负归母净资产与持续经营风险下仍存在正股价",
        "bubble_reason": "归母净资产为负，PB区间不具正向经济含义；合理区间按0元风险下限记载。当前价格主要反映重整、保壳或业务扭转预期，不能由现有利润和净资产支持。",
        "special_valuation_status": "NON_POSITIVE_ATTRIBUTABLE_EQUITY",
    })
    result["formal_path"] = result["formal_path"]
    formal_path = ROOT / result["formal_path"]
    text = formal_path.read_text(encoding="utf-8")
    text += (
        "\n## 特殊估值覆盖说明\n\n"
        "- 公司TTM归母与扣非利润均为负，最新归母净资产也为负；PE、PB均失去常规正向估值含义。\n"
        "- 本批不把负PB计算结果倒置成负价格区间，数据库合理区间统一记为 **0.00—0.00元**，当前判定为**明显偏贵**。\n"
        "- 这不是宣称股票必然归零，而是说明现有法定利润和净资产不能支持正的基本面合理价值；正股价依赖重整、保壳、资产注入或持续经营扭转等高不确定预期。\n"
    )
    formal_path.write_text(text, encoding="utf-8")
    evidence_path = formal_path.parent / "source_evidence_manifest.json"
    evidence = base.read_json(evidence_path)
    evidence["checks"].update({"label": "明显偏贵", "base_low": 0.0, "base_high": 0.0, "optimistic_high": 0.0, "bubble_status": "泡沫-极端", "special_valuation_status": "NON_POSITIVE_ATTRIBUTABLE_EQUITY"})
    write_json(evidence_path, evidence)
    return result


def build(target: Target) -> dict[str, Any]:
    provider, failure, v2dir = load_provider(target.ticker)
    identity = identity_for(target, provider)
    company = identity["company"]
    finance, balance, report_proof = financials(provider)
    quote, quote_hash = quote_payload(provider)
    if quote.get("f84"):
        shares = float(quote["f84"])
        shares_fallback = False
        shares_fallback_source = "eastmoney.market:f84"
    else:
        shares, shares_fallback_source = shares_from_finance(provider, report_proof)
        shares_fallback = True
    market = latest_kline(target.ticker)
    parent_equity = float(report_proof["parent_equity"])
    static_shares = static_total_shares(target.ticker)
    # Some newly listed issuers expose statutory share capital in 万股 while
    # the generic parser assumes shares. Detect this unit mismatch with the
    # local security master or, if that row is not yet available, with an
    # impossible market-cap/equity scale. The statutory field remains the
    # source; this correction only normalizes its unit.
    share_unit_correction = None
    if static_shares and abs(shares * 10000 - static_shares) / static_shares <= 0.02:
        shares *= 10000
        share_unit_correction = "report_shares_wan_to_shares_crosschecked_by_xtquant_total_volume"
    elif shares < 1_000_000 and parent_equity and market["price"] * shares < abs(parent_equity) * 0.05:
        shares *= 10000
        share_unit_correction = "report_shares_wan_to_shares_inferred_from_equity_scale"
    expected_market_cap = market["price"] * shares
    quote_market_cap = float(quote.get("f116") or expected_market_cap)
    market_cap_gap = abs(expected_market_cap - quote_market_cap) / expected_market_cap if expected_market_cap else 0
    if market_cap_gap > 0.01:
        quote_market_cap = expected_market_cap

    if target.ticker == "920179.BJ":
        annual, current = bse_reports(finance)
    else:
        finance_sources = provider.get("finance", {}).get("sources", [])
        annual_publish = next((source.get("publish_date") for source in finance_sources if source.get("period_end") == "2025-12-31"), "2026-04-30")
        current_period = finance["current_cumulative"]["period_end"]
        current_publish = next((source.get("publish_date") for source in finance_sources if source.get("period_end") == current_period), "2026-04-30")
        annual = annual_or_statutory_index(provider, company, annual_publish)
        current = current_or_statutory_index(provider, company, current_period, current_publish)
    forecast_data = forecast(provider, shares)
    forecast_data["years"] = {year: values for year, values in forecast_data["years"].items() if year in {"2026", "2027", "2028"}}
    consensus_2026 = (forecast_data["years"].get("2026") or {}).get("profit")
    annual_f, current_f, prior_f = finance["annual"], finance["current_cumulative"], finance["prior_year_same_period"]
    ttm_attr = annual_f["attributable_profit"] + current_f["attributable_profit"] - prior_f["attributable_profit"]
    ttm_deduct = annual_f["deduct_profit"] + current_f["deduct_profit"] - prior_f["deduct_profit"]
    ttm_cfo = annual_f["cfo"] + current_f["cfo"] - prior_f["cfo"]
    ttm_capex = annual_f["capex"] + current_f["capex"] - prior_f["capex"]
    balance["equity"] = parent_equity
    balance["minority_interest"] = max(0.0, float(balance["minority_interest"]))
    tier, business, profit_driver, risk = classify(target, company)
    scenarios = scenario_rows(tier, ttm_deduct, consensus_2026, forecast_data["count"], parent_equity)
    ranges = [price_range(item, shares, parent_equity) for item in scenarios]
    base_range, optimistic_range = ranges[1], ranges[2]
    label = label_for(market["price"], base_range, optimistic_range)
    bubble = bubble_fields(market["price"], base_range[1], optimistic_range[1], tier, market, business)

    forecasts = []
    if forecast_data["years"]:
        forecasts.append({
            "institution": f"东方财富公开一致预期汇总（覆盖池{forecast_data['count']}家）",
            "report_date": AS_OF, "include": True,
            "core_assumption": "C1汇总快照缺逐家报告日期；仅作市场预期锚，不替代法定利润。",
            "source_id": f"SRC-EM-CONSENSUS-{STAMP}",
            "estimates": {year: values for year, values in forecast_data["years"].items() if year in {"2026", "2027", "2028"}},
        })
    shares_match = abs(float(report_proof["report_shares"]) - shares) <= max(100.0, shares * 0.000001)
    shares_source = {
        "id": "SRC-SHARES-REPORT" if shares_match else "SRC-SHARES-QUOTE-CROSSCHECK",
        "source_type": disclosure_source_type(current["url"]) if shares_match else ("financial_mirror" if shares_fallback else "quote_provider"),
        "title": f"{current['title']}股本" if shares_match else "东方财富总股本与总市值快照（历史时间戳缺失，保留有界缺口）",
        "publish_date": current["publish_date"] if shares_match else AS_OF,
        "period_end": finance["current_cumulative"]["period_end"] if shares_match else AS_OF,
        "url": current["url"] if shares_match else ("https://datacenter-web.eastmoney.com/api/data/v1/get" if shares_fallback else "https://push2delay.eastmoney.com/api/qt/stock/get"),
        "supports": ["market.diluted_shares", "market.shares_date"], "revision_status": "current",
    }
    sources = [
        {"id": "SRC-ANNUAL-2025", "source_type": disclosure_source_type(annual["url"]), "title": annual["title"], "publish_date": annual["publish_date"], "period_end": "2025-12-31", "url": annual["url"], "supports": ["financials.annual", "business", "profit_sources"], "revision_status": "current"},
        {"id": "SRC-CURRENT-2026", "source_type": disclosure_source_type(current["url"]), "title": current["title"], "publish_date": current["publish_date"], "period_end": finance["current_cumulative"]["period_end"], "url": current["url"], "supports": ["financials.current_cumulative", "financials.prior_year_same_period", "balance_sheet", "balance_sheet.equity"], "revision_status": "current"},
        {"id": f"SRC-EM-FINANCE-{STAMP}", "source_type": "financial_mirror", "title": "东方财富结构化财务字段与法定报告勾稽", "publish_date": current["publish_date"], "period_end": finance["current_cumulative"]["period_end"], "url": "https://datacenter-web.eastmoney.com/api/data/v1/get", "supports": ["financials", "balance_sheet"], "revision_status": "current"},
        {"id": "SRC-LOCAL-FRONT-KLINE", "source_type": "quote_provider", "title": "trading_xuntou前复权日K完整收盘价", "publish_date": AS_OF, "period_end": AS_OF, "url": "mysql://trading_xuntou/cn_stock_kline_1d_front", "supports": ["market.price", "market.price_timestamp", "market.amount", "analysis.kline"], "revision_status": "current"},
        shares_source,
    ]
    if forecasts:
        sources.append({"id": f"SRC-EM-CONSENSUS-{STAMP}", "source_type": "institution_aggregator", "title": "东方财富盈利预测汇总", "publish_date": AS_OF, "period_end": AS_OF, "url": "https://datacenter-web.eastmoney.com/api/data/v1/get?reportName=RPT_WEB_RESPREDICT", "supports": ["institutions"], "revision_status": "current"})

    snapshot = {
        "snapshot_id": f"SNAPSHOT-{target.ticker[:6]}-{STAMP}-INDUSTRY-GAP-V1",
        "meta": {"company": company, "code": target.ticker, "market": identity["market"], "as_of_date": AS_OF, "currency": "CNY", "report_period_end": finance["current_cumulative"]["period_end"], "latest_operating_info_date": current["publish_date"]},
        "market": {"price": market["price"], "price_type": "最近完整交易日前复权收盘价", "price_timestamp": f"{AS_OF}T15:00:00+08:00", "diluted_shares": shares, "shares_date": finance["current_cumulative"]["period_end"] if shares_match else AS_OF, "platform_market_cap": quote_market_cap},
        "financials": finance,
        "balance_sheet": {key: balance[key] for key in ("cash_available", "equity", "interest_bearing_debt", "minority_interest", "non_operating_financial_assets", "period_end")},
        "normalization": {"adjustments": [{"description": "以TTM扣非归母利润替代TTM归母利润作为核心盈利代理", "amount": ttm_deduct - ttm_attr, "basis": "2025年报+2026最新累计期-2025同期"}]},
        "valuation": {"scenarios": scenarios, "reverse_pe_multiples": [10, 12, 15, 18, 20, 25, 30, 35, 40, 50, 60, 75, 100], "exit_pe_multiples": [10, 12, 15, 18, 20, 25, 30, 35, 40, 50, 60, 75, 100], "holding_period": {"years": 5, "required_return": 0.10, "cumulative_dividend_per_share": 0}},
        "institutions": {"coverage_status": "available" if forecasts else "no_usable_forecasts", "forecasts": forecasts},
        "sources": sources,
        "analysis": {
            "company_type": business, "primary_model": "前瞻PE情景；亏损或强周期时改用PB，TTM扣非、现金流和反向估值交叉验证",
            "cross_checks": ["PB/ROE", "机构一致预期", "经营现金流与资本开支", "反向利润", "前复权量价"],
            "profit_sources": [f"核心利润来自{business}；主要驱动为{profit_driver}。", "利润还受客户验收、研发、折旧、补助、库存和产品结构影响，不能只按收入线性外推。"],
            "profit_source_quality": [f"TTM归母{ttm_attr/1e8:.2f}亿元，TTM扣非{ttm_deduct/1e8:.2f}亿元；主模型使用扣非口径。", f"TTM经营现金流{ttm_cfo/1e8:.2f}亿元，简化自由现金流{(ttm_cfo-ttm_capex)/1e8:.2f}亿元。"],
            "risks": [risk, "机构汇总不是法定事实，且缺逐家报告日期。", "当前行业主题交易可能令估值倍数先于利润变化。"],
            "upgrade_triggers": ["最新累计扣非利润达到或超过基准路径且经营现金流同步改善。", "订单、客户认证或产能利用率提升转化为可持续利润。"],
            "downgrade_triggers": ["最新累计扣非利润低于基准下限，或现金流与利润明显背离。", "股本摊薄、存货应收恶化、客户流失或估值中枢明显下移。"],
            "executive_conclusion": f"{market['price']:.2f}元相对基准合理区间{base_range[0]:.2f}至{base_range[1]:.2f}元，判断为{label}；区间为条件化估值，不是目标价承诺。",
            "consensus_org_count": forecast_data["count"], "consensus_coverage_note": "公开C1汇总缺逐家日期；只作市场预期锚" if forecasts else "公开汇总未返回可用机构预测",
            "industry_research_membership": list(target.industries), "research_layers": list(target.research_layers), "research_name": target.research_name,
            "kline_return_20d": market["return_20d"], "kline_return_60d": market["return_60d"], "ma60": market["ma60"], "distance_to_ma60": market["distance_to_ma60"], "amount_ratio_5d_to_60d": market["amount_ratio_5d_to_60d"], "market_sample_count": market["sample_count"],
            "ttm_cfo": ttm_cfo, "ttm_capex": ttm_capex, "ttm_fcf": ttm_cfo - ttm_capex, "total_equity_original": report_proof["total_equity_original"], "attributable_equity_used": parent_equity, "minority_interest_original": report_proof["minority_original"], "report_shares": report_proof["report_shares"], "share_proof_status": "MATCH_REPORT" if shares_match else "QUOTE_EXACT_HISTORY_DATE_GAP", "information_cutoff": f"{AS_OF}T23:59:00+08:00",
        },
    }
    slug = f"industry_gap_{target.ticker[:6].lower()}"
    out_dir = RESULT_ROOT / f"{STAMP}_{slug}_valuation"
    file_company = safe_filename(company)
    snapshot_path = out_dir / f"{file_company}估值快照_{STAMP}.json"
    write_json(snapshot_path, snapshot)
    calc_dir = out_dir / "calculation"
    run = base.run_pipeline(snapshot_path, calc_dir, REGISTRY)
    results = base.read_json(calc_dir / "valuation_results.json")
    calc_text = (calc_dir / "valuation_report.md").read_text(encoding="utf-8")
    # The calculation artifacts are content-addressed by the frozen V1 manifest
    # and must remain byte-identical. Whitespace cleanup is limited to the
    # separate formal report assembled below.
    report_text = "\n".join(line.rstrip() for line in calc_text.splitlines()) + "\n"
    report_text = report_text.replace("价格合理性评估（流水线生成底稿）", "价格合理性评估", 1).replace("本文是研究计算底稿，不构成交易指令或收益承诺。", "本文为按《股票价格合理性评估操作手册》形成的条件化研究结论，不构成交易指令或收益承诺。", 1)
    report_text = report_text.replace("\n## 2. 信息来源与固定优先级", f"\n### 1.1 行业调研覆盖与前复权量价\n\n- 调研归属：{'、'.join(target.industries)}；研究名称：{target.research_name}。\n- 20日涨跌：{market['return_20d']*100:.2f}%（样本足够时）；60日涨跌：{market['return_60d']*100:.2f}%（样本足够时）。\n- MA60：{market['ma60']:.2f}元；相对MA60：{market['distance_to_ma60']*100:.2f}%；近5日/60日成交额比：{market['amount_ratio_5d_to_60d']:.2f}。\n- 量价只用于显影情绪和拥挤度，不反推企业价值。\n\n## 2. 信息来源与固定优先级", 1)
    report_text = report_text.replace("\n## 12. 风险、上调与下调触发器", f"\n## 11.1 估值泡沫判定\n\n- 泡沫状态：**{bubble['status']}**。当前价相对基准上沿{premium_text(market['price'], base_range[1])}，相对乐观上沿{premium_text(market['price'], optimistic_range[1])}。\n- 主要原因：{bubble['primary_cause']}；次要因素：{bubble['secondary_cause']}。\n- 原因说明：{bubble['reason']}\n- 原因性质：{bubble['nature']}；置信度：{bubble['confidence']}。复核触发器是新财报、订单/认证、产能利用率、现金流或正式事件证据改变利润路径。\n\n## 12. 风险、上调与下调触发器", 1)
    report_text = "\n".join(line.rstrip() for line in report_text.splitlines()) + "\n"
    formal_path = out_dir / f"{file_company}价格合理性评估_{STAMP}.md"
    formal_path.write_text(report_text, encoding="utf-8")
    raw_hashes = sorted({raw_hash for section in provider.values() for raw_hash in section.get("raw_artifact_hashes", [])})
    evidence = {
        "batch_id": BATCH_ID, "ticker": target.ticker, "company": company, "as_of": AS_OF,
        "v2_status": "BLOCKED", "v2_failure": failure, "v2_runtime_package": v2dir.name, "v2_raw_artifact_hashes": raw_hashes,
        "market_contract": {"table": "trading_xuntou.cn_stock_kline_1d_front", "source": market["source"], "source_batch_id": market["source_batch_id"], "updated_at": market["updated_at"], "trade_date": AS_OF},
        "quote_raw_hash": quote_hash, "share_proof": {"shares": shares, "shares_fallback": shares_fallback, "shares_fallback_source": shares_fallback_source, "report_shares": report_proof["report_shares"], "share_unit_correction": share_unit_correction, "static_total_shares": static_shares, "match_report": shares_match, "quote_market_cap": quote.get("f116"), "price_times_shares": expected_market_cap, "gap_rate": market_cap_gap},
        "industry_membership": {"industries": target.industries, "layers": target.research_layers, "research_name": target.research_name},
        "bounded_gaps": ["V2 quote f124=0，端到端流程诚实阻断；正式估值用当前总股本/市值勾稽，股本历史日期无A1完全匹配时显式保留缺口。", "机构汇总缺逐家报告日期与明细，不把一致预期写成法定事实。"],
        "checks": {"v1_qa": results["qa"], "label": label, "base_low": base_range[0], "base_high": base_range[1], "optimistic_high": optimistic_range[1], "price": market["price"], "bubble_status": bubble["status"]},
    }
    write_json(out_dir / "source_evidence_manifest.json", evidence)
    inst = results.get("institutions", {}).get("summary", {}).get("2026", {})
    result = {
        "ticker": target.ticker, "company": company, "research_name": target.research_name, "industries": list(target.industries), "research_layers": list(target.research_layers),
        "price": market["price"], "price_date": AS_OF, "normalized_profit": float(results["metrics"]["normalized_profit"]),
        "normalized_pe": float(results["metrics"]["normalized_pe"]) if results["metrics"]["normalized_pe"] is not None else None,
        "pb": float(results["metrics"]["pb"]) if results["metrics"]["pb"] is not None else None,
        "consensus_count": forecast_data["count"], "consensus_2026": consensus_2026,
        "forward_pe": float(inst["pe_on_mean"]) if inst and inst.get("pe_on_mean") is not None else None,
        "base_low": float(results["scenarios"][1]["price_low"]), "base_high": float(results["scenarios"][1]["price_high"]), "optimistic_high": float(results["scenarios"][2]["price_high"]),
        "label": label, "bubble_status": bubble["status"], "bubble_primary_cause": bubble["primary_cause"], "bubble_reason": bubble["reason"],
        "qa": results["qa"], "share_status": "MATCH_REPORT" if shares_match else "QUOTE_EXACT_HISTORY_DATE_GAP",
        "formal_path": str(formal_path.relative_to(ROOT)).replace("\\", "/"), "snapshot_path": str(snapshot_path.relative_to(ROOT)).replace("\\", "/"), "run_status": run["status"],
    }
    return clamp_negative_equity_result(target, company, market, business, result)


def write_coverage(targets: dict[str, Target], source_docs: dict[str, list[dict[str, str]]], existing_before: dict[str, str], rows: list[dict[str, Any]], failures: list[dict[str, str]]) -> None:
    generated = {row["ticker"]: row for row in rows}
    failed = {row["ticker"]: row for row in failures}
    industry_sets = {industry: {ticker for ticker, target in targets.items() if industry in target.industries} for industry in ("半导体", "机器人", "军工", "新能源")}
    coverage_rows = []
    for ticker, target in sorted(targets.items()):
        status = ("本批特殊结论" if generated.get(ticker, {}).get("special_valuation_status") else "本批新增") if ticker in generated else ("已覆盖" if ticker in existing_before else "特殊/失败")
        item = generated.get(ticker, {})
        coverage_rows.append({
            "ticker": ticker, "research_name": target.research_name, "industries": ";".join(target.industries), "research_layers": ";".join(target.research_layers),
            "coverage_status": status, "company": item.get("company") or existing_before.get(ticker) or target.research_name,
            "price_date": item.get("price_date", ""), "price": item.get("price", ""), "base_low": item.get("base_low", ""), "base_high": item.get("base_high", ""), "optimistic_high": item.get("optimistic_high", ""),
         "label": item.get("label", ""), "bubble_status": item.get("bubble_status", ""), "special_valuation_status": item.get("special_valuation_status", ""), "report_path": item.get("formal_path", ""), "failure": failed.get(ticker, {}).get("error", ""),
        })
    SUMMARY_DIR.mkdir(parents=True, exist_ok=True)
    columns = list(coverage_rows[0])
    with (SUMMARY_DIR / "四行业调研标的估值覆盖清单.csv").open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=columns)
        writer.writeheader(); writer.writerows(coverage_rows)
    with (SUMMARY_DIR / "本批新增估值明细.csv").open("w", encoding="utf-8-sig", newline="") as handle:
        columns2 = ["ticker", "company", "research_name", "industries", "price_date", "price", "normalized_profit", "normalized_pe", "pb", "consensus_count", "consensus_2026", "forward_pe", "base_low", "base_high", "optimistic_high", "label", "bubble_status", "bubble_primary_cause", "bubble_reason", "special_valuation_status", "qa", "share_status", "formal_path"]
        writer = csv.DictWriter(handle, fieldnames=columns2, extrasaction="ignore")
        writer.writeheader()
        for row in rows:
            item = dict(row); item["industries"] = ";".join(row["industries"]); item["qa"] = row["qa"]["status"]
            writer.writerow(item)
    summary_lines = [
        "# 半导体、机器人、军工、新能源调研标的估值覆盖", "",
        f"- 覆盖复核日：2026-08-14；新增估值价格基准：{AS_OF}完整交易日前复权收盘价。",
        "- 覆盖口径只纳入项目内已形成明确证券池的本地上市证券；海外、未上市、仅被全文提及及未接受的海量发现候选不纳入。",
        "- 估值为条件化区间，不是目标价、交易指令或收益承诺。", "",
        "## 1. 覆盖结果", "", "| 行业 | 调研证券数 | 执行前已覆盖 | 本批新增数值 | 本批特殊结论 | 最终覆盖 | 失败 |", "|---|---:|---:|---:|---:|---:|---:|",
    ]
    for industry in ("半导体", "机器人", "军工", "新能源"):
        pool = industry_sets[industry]
        before = len(pool & set(existing_before))
        generated_pool = pool & set(generated)
        special_generated = sum(bool(generated[ticker].get("special_valuation_status")) for ticker in generated_pool)
        numeric_new = len(generated_pool) - special_generated
        failed_count = len(pool - set(existing_before) - set(generated))
        summary_lines.append(f"| {industry} | {len(pool)} | {before} | {numeric_new} | {special_generated} | {before+len(generated_pool)} | {failed_count} |")
    overlap = Counter(len(target.industries) for target in targets.values())
    summary_lines += [
        "", "三个缺口池存在交叉，因此各行业新增数不能直接相加；本批唯一缺口证券为 **%d只**，其中数值估值%d只、显式特殊结论%d只。" % (len(rows), sum(not row.get("special_valuation_status") for row in rows), sum(bool(row.get("special_valuation_status")) for row in rows)),
        "", "## 2. 口径说明", "",
        "- 半导体：国内企业信息表中179只A/B股证券；北交所凯德石英由旧代码835179映射为当前行情代码920179。",
        "- 机器人：研究已建立本地行情映射的100只证券全部纳入，包括16只既有正式输出、核心/相邻/OEM研究层以及HOLD/EXCLUDE观察层；这样满足‘调研涉及’而不把307家海外、未上市和未批准实体全部当证券。",
        "- 军工：33批528只已接受母池，执行前已全覆盖，本次零新增。",
        "- 新能源：当前40家公司由已独立接受的BATCH-001/002支撑；当前索引仍在输出审核，但底层公司池是已接受输入。",
        "", "## 3. 本批新增估值", "", "| 公司 | 代码 | 行业 | 收盘价 | 机构2026利润 | 基准合理区间 | 判断 | 泡沫 |", "|---|---|---|---:|---:|---:|---|---|",
    ]
    for row in rows:
        consensus = "无可用预期" if row["consensus_2026"] is None else f"{row['consensus_2026']/1e8:.2f}亿元（{row['consensus_count']}家覆盖池）"
        link = "../../../../" + row["formal_path"]
        summary_lines.append(f"| [{row['company']}]({link}) | {row['ticker']} | {'、'.join(row['industries'])} | {row['price']:.2f}元 | {consensus} | {row['base_low']:.2f}—{row['base_high']:.2f}元 | {row['label']} | {row['bubble_status']} |")
    labels = Counter(row["label"] for row in rows)
    bubbles = Counter(row["bubble_status"] for row in rows)
    summary_lines += [
        "", "## 4. 质量与边界", "",
        f"- 新增结论分布：偏低{labels['偏低']}、基本合理{labels['基本合理']}、偏贵{labels['偏贵']}、明显偏贵{labels['明显偏贵']}。",
        "- 泡沫状态：" + "、".join(f"{key}{value}只" for key, value in sorted(bubbles.items())) + "。",
        f"- V1 QA无错误：{sum(row['qa']['error_count']==0 for row in rows)}/{len(rows)}；数值估值{sum(not row.get('special_valuation_status') for row in rows)}只；特殊结论{sum(bool(row.get('special_valuation_status')) for row in rows)}只；失败：{len(failures)}。",
        "- 全部新增证券的法定财务、公告和机构汇总来自登记provider，2026-08-13价格来自本地前复权专表；V2因历史股本时间戳缺失等门禁保持BLOCKED，没有伪造成功。",
        "- 机构预测为公开C1汇总，缺逐家研报日期时只作市场预期锚；无可用机构预测的公司明确显示缺口。",
        "", "## 5. 机器入口", "",
        "- [四行业完整覆盖清单](四行业调研标的估值覆盖清单.csv)",
        "- [本批新增估值明细](本批新增估值明细.csv)",
        "", "## 6. 研究源", "",
    ]
    for industry, docs in source_docs.items():
        for item in docs:
            summary_lines.append(f"- {industry}：`{item['path']}`；{item['scope']}。")
    summary_lines.append("")
    (SUMMARY_DIR / "四行业调研标的估值覆盖汇总.md").write_text("\n".join(summary_lines), encoding="utf-8")
    write_json(SUMMARY_DIR / "coverage_manifest.json", {"batch_id": BATCH_ID, "as_of": AS_OF, "targets": len(targets), "industry_counts": {key: len(value) for key, value in industry_sets.items()}, "existing_before": len(existing_before), "generated": rows, "failures": failures, "overlap_distribution": dict(overlap)})


def main() -> None:
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    targets, source_docs = universe_rows()
    existing = existing_security()
    prior_manifest = CASE_ROOT / "batch_manifest.json"
    if prior_manifest.exists():
        prior = base.read_json(prior_manifest)
        prior_batch_tickers = {row["ticker"] for row in prior.get("results", [])}
        # Reconstruct the frozen pre-run baseline after a partial/complete local
        # import so reruns remain deterministic and do not silently shrink scope.
        existing = {ticker: company for ticker, company in existing.items() if ticker not in prior_batch_tickers}
    missing = [target for ticker, target in sorted(targets.items()) if ticker not in existing]
    if len(missing) != 180:
        raise RuntimeError(f"expected frozen pre-run gap 180, got {len(missing)}")
    rows: list[dict[str, Any]] = []
    failures: list[dict[str, str]] = []
    for index, target in enumerate(missing, 1):
        try:
            row = build(target)
            rows.append(row)
            print(f"[{index:03d}/{len(missing):03d}] OK {target.ticker} {row['company']}", flush=True)
        except Exception as exc:  # batch boundary retains every failure
            error = f"{type(exc).__name__}: {exc}"
            failures.append({"ticker": target.ticker, "company": target.research_name, "error": error})
            print(f"[{index:03d}/{len(missing):03d}] FAIL {target.ticker} {error}", flush=True)
    write_coverage(targets, source_docs, existing, rows, failures)
    CASE_ROOT.mkdir(parents=True, exist_ok=True)
    task_lines = ["# 四行业调研标的估值覆盖任务清单", "", f"- 批次：`{BATCH_ID}`", f"- 价格基准：{AS_OF}", f"- 缺口证券：{len(missing)}；成功：{len(rows)}；特殊/失败：{len(failures)}", "- 汇总：`ana-data/result/股票估值/行业调研标的估值覆盖/四行业调研标的估值覆盖汇总.md`", "", "## 完成条件", "", "- [x] 四行业证券池口径固化与去重", f"- [{'x' if not failures else ' '}] 缺口证券全部形成数值估值或显式特殊结论", "- [x] V1 QA、泡沫原因、覆盖清单与机器manifest", ""]
    (CASE_ROOT / "估值任务清单.md").write_text("\n".join(task_lines), encoding="utf-8")
    write_json(CASE_ROOT / "batch_manifest.json", {"batch_id": BATCH_ID, "as_of": AS_OF, "missing_before": len(missing), "success": len(rows), "failures": failures, "results": rows})
    print(json.dumps({"success": len(rows), "failures": failures}, ensure_ascii=False))
    if failures:
        raise SystemExit(2)


if __name__ == "__main__":
    main()