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