"""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())
|
