"""Build a same-day peer-valuation comparison for the current valuation universe. This is the analyst-side one-off/reference implementation. It does not change valuation ranges or MySQL facts. The project ledger implementation owns the long-lived database integration. """ from __future__ import annotations import argparse import csv import hashlib import json import math import os import statistics import tempfile from collections import Counter from pathlib import Path from typing import Iterable PEER_FIELDS = [ "industry", "peer_group_id", "peer_group_name", "peer_group_basis", "peer_tickers", "peer_metric", "target_multiple", "peer_count", "peer_mean", "peer_median", "peer_p25", "peer_p75", "peer_premium_pct", "peer_vs_mean_pct", "peer_raw_label", "peer_adjusted_label", "peer_adjustment_reason", "peer_confidence", "peer_as_of", "peer_gap_reason", ] GROUP_RULES: list[tuple[str, str, tuple[str, ...]]] = [ ("SOLAR", "光伏材料、设备、组件与电站", ("光伏", "光伏组件", "太阳能电池", "太阳能发电", "逆变器", "光伏电站")), ("LITHIUM_BATTERY", "锂电材料、电池与设备", ("锂电", "锂离子", "电池材料", "动力电池", "负极", "正极材料")), ("SEMICONDUCTOR_EQUIPMENT", "半导体设备、厂务与EDA", ("半导体设备", "半导体测试", "半导体及激光设备", "厂务", "eda")), ("SEMICONDUCTOR_MATERIAL", "半导体材料与电子化学品", ("半导体材料", "电子化学", "电子特种气体", "光刻胶", "硅片", "抛光材料")), ("WAFER_FOUNDRY", "晶圆制造、硅片与化合物半导体", ("晶圆制造", "化合物半导体", "晶圆代工", "硅片")), ("SEMICONDUCTOR_PACKAGING", "半导体封装测试", ("封装测试", "封测")), ("MEMORY", "存储模组与存储产品", ("存储模组", "存储控制器", "存储产品")), ("IC_DESIGN", "集成电路设计与芯片产品", ("集成电路设计", "芯片产品", "芯片设计", "soc")), ("DEFENSE", "军工电子、航空航天与特种装备", ("军工", "航空航天", "航空发动机", "航空器材", "航材", "军需", "特种装备", "雷达", "卫星通信", "高精度卫星导航", "武器系统", "民爆", "海洋防务", "光电防务", "防务装备", "军用", "军事", "导弹", "弹药", "红外", "低空经济", "固态微波", "高波段")), ("AI_OPTICAL", "AI算力、服务器、光通信与散热", ("ai算力", "服务器", "光通信", "cpo", "散热基础设施", "通信模组", "企业通信", "无线通信", "物联网", "光芯片", "光纤器件", "光纤环", "光测试仪器", "精密光学元组件", "光电元器件")), ("PCB_COMPONENT", "PCB、覆铜板与电子元件", ("pcb", "覆铜板", "电子铜箔", "电子元件", "被动元件", "印制电路")), ("ROBOT_AUTOMATION", "机器人、自动化与工业控制", ("机器人", "自动化", "工业控制", "运动控制", "伺服")), ("STORAGE_POWER", "储能与电力系统", ("储能", "新型电力系统", "电网", "电力设备")), ("WIND_NUCLEAR", "风电与核电", ("风电", "核电")), ("FERTILIZER", "氮磷钾肥、复合肥与化工联产", ("化肥", "复合肥", "磷肥", "钾肥", "氮磷钾", "肥料")), ("AGRICULTURE", "农业、农资、农药与养殖", ("农业", "农资", "农药", "养殖", "种业", "食品原料")), ("HEALTHCARE", "医药、医疗器械与生命科学服务", ("医药", "创新药", "疫苗", "医疗器械", "生命科学", "生物制品", "poct", "体外诊断", "诊断试剂", "过敏原", "脱敏治疗", "细胞培养")), ("SOFTWARE", "软件、数据服务、安全与信创", ("软件", "数据服务", "安全", "信创", "ai应用", "网络游戏", "军事仿真", "嵌入式系统测试", "cax", "cad", "cae", "cam")), ("TRADITIONAL_ENERGY", "煤炭、油气与传统能源", ("煤炭", "煤化工", "油气", "海洋石油", "传统能源")), ("METALS_RESOURCES", "有色金属、矿产与资源品", ("有色金属", "资源开采", "矿产", "锂资源", "镍钴", "稀土", "黄金", "铜矿", "铝加工", "铝合金", "稀有金属", "不锈钢", "合金管", "金属材料")), ("CHEMICAL", "化工、氟化工与化学材料", ("化工", "氟化工", "化学材料", "化学品", "原药", "制剂及中间体", "有机新材料", "催化剂", "分子筛", "高分子", "pvc", "烧碱")), ("AUTO", "汽车、零部件与智能驾驶", ("汽车", "汽车零部件", "智能驾驶", "车载", "客车", "轮胎", "摩托车", "全地形车", "车轮")), ("CONSTRUCTION", "地产、建筑与基础设施", ("地产", "建筑", "基础设施")), ("CONSUMER", "消费、商业服务与文旅", ("消费", "商业服务", "商务服务", "文旅", "旅游", "影视", "珠宝", "金银", "广告", "营销", "品牌传播")), ("TRANSPORT_UTILITY", "交通运输、物流与公用运营", ("交通运输", "物流", "公用运营", "港口", "机场")), ("FINANCIAL", "银行、保险与其他金融", ("银行", "保险", "证券", "金融")), ("INDUSTRIAL", "通用制造、机械设备与工业材料", ("通用制造", "机械设备", "工业材料", "仪器仪表", "检测", "矿用车", "高空作业平台", "工程机械", "高端装备", "重工装备", "轴承", "机械密封", "轨道交通", "刀具", "包装设备", "水泵", "减速机", "试验设备", "铸件", "管材", "密封")), ] ASSET_GROUPS = {"METALS_RESOURCES", "TRADITIONAL_ENERGY", "FERTILIZER", "FINANCIAL"} PS_FRIENDLY_GROUPS = { "SEMICONDUCTOR_EQUIPMENT", "SEMICONDUCTOR_MATERIAL", "WAFER_FOUNDRY", "SEMICONDUCTOR_PACKAGING", "MEMORY", "IC_DESIGN", "AI_OPTICAL", "SOFTWARE", "ROBOT_AUTOMATION", "HEALTHCARE", } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Build peer valuation comparisons for all current targets.") parser.add_argument("--project-root", type=Path, default=Path(".")) parser.add_argument( "--valuation-csv", type=Path, default=Path("ana-data/result/股票估值/全量中报重估/全部已评估公司最新估值.csv"), ) parser.add_argument( "--latest-csv", type=Path, default=Path("ana-data/result/股票估值/估值台账/latest.csv"), ) parser.add_argument( "--gaps-csv", type=Path, default=Path("ana-data/result/股票估值/估值台账/latest_gaps.csv"), ) parser.add_argument( "--official-industry-csv", type=Path, default=Path("ana-data/cases/农业案例/extracted/candidate_disposition.csv"), ) parser.add_argument( "--semiconductor-map-csv", type=Path, default=Path("ana-data/cases/半导体案例/核心文档/企业子行业映射.csv"), ) parser.add_argument( "--output-dir", type=Path, default=Path("ana-data/result/股票估值/同行估值比较"), ) parser.add_argument("--generated-at", help="同行输入生成时间；默认使用价格日18:00+08:00") parser.add_argument("--available-at", help="同行输入可用时间；默认等于generated-at") parser.add_argument("--effective-from", help="同行组生效日；默认等于价格日") parser.add_argument("--write", action="store_true") return parser.parse_args() def under(root: Path, value: Path) -> Path: path = value if value.is_absolute() else root / value return path.resolve() 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) return list(reader.fieldnames or []), list(reader) def number(value: object) -> float | None: try: result = float(str(value).strip()) except (TypeError, ValueError): return None return result if math.isfinite(result) else None def fmt(value: float | None, digits: int = 4) -> str: if value is None or not math.isfinite(value): return "" return f"{value:.{digits}f}" def quantile(values: list[float], q: float) -> float: ordered = sorted(values) if len(ordered) == 1: return ordered[0] index = (len(ordered) - 1) * q low = math.floor(index) high = math.ceil(index) if low == high: return ordered[low] return ordered[low] + (ordered[high] - ordered[low]) * (index - low) def classify_text(text: str) -> tuple[str, str] | None: lowered = text.lower() for group_id, name, keys in GROUP_RULES: if any(key.lower() in lowered for key in keys): return group_id, name return None def classify_group( row: dict[str, str], official_industry: str, semiconductor_override: tuple[str, str, str] | None, business_text: str, ) -> tuple[str, str, str]: if semiconductor_override is not None: return semiconductor_override current = (row.get("company_type") or "").strip() old = (row.get("old_company_type") or "").strip() official_direct = [ ("金融", "FINANCIAL", "银行、保险与其他金融"), ("建筑", "CONSTRUCTION", "地产、建筑与基础设施"), ("房地产", "CONSTRUCTION", "地产、建筑与基础设施"), ("交通运输", "TRANSPORT_UTILITY", "交通运输、物流与公用运营"), ] for key, group_id, name in official_direct: if key in official_industry: return group_id, name, f"交易所正式行业：{official_industry}" business_match = classify_text(business_text) if business_match: excerpt = business_text[:120] + ("…" if len(business_text) > 120 else "") return business_match[0], business_match[1], f"正式估值快照主营映射：{excerpt}" official_fallback = [ ("农、林、牧、渔", "AGRICULTURE", "农业、农资、农药与养殖"), ("信息传输、软件", "SOFTWARE", "软件、数据服务、安全与信创"), ("卫生和社会工作", "HEALTHCARE", "医药、医疗器械与生命科学服务"), ("采矿", "METALS_RESOURCES", "有色金属、矿产与资源品"), ("电力、热力", "TRANSPORT_UTILITY", "交通运输、物流与公用运营"), ("批发和零售", "CONSUMER", "消费、商业服务与文旅"), ("商务服务", "CONSUMER", "消费、商业服务与文旅"), ("住宿和餐饮", "CONSUMER", "消费、商业服务与文旅"), ("文化、体育", "CONSUMER", "消费、商业服务与文旅"), ] for key, group_id, name in official_fallback: if key in official_industry: return group_id, name, f"交易所正式行业兜底：{official_industry}" for source_name, text in (("原主营类型", old), ("当前主营类型", current)): match = classify_text(text) if match: return match[0], match[1], f"{source_name}映射：{text}" basis = old or current or official_industry or "主营类型缺失" return "OTHER_DIVERSIFIED", "其他或多元化公司", f"宽口径兜底：{basis}" def official_ticker(security_id: str) -> str | None: try: exchange, code = security_id.split(":", 1) except ValueError: return None suffix = {"SSE": "SH", "SZSE": "SZ", "BSE": "BJ"}.get(exchange) return f"{code}.{suffix}" if suffix else None def normalized_company_name(value: str) -> str: name = value.strip().replace("*", "") if name.upper().startswith("ST"): name = name[2:] for suffix in ("股份有限公司", "有限责任公司", "-U", "-W"): if name.endswith(suffix): name = name[: -len(suffix)] return name.strip() def semiconductor_group(subindustry_id: str) -> tuple[str, str] | None: if subindustry_id in {"01", "04", "06"}: return "SEMICONDUCTOR_EQUIPMENT", "半导体设备、厂务与EDA" if subindustry_id == "05": return "SEMICONDUCTOR_MATERIAL", "半导体材料与电子化学品" if subindustry_id in {"03", "14"}: return "WAFER_FOUNDRY", "晶圆制造、硅片与化合物半导体" if subindustry_id == "07": return "SEMICONDUCTOR_PACKAGING", "半导体封装测试" if subindustry_id == "08": return "MEMORY", "存储模组与存储产品" if subindustry_id in {"02", "09", "10", "11", "12", "13"}: return "IC_DESIGN", "集成电路设计与芯片产品" if subindustry_id == "A01": return "AI_OPTICAL", "AI算力、服务器、光通信与散热" return None def business_text_from_snapshot(root: Path, row: dict[str, str]) -> str: snapshot_value = (row.get("snapshot_path") or "").strip() if not snapshot_value: return "" snapshot_path = under(root, Path(snapshot_value)) if not snapshot_path.is_file(): return "" try: payload = json.loads(snapshot_path.read_text(encoding="utf-8-sig")) except (OSError, UnicodeError, json.JSONDecodeError): return "" analysis = payload.get("analysis") if isinstance(payload, dict) else None identity = analysis.get("business_identity") if isinstance(analysis, dict) else None if not isinstance(identity, dict): return "" value = identity.get("main_business") return str(value).strip() if value else "" def scaled_multiple(row: dict[str, str], field: str, current_close: float) -> float | None: old_close = number(row.get("close")) old_value = number(row.get(field)) if not old_close or old_close <= 0 or old_value is None or old_value <= 0: return None value = old_value * current_close / old_close return value if math.isfinite(value) else None def metric_value(row: dict[str, str], close: float, metric: str) -> float | None: shares = number(row.get("shares")) profit = number(row.get("normalized_profit")) if metric.startswith("FORWARD_PE_"): expected_year = metric.removeprefix("FORWARD_PE_") consensus_year = (row.get("consensus_year") or "").strip() consensus_count = number(row.get("consensus_count")) consensus_profit = number(row.get("consensus_profit")) if ( consensus_year != expected_year or consensus_count is None or consensus_count < 3 or not shares or shares <= 0 or not consensus_profit or consensus_profit <= 0 ): return None value = close * shares / consensus_profit elif metric == "NORMALIZED_PE": if not shares or shares <= 0 or not profit or profit <= 0: return None value = close * shares / profit elif metric == "PB": return scaled_multiple(row, "pb", close) elif metric == "PS": return scaled_multiple(row, "ps", close) else: return None return value if math.isfinite(value) and value > 0 else None def metric_denominator(row: dict[str, str], metric: str) -> float | None: """Return the major-currency denominator used by the MySQL peer ledger.""" shares = number(row.get("shares")) old_close = number(row.get("close")) if not shares or shares <= 0: return None if metric.startswith("FORWARD_PE_"): expected_year = metric.removeprefix("FORWARD_PE_") if (row.get("consensus_year") or "").strip() != expected_year: return None value = number(row.get("consensus_profit")) elif metric == "NORMALIZED_PE": value = number(row.get("normalized_profit")) elif metric == "PB": multiple = number(row.get("pb")) value = old_close * shares / multiple if old_close and multiple and multiple > 0 else None elif metric == "PS": multiple = number(row.get("ps")) value = old_close * shares / multiple if old_close and multiple and multiple > 0 else None else: value = None return value if value is not None and math.isfinite(value) and value > 0 else None def decimal_number(value: float) -> float | int: """Limit strict peer-input JSON numbers to at most six decimal places.""" rounded = round(value, 6) return int(rounded) if float(rounded).is_integer() else rounded def choose_metric(row: dict[str, str], close: float, group_id: str) -> tuple[str, float | None, str]: if group_id == "OTHER_DIVERSIFIED": return "NOT_APPLICABLE", None, "主营分类过宽或多元化，不能把兜底组伪装成可比行业" pe = metric_value(row, close, "NORMALIZED_PE") pb = metric_value(row, close, "PB") ps = metric_value(row, close, "PS") if group_id in ASSET_GROUPS and pb is not None and 0.1 <= pb <= 20: return "PB", pb, "资源/周期或资产型公司优先PB，避免峰值利润机械套PE" consensus_year = (row.get("consensus_year") or "").strip() if consensus_year: forward_metric = f"FORWARD_PE_{consensus_year}" forward_pe = metric_value(row, close, forward_metric) if forward_pe is not None and 2 <= forward_pe <= 200: return forward_metric, forward_pe, "至少3家机构且年度一致，优先使用同年度前瞻PE" if pe is not None and 2 <= pe <= 200: return "NORMALIZED_PE", pe, "正利润且正常化PE处于可解释范围" if group_id in PS_FRIENDLY_GROUPS and ps is not None and 0.05 <= ps <= 50: return "PS", ps, "亏损或PE极端敏感，切换PS" if pb is not None and 0.1 <= pb <= 20: return "PB", pb, "PE不可用，使用可复算PB交叉比较" if ps is not None and 0.05 <= ps <= 50: return "PS", ps, "PE/PB不可用，使用PS交叉比较" return "NOT_APPLICABLE", None, "没有可用且有限的同口径倍数" def valid_peer_metric(metric: str, value: float | None) -> bool: if value is None: return False if metric == "NORMALIZED_PE" or metric.startswith("FORWARD_PE_"): return 2 <= value <= 200 if metric == "PB": return 0.1 <= value <= 20 if metric == "PS": return 0.05 <= value <= 50 return False def label(premium: float) -> str: if premium <= -0.30: return "显著低于同行" if premium <= -0.10: return "低于同行" if premium < 0.10: return "接近同行" if premium < 0.30: return "高于同行" return "显著高于同行" def refresh_absolute_price_fields(row: dict[str, str], close: float) -> None: base_low = number(row.get("base_low")) base_high = number(row.get("base_high")) optimistic_high = number(row.get("optimistic_high")) if not base_low or not base_high or base_low <= 0 or base_high < base_low: return row["distance_to_base_low"] = fmt(close / base_low - 1, 8) row["distance_to_base_high"] = fmt(close / base_high - 1, 8) row["premium_to_base_high"] = fmt(close / base_high - 1, 8) if optimistic_high and optimistic_high > 0: row["vs_optimistic_high"] = fmt(close / optimistic_high - 1, 8) midpoint = (base_low + base_high) / 2 if close < base_low: row["label"] = "偏低" row["valuation_position_pct"] = fmt(close / base_low - 1, 8) elif close <= base_high: row["label"] = "基本合理" row["valuation_position_pct"] = fmt(close / midpoint - 1, 8) elif optimistic_high and close <= optimistic_high: row["label"] = "偏贵" row["valuation_position_pct"] = fmt(close / base_high - 1, 8) else: row["label"] = "明显偏贵" row["valuation_position_pct"] = fmt(close / base_high - 1, 8) premium = close / base_high - 1 if premium <= 0: row["bubble_status"] = "未识别估值泡沫" row["bubble_reason"] = "当前收盘价未高于基准合理区间上沿，按统一规则不认定估值泡沫。" elif optimistic_high and close > optimistic_high: row["bubble_status"] = "泡沫-极端（超过乐观上沿）" row["bubble_reason"] = f"当前收盘价高于基准上沿{premium:.1%}且超过乐观上沿，需复核额外叙事和基本面兑现条件。" elif premium <= 0.15: row["bubble_status"] = "泡沫-轻" row["bubble_reason"] = f"当前收盘价高于基准上沿{premium:.1%}，处于估值误差与乐观预期交界。" elif premium <= 0.50: row["bubble_status"] = "泡沫-中" row["bubble_reason"] = f"当前收盘价高于基准上沿{premium:.1%}，已明显提前支付乐观增长。" else: row["bubble_status"] = "泡沫-高" row["bubble_reason"] = f"当前收盘价高于基准上沿{premium:.1%}，大幅提前资本化乐观情景。" def confidence(peer_count: int, group_id: str, basis: str, spread: float | None) -> str: if peer_count < 3: return "不可用" broad = basis.startswith("宽口径兜底") if group_id in {"OTHER_DIVERSIFIED", "INDUSTRIAL", "CONSUMER"}: return "低" if group_id in {"AI_OPTICAL", "DEFENSE", "ROBOT_AUTOMATION", "AUTO", "HEALTHCARE", "SOFTWARE"}: return "中" if peer_count >= 5 else "低" if peer_count >= 10 and not broad and spread is not None and spread <= 1.0: return "较高" if peer_count >= 5 and not broad: return "中" return "低" def quality_interpretation( raw_label: str, target: dict[str, object], peers: list[dict[str, object]], ) -> tuple[str, str]: if target["metric"] != "NORMALIZED_PE" and not str(target["metric"]).startswith("FORWARD_PE_"): return raw_label, "缺少统一的增长、ROE与现金流横截面，本轮不作机械质量调整" target_pb = target.get("pb") target_pe = target.get("value") if not isinstance(target_pb, float) or not isinstance(target_pe, float) or target_pe <= 0: return raw_label, "缺少可比ROE代理，本轮不作机械质量调整" target_roe = target_pb / target_pe peer_roes: list[float] = [] for peer in peers: peer_pb = peer.get("pb") peer_pe = peer.get("value") if isinstance(peer_pb, float) and isinstance(peer_pe, float) and peer_pe > 0: peer_roes.append(peer_pb / peer_pe) if len(peer_roes) < 3: return raw_label, "可比ROE代理少于3个，本轮不作机械质量调整" median_roe = statistics.median(peer_roes) if median_roe <= 0: return raw_label, "同行ROE代理不可解释，本轮不作机械质量调整" gap = target_roe / median_roe - 1 if "低于" in raw_label and gap <= -0.20: return "低倍数但质量偏弱", f"正常化ROE代理较同行中位数低{abs(gap):.1%}，折价可能有基本面原因" if "低于" in raw_label and gap >= 0.20: return "低倍数且质量较强", f"正常化ROE代理较同行中位数高{gap:.1%}，同行相对估值更有利" if "高于" in raw_label and gap >= 0.20: return "高倍数但部分有质量支撑", f"正常化ROE代理较同行中位数高{gap:.1%}，部分溢价有质量支撑" if "高于" in raw_label and gap <= -0.20: return "高倍数且质量偏弱", f"正常化ROE代理较同行中位数低{abs(gap):.1%}，溢价缺乏质量支撑" return raw_label, f"正常化ROE代理相对同行偏离{gap:.1%}，不足以改变原始倍数位置解释" def atomic_write(path: Path, data: bytes) -> bool: path.parent.mkdir(parents=True, exist_ok=True) if path.exists() and path.read_bytes() == data: return False fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) temp = Path(temp_name) try: with os.fdopen(fd, "wb") as handle: handle.write(data) handle.flush() os.fsync(handle.fileno()) os.replace(temp, path) finally: if temp.exists(): temp.unlink() return True def csv_bytes(fieldnames: list[str], rows: Iterable[dict[str, str]]) -> bytes: import io stream = io.StringIO(newline="") writer = csv.DictWriter(stream, fieldnames=fieldnames, extrasaction="ignore", lineterminator="\n") writer.writeheader() writer.writerows(rows) return stream.getvalue().encode("utf-8") def md_escape(value: object) -> str: return str(value or "").replace("|", "\\|").replace("\n", " ") def render_summary(rows: list[dict[str, str]], gaps: list[dict[str, str]], as_of: str) -> bytes: counts = Counter(row["peer_raw_label"] or "不可用" for row in rows) metrics = Counter(row["peer_metric"] for row in rows) lines = [ "# 全部已评估公司同行估值比较", "", f"- 价格日期：`{as_of}`", f"- 已形成同行比较：`{sum(1 for row in rows if row['peer_raw_label'])}`只；不可用：`{sum(1 for row in rows if not row['peer_raw_label'])}`只；价格/口径缺口：`{len(gaps)}`只。", "- 主判定使用目标公司剔除自身后的同行中位数；算术均值仅辅助展示，不参与主分档。", "- 本表的合理区间来自全量中报重估当前基线，同行可比池限定为已完成正式估值且具有同日同口径倍数的证券，不冒充全A股行业总体。", "- MySQL每日台账仍对有效区间重叠证券失败关闭；本表的逐股直接复算不等于这些证券已经通过每日台账生效门禁。", "- 同行比较不替代三情景合理区间，不构成目标价或交易指令。", "", "## 分布", "", "| 项目 | 数量 |", "|---|---:|", ] for name in ["显著低于同行", "低于同行", "接近同行", "高于同行", "显著高于同行", "不可用"]: lines.append(f"| {name} | {counts.get(name, 0)} |") lines.extend(["", "## 指标使用", "", "| 指标 | 数量 |", "|---|---:|"]) for name, count in sorted(metrics.items()): lines.append(f"| {md_escape(name)} | {count} |") lines.extend( [ "", "## 全量排序", "", "同一原始标签内按相对同行中位数溢折价从低到高排列。同行分组为当前估值主营类型的规则化映射；宽口径兜底组和小样本已降低置信度。", "", "| 排名 | 代码 | 公司 | 所属可比行业 | 收盘价 | 指标 | 目标倍数 | 同行样本 | 同行均值 | 同行中位数 | 相对中位数 | 原始标签 | 调整后解释 | 置信度 | 基准区间 | 原价格判定 | 正式报告 |", "|---:|---|---|---|---:|---|---:|---:|---:|---:|---:|---|---|---|---|---|---|", ] ) for index, row in enumerate(rows, start=1): base = f"{row.get('base_low','')}—{row.get('base_high','')}" premium = f"{float(row['peer_premium_pct']):.1%}" if row.get("peer_premium_pct") else "—" report = row.get("report_path", "") report_link = f"`{report}`" if report else "—" values = [ index, row.get("ticker"), row.get("company"), row.get("industry"), row.get("close"), row.get("peer_metric"), row.get("target_multiple") or "—", row.get("peer_count") or "0", row.get("peer_mean") or "—", row.get("peer_median") or "—", premium, row.get("peer_raw_label") or "不可用", row.get("peer_adjusted_label") or "—", row.get("peer_confidence"), base, row.get("label"), report_link, ] lines.append("| " + " | ".join(md_escape(value) for value in values) + " |") if gaps: lines.extend(["", "## 价格或口径缺口", "", "详见同目录 `同行估值比较缺口.csv`。"]) return ("\n".join(lines) + "\n").encode("utf-8") def render_latest(rows: list[dict[str, str]], as_of: str) -> bytes: counts = Counter(row.get("label", "") for row in rows) peer_counts = Counter(row.get("peer_raw_label", "") or "不可用" for row in rows) lines = [ "# 股票估值最新判定", "", f"- 价格日期：`{as_of}`", f"- 数值判定：`{len(rows)}`只；四档分布：偏低{counts.get('偏低',0)}、基本合理{counts.get('基本合理',0)}、偏贵{counts.get('偏贵',0)}、明显偏贵{counts.get('明显偏贵',0)}。", f"- 同行分布：显著低于{peer_counts.get('显著低于同行',0)}、低于{peer_counts.get('低于同行',0)}、接近{peer_counts.get('接近同行',0)}、高于{peer_counts.get('高于同行',0)}、显著高于{peer_counts.get('显著高于同行',0)}、不可用{peer_counts.get('不可用',0)}。", "- 同行主统计量为剔除自身后的中位数；均值仅辅助。同行倍数位置不替代绝对合理区间。", "", "| 代码 | 公司 | 所属可比行业 | 收盘价 | 基准区间 | 原判定 | 泡沫判定 | 高于基准上沿 | 主要泡沫原因 | 原因说明 | 泡沫置信度 | 60日涨幅 | 相对MA60 | 同行指标 | 目标倍数 | 同行均值 | 同行中位数 | 相对中位数 | 同行原始标签 | 同行解释 | 同行置信度 | 正式报告 |", "|---|---|---|---:|---|---|---|---:|---|---|---|---:|---:|---|---:|---:|---:|---:|---|---|---|---|", ] for row in rows: premium = f"{float(row['peer_premium_pct']):.1%}" if row.get("peer_premium_pct") else "—" report = row.get("report_path", "") link = f"`{report}`" if report else "—" bubble_premium = row.get("bubble_premium_pct", "") return_60d = row.get("return_60d_pct", "") distance_ma60 = row.get("distance_to_ma60_pct", "") values = [ row.get("ticker"), row.get("company"), row.get("industry"), row.get("close"), f"{row.get('base_low','')}—{row.get('base_high','')}", row.get("label"), row.get("bubble_status", ""), f"{bubble_premium}%" if bubble_premium else "—", row.get("bubble_primary_cause", ""), row.get("bubble_reason", ""), row.get("bubble_confidence", ""), f"{return_60d}%" if return_60d else "—", f"{distance_ma60}%" if distance_ma60 else "—", row.get("peer_metric"), row.get("target_multiple") or "—", row.get("peer_mean") or "—", row.get("peer_median") or "—", premium, row.get("peer_raw_label") or "不可用", row.get("peer_adjusted_label") or "不可用", row.get("peer_confidence"), link, ] lines.append("| " + " | ".join(md_escape(value) for value in values) + " |") return ("\n".join(lines) + "\n").encode("utf-8") def main() -> int: args = parse_args() root = args.project_root.resolve() valuation_path = under(root, args.valuation_csv) latest_path = under(root, args.latest_csv) gaps_path = under(root, args.gaps_csv) official_industry_path = under(root, args.official_industry_csv) semiconductor_map_path = under(root, args.semiconductor_map_csv) output_dir = under(root, args.output_dir) valuation_fields, valuation_rows = read_csv(valuation_path) latest_fields, latest_rows = read_csv(latest_path) _, gap_rows = read_csv(gaps_path) _, official_rows = read_csv(official_industry_path) _, semiconductor_rows = read_csv(semiconductor_map_path) official_by_ticker: dict[str, str] = {} for official_row in official_rows: ticker = official_ticker(official_row.get("security_id", "")) industry = official_row.get("classification_text", "").strip() if not ticker or not industry: continue previous = official_by_ticker.setdefault(ticker, industry) if previous != industry: raise RuntimeError(f"Conflicting official industries for {ticker}: {previous!r} vs {industry!r}") semiconductor_candidates: dict[str, set[tuple[str, str]]] = {} for semiconductor_row in semiconductor_rows: name = normalized_company_name( semiconductor_row.get("canonical_name", "") or semiconductor_row.get("display_name", "") ) mapped = semiconductor_group(semiconductor_row.get("subindustry_id", "").strip()) if name and mapped: semiconductor_candidates.setdefault(name, set()).add(mapped) semiconductor_by_company = { name: (next(iter(values))[0], next(iter(values))[1], "半导体研究正式子行业映射") for name, values in semiconductor_candidates.items() if len(values) == 1 } valuation_by_ticker = {row["ticker"]: row for row in valuation_rows} if len(valuation_by_ticker) != len(valuation_rows): raise RuntimeError("Duplicate tickers in valuation CSV") trade_dates = {row.get("trade_date", "") for row in latest_rows if row.get("trade_date")} if len(trade_dates) != 1: raise RuntimeError(f"latest.csv must contain one trade date, got {sorted(trade_dates)}") as_of = next(iter(trade_dates)) generated_at = args.generated_at or f"{as_of}T18:00:00+08:00" available_at = args.available_at or generated_at effective_from = args.effective_from or as_of prices = {row["ticker"]: (number(row.get("close")), row.get("trade_date", "")) for row in latest_rows} for row in gap_rows: if row.get("price_date") == as_of and number(row.get("close")): prices.setdefault(row["ticker"], (number(row.get("close")), as_of)) prepared: list[dict[str, object]] = [] result_gaps: list[dict[str, str]] = [] for ticker, row in valuation_by_ticker.items(): close, price_date = prices.get(ticker, (None, "")) if close is None or close <= 0 or price_date != as_of: result_gaps.append({ "ticker": ticker, "company": row.get("company", ""), "reason": "缺少与全量基准一致的同日完整收盘价", "price_date": price_date, "close": fmt(close, 4), }) continue official_industry = official_by_ticker.get(ticker, "") semiconductor_override = semiconductor_by_company.get(normalized_company_name(row.get("company", ""))) business_text = business_text_from_snapshot(root, row) group_id, group_name, basis = classify_group(row, official_industry, semiconductor_override, business_text) metric, value, route_reason = choose_metric(row, close, group_id) pb = metric_value(row, close, "PB") prepared.append({ "ticker": ticker, "row": row, "close": close, "group_id": group_id, "group_name": group_name, "basis": basis, "metric": metric, "value": value, "pb": pb, "route_reason": route_reason, }) for latest_row in latest_rows: if latest_row["ticker"] not in valuation_by_ticker: result_gaps.append({ "ticker": latest_row["ticker"], "company": latest_row.get("company", ""), "reason": "当前台账有价格判定，但缺少本轮统一财务估值基线，同行倍数不可算", "price_date": latest_row.get("trade_date", ""), "close": latest_row.get("close", ""), }) for gap_row in gap_rows: if gap_row["ticker"] in valuation_by_ticker or gap_row.get("status") == "缺少有效估值版本": continue result_gaps.append({ "ticker": gap_row["ticker"], "company": gap_row.get("company", ""), "reason": gap_row.get("reason", "") or gap_row.get("status", ""), "price_date": gap_row.get("price_date", ""), "close": gap_row.get("close", ""), }) by_group: dict[str, list[dict[str, object]]] = {} for item in prepared: by_group.setdefault(str(item["group_id"]), []).append(item) comparisons: list[dict[str, str]] = [] comparison_by_ticker: dict[str, dict[str, str]] = {} peer_groups: list[dict[str, object]] = [] for item in prepared: row = dict(item["row"]) row["close"] = fmt(item["close"], 4) row["price_date"] = as_of refresh_absolute_price_fields(row, float(item["close"])) metric = str(item["metric"]) target_value = item.get("value") if isinstance(item.get("value"), float) else None route_reason = str(item["route_reason"]) def peers_for(selected_metric: str) -> list[dict[str, object]]: selected: list[dict[str, object]] = [] for peer in by_group.get(str(item["group_id"]), []): if peer["ticker"] == item["ticker"]: continue value = metric_value(dict(peer["row"]), float(peer["close"]), selected_metric) if valid_peer_metric(selected_metric, value): enriched_peer = dict(peer) enriched_peer["value"] = value selected.append(enriched_peer) return selected peers = peers_for(metric) if metric != "NOT_APPLICABLE" else [] if len(peers) < 3 and metric.startswith("FORWARD_PE_"): fallback_value = metric_value(dict(item["row"]), float(item["close"]), "NORMALIZED_PE") fallback_peers = peers_for("NORMALIZED_PE") if valid_peer_metric("NORMALIZED_PE", fallback_value) and len(fallback_peers) >= 3: metric = "NORMALIZED_PE" target_value = fallback_value peers = fallback_peers route_reason += "；同年度前瞻PE同行少于3只，降级为正常化PE" peer_values = [float(peer["value"]) for peer in peers if isinstance(peer.get("value"), float)] peer_gap = "" raw = "" adjusted = "" adjustment_reason = route_reason mean = median = p25 = p75 = premium = vs_mean = None if metric == "NOT_APPLICABLE" or not isinstance(target_value, float): peer_gap = route_reason elif len(peer_values) < 3: peer_gap = f"目标剔除自身后只有{len(peer_values)}个同口径有效同行，少于3个" else: mean = statistics.fmean(peer_values) median = statistics.median(peer_values) p25 = quantile(peer_values, 0.25) p75 = quantile(peer_values, 0.75) premium = target_value / median - 1 vs_mean = target_value / mean - 1 raw = label(premium) target_for_quality = dict(item) target_for_quality["metric"] = metric target_for_quality["value"] = target_value adjusted, quality_reason = quality_interpretation(raw, target_for_quality, peers) adjustment_reason = f"{route_reason}；{quality_reason}" spread = (p75 - p25) / median if p75 is not None and p25 is not None and median else None peer_confidence = confidence(len(peer_values), str(item["group_id"]), str(item["basis"]), spread) peer_fields = { "industry": str(item["group_name"]), "peer_group_id": str(item["group_id"]), "peer_group_name": str(item["group_name"]), "peer_group_basis": str(item["basis"]), "peer_tickers": ";".join(sorted(str(peer["ticker"]) for peer in peers)), "peer_metric": metric, "target_multiple": fmt(target_value, 8), "peer_count": str(len(peer_values)), "peer_mean": fmt(mean, 8), "peer_median": fmt(median, 8), "peer_p25": fmt(p25, 8), "peer_p75": fmt(p75, 8), "peer_premium_pct": fmt(premium, 6), "peer_vs_mean_pct": fmt(vs_mean, 6), "peer_raw_label": raw, "peer_adjusted_label": adjusted, "peer_adjustment_reason": adjustment_reason, "peer_confidence": peer_confidence, "peer_as_of": as_of, "peer_gap_reason": peer_gap, } row.update(peer_fields) comparisons.append(row) comparison_by_ticker[str(item["ticker"])] = peer_fields if raw and len(peers) >= 3 and isinstance(target_value, float): member_items = [item, *peers] members: list[dict[str, object]] = [] for member_item in member_items: member_row = dict(member_item["row"]) denominator = metric_denominator(member_row, metric) shares = number(member_row.get("shares")) if denominator is None or shares is None or shares <= 0: raise RuntimeError( f"Missing strict peer denominator for {item['ticker']} member {member_item['ticker']} metric {metric}" ) members.append({ "ticker": str(member_item["ticker"]), "role": "TARGET" if member_item["ticker"] == item["ticker"] else "PEER", "equity_units": decimal_number(shares), "enterprise_value_adjustment": 0, "metric_denominator": decimal_number(denominator), "eligible": True, "exclusion_reason": None, "selection_reason": ( f"同属{item['group_name']}，使用{metric}同日横向比较；目标股不进入自身同行样本" ), "source_id": "stock_valuation_peer_reference_v1", "evidence_ref": member_row.get("report_path", "") or member_row.get("snapshot_path", ""), }) configured_confidence = {"较高": "HIGH", "中": "MEDIUM"}.get(peer_confidence, "LOW") metric_basis_period = ( f"{metric.removeprefix('FORWARD_PE_')}E" if metric.startswith("FORWARD_PE_") else ("NORMALIZED_TTM" if metric == "NORMALIZED_PE" else (row.get("latest_period") or "LATEST_REPORTED")) ) peer_groups.append({ "peer_group_id": str(item["group_id"]), "target_ticker": str(item["ticker"]), "industry": str(item["group_name"]), "peer_group_name": str(item["group_name"]), "metric": "FORWARD_PE" if metric.startswith("FORWARD_PE_") else metric, "metric_basis_period": metric_basis_period, "currency": str(row.get("currency") or "CNY"), "monetary_unit": f"{str(row.get('currency') or 'CNY')}_MAJOR", "basis_as_of": str(row.get("valuation_date") or as_of), "effective_from": effective_from, "configured_confidence": configured_confidence, "quality_adjusted_label": adjusted or None, "adjustment_reason": adjustment_reason, "selection_basis": ( f"{item['basis']}；剔除目标自身；仅使用当前正式估值池中同日、同币种、同指标且倍数有效的公司" ), "evidence_ref": "ana-data/result/股票估值/同行估值比较/全部已评估公司同行估值比较.csv", "members": members, }) label_order = {"显著低于同行": 0, "低于同行": 1, "接近同行": 2, "高于同行": 3, "显著高于同行": 4, "": 5} comparisons.sort(key=lambda row: (label_order.get(row.get("peer_raw_label", ""), 5), number(row.get("peer_premium_pct")) or 0, row["ticker"])) enriched_latest: list[dict[str, str]] = [] for row in latest_rows: enriched = dict(row) peer = comparison_by_ticker.get(row["ticker"]) if peer is None: peer = {field: "" for field in PEER_FIELDS} peer["peer_gap_reason"] = "当前正式估值基线缺失" enriched.update(peer) enriched_latest.append(enriched) output_fields = list(latest_fields) for field in PEER_FIELDS: if field not in output_fields: output_fields.append(field) comparison_fields = list(valuation_fields) if "trade_date" not in comparison_fields: comparison_fields.append("trade_date") for field in PEER_FIELDS: if field not in comparison_fields: comparison_fields.append(field) for row in comparisons: row["trade_date"] = as_of peer_input = { "schema_version": 1, "source_id": "stock_valuation_peer_reference_v1", "generated_at": generated_at, "available_at": available_at, "groups": sorted(peer_groups, key=lambda group: str(group["target_ticker"])), } files = { latest_path: csv_bytes(output_fields, enriched_latest), latest_path.with_suffix(".md"): render_latest(enriched_latest, as_of), output_dir / "全部已评估公司同行估值比较.csv": csv_bytes(comparison_fields, comparisons), output_dir / "全部已评估公司同行估值比较.md": render_summary(comparisons, result_gaps, as_of), output_dir / "同行估值比较缺口.csv": csv_bytes(["ticker", "company", "reason", "price_date", "close"], result_gaps), output_dir / "同行组输入.json": ( json.dumps(peer_input, ensure_ascii=False, indent=2, sort_keys=True) + "\n" ).encode("utf-8"), } base_latest_fields = [field for field in latest_fields if field not in PEER_FIELDS] manifest_payload = { "schema": "stock_valuation_peer_comparison_manifest_v1", "as_of": as_of, "scope": "current_formally_valued_securities_with_same-day-comparable-multiples", "universe": { "current_valuation": len(comparisons) + len(result_gaps), "same_day_priced_and_baselined": len(comparisons), "peer_comparable": sum(1 for row in comparisons if row["peer_raw_label"]), "peer_not_applicable": sum(1 for row in comparisons if not row["peer_raw_label"]), "gaps": len(result_gaps), "mysql_peer_groups": len(peer_groups), }, "contract": { "target_excluded": True, "minimum_peer_count": 3, "primary_statistic": "median", "auxiliary_statistics": ["mean", "p25", "p75"], "premium_formula": "target_multiple / peer_median - 1", "full_a_share_industry_claim": False, }, "inputs": { str(valuation_path.relative_to(root)): hashlib.sha256(valuation_path.read_bytes()).hexdigest().upper(), str(official_industry_path.relative_to(root)): hashlib.sha256(official_industry_path.read_bytes()).hexdigest().upper(), str(semiconductor_map_path.relative_to(root)): hashlib.sha256(semiconductor_map_path.read_bytes()).hexdigest().upper(), f"{latest_path.relative_to(root)}#without_peer_fields": hashlib.sha256( csv_bytes(base_latest_fields, latest_rows) ).hexdigest().upper(), }, "outputs": { str(path.relative_to(root)): hashlib.sha256(data).hexdigest().upper() for path, data in files.items() }, } files[output_dir / "manifest.json"] = ( json.dumps(manifest_payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n" ).encode("utf-8") changed: dict[str, bool] = {} if args.write: for path, data in files.items(): changed[str(path.relative_to(root))] = atomic_write(path, data) summary = { "mode": "WRITE" if args.write else "DRY_RUN", "as_of": as_of, "valuation_universe": len(valuation_rows), "current_valuation_universe": len(comparisons) + len(result_gaps), "same_day_priced": len(prepared), "peer_comparable": sum(1 for row in comparisons if row["peer_raw_label"]), "peer_not_applicable": sum(1 for row in comparisons if not row["peer_raw_label"]), "price_or_baseline_gaps": len(result_gaps), "latest_rows": len(enriched_latest), "labels": dict(Counter(row["peer_raw_label"] or "不可用" for row in comparisons)), "metrics": dict(Counter(row["peer_metric"] for row in comparisons)), "files_changed": changed, "output_sha256": { str(path.relative_to(root)): hashlib.sha256(data).hexdigest().upper() for path, data in files.items() }, } print(json.dumps(summary, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())