from __future__ import annotations import csv import hashlib import json import os import tempfile from collections import Counter from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[2] COVERAGE_PATH = PROJECT_ROOT / "ana-data/result/股票估值/行业调研标的估值覆盖/全部行业调研标的估值覆盖清单.csv" LATEST_PATH = PROJECT_ROOT / "ana-data/result/股票估值/估值台账/latest.csv" OUTPUT_DIR = PROJECT_ROOT / "ana-data/result/股票估值/行业调研标的估值覆盖" RESULT_ROOT = PROJECT_ROOT / "ana-data/result/股票估值" OUTPUT_CSV = OUTPUT_DIR / "半导体估值排序.csv" OUTPUT_MD = OUTPUT_DIR / "半导体估值排序.md" OUTPUT_MANIFEST = OUTPUT_DIR / "全部行业调研标的估值覆盖manifest.json" FIELDS = [ "代码", "公司", "所属行业", "收盘价", "基准区间", "估值位置偏离", "原判定", "泡沫判定", "高于基准上沿", "主要泡沫原因", "原因说明", "置信度", "60日涨幅", "相对MA60", "正式报告", ] def read_csv(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as handle: return list(csv.DictReader(handle)) def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest().upper() def as_float(value: str) -> float | None: try: return float(value) except (TypeError, ValueError): return None def signed_pct(value: str) -> str: number = as_float(value) return "—" if number is None else f"{number:+.2f}%" def valuation_position(row: dict[str, str]) -> tuple[int, float, str] | None: close = as_float(row.get("close", "")) low = as_float(row.get("base_low", "")) high = as_float(row.get("base_high", "")) if close is None or low is None or high is None or low <= 0 or high <= low: return None if close < low: discount = (low - close) / low * 100.0 return 0, discount, f"低于下沿 {discount:.2f}%" if close <= high: midpoint = (low + high) / 2.0 deviation = (close - midpoint) / midpoint * 100.0 if deviation < -0.005: display = f"低于中枢 {abs(deviation):.2f}%" elif deviation > 0.005: display = f"高于中枢 {deviation:.2f}%" else: display = "位于中枢 0.00%" return 1, deviation, display premium = (close - high) / high * 100.0 return 2, premium, f"高于上沿 {premium:.2f}%" def display_industries(value: str) -> str: return value.replace(";", "、") def resolve_report_path(source: dict[str, str]) -> str: if source.get("report_path"): return source["report_path"] code = source["ticker"].split(".", 1)[0] candidates = sorted(RESULT_ROOT.glob(f"*_{code}_valuation/*价格合理性评估*.md")) if len(candidates) != 1: raise RuntimeError(f"Expected one formal report for {source['ticker']}, got {len(candidates)}") return candidates[0].relative_to(PROJECT_ROOT).as_posix() def escape_markdown(value: str) -> str: return value.replace("\\", "\\\\").replace("|", "\\|").replace("\n", " ") def atomic_write_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) try: with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle: handle.write(text) os.replace(temp_name, path) except BaseException: try: os.unlink(temp_name) except FileNotFoundError: pass raise def atomic_write_csv(path: Path, rows: list[dict[str, str]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) try: with os.fdopen(fd, "w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=FIELDS, lineterminator="\n") writer.writeheader() writer.writerows(rows) os.replace(temp_name, path) except BaseException: try: os.unlink(temp_name) except FileNotFoundError: pass raise def main() -> int: coverage = [row for row in read_csv(COVERAGE_PATH) if "半导体" in row["industries"].split(";")] latest = {row["ticker"]: row for row in read_csv(LATEST_PATH)} ranked: list[tuple[tuple[int, float, str], dict[str, str]]] = [] for source in coverage: current = latest.get(source["ticker"]) report_path = resolve_report_path(source) if current is None: output = { "代码": source["ticker"], "公司": source["company"], "所属行业": display_industries(source["industries"]), "收盘价": "—", "基准区间": "—", "估值位置偏离": "—", "原判定": "特殊结论", "泡沫判定": "不适用", "高于基准上沿": "—", "主要泡沫原因": "常规PE/PB失效", "原因说明": source["coverage_status"], "置信度": "—", "60日涨幅": "—", "相对MA60": "—", "正式报告": report_path, } ranked.append(((4, float("inf"), source["ticker"]), output)) continue close = as_float(current["close"]) low = as_float(current["base_low"]) high = as_float(current["base_high"]) premium = as_float(current["bubble_premium_pct"]) position = valuation_position(current) output = { "代码": source["ticker"], "公司": source["company"], "所属行业": display_industries(source["industries"]), "收盘价": "—" if close is None else f"{close:.2f}", "基准区间": "—" if low is None or high is None else f"{low:.2f}—{high:.2f}", "估值位置偏离": "—" if position is None else position[2], "原判定": current["label"], "泡沫判定": current["bubble_status"], "高于基准上沿": "—" if premium is None else f"{max(0.0, premium):.2f}%", "主要泡沫原因": current["bubble_primary_cause"], "原因说明": current["bubble_reason"], "置信度": current["bubble_confidence"], "60日涨幅": signed_pct(current["return_60d_pct"]), "相对MA60": signed_pct(current["distance_to_ma60_pct"]), "正式报告": report_path, } if position is None: key = (3, float("inf"), source["ticker"]) else: zone, metric, _ = position key = (zone, -metric if zone == 0 else metric, source["ticker"]) ranked.append((key, output)) ranked.sort(key=lambda item: item[0]) rows = [item[1] for item in ranked] if len(rows) != 179 or len({row["代码"] for row in rows}) != 179: raise RuntimeError(f"Expected 179 unique semiconductor securities, got {len(rows)}") atomic_write_csv(OUTPUT_CSV, rows) label_counts = Counter(row["原判定"] for row in rows) trade_dates = sorted({latest[row["代码"]]["trade_date"] for row in rows if row["代码"] in latest}) lines = [ "# 半导体股票估值排序", "", f"- 价格日：{'、'.join(trade_dates)}；半导体研究证券共{len(rows)}只。", "- 排序：低于下沿 → 基准区间内 → 高于上沿 → 特殊结论。低于下沿按折价从大到小；区间内按相对中枢偏离从低到高；高于上沿按溢价从小到大。", f"- 分布：偏低{label_counts['偏低']}只、基本合理{label_counts['基本合理']}只、偏贵{label_counts['偏贵']}只、明显偏贵{label_counts['明显偏贵']}只、特殊结论{label_counts['特殊结论']}只。", "- “估值位置偏离”分段计算：低于下沿用 `(下沿-收盘价)/下沿`；区间内用 `(收盘价-区间中枢)/区间中枢`；高于上沿用 `(收盘价-上沿)/上沿`。", "- 该排序只比较当前价格在既定合理区间中的位置，不代表经营风险已经消除，也不把不同公司的估值区间置信度视为完全相同。", "- “高于基准上沿”为0表示没有超过基准合理区间上沿；泡沫原因除价格位置外属于最可能解释，不是已经证实的资金因果。", f"- 输入：`全部行业调研标的估值覆盖清单.csv` SHA-256 `{sha256(COVERAGE_PATH)}`；`../估值台账/latest.csv` SHA-256 `{sha256(LATEST_PATH)}`。", "- 合理区间是条件化估值，不是目标价、交易指令或收益承诺。", "", "| " + " | ".join(FIELDS) + " |", "|" + "|".join(["---"] * len(FIELDS)) + "|", ] for row in rows: values = [] for field in FIELDS: value = row[field] if field == "正式报告" and value: value = f"[报告](../../../../{value})" values.append(escape_markdown(value)) lines.append("| " + " | ".join(values) + " |") lines.append("") atomic_write_text(OUTPUT_MD, "\n".join(lines)) manifest = json.loads(OUTPUT_MANIFEST.read_text(encoding="utf-8")) manifest["semiconductor_ranking_trade_date"] = "、".join(trade_dates) manifest["semiconductor_ranking_md_sha256"] = sha256(OUTPUT_MD).lower() manifest["semiconductor_ranking_csv_sha256"] = sha256(OUTPUT_CSV).lower() atomic_write_text( OUTPUT_MANIFEST, json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", ) print(json.dumps({ "rows": len(rows), "label_counts": dict(label_counts), "trade_dates": trade_dates, "csv": str(OUTPUT_CSV.relative_to(PROJECT_ROOT)), "csv_sha256": sha256(OUTPUT_CSV), "markdown": str(OUTPUT_MD.relative_to(PROJECT_ROOT)), "markdown_sha256": sha256(OUTPUT_MD), "manifest": str(OUTPUT_MANIFEST.relative_to(PROJECT_ROOT)), "manifest_sha256": sha256(OUTPUT_MANIFEST), }, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())