from __future__ import annotations import csv import hashlib import json import math import statistics from collections import Counter from datetime import datetime from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[2] RESULT_ROOT = ROOT / "ana-data/result/股票估值" MASTER_ROOT = RESULT_ROOT / "全量中报重估" CURRENT_ROOT = RESULT_ROOT / "当前估值" LEDGER_ROOT = RESULT_ROOT / "估值台账" BATCH_ROOT = ROOT / "ana-data/tmp/股票估值/BATCH-STOCK-VALUATION-H1-GAP-CLOSURE-20260903-001" CASE_ROOT = ROOT / "ana-data/cases/股票估值/BATCH-STOCK-VALUATION-H1-GAP-CLOSURE-20260903-001" OLD_MASTER = MASTER_ROOT / "全部已评估公司最新估值.csv" INCREMENTAL = BATCH_ROOT / "revaluation_output/全部已评估公司最新估值.csv" UNIVERSE = BATCH_ROOT / "universe.csv" LATEST = LEDGER_ROOT / "latest.csv" LATEST_GAPS = LEDGER_ROOT / "latest_gaps.csv" WEEKLY = RESULT_ROOT / "周度增量复评/latest.csv" MODEL_GAPS = { "600892.SH": ("大晟文化", "最新归母净资产不为正且TTM扣非亏损，PE/PB均无有效锚点"), "603398.SH": ("*ST沐邦", "最新归母净资产不为正且TTM扣非亏损，PE/PB均无有效锚点"), } HK_GAPS = { "03888.HK": "金山软件", "09880.HK": "优必选", } SUSPENDED = {"600929.SH", "688432.SH"} CURRENT_OVERRIDES = { "002158.SZ": "VAL-be8591b5946a328183125299", "300223.SZ": "VAL-f43dd629fde5197ab61e2f1f", "300623.SZ": "VAL-e5fea1c78750cdafffb8d7e8", "600745.SH": "VAL-948f52a9cc02658f919e1b0a", "603163.SH": "VAL-c2739ad9efcd0971aa6ba773", "688018.SH": "VAL-76f429dabd366d04b0d8720f", "688019.SH": "VAL-18f847d08045478ed367f5a1", "688138.SH": "VAL-3cde42baa99030febd11c676", "688187.SH": "VAL-409ffcd54b4fae5ba10826d5", "688213.SH": "VAL-9ef621439025dbdc5199b80e", "688401.SH": "VAL-0d030950a6ae9ed8bc573c94", } 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 read_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8-sig")) def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest().upper() def f(value: Any) -> float | None: if value in (None, "", "None"): return None return float(value) def valuation_position(price: float, low: float, high: float) -> float: if price < low: return price / low - 1 if price <= high: return price / ((low + high) / 2) - 1 return price / high - 1 def classify(price: float, base_low: float, base_high: float, optimistic_high: float) -> tuple[str, str, str]: if price < base_low: label = "偏低" elif price <= base_high: label = "基本合理" elif price <= optimistic_high: label = "偏贵" else: label = "明显偏贵" premium = price / base_high - 1 vs_optimistic = price / optimistic_high - 1 if premium <= 0: return label, "未识别估值泡沫", "当前价格未高于基准合理区间上沿，按统一规则不认定估值泡沫。" if vs_optimistic > 0: status = "泡沫-极端" elif premium <= 0.15: status = "泡沫-轻" elif premium <= 0.50: status = "泡沫-中" else: status = "泡沫-高" reason = ( f"价格高于基准上沿{premium * 100:.1f}%，市场已提前交易行业增长、盈利修复或份额提升预期" + (f"；同时高于乐观上沿{vs_optimistic * 100:.1f}%，包含模型外叙事溢价" if vs_optimistic > 0 else "") + "。除价格位置外，原因属于基于估值状态的解释性推断。" ) return label, status, reason def current_file(ticker: str, suffix: str) -> Path: candidates = sorted((CURRENT_ROOT / ticker).glob(f"*{suffix}_当前.*")) if candidates: return candidates[0] candidates = sorted((CURRENT_ROOT / ticker).glob(f"*{suffix}.*")) if len(candidates) != 1: raise RuntimeError(f"expected one {suffix} for {ticker}, got {len(candidates)}") return candidates[0] def snapshot_row(ticker: str, price_row: dict[str, str], valuation_id: str) -> dict[str, Any]: snapshot_path = current_file(ticker, "估值快照") report_path = current_file(ticker, "价格合理性评估") snapshot = read_json(snapshot_path) results = read_json(snapshot_path.parent / "calculation/valuation_results.json") scenarios = {row["role"]: row for row in results["scenarios"]} metrics = results["metrics"] meta = snapshot["meta"] market = snapshot["market"] analysis = snapshot.get("analysis", {}) forecasts = snapshot.get("institutions", {}).get("forecasts", []) profits = [] for forecast in forecasts: estimate = forecast.get("estimates", {}).get("2026") or {} if isinstance(estimate, dict) and f(estimate.get("profit")) is not None: profits.append(float(estimate["profit"])) price_date = price_row.get("trade_date") or price_row.get("price_date") close = price_row["close"] return { "ticker": ticker, "company": meta["company"], "market": meta["market"], "currency": meta["currency"], "status": "COMPLETE", "valuation_id": valuation_id, "valuation_date": meta["as_of_date"], "price_date": price_date, "close": close, "shares": metrics["diluted_shares"], "market_cap": float(close) * float(metrics["diluted_shares"]), "latest_period": meta["report_period_end"], "latest_notice_date": meta["latest_operating_info_date"], "old_valuation_date": analysis.get("old_valuation_date", ""), "financial_update": analysis.get("financial_update", ""), "model_reclassified": analysis.get("model_reclassified", ""), "old_company_type": analysis.get("old_company_type", ""), "company_type": analysis.get("company_type", ""), "method": scenarios["base"]["method"].upper(), "normalized_profit": metrics.get("normalized_profit"), "normalized_pe": metrics.get("normalized_pe"), "pb": metrics.get("pb"), "ps": metrics.get("ps"), "consensus_year": 2026 if profits else "", "consensus_profit": statistics.median(profits) if profits else "", "consensus_count": len(profits), "excluded_forecast_count": len(analysis.get("excluded_institution_forecasts", [])), "pessimistic_low": scenarios["pessimistic"]["price_low"], "pessimistic_high": scenarios["pessimistic"]["price_high"], "base_low": scenarios["base"]["price_low"], "base_high": scenarios["base"]["price_high"], "optimistic_low": scenarios["optimistic"]["price_low"], "optimistic_high": scenarios["optimistic"]["price_high"], "confidence": analysis.get("classification_confidence", "中"), "qa_status": results["qa"]["status"], "qa_warning_count": results["qa"]["warning_count"], "report_path": report_path.relative_to(ROOT).as_posix(), "snapshot_path": snapshot_path.relative_to(ROOT).as_posix(), "source_hash": sha256(snapshot_path), } def write_csv(path: Path, rows: list[dict[str, Any]], fields: list[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore", lineterminator="\n") writer.writeheader() writer.writerows(rows) def main() -> None: old_rows = read_csv(OLD_MASTER) incremental_rows = read_csv(INCREMENTAL) universe_rows = read_csv(UNIVERSE) latest_rows = read_csv(LATEST) gap_rows = read_csv(LATEST_GAPS) weekly_rows = read_csv(WEEKLY) latest_by_ticker = {row["ticker"]: row for row in latest_rows} gap_by_ticker = {row["ticker"]: row for row in gap_rows} weekly_by_ticker = {row["ticker"]: row for row in weekly_rows} old_by_ticker = {row["ticker"]: row for row in old_rows} universe_by_ticker = {row["ticker"]: row for row in universe_rows} combined = {row["ticker"]: dict(row) for row in old_rows} for row in incremental_rows: combined[row["ticker"]] = dict(row) for ticker, valuation_id in CURRENT_OVERRIDES.items(): price_row = latest_by_ticker.get(ticker) or gap_by_ticker.get(ticker) if not price_row: raise RuntimeError(f"missing price row for current override {ticker}") combined[ticker] = snapshot_row(ticker, price_row, valuation_id) for ticker in MODEL_GAPS: combined.pop(ticker, None) for ticker in SUSPENDED: combined[ticker] = snapshot_row(ticker, gap_by_ticker[ticker], weekly_by_ticker[ticker]["valuation_id"]) if len(combined) != 1225: raise RuntimeError(f"expected 1225 numeric valuations, got {len(combined)}") final_rows: list[dict[str, Any]] = [] for ticker, row in combined.items(): price_row = latest_by_ticker.get(ticker) or gap_by_ticker.get(ticker) if not price_row or not price_row.get("close"): raise RuntimeError(f"missing latest available price for {ticker}") price = float(price_row["close"]) base_low = float(row["base_low"]) base_high = float(row["base_high"]) optimistic_high = float(row["optimistic_high"]) label, bubble, reason = classify(price, base_low, base_high, optimistic_high) row.update({ "close": price, "price_date": price_row.get("trade_date") or price_row.get("price_date"), "market_cap": price * float(row["shares"]), "label": label, "bubble_status": bubble, "bubble_reason": reason, "valuation_position_pct": valuation_position(price, base_low, base_high), "premium_to_base_high": price / base_high - 1, "vs_optimistic_high": price / optimistic_high - 1, "industry": row.get("company_type", ""), }) if row.get("latest_period") != "2026-06-30": raise RuntimeError(f"non-H1 numeric valuation remains: {ticker} {row.get('latest_period')}") final_rows.append(row) final_rows.sort(key=lambda row: (float(row["valuation_position_pct"]), row["ticker"])) fields = list(final_rows[0]) if "industry" not in fields: fields.insert(fields.index("company_type") + 1, "industry") csv_path = MASTER_ROOT / "全部已评估公司最新估值.csv" write_csv(csv_path, final_rows, fields) formal_gaps = [] for ticker, (company, reason) in MODEL_GAPS.items(): report_path = current_file(ticker, "价格合理性评估") snapshot_path = current_file(ticker, "估值快照") formal_gaps.append({ "ticker": ticker, "company": company, "status": "VALUATION_MODEL_GAP", "latest_period": "2026-06-30", "price_date": latest_by_ticker[ticker]["trade_date"], "close": latest_by_ticker[ticker]["close"], "reason": reason, "report_path": report_path.relative_to(ROOT).as_posix(), "snapshot_path": snapshot_path.relative_to(ROOT).as_posix(), }) for ticker, company in HK_GAPS.items(): formal_gaps.append({ "ticker": ticker, "company": company, "status": "UNSUPPORTED_HK", "latest_period": "", "price_date": "", "close": "", "reason": "现有登记行情与中报批处理合同仅覆盖A股，未伪造港股数据或回退未审核来源", "report_path": weekly_by_ticker[ticker].get("valuation_id", ""), "snapshot_path": "", }) formal_gaps.sort(key=lambda row: row["ticker"]) gaps_path = MASTER_ROOT / "全部已评估公司最新估值_缺口.csv" write_csv(gaps_path, formal_gaps, list(formal_gaps[0])) share_audit = [] valuation_change_audit = [] for row in incremental_rows: ticker = row["ticker"] static_shares = f(universe_by_ticker[ticker].get("static_total_shares")) h1_shares = f(row.get("shares")) share_delta = h1_shares / static_shares - 1 if static_shares and h1_shares else None if share_delta is not None and abs(share_delta) > 0.02: share_audit.append({ "ticker": ticker, "company": row["company"], "static_shares": static_shares, "h1_reported_shares": h1_shares, "difference_pct": share_delta, "selected_basis": "2026H1法定报告TOTAL_SHARE", }) old = old_by_ticker.get(ticker) if old: old_mid = (float(old["base_low"]) + float(old["base_high"])) / 2 new_mid = (float(row["base_low"]) + float(row["base_high"])) / 2 valuation_change_audit.append({ "ticker": ticker, "company": row["company"], "old_period": old.get("latest_period", ""), "new_period": row["latest_period"], "old_base_low": old["base_low"], "old_base_high": old["base_high"], "new_base_low": row["base_low"], "new_base_high": row["base_high"], "midpoint_change_pct": new_mid / old_mid - 1 if old_mid else "", "method_changed": str(old.get("method")) != str(row.get("method")), }) share_audit.sort(key=lambda row: (-abs(float(row["difference_pct"])), row["ticker"])) valuation_change_audit.sort( key=lambda row: (-abs(float(row["midpoint_change_pct"])), row["ticker"]) ) share_audit_path = CASE_ROOT / "share_basis_audit.csv" change_audit_path = CASE_ROOT / "valuation_change_audit.csv" write_csv(share_audit_path, share_audit, list(share_audit[0])) write_csv(change_audit_path, valuation_change_audit, list(valuation_change_audit[0])) labels = Counter(row["label"] for row in final_rows) bubbles = Counter(row["bubble_status"] for row in final_rows) price_dates = Counter(row["price_date"] for row in final_rows) lines = [ "# 全部已评估公司中报期最新估值", "", f"> 估值信息截止：2026-09-03；统一价格日：2026-09-02（{price_dates['2026-09-02']}家）；因停牌无9月2日日K而使用最新可得2026-08-28收盘：{price_dates['2026-08-28']}家。", "> 合理价值区间是条件化估值，不等同目标价；本表不构成交易指令或收益承诺。", "", "## 覆盖结论", "", "- 已评估证券总数：1,229家。", "- A股中报已逐家处理：1,227/1,227（100%）。其中1,225家形成数值合理区间，2家转为显式估值模型缺口，旧区间已停止作为当前判断依据。", "- 港股显式缺口：2家；现有登记行情和批处理合同不支持，未使用未经审核来源补数。", f"- 数值估值标签：偏低{labels['偏低']}、基本合理{labels['基本合理']}、偏贵{labels['偏贵']}、明显偏贵{labels['明显偏贵']}。", f"- 泡沫标签：未识别{bubbles['未识别估值泡沫']}、轻{bubbles['泡沫-轻']}、中{bubbles['泡沫-中']}、高{bubbles['泡沫-高']}、极端{bubbles['泡沫-极端']}。", "", "## 估值最低的前30家公司", "", "| 排名 | 代码 | 公司 | 所属行业 | 收盘价 | 价格日 | 基准区间 | 估值位置 | 判定 | 泡沫 | 正式报告 |", "|---:|---|---|---|---:|---|---:|---:|---|---|---|", ] for rank, row in enumerate(final_rows[:30], 1): link = "../" + row["report_path"].split("股票估值/", 1)[-1].replace(" ", "%20") lines.append( f"| {rank} | {row['ticker']} | {row['company']} | {row['industry']} | {float(row['close']):.2f} | {row['price_date']} | " f"{float(row['base_low']):.2f}—{float(row['base_high']):.2f} | {float(row['valuation_position_pct'])*100:.2f}% | " f"{row['label']} | {row['bubble_status']} | [查看]({link}) |" ) lines += [ "", "## 全量结果", "", "排序口径：低于下沿时计算相对下沿折价；区间内计算相对区间中枢偏离；高于上沿时计算相对上沿溢价。数值越低，估值位置越靠前。", "", "| 排名 | 代码 | 公司 | 所属行业 | 收盘价 | 价格日 | 基准区间 | 乐观上沿 | 估值位置 | 判定 | 泡沫 | 置信度 | 正式报告 |", "|---:|---|---|---|---:|---|---:|---:|---:|---|---|---|---|", ] for rank, row in enumerate(final_rows, 1): link = "../" + row["report_path"].split("股票估值/", 1)[-1].replace(" ", "%20") lines.append( f"| {rank} | {row['ticker']} | {row['company']} | {row['industry']} | {float(row['close']):.2f} | {row['price_date']} | " f"{float(row['base_low']):.2f}—{float(row['base_high']):.2f} | {float(row['optimistic_high']):.2f} | " f"{float(row['valuation_position_pct'])*100:.2f}% | {row['label']} | {row['bubble_status']} | {row['confidence']} | [查看]({link}) |" ) lines += [ "", "## 显式缺口", "", "| 代码 | 公司 | 状态 | 最新经营期 | 最新价格 | 原因 |", "|---|---|---|---|---:|---|", ] for row in formal_gaps: price = f"{row['close']}（{row['price_date']}）" if row["close"] else "—" lines.append(f"| {row['ticker']} | {row['company']} | {row['status']} | {row['latest_period'] or '—'} | {price} | {row['reason']} |") md_path = MASTER_ROOT / "全部已评估公司最新估值.md" md_path.write_text("\n".join(lines) + "\n", encoding="utf-8") evidence_path = MASTER_ROOT / "evidence.json" generated_at = datetime.now().astimezone().isoformat(timespec="seconds") if evidence_path.exists(): try: previous_evidence = read_json(evidence_path) if previous_evidence.get("schema") == "stock_valuation_all_h1_refresh_v1": generated_at = previous_evidence.get("generated_at") or generated_at except (OSError, UnicodeDecodeError, json.JSONDecodeError): pass evidence = { "schema": "stock_valuation_all_h1_refresh_v1", "generated_at": generated_at, "information_cutoff": "2026-09-03", "latest_complete_trade_date": "2026-09-02", "scope": {"total": 1229, "a_share": 1227, "numeric": 1225, "model_gaps": 2, "unsupported_hk": 2}, "h1_coverage": {"a_share_processed": 1227, "a_share_total": 1227, "numeric_h1": 1225, "model_gap_h1": 2}, "price_dates": dict(price_dates), "labels": dict(labels), "bubble_status": dict(bubbles), "new_h1_fetch": {"requested": 500, "a_share_ok": 498, "unsupported_hk": 2}, "new_valuations": {"numeric": 496, "model_gap": 2, "mysql_new_versions": 496, "effective_from": "2026-09-04"}, "database_model_gap_disposition": { "tickers": ["600892.SH", "603398.SH"], "security_active": 0, "reason": "防止每日任务继续引用已经失效的旧数值区间；历史版本和历史判断均保留", }, "quality_audits": { "h1_share_difference_over_2pct": len(share_audit), "valuation_change_rows": len(valuation_change_audit), "share_basis_rule": "2026H1法定报告期末总股本优先于旧证券静态股本", }, "historical_policy": "2026-09-02 daily_price/daily_judgement preserved; no historical daily rerun or rewrite", "inputs": { str(INCREMENTAL.relative_to(ROOT)): sha256(INCREMENTAL), str(LATEST.relative_to(ROOT)): sha256(LATEST), str(LATEST_GAPS.relative_to(ROOT)): sha256(LATEST_GAPS), }, "outputs": {}, } for path in (csv_path, md_path, gaps_path, share_audit_path, change_audit_path): evidence["outputs"][str(path.relative_to(ROOT))] = {"bytes": path.stat().st_size, "sha256": sha256(path)} evidence_path.write_text(json.dumps(evidence, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") print(json.dumps({"status": "PASS", "rows": len(final_rows), "gaps": len(formal_gaps), "labels": dict(labels), "prices": dict(price_dates), "outputs": evidence["outputs"]}, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()