#!/usr/bin/env python3
|
"""Rebuild current valuation snapshots after the 2026 interim-report refresh.
|
|
Judgment (business classification, scenario profit bands and multiples) lives
|
here; all numeric valuation formulae are executed by the frozen V1 engine.
|
Files are written to one stable ``当前估值`` tree and are overwritten on rerun.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import csv
|
import hashlib
|
import json
|
import math
|
import re
|
import statistics
|
import subprocess
|
import sys
|
import time
|
from collections import Counter
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
from datetime import datetime
|
from pathlib import Path
|
from typing import Any
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
sys.path.insert(0, str(ROOT / "dev/ana-dev"))
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
from stock_valuation_pipeline.core import run_pipeline # noqa: E402
|
import full_midyear_refresh_20260826 as fetch # noqa: E402
|
|
|
BATCH_ID = fetch.BATCH_ID
|
AS_OF = fetch.AS_OF
|
PRICE_DATE = fetch.PRICE_DATE
|
TMP_ROOT = fetch.TMP_ROOT
|
RAW_ROOT = fetch.RAW_ROOT
|
CASE_ROOT = fetch.CASE_ROOT
|
CURRENT_ROOT = ROOT / "ana-data/result/股票估值/当前估值"
|
MASTER_ROOT = ROOT / "ana-data/result/股票估值/全量中报重估"
|
REGISTRY = ROOT / "dev/ana-dev/stock_valuation_pipeline/source_registry.json"
|
SCAN_PATH = ROOT / "ana-data/cases/股票估值/BATCH-STOCK-VALUATION-QINGFENGPU-TIMELINE-20260825-001/估值口径异常扫描.csv"
|
|
|
TIER_RULES: dict[str, dict[str, Any]] = {
|
"bank": {"name": "商业银行", "method": "pb", "bands": ((0.45, 0.70), (0.70, 1.05), (1.05, 1.40))},
|
"financial": {"name": "证券、保险或综合金融", "method": "pb", "bands": ((0.55, 0.90), (0.90, 1.45), (1.45, 2.10))},
|
"resource": {"name": "资源开采与有色金属", "method": "pe", "bands": ((6, 10), (10, 16), (16, 24))},
|
"coal_energy": {"name": "煤炭、油气与传统能源", "method": "pe", "bands": ((5, 8), (8, 13), (13, 20))},
|
"chemical": {"name": "化工、氟化工或化学材料", "method": "pe", "bands": ((8, 13), (13, 21), (21, 30))},
|
"agriculture": {"name": "农业、农资、养殖或食品原料", "method": "pe", "bands": ((8, 13), (13, 21), (21, 30))},
|
"memory_module": {"name": "存储模组、存储控制器或存储产品（强周期）", "method": "pe", "bands": ((5, 8), (10, 15), (15, 20))},
|
"semi_equipment": {"name": "半导体设备、厂务或EDA", "method": "pe", "bands": ((22, 32), (32, 48), (48, 68))},
|
"semi_material": {"name": "半导体材料与电子化学品", "method": "pe", "bands": ((18, 28), (28, 42), (42, 60))},
|
"semi_foundry": {"name": "晶圆制造、硅片或化合物半导体", "method": "pe", "bands": ((20, 30), (30, 48), (48, 68))},
|
"semi_packaging": {"name": "半导体封装测试", "method": "pe", "bands": ((15, 23), (23, 34), (34, 48))},
|
"semi_design": {"name": "集成电路设计与芯片产品", "method": "pe", "bands": ((20, 30), (30, 45), (45, 65))},
|
"optical_ai": {"name": "AI算力、服务器、通信或光通信", "method": "pe", "bands": ((18, 28), (28, 42), (42, 60))},
|
"software": {"name": "软件、数据服务、安全或信创", "method": "pe", "bands": ((18, 28), (28, 42), (42, 60))},
|
"robotics": {"name": "机器人核心零部件、自动化或工业控制", "method": "pe", "bands": ((15, 23), (23, 35), (35, 50))},
|
"military": {"name": "军工电子、航空航天或特种装备", "method": "pe", "bands": ((18, 28), (28, 42), (42, 60))},
|
"battery_solar": {"name": "光伏、锂电、储能材料或设备", "method": "pe", "bands": ((8, 13), (13, 21), (21, 30))},
|
"wind_power": {"name": "风电、核电、电网或电力设备", "method": "pe", "bands": ((12, 18), (18, 27), (27, 38))},
|
"pharma": {"name": "医药、医疗器械或生命科学服务", "method": "pe", "bands": ((15, 23), (23, 35), (35, 50))},
|
"consumer": {"name": "消费品、商业服务或文旅", "method": "pe", "bands": ((10, 16), (16, 25), (25, 36))},
|
"property": {"name": "地产、建筑或基础设施", "method": "pb", "bands": ((0.45, 0.75), (0.75, 1.20), (1.20, 1.80))},
|
"transport": {"name": "交通运输、物流或公用运营", "method": "pe", "bands": ((9, 14), (14, 22), (22, 32))},
|
"industrial": {"name": "通用制造、机械设备或工业材料", "method": "pe", "bands": ((10, 16), (16, 25), (25, 36))},
|
}
|
|
|
OLD_TYPE_KEYWORDS: dict[str, tuple[str, ...]] = {
|
"集成电路设计与芯片产品": ("芯片", "集成电路", "半导体", "处理器", "fpga", "soc"),
|
"AI算力、网络、光通信或散热基础设施": ("服务器", "算力", "通信", "网络", "光模块", "光通信", "散热", "数据中心"),
|
"机器人核心零部件、自动化或关联制造": ("机器人", "自动化", "伺服", "减速", "机床", "工业控制", "传感器", "电机"),
|
"风电、核电设备或运营": ("风电", "核电", "风机", "叶片", "电力", "发电"),
|
"光伏或锂电材料、设备、组件与系统": ("光伏", "锂电", "电池", "储能", "逆变器", "硅片", "组件"),
|
"农业、养殖、化肥或化工": ("农业", "种业", "养殖", "饲料", "农药", "化肥", "化工"),
|
"AI应用、软件、安全或信创": ("软件", "信息安全", "网络安全", "数据服务", "云计算", "信息技术"),
|
"消费、地产、旅游、影视或事件驱动公司": ("消费", "食品", "饮料", "旅游", "酒店", "影视", "地产", "商业"),
|
"医药、创新药、疫苗或医疗器械": ("医药", "药品", "疫苗", "医疗", "生物制品", "诊断"),
|
"军工电子、航空航天或特种装备": ("军工", "航空", "航天", "雷达", "特种装备", "军用", "无人机"),
|
}
|
|
|
def read_json(path: Path) -> dict[str, Any]:
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
def write_json(path: Path, value: Any) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
def normalize_pipeline_report(calc_dir: Path) -> None:
|
"""Remove generated Markdown trailing spaces and keep its manifest exact."""
|
report_path = calc_dir / "valuation_report.md"
|
manifest_path = calc_dir / "run_manifest.json"
|
content = report_path.read_text(encoding="utf-8")
|
normalized = "\n".join(line.rstrip() for line in content.splitlines()) + "\n"
|
report_path.write_text(normalized, encoding="utf-8")
|
manifest = read_json(manifest_path)
|
payload = report_path.read_bytes()
|
manifest["artifacts"]["valuation_report.md"] = {
|
"bytes": len(payload),
|
"sha256": hashlib.sha256(payload).hexdigest(),
|
}
|
write_json(manifest_path, manifest)
|
|
|
def safe_name(value: str) -> str:
|
return re.sub(r'[<>:"/\\|?*]', "", value).strip() or "公司"
|
|
|
def valuation_position_pct(price: float, base_low: float, base_high: float) -> float:
|
"""Comparable valuation-position score requested by the operating manual.
|
|
Below the range it is the discount to the lower bound; inside the range it
|
is the deviation from the midpoint; above the range it is the premium to
|
the upper bound. Ascending order therefore consistently means cheaper.
|
"""
|
if price < base_low:
|
return price / base_low - 1
|
if price <= base_high:
|
midpoint = (base_low + base_high) / 2
|
return price / midpoint - 1
|
return price / base_high - 1
|
|
|
def number(row: dict[str, Any], *keys: str, default: float | None = None) -> float | None:
|
for key in keys:
|
value = row.get(key)
|
if value not in (None, ""):
|
try:
|
return float(value)
|
except (TypeError, ValueError):
|
pass
|
return default
|
|
|
def force_reclass_tickers() -> set[str]:
|
if not SCAN_PATH.exists():
|
return set()
|
with SCAN_PATH.open(encoding="utf-8", newline="") as handle:
|
return {
|
row["ticker"]
|
for row in csv.DictReader(handle)
|
if "RESEARCH_TAG_MODEL_ROUTE" in row.get("flags", "")
|
}
|
|
|
def classify(business: dict[str, str]) -> str:
|
text = " ".join(business.values()).lower()
|
rules = (
|
("bank", r"银行业务|发放贷款|公司金融业务|个人金融业务"),
|
("financial", r"证券经纪|证券投资|保险业务|期货经纪|信托业务|融资租赁"),
|
("military", r"军工|军用|航空装备|航天|雷达|导弹|弹药|火工品|特种装备|无人机"),
|
("memory_module", r"存储器|存储控制|固态硬盘|内存条|移动存储|存储卡模组|存储盘模组"),
|
("semi_equipment", r"半导体设备|刻蚀设备|薄膜沉积|离子注入|光刻设备|晶圆制造设备|eda|掩膜版"),
|
("semi_packaging", r"封装测试|封装、测试|集成电路封测"),
|
("semi_foundry", r"晶圆代工|晶圆制造|硅片|外延片|化合物半导体|砷化镓|氮化镓"),
|
("semi_material", r"半导体材料|电子化学品|光刻胶|抛光液|靶材|电子特气|高纯气体"),
|
("semi_design", r"集成电路设计|芯片设计|芯片研发|芯片产品|处理器|fpga|soc芯片|存储芯片"),
|
("optical_ai", r"光模块|光通信|服务器|数据中心|算力|网络设备|通信设备|射频器件|散热"),
|
("software", r"软件开发|软件产品|网络安全|信息安全|数据服务|云计算|信息技术服务|系统集成"),
|
("robotics", r"机器人|自动化设备|工业自动化|伺服|减速器|减速机|工业控制|数控机床|传感器|机器视觉"),
|
("battery_solar", r"光伏|太阳能|锂电|动力电池|储能电池|电池材料|正极材料|负极材料|隔膜|电解液|逆变器"),
|
("wind_power", r"风电|风力发电|核电|电网|输配电|电力设备|变压器|电气设备"),
|
("resource", r"金矿|银矿|铜矿|铝矿|锌矿|铅矿|锂矿|稀土|钨矿|钼矿|矿产开采|有色金属采选"),
|
("coal_energy", r"煤炭开采|煤炭生产|油气开采|石油开采|天然气开采|煤化工"),
|
("pharma", r"药品|医药|原料药|制剂|疫苗|医疗器械|体外诊断|生命科学|cro|cdmo"),
|
("agriculture", r"种业|种子|养殖|饲料|农药|化肥|复合肥|农业服务|农产品"),
|
("chemical", r"化工产品|化学品|氟化物|农化|涂料|树脂|橡胶|塑料助剂|化学材料"),
|
("property", r"房地产开发|建筑施工|工程承包|基础设施建设|园林工程"),
|
("transport", r"航空运输|港口|高速公路|铁路运输|物流服务|水务|燃气供应"),
|
("consumer", r"食品|饮料|酒类|服装|家居|家电|旅游|酒店|影视|零售|珠宝"),
|
)
|
for tier, pattern in rules:
|
if re.search(pattern, text, re.I):
|
return tier
|
return "industrial"
|
|
|
def old_type_valid(old_type: str, business: dict[str, str], forced: bool) -> bool:
|
if forced:
|
return False
|
keywords = OLD_TYPE_KEYWORDS.get(old_type)
|
if keywords is None:
|
return True
|
text = " ".join(business.values()).lower()
|
return any(keyword.lower() in text for keyword in keywords)
|
|
|
def financial_period(row: dict[str, Any], cash: dict[str, Any] | None, period_end: str) -> dict[str, Any]:
|
attributable = number(row, "PARENT_NETPROFIT", "PARENTNETPROFIT", default=0.0) or 0.0
|
deduct = number(row, "DEDUCT_PARENT_NETPROFIT", "KCFJCXSYJLR", default=attributable)
|
return {
|
"period_end": period_end,
|
"basis": "audited" if period_end.endswith("12-31") else "interim_report_unaudited",
|
"revenue": number(row, "TOTAL_OPERATE_INCOME", "TOTALOPERATEREVE", default=0.0) or 0.0,
|
"attributable_profit": attributable,
|
"deduct_profit": deduct if deduct is not None else attributable,
|
"cfo": number(cash or {}, "NETCASH_OPERATE"),
|
"capex": abs(number(cash or {}, "CONSTRUCT_LONG_ASSET")) if number(cash or {}, "CONSTRUCT_LONG_ASSET") is not None else None,
|
}
|
|
|
def scenario_bands(old: dict[str, Any], tier: str, preserve: bool, method: str) -> tuple[tuple[float, float], ...]:
|
if preserve:
|
scenarios = sorted(old["valuation"]["scenarios"], key=lambda x: {"pessimistic": 0, "base": 1, "optimistic": 2}.get(x["role"], 9))
|
if len(scenarios) >= 3 and all(item.get("method") == method for item in scenarios[:3]):
|
return tuple((float(item["multiple_low"]), float(item["multiple_high"])) for item in scenarios[:3])
|
rule = TIER_RULES[tier]
|
if method == rule["method"]:
|
return rule["bands"]
|
if method == "pb":
|
return ((0.5, 0.9), (0.9, 1.5), (1.5, 2.4))
|
return ((10, 16), (16, 25), (25, 36))
|
|
|
def build_scenarios(
|
*, method: str, bands: tuple[tuple[float, float], ...], normalized: float,
|
consensus: float | None, consensus_count: int, tier: str,
|
) -> list[dict[str, Any]]:
|
roles = (("悲观", "pessimistic"), ("基准", "base"), ("乐观", "optimistic"))
|
if method == "pb":
|
assumptions = (
|
"盈利或资产回报继续承压，按净资产折价情景定价。",
|
"盈利、资产质量和回报率处于中性路径，按基准PB区间定价。",
|
"盈利修复、资产周转或ROE明显改善，按乐观PB区间定价。",
|
)
|
return [
|
{"name": name, "role": role, "method": "pb", "multiple_low": band[0], "multiple_high": band[1], "assumption": assumption}
|
for (name, role), band, assumption in zip(roles, bands, assumptions)
|
]
|
anchor = consensus if consensus is not None and consensus > 0 else normalized
|
anchor = max(anchor, 1.0)
|
base_low = max(anchor * 0.78, normalized * 0.85 if normalized > 0 else 0.0, 1.0)
|
base_high = max(anchor * (1.18 if consensus_count >= 2 else 1.25), normalized * 1.15 if normalized > 0 else 0.0, base_low)
|
profits = ((base_low * 0.62, base_low * 0.88), (base_low, base_high), (base_high * 1.10, base_high * 1.38))
|
assumptions = (
|
"需求、价格、订单或验收低于预期，利润与估值倍数同步回落。",
|
f"以最新TTM扣非利润和{consensus_count}家有效机构预测为锚，按{TIER_RULES[tier]['name']}的周期与质量定价。",
|
"经营兑现显著超预期，但避免利润和估值倍数无限双重上调。",
|
)
|
return [
|
{
|
"name": name, "role": role, "method": "pe",
|
"profit_low": profit[0], "profit_high": profit[1],
|
"multiple_low": band[0], "multiple_high": band[1], "assumption": assumption,
|
}
|
for (name, role), profit, band, assumption in zip(roles, profits, bands, assumptions)
|
]
|
|
|
def source_url(code: str, kind: str) -> str:
|
if kind == "finance":
|
return "https://datacenter-web.eastmoney.com/api/data/v1/get"
|
if kind == "forecast":
|
return f"https://basic.10jqka.com.cn/{code}/worth.html"
|
if kind == "business":
|
return f"https://basic.10jqka.com.cn/{code}/operate.html"
|
return ""
|
|
|
def included_forecasts(raw: dict[str, Any], notice_date: str, current_profit: float, shares: float) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
included: list[dict[str, Any]] = []
|
excluded: list[dict[str, Any]] = []
|
for item in raw.get("forecasts", []):
|
profit = number(item, "profit_2026")
|
is_current = item.get("report_date", "") >= notice_date
|
clears_achieved = profit is not None and (current_profit <= 0 or profit >= current_profit * 0.95)
|
use = bool(is_current and clears_achieved and profit is not None and profit > 0)
|
estimates = {}
|
for year in (2026, 2027, 2028):
|
year_profit = number(item, f"profit_{year}")
|
eps = number(item, f"eps_{year}")
|
if year_profit is not None and year_profit > 0:
|
estimates[str(year)] = {"profit": year_profit, "eps": eps if eps is not None else year_profit / shares}
|
record = {
|
"institution": item.get("institution") or "未命名机构",
|
"report_date": item["report_date"],
|
"include": use,
|
"core_assumption": "研报日期不早于最新经营披露，且全年利润不低于已实现累计利润。" if use else "早于最新披露、全年利润低于已实现累计利润或关键值缺失，仅保留作预期差证据。",
|
"source_id": "SRC-THS-FORECAST-DETAIL-20260826",
|
"estimates": estimates,
|
}
|
if estimates:
|
(included if use else excluded).append(record)
|
return included, excluded
|
|
|
def balance_sheet(raw: dict[str, Any], old: dict[str, Any]) -> dict[str, Any]:
|
bal = raw["selected"].get("current_balance") or {}
|
main = raw["selected"].get("current_main") or {}
|
old_bal = old["balance_sheet"]
|
parent_equity = number(bal, "TOTAL_EQUITY_PARENT")
|
if parent_equity is None:
|
total_equity = number(bal, "TOTAL_EQUITY")
|
minority = number(bal, "MINORITY_EQUITY", default=0.0) or 0.0
|
parent_equity = total_equity - minority if total_equity is not None else number(main, "TOTAL_EQUITY_PARENT", default=float(old_bal["equity"]))
|
equity = parent_equity
|
monetary = max(0.0, number(bal, "MONETARYFUNDS", default=float(old_bal["cash_available"])) or 0.0)
|
old_cash = max(0.0, float(old_bal["cash_available"]))
|
cash_available = min(monetary, old_cash) if old_cash < monetary else monetary
|
fv_keys = (
|
"TRADE_FINASSET_NOTFVTPL", "TRADE_FINASSET", "FVTPL_FINASSET", "APPOINT_FVTPL_FINASSET",
|
"AVAILABLE_SALE_FINASSET", "DERIVE_FINASSET", "BUY_RESALE_FINASSET",
|
)
|
debt_keys = ("SHORT_LOAN", "NONCURRENT_LIAB_1YEAR", "LONG_LOAN", "BOND_PAYABLE", "LEASE_LIAB", "SHORT_BOND_PAYABLE")
|
financial_assets = sum(max(0.0, number(bal, key, default=0.0) or 0.0) for key in fv_keys)
|
debt = sum(max(0.0, number(bal, key, default=0.0) or 0.0) for key in debt_keys)
|
return {
|
"period_end": raw["periods"]["current"],
|
"equity": equity if equity is not None else float(old_bal["equity"]),
|
"cash_available": cash_available,
|
"non_operating_financial_assets": financial_assets,
|
"interest_bearing_debt": debt,
|
"minority_interest": max(0.0, number(bal, "MINORITY_EQUITY", default=float(old_bal["minority_interest"])) or 0.0),
|
}
|
|
|
def build_one(item: dict[str, Any], forced: set[str]) -> dict[str, Any]:
|
ticker = item["ticker"]
|
if ticker.endswith(".HK"):
|
return {"ticker": ticker, "company": item["company"], "status": "UNSUPPORTED_HK"}
|
old = read_json(ROOT / item["snapshot_path"])
|
if item.get("close") is None:
|
out_dir = CURRENT_ROOT / ticker
|
out_dir.mkdir(parents=True, exist_ok=True)
|
last_price = float(old["market"]["price"])
|
last_price_date = str(old["market"].get("price_timestamp") or "")[:10]
|
gap_snapshot = {
|
"status": "PRICE_GAP",
|
"batch_id": BATCH_ID,
|
"meta": {"company": item["company"], "code": ticker, "as_of_date": AS_OF},
|
"required_price_date": PRICE_DATE,
|
"last_valid_price": last_price,
|
"last_valid_price_date": last_price_date,
|
"retained_valuation_id": item["valuation_id"],
|
"retained_base_low": item["base_low"],
|
"retained_base_high": item["base_high"],
|
}
|
write_json(out_dir / f"{safe_name(item['company'])}估值快照_当前.json", gap_snapshot)
|
calc_dir = out_dir / "calculation"
|
calc_dir.mkdir(parents=True, exist_ok=True)
|
for stale_name in ("valuation_results.json", "valuation_report.md", "run_manifest.json"):
|
stale_path = calc_dir / stale_name
|
if stale_path.exists():
|
stale_path.unlink()
|
formal_path = out_dir / f"{safe_name(item['company'])}价格合理性评估_当前.md"
|
formal_path.write_text(
|
"\n".join(
|
[
|
f"# {item['company']}当前价格合理性评估", "", "## 结论", "",
|
f"- {PRICE_DATE}是本轮统一价格日，但前复权专表没有该股当日日K，因此**不生成或伪造当日价格判定**。",
|
f"- 最近可核实价格为{last_price:.2f}元（{last_price_date}）；原基准合理区间{float(item['base_low']):.2f}—{float(item['base_high']):.2f}元继续作为最近有效价值版本，但不能冒充{PRICE_DATE}的新判定。",
|
"- 后续前复权专表补齐复牌/有效交易日数据后，再按每日流程恢复价格判定。", "",
|
"## 边界", "",
|
"- 本报告是数据缺口说明；不构成交易指令或收益承诺。", "",
|
]
|
),
|
encoding="utf-8",
|
)
|
write_json(
|
out_dir / "source_evidence_manifest.json",
|
{"batch_id": BATCH_ID, "ticker": ticker, "company": item["company"], "as_of": AS_OF,
|
"status": "PRICE_GAP", "required_price_date": PRICE_DATE, "last_valid_price_date": last_price_date,
|
"retained_valuation_id": item["valuation_id"]},
|
)
|
return {
|
"ticker": ticker, "company": item["company"], "status": "PRICE_GAP",
|
"errors": [f"{PRICE_DATE}前复权专表无当日日K，不回退旧价"],
|
"last_price": last_price, "last_price_date": last_price_date,
|
"report_path": formal_path.relative_to(ROOT).as_posix(),
|
}
|
raw = read_json(RAW_ROOT / f"{ticker.replace('.', '_')}.json")
|
if not raw.get("selected"):
|
return {"ticker": ticker, "company": item["company"], "status": "DATA_GAP", "errors": raw.get("errors", [])}
|
annual_end = raw["periods"]["annual"]
|
current_end = raw["periods"]["current"]
|
prior_end = raw["periods"]["prior"]
|
selected = raw["selected"]
|
financials = {
|
"annual": financial_period(selected["annual_income"], selected.get("annual_cashflow"), annual_end),
|
"current_cumulative": financial_period(selected["current_income"], selected.get("current_cashflow"), current_end),
|
"prior_year_same_period": financial_period(selected["prior_income"], selected.get("prior_cashflow"), prior_end),
|
}
|
annual = financials["annual"]
|
current = financials["current_cumulative"]
|
prior = financials["prior_year_same_period"]
|
ttm_attr = annual["attributable_profit"] + current["attributable_profit"] - prior["attributable_profit"]
|
ttm_deduct = annual["deduct_profit"] + current["deduct_profit"] - prior["deduct_profit"]
|
ttm_revenue = annual["revenue"] + current["revenue"] - prior["revenue"]
|
# Latest statutory report equity units take precedence over the older
|
# instrument-static snapshot. This matters after bonus issues, splits,
|
# conversions and other H1 capital changes.
|
reported_shares = number(selected.get("current_main") or {}, "TOTAL_SHARE")
|
shares = reported_shares or item.get("static_total_shares") or float(old["market"]["diluted_shares"])
|
shares = float(shares)
|
if shares < 1_000_000 and float(old["market"]["diluted_shares"]) > 10_000_000:
|
shares *= 10000
|
shares_date = current_end if reported_shares else PRICE_DATE
|
price = item.get("close") or float(old["market"]["price"])
|
notice_date = str(selected["current_income"].get("NOTICE_DATE") or selected["current_income"].get("UPDATE_DATE"))[:10]
|
included, excluded = included_forecasts(raw, notice_date, current["attributable_profit"], shares)
|
consensus_values = [forecast["estimates"]["2026"]["profit"] for forecast in included if "2026" in forecast["estimates"]]
|
consensus = statistics.median(consensus_values) if consensus_values else None
|
tier = classify(raw["business"])
|
old_type = str(old.get("analysis", {}).get("company_type") or "")
|
preserve = old_type_valid(old_type, raw["business"], ticker in forced)
|
if tier == "memory_module" and "强周期" not in old_type:
|
preserve = False
|
old_method = str(old["valuation"]["scenarios"][0]["method"])
|
default_method = str(TIER_RULES[tier]["method"])
|
method = old_method if preserve else default_method
|
quality_override_reason = ""
|
ttm_margin = ttm_deduct / ttm_revenue if ttm_revenue > 0 else None
|
if method == "pe" and ttm_deduct <= 0:
|
method = "pb"
|
preserve = False
|
quality_override_reason = "TTM扣非亏损，停止用尚未兑现的远期预测机械套PE，改用资产口径"
|
elif method == "pe" and ttm_margin is not None and ttm_margin < 0.01 and len(consensus_values) < 3:
|
method = "pb"
|
preserve = False
|
quality_override_reason = "TTM扣非净利率低于1%且机构覆盖不足，PE对微小利润过度敏感"
|
elif method == "pe" and ttm_margin is not None and ttm_margin > 1.0:
|
method = "pb"
|
preserve = False
|
quality_override_reason = "TTM扣非利润超过TTM营业收入，盈利锚异常，改用资产口径并下调置信度"
|
normalized_anchor = ttm_deduct
|
if (
|
method == "pe" and consensus is not None and len(consensus_values) >= 3
|
and ttm_deduct > consensus * 1.5
|
):
|
normalized_anchor = consensus
|
quality_override_reason = (
|
f"TTM扣非利润显著高于最新{len(consensus_values)}家机构预测中位数，"
|
"按最新一致预期压低正常化盈利，避免一次性高利润外推"
|
)
|
old_info_for_forecast = str(old["meta"]["latest_operating_info_date"])
|
old_current_profit = float(old["financials"]["current_cumulative"]["attributable_profit"])
|
old_included = {
|
(
|
str(row.get("institution") or ""), str(row.get("report_date") or ""),
|
round(float(((row.get("estimates") or {}).get("2026") or {}).get("profit") or 0.0), 2),
|
)
|
for row in old.get("institutions", {}).get("forecasts", [])
|
if row.get("include", True)
|
and str(row.get("report_date") or "") >= old_info_for_forecast
|
and float(((row.get("estimates") or {}).get("2026") or {}).get("profit") or 0.0) >= (old_current_profit * 0.95 if old_current_profit > 0 else 0.0)
|
}
|
new_included = {
|
(
|
str(row.get("institution") or ""), str(row.get("report_date") or ""),
|
round(float(((row.get("estimates") or {}).get("2026") or {}).get("profit") or 0.0), 2),
|
)
|
for row in included
|
}
|
financial_update = notice_date > str(old["meta"]["latest_operating_info_date"])
|
reuse_prior_judgment = bool(preserve and not financial_update and old_included == new_included)
|
if reuse_prior_judgment:
|
scenarios = json.loads(json.dumps(old["valuation"]["scenarios"], ensure_ascii=False))
|
method = str(scenarios[0]["method"])
|
else:
|
bands = scenario_bands(old, tier, preserve, method)
|
scenarios = build_scenarios(method=method, bands=bands, normalized=normalized_anchor, consensus=consensus, consensus_count=len(consensus_values), tier=tier)
|
bal = balance_sheet(raw, old)
|
if method == "pb" and bal["equity"] <= 0:
|
bal["equity"] = 0.0
|
company_type = old_type if preserve and old_type else TIER_RULES[tier]["name"]
|
code = ticker.split(".")[0]
|
sources = [
|
{
|
"id": "SRC-EM-FINANCE-ANNUAL-20260826", "source_type": "financial_mirror",
|
"title": f"东方财富结构化财务：{annual_end}年度/比较基础", "publish_date": str(selected["annual_income"].get("NOTICE_DATE") or selected["annual_income"].get("UPDATE_DATE"))[:10],
|
"period_end": annual_end, "url": source_url(code, "finance"),
|
"supports": ["financials.annual"], "revision_status": "current",
|
},
|
{
|
"id": "SRC-EM-FINANCE-CURRENT-20260826", "source_type": "financial_mirror",
|
"title": f"东方财富结构化财务：{current_end}最新累计期", "publish_date": notice_date,
|
"period_end": current_end, "url": source_url(code, "finance"),
|
"supports": ["financials.current_cumulative", "balance_sheet.equity"], "revision_status": "current",
|
},
|
{
|
"id": "SRC-EM-FINANCE-PRIOR-20260826", "source_type": "financial_mirror",
|
"title": f"东方财富结构化财务：{prior_end}同期比较", "publish_date": notice_date,
|
"period_end": prior_end, "url": source_url(code, "finance"),
|
"supports": ["financials.prior_year_same_period"], "revision_status": "current",
|
},
|
{
|
"id": "SRC-LOCAL-FRONT-CLOSE-20260825", "source_type": "quote_provider",
|
"title": "trading_xuntou前复权完整交易日收盘与当前股本勾稽", "publish_date": PRICE_DATE,
|
"period_end": PRICE_DATE, "url": None,
|
"supports": ["market.price", "market.diluted_shares", "market.platform_market_cap"], "revision_status": "current",
|
},
|
]
|
if included or excluded:
|
sources.append(
|
{
|
"id": "SRC-THS-FORECAST-DETAIL-20260826", "source_type": "institution_aggregator",
|
"title": "同花顺盈利预测逐家明细（使用真实研报日期）", "publish_date": max(item["report_date"] for item in included + excluded),
|
"period_end": max(item["report_date"] for item in included + excluded), "url": source_url(code, "forecast"),
|
"supports": ["institutions"], "revision_status": "current",
|
}
|
)
|
old_info_date = str(old["meta"]["latest_operating_info_date"])
|
reclassified = not preserve
|
snapshot = {
|
"snapshot_id": f"{ticker.replace('.', '-')}-midyear-current",
|
"meta": {
|
"company": item["company"], "code": ticker, "market": item["market"], "as_of_date": AS_OF,
|
"currency": item["currency"], "report_period_end": current_end, "latest_operating_info_date": notice_date,
|
},
|
"market": {
|
"price": price, "price_type": "前复权完整交易日收盘价", "price_timestamp": f"{PRICE_DATE}T15:00:00+08:00",
|
"diluted_shares": shares, "shares_date": shares_date, "platform_market_cap": price * shares,
|
},
|
"financials": financials,
|
"balance_sheet": bal,
|
"normalization": {
|
"adjustments": [{
|
"description": (
|
"以最新机构一致预期对异常高TTM扣非利润做正常化约束"
|
if normalized_anchor != ttm_deduct
|
else "以最新TTM扣非归母利润替代TTM归母利润作为核心盈利代理"
|
),
|
"amount": normalized_anchor - ttm_attr,
|
"basis": f"{annual_end}+{current_end}-{prior_end}",
|
}],
|
},
|
"valuation": {
|
"scenarios": scenarios,
|
"reverse_pe_multiples": [6, 8, 10, 12, 15, 18, 20, 25, 30, 35, 40, 50, 60, 75, 100],
|
"exit_pe_multiples": [8, 10, 12, 15, 18, 20, 25, 30, 35, 40, 50],
|
"holding_period": {"years": 5, "required_return": 0.10, "cumulative_dividend_per_share": 0},
|
},
|
"institutions": {"coverage_status": "available" if included else "no_usable_forecasts", "forecasts": included},
|
"sources": sources,
|
"analysis": {
|
"company_type": company_type,
|
"business_identity": raw["business"],
|
"business_identity_source": source_url(code, "business"),
|
"primary_model": f"{method.upper()}三情景；最新TTM扣非、真实日期机构预测与资产负债交叉验证",
|
"cross_checks": ["最新TTM扣非利润", "机构预测真实日期", "PB/PS", "经营现金流", "反向估值"],
|
"profit_sources": [f"主营业务：{raw['business'].get('main_business') or '公开经营分析未返回'}", f"产品类型：{raw['business'].get('product_types') or '未披露'}"],
|
"profit_source_quality": [f"TTM归母{ttm_attr/1e8:.2f}亿元，TTM扣非{ttm_deduct/1e8:.2f}亿元。", f"有效机构2026预测{len(consensus_values)}家；剔除滞后或低于已实现利润的预测{len(excluded)}家。"],
|
"risks": ["中报仍可能未经审计，全年季节性、价格周期和一次性因素可能令TTM失真。", "结构化财务为C1公告镜像，正式复核时优先回到交易所/巨潮原文。", "合理区间是条件化估值，不是目标价。"],
|
"upgrade_triggers": ["后续财报扣非利润和经营现金流同时高于基准路径。", "机构在最新披露后上调盈利且经营数据验证。"],
|
"downgrade_triggers": ["单季利润显著回落、扣非与现金流背离或机构预测再次下修。", "主营身份、股本、会计口径或重大事项改变。"],
|
"executive_conclusion": "由冻结V1内核计算后生成。",
|
"old_valuation_date": item["valuation_date"], "old_latest_operating_info_date": old_info_date,
|
"financial_update": financial_update, "model_reclassified": reclassified, "old_company_type": old_type,
|
"reused_prior_judgment": reuse_prior_judgment,
|
"classification_tier": tier, "classification_confidence": "中低" if reclassified else "中",
|
"quality_model_override_reason": quality_override_reason,
|
"excluded_institution_forecasts": excluded, "information_cutoff": f"{AS_OF}T10:19:24+08:00",
|
"raw_sha256": raw.get("raw_sha256", {}),
|
},
|
}
|
out_dir = CURRENT_ROOT / ticker
|
out_dir.mkdir(parents=True, exist_ok=True)
|
snapshot_path = out_dir / f"{safe_name(item['company'])}估值快照_当前.json"
|
write_json(snapshot_path, snapshot)
|
calc_dir = out_dir / "calculation"
|
if method == "pb" and bal["equity"] <= 0:
|
calc_dir.mkdir(parents=True, exist_ok=True)
|
for stale_name in ("valuation_results.json", "valuation_report.md", "run_manifest.json"):
|
stale_path = calc_dir / stale_name
|
if stale_path.exists():
|
stale_path.unlink()
|
formal_path = out_dir / f"{safe_name(item['company'])}价格合理性评估_当前.md"
|
formal_path.write_text(
|
"\n".join(
|
[
|
f"# {item['company']}当前价格合理性评估",
|
"",
|
"## 结论",
|
"",
|
f"- 当前价格：{price:.2f}元（{PRICE_DATE}前复权完整交易日收盘）。",
|
f"- 最新经营期：{current_end}；TTM归母/扣非：{ttm_attr/1e8:.2f}/{ttm_deduct/1e8:.2f}亿元。",
|
"- **本轮不输出虚假的数值合理区间。** 公司最新归母净资产不为正，同时TTM扣非利润为负，冻结V1内核支持的PE与PB都失去有效锚点。",
|
"- 当前状态：估值模型缺口；旧合理区间停止作为当前判断依据。若后续净资产转正、形成可持续盈利或完成可核实资产重组，再重新建立区间。",
|
"",
|
"## 风险与泡沫解释",
|
"",
|
"- 在负净资产、持续亏损条件下，价格主要反映保壳、重整、资产注入或经营扭转的期权价值；这些预期无法由当前盈利和净资产验证。",
|
"- 这不是‘低估’，也不能因旧PB区间很低而机械判成‘明显偏贵’；应单列为高不确定性、无法可靠数值估值。",
|
"",
|
"## 证据与边界",
|
"",
|
f"- 财务镜像更新日：{notice_date}；信息截止：{AS_OF}。",
|
"- 财务数字来自登记C1结构化公告镜像，正式决策应回到交易所/巨潮原文复核。",
|
"- 不构成交易指令或收益承诺。",
|
"",
|
]
|
),
|
encoding="utf-8",
|
)
|
write_json(
|
out_dir / "source_evidence_manifest.json",
|
{
|
"batch_id": BATCH_ID,
|
"ticker": ticker,
|
"company": item["company"],
|
"as_of": AS_OF,
|
"price_date": PRICE_DATE,
|
"status": "VALUATION_MODEL_GAP",
|
"reason": "latest parent equity <= 0 and TTM deduct profit <= 0; frozen V1 PE/PB cannot form a valid range",
|
"raw_sha256": raw.get("raw_sha256", {}),
|
},
|
)
|
return {
|
"ticker": ticker,
|
"company": item["company"],
|
"status": "VALUATION_MODEL_GAP",
|
"errors": ["最新归母净资产不为正且TTM扣非亏损，PE/PB均无有效锚点，旧区间停止使用"],
|
"report_path": formal_path.relative_to(ROOT).as_posix(),
|
}
|
run = run_pipeline(snapshot_path, calc_dir, REGISTRY)
|
normalize_pipeline_report(calc_dir)
|
results = read_json(calc_dir / "valuation_results.json")
|
if int(results.get("qa", {}).get("error_count") or 0) > 0:
|
issues = results.get("qa", {}).get("issues") or []
|
raise RuntimeError(f"V1 QA blocked: {issues[:3]}")
|
base_result = next(row for row in results["scenarios"] if row["role"] == "base")
|
optimistic_result = next(row for row in results["scenarios"] if row["role"] == "optimistic")
|
base_low = float(base_result["price_low"])
|
base_high = float(base_result["price_high"])
|
optimistic_high = float(optimistic_result["price_high"])
|
if price < base_low:
|
label = "偏低"
|
elif price <= base_high:
|
label = "基本合理"
|
elif price <= optimistic_high:
|
label = "偏贵"
|
else:
|
label = "明显偏贵"
|
premium = price / base_high - 1 if base_high > 0 else math.inf
|
vs_optimistic = price / optimistic_high - 1 if optimistic_high > 0 else math.inf
|
if premium <= 0:
|
bubble = "未识别估值泡沫"
|
elif vs_optimistic > 0:
|
bubble = "泡沫-极端"
|
elif premium <= 0.15:
|
bubble = "泡沫-轻"
|
elif premium <= 0.50:
|
bubble = "泡沫-中"
|
else:
|
bubble = "泡沫-高"
|
yoy_profit = None
|
if prior["attributable_profit"] not in (None, 0):
|
yoy_profit = current["attributable_profit"] / prior["attributable_profit"] - 1
|
if bubble == "未识别估值泡沫":
|
bubble_reason = "当前价格未高于基准合理区间上沿，按统一规则不认定估值泡沫。"
|
else:
|
signals = [
|
f"价格高于基准上沿{premium*100:.1f}%",
|
f"市场提前交易{company_type}的增长、修复或份额提升预期",
|
]
|
if vs_optimistic > 0:
|
signals.append(f"且高于乐观上沿{vs_optimistic*100:.1f}%，包含模型之外的额外叙事溢价")
|
if yoy_profit is not None:
|
signals.append(f"最新累计归母利润同比{'增长' if yoy_profit >= 0 else '下降'}{abs(yoy_profit)*100:.1f}%")
|
if current.get("cfo") is not None and float(current["cfo"]) < 0:
|
signals.append("经营现金流为负，利润兑现质量尚未完全验证")
|
if not consensus_values:
|
signals.append("最新披露后没有可用机构盈利预测，乐观预期缺少外部预测交叉验证")
|
elif len(consensus_values) <= 2:
|
signals.append(f"有效机构预测仅{len(consensus_values)}家，样本偏少")
|
if reclassified:
|
signals.append("原估值模型与真实主营不完全匹配，本轮已重分类，历史题材溢价需折价看待")
|
bubble_reason = "；".join(signals) + "。"
|
position_pct = valuation_position_pct(price, base_low, base_high)
|
calc_text = (calc_dir / "valuation_report.md").read_text(encoding="utf-8")
|
intro = (
|
"\n## 本轮中报更新\n\n"
|
f"- 当前价格：{price:.2f}元（{PRICE_DATE}前复权完整交易日收盘）；信息截止：{AS_OF} 10:19:24。\n"
|
f"- 最新经营期：{current_end}，披露/镜像更新日：{notice_date}；上次经营信息日：{old_info_date}。\n"
|
f"- 主营身份：{company_type}；原模型：{old_type or '未记录'}；模型{'已重分类' if reclassified else '延续已核实模型'}。\n"
|
f"- TTM归母/扣非：{ttm_attr/1e8:.2f}/{ttm_deduct/1e8:.2f}亿元；有效机构2026预测：{len(consensus_values)}家，中位数：{('—' if consensus is None else f'{consensus/1e8:.2f}亿元')}。\n"
|
f"- 盈利质量模型修正：{quality_override_reason or '未触发异常盈利锚修正'}。\n"
|
f"- 基准合理区间：**{base_low:.2f}—{base_high:.2f}元**；乐观上沿：{optimistic_high:.2f}元；当前判断：**{label}**；泡沫状态：**{bubble}**。\n"
|
f"- 相对基准上沿：{premium*100:.2f}%；相对乐观上沿：{vs_optimistic*100:.2f}%。\n"
|
"- 财务核心数字来自登记C1结构化公告镜像；机构预测只使用真实研报日期，早于最新披露或低于已实现利润的记录已排除。\n"
|
"\n## 泡沫识别与原因\n\n"
|
f"- 判定：**{bubble}**。{bubble_reason}\n"
|
"- 原因性质：除价格位置外均为基于最新财务、真实主营与机构覆盖的解释性推断，不等同于已证实的资金流因果。\n"
|
)
|
formal = calc_text.replace("\n## 1. 结论先行", intro + "\n## 1. 结论先行", 1)
|
formal = formal.replace("价格合理性评估（流水线生成底稿）", "当前价格合理性评估", 1)
|
formal_path = out_dir / f"{safe_name(item['company'])}价格合理性评估_当前.md"
|
formal_path.write_text("\n".join(line.rstrip() for line in formal.splitlines()) + "\n", encoding="utf-8")
|
evidence = {
|
"batch_id": BATCH_ID, "ticker": ticker, "company": item["company"], "as_of": AS_OF,
|
"price_date": PRICE_DATE, "old_valuation_id": item["valuation_id"], "old_valuation_date": item["valuation_date"],
|
"financial_update": financial_update, "model_reclassified": reclassified, "preserved_old_multiples": preserve,
|
"reused_prior_judgment": reuse_prior_judgment,
|
"raw_sha256": raw.get("raw_sha256", {}), "qa": results["qa"], "run_status": run["status"],
|
"excluded_forecast_count": len(excluded), "included_forecast_count": len(included),
|
}
|
write_json(out_dir / "source_evidence_manifest.json", evidence)
|
source_hash = hashlib.sha256(snapshot_path.read_bytes()).hexdigest().upper()
|
valuation_id = "VAL-" + hashlib.sha256(f"{ticker}|{AS_OF}|{source_hash}".encode()).hexdigest()[:24]
|
return {
|
"ticker": ticker, "company": item["company"], "market": item["market"], "currency": item["currency"],
|
"status": "COMPLETE", "valuation_id": valuation_id, "valuation_date": AS_OF, "price_date": PRICE_DATE,
|
"close": price, "shares": shares, "market_cap": price * shares, "latest_period": current_end,
|
"latest_notice_date": notice_date, "old_valuation_date": item["valuation_date"], "financial_update": financial_update,
|
"model_reclassified": reclassified, "old_company_type": old_type, "company_type": company_type,
|
"method": method.upper(), "normalized_profit": float(results["metrics"]["normalized_profit"]),
|
"normalized_pe": float(results["metrics"]["normalized_pe"]) if results["metrics"].get("normalized_pe") is not None else None,
|
"pb": float(results["metrics"]["pb"]) if results["metrics"].get("pb") is not None else None,
|
"ps": float(results["metrics"]["ps"]) if results["metrics"].get("ps") is not None else None,
|
"consensus_year": 2026 if consensus is not None else None, "consensus_profit": consensus,
|
"consensus_count": len(consensus_values), "excluded_forecast_count": len(excluded),
|
"pessimistic_low": float(next(row for row in results["scenarios"] if row["role"] == "pessimistic")["price_low"]),
|
"pessimistic_high": float(next(row for row in results["scenarios"] if row["role"] == "pessimistic")["price_high"]),
|
"base_low": base_low, "base_high": base_high,
|
"optimistic_low": float(optimistic_result["price_low"]), "optimistic_high": optimistic_high,
|
"label": label, "bubble_status": bubble, "bubble_reason": bubble_reason,
|
"valuation_position_pct": position_pct,
|
"premium_to_base_high": premium, "vs_optimistic_high": vs_optimistic,
|
"confidence": "中低" if reclassified or raw["status"] != "OK" or quality_override_reason else ("中高" if current_end == "2026-06-30" and len(consensus_values) >= 2 else "中"),
|
"qa_status": results["qa"]["status"], "qa_warning_count": results["qa"]["warning_count"],
|
"report_path": formal_path.relative_to(ROOT).as_posix(), "snapshot_path": snapshot_path.relative_to(ROOT).as_posix(),
|
"source_hash": source_hash,
|
}
|
|
|
def load_universe() -> list[dict[str, Any]]:
|
with (TMP_ROOT / "universe.csv").open(encoding="utf-8", newline="") as handle:
|
rows = list(csv.DictReader(handle))
|
for row in rows:
|
for key in ("base_low", "base_high", "optimistic_low", "optimistic_high", "normalized_profit", "normalized_pe", "pb", "ps", "consensus_profit", "close", "static_total_shares"):
|
row[key] = float(row[key]) if row.get(key) not in (None, "") else None
|
row["consensus_count"] = int(row["consensus_count"]) if row.get("consensus_count") else 0
|
return rows
|
|
|
def write_outputs(rows: list[dict[str, Any]], failures: list[dict[str, Any]], wall_seconds: float) -> None:
|
MASTER_ROOT.mkdir(parents=True, exist_ok=True)
|
columns = list(rows[0])
|
csv_path = MASTER_ROOT / "全部已评估公司最新估值.csv"
|
with csv_path.open("w", encoding="utf-8", newline="") as handle:
|
writer = csv.DictWriter(handle, fieldnames=columns)
|
writer.writeheader()
|
writer.writerows(sorted(rows, key=lambda row: (row["valuation_position_pct"], row["ticker"])))
|
labels = Counter(row["label"] for row in rows)
|
updates = Counter("半年报" if row["latest_period"] == "2026-06-30" else "一季报" for row in rows)
|
lines = [
|
"# 全部已评估公司中报期最新估值",
|
"",
|
f"- 信息截止：{AS_OF} 10:19:24；价格：{PRICE_DATE}前复权完整交易日收盘。",
|
f"- 有效数值估值：{len(rows)}只；显式缺口：{len(failures)}只；本轮计算墙钟：{wall_seconds:.1f}秒。",
|
f"- 最新财务覆盖：半年报{updates['半年报']}只、一季报{updates['一季报']}只；相对旧经营信息新增披露{sum(row['financial_update'] for row in rows)}只。",
|
f"- 主营模型重分类：{sum(row['model_reclassified'] for row in rows)}只；机构预测使用真实研报日期，旧预测不会以抓取日代替。",
|
f"- 判断分布：偏低{labels['偏低']}、基本合理{labels['基本合理']}、偏贵{labels['偏贵']}、明显偏贵{labels['明显偏贵']}。",
|
"- 当前稳定逐股报告位于`../当前估值/<ticker>/`，每次覆盖同名文件，不再新增日期版当前报告。",
|
"- 合理价值是条件化区间，不构成交易指令或收益承诺。",
|
"",
|
"## 估值由低到高",
|
"",
|
"| 公司 | 代码 | 收盘价 | 基准区间 | 估值位置 | 判断 | 泡沫判定 | 最新财务 | 有效机构 | 模型变化 | 置信度 | 报告 |",
|
"|---|---|---:|---:|---:|---|---|---|---:|---|---|---|",
|
]
|
for row in sorted(rows, key=lambda item: (item["valuation_position_pct"], item["ticker"])):
|
report = "../" + row["report_path"].split("ana-data/result/股票估值/", 1)[1]
|
lines.append(
|
f"| {row['company']} | {row['ticker']} | {row['close']:.2f} | {row['base_low']:.2f}—{row['base_high']:.2f} | "
|
f"{row['valuation_position_pct']*100:.2f}% | {row['label']} | {row['bubble_status']} | {row['latest_period']} | {row['consensus_count']} | "
|
f"{'重分类' if row['model_reclassified'] else '延续'} | {row['confidence']} | [报告]({report}) |"
|
)
|
if failures:
|
lines += ["", "## 显式缺口", "", "| 公司 | 代码 | 状态 | 原因 |", "|---|---|---|---|"]
|
for item in failures:
|
lines.append(f"| {item.get('company','')} | {item['ticker']} | {item['status']} | {'；'.join(item.get('errors', [])) or '公开源不支持'} |")
|
md_path = MASTER_ROOT / "全部已评估公司最新估值.md"
|
md_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
write_json(
|
CASE_ROOT / "revaluation_manifest.json",
|
{
|
"batch_id": BATCH_ID, "as_of": AS_OF, "price_date": PRICE_DATE,
|
"completed_at": datetime.now().astimezone().isoformat(), "wall_seconds": round(wall_seconds, 3),
|
"success": len(rows), "failures": failures, "label_counts": dict(labels),
|
"financial_update_count": sum(row["financial_update"] for row in rows),
|
"reclassified_count": sum(row["model_reclassified"] for row in rows),
|
"csv_sha256": hashlib.sha256(csv_path.read_bytes()).hexdigest().upper(),
|
"md_sha256": hashlib.sha256(md_path.read_bytes()).hexdigest().upper(),
|
},
|
)
|
|
|
def sql_text(value: Any) -> str:
|
if value is None or value == "":
|
return "NULL"
|
return f"CONVERT(0x{str(value).encode('utf-8').hex()} USING utf8mb4)"
|
|
|
def sql_number(value: Any) -> str:
|
if value in (None, "", "NULL"):
|
return "NULL"
|
result = float(value)
|
if not math.isfinite(result):
|
return "NULL"
|
return format(result, ".12g")
|
|
|
def mysql_execute(sql: str) -> None:
|
proc = subprocess.run(
|
[
|
"mysql",
|
"--login-path=ana_semi_admin_preflight",
|
"--default-character-set=utf8mb4",
|
"--binary-mode",
|
],
|
cwd=ROOT,
|
input=sql,
|
text=True,
|
encoding="utf-8",
|
errors="strict",
|
capture_output=True,
|
check=False,
|
)
|
if proc.returncode != 0:
|
raise RuntimeError(f"mysql publish failed: {proc.stderr.strip()}")
|
|
|
def write_csv_rows(path: Path, rows: list[dict[str, Any]], fieldnames: list[str] | None = None) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
if fieldnames is None:
|
fieldnames = list(rows[0]) if rows else []
|
with path.open("w", encoding="utf-8", newline="") as handle:
|
writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore", lineterminator="\n")
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def action_publish() -> None:
|
master_path = MASTER_ROOT / "全部已评估公司最新估值.csv"
|
manifest_path = CASE_ROOT / "revaluation_manifest.json"
|
with master_path.open(encoding="utf-8", newline="") as handle:
|
rows = list(csv.DictReader(handle))
|
manifest = read_json(manifest_path)
|
failures = list(manifest["failures"])
|
if len(rows) + len(failures) != 1229:
|
raise RuntimeError(f"unexpected publish scope: rows={len(rows)}, failures={len(failures)}")
|
tickers = [row["ticker"] for row in rows]
|
if len(set(tickers)) != len(tickers):
|
raise RuntimeError("duplicate master tickers")
|
base_universe_map = {row["ticker"]: row for row in load_universe()}
|
price_rows = fetch.mysql_rows(
|
f"SELECT ticker,close FROM stock_valuation.daily_price WHERE trade_date='{PRICE_DATE}' ORDER BY ticker"
|
)
|
price_map = {row["ticker"]: float(row["close"]) for row in price_rows}
|
mismatches = [
|
row["ticker"] for row in rows
|
if row["ticker"] not in price_map or abs(float(row["close"]) - price_map[row["ticker"]]) > 0.000001
|
]
|
if mismatches:
|
raise RuntimeError(f"daily price mismatch: {mismatches[:10]}")
|
|
CASE_ROOT.mkdir(parents=True, exist_ok=True)
|
active_before = fetch.mysql_rows(
|
"SELECT * FROM stock_valuation.valuation_version WHERE active_to IS NULL ORDER BY ticker,valuation_id"
|
)
|
judgement_before = fetch.mysql_rows(
|
f"SELECT * FROM stock_valuation.daily_judgement WHERE trade_date='{PRICE_DATE}' ORDER BY ticker"
|
)
|
active_backup_path = CASE_ROOT / "pre_publish_active_versions.csv"
|
judgement_backup_path = CASE_ROOT / f"pre_publish_daily_judgement_{PRICE_DATE.replace('-', '')}.csv"
|
if not active_backup_path.exists():
|
write_csv_rows(active_backup_path, active_before)
|
if not judgement_backup_path.exists():
|
write_csv_rows(judgement_backup_path, judgement_before)
|
|
valuation_values: list[str] = []
|
judgement_values: list[str] = []
|
for row in rows:
|
valuation_values.append(
|
"(" + ",".join(
|
[
|
sql_text(row["valuation_id"]), sql_text(row["ticker"]), sql_text(AS_OF), sql_text(row["method"]),
|
sql_number(row["pessimistic_low"]), sql_number(row["pessimistic_high"]),
|
sql_number(row["base_low"]), sql_number(row["base_high"]),
|
sql_number(row["optimistic_low"]), sql_number(row["optimistic_high"]),
|
sql_number(row["normalized_profit"]), sql_number(row["normalized_pe"]),
|
sql_number(row["pb"]), sql_number(row["ps"]),
|
sql_number(row["consensus_year"]), sql_number(row["consensus_profit"]), sql_number(row["consensus_count"]),
|
sql_text(row["report_path"]), sql_text(row["snapshot_path"]), sql_text(row["source_hash"]),
|
sql_text(AS_OF), "NULL",
|
]
|
) + ")"
|
)
|
close = float(row["close"])
|
base_low = float(row["base_low"])
|
base_high = float(row["base_high"])
|
judgement_values.append(
|
"(" + ",".join(
|
[
|
sql_text(row["ticker"]), sql_text(PRICE_DATE), sql_text(row["valuation_id"]),
|
sql_number(close), sql_number(base_low), sql_number(base_high), sql_number(row["optimistic_high"]),
|
sql_text(row["label"]), sql_number(close / base_low - 1), sql_number(close / base_high - 1),
|
]
|
) + ")"
|
)
|
all_a_tickers = tickers + [item["ticker"] for item in failures if not item["ticker"].endswith(".HK")]
|
close_tickers = tickers + [item["ticker"] for item in failures if item["status"] == "VALUATION_MODEL_GAP"]
|
ticker_sql = ",".join(sql_text(value) for value in all_a_tickers)
|
close_ticker_sql = ",".join(sql_text(value) for value in close_tickers)
|
valuation_id_sql = ",".join(sql_text(row["valuation_id"]) for row in rows)
|
retained_gap_ids = [
|
base_universe_map[item["ticker"]]["valuation_id"] for item in failures if item["status"] == "PRICE_GAP"
|
]
|
reopen_gap_sql = "\n".join(
|
f"UPDATE valuation_version SET active_to=NULL WHERE valuation_id={sql_text(value)};" for value in retained_gap_ids
|
)
|
sql = f"""
|
SET NAMES utf8mb4;
|
USE stock_valuation;
|
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;
|
START TRANSACTION;
|
UPDATE valuation_version
|
SET active_to=GREATEST(active_from,{sql_text(PRICE_DATE)})
|
WHERE active_to IS NULL
|
AND ticker IN ({close_ticker_sql})
|
AND valuation_id NOT IN ({valuation_id_sql});
|
INSERT INTO valuation_version
|
(valuation_id,ticker,valuation_date,method,pessimistic_low,pessimistic_high,base_low,base_high,
|
optimistic_low,optimistic_high,normalized_profit,normalized_pe,pb,ps,consensus_year,consensus_profit,
|
consensus_count,report_path,snapshot_path,source_hash,active_from,active_to)
|
VALUES {','.join(valuation_values)}
|
ON DUPLICATE KEY UPDATE
|
method=VALUES(method),pessimistic_low=VALUES(pessimistic_low),pessimistic_high=VALUES(pessimistic_high),
|
base_low=VALUES(base_low),base_high=VALUES(base_high),optimistic_low=VALUES(optimistic_low),
|
optimistic_high=VALUES(optimistic_high),normalized_profit=VALUES(normalized_profit),normalized_pe=VALUES(normalized_pe),
|
pb=VALUES(pb),ps=VALUES(ps),consensus_year=VALUES(consensus_year),consensus_profit=VALUES(consensus_profit),
|
consensus_count=VALUES(consensus_count),report_path=VALUES(report_path),snapshot_path=VALUES(snapshot_path),
|
active_from=VALUES(active_from),active_to=NULL;
|
DELETE FROM daily_judgement WHERE trade_date={sql_text(PRICE_DATE)} AND ticker IN ({ticker_sql});
|
INSERT INTO daily_judgement
|
(ticker,trade_date,valuation_id,close,base_low,base_high,optimistic_high,label,distance_to_base_low,distance_to_base_high)
|
VALUES {','.join(judgement_values)};
|
{reopen_gap_sql}
|
DELETE v FROM valuation_version v
|
LEFT JOIN daily_judgement d ON d.valuation_id=v.valuation_id
|
WHERE v.valuation_date={sql_text(AS_OF)}
|
AND v.valuation_id NOT IN ({valuation_id_sql})
|
AND d.valuation_id IS NULL;
|
COMMIT;
|
"""
|
mysql_execute(sql)
|
|
latest_root = ROOT / "ana-data/result/股票估值/估值台账"
|
base_fields = [
|
"ticker", "company", "market", "currency", "close", "trade_date", "base_low", "base_high",
|
"optimistic_high", "label", "distance_to_base_low", "distance_to_base_high", "valuation_date",
|
"consensus_year", "consensus_profit", "consensus_count", "report_path",
|
]
|
latest_rows: list[dict[str, Any]] = []
|
for row in sorted(rows, key=lambda item: (float(item["valuation_position_pct"]), item["ticker"])):
|
close = float(row["close"])
|
base_low = float(row["base_low"])
|
base_high = float(row["base_high"])
|
latest_rows.append(
|
{
|
"ticker": row["ticker"], "company": row["company"], "market": row["market"], "currency": row["currency"],
|
"close": row["close"], "trade_date": PRICE_DATE, "base_low": row["base_low"], "base_high": row["base_high"],
|
"optimistic_high": row["optimistic_high"], "label": row["label"],
|
"distance_to_base_low": f"{close/base_low-1:.8f}", "distance_to_base_high": f"{close/base_high-1:.8f}",
|
"valuation_date": AS_OF, "consensus_year": row["consensus_year"], "consensus_profit": row["consensus_profit"],
|
"consensus_count": row["consensus_count"], "report_path": row["report_path"],
|
}
|
)
|
write_csv_rows(latest_root / "latest.csv", latest_rows, base_fields)
|
gap_fields = ["ticker", "company", "market", "currency", "status", "reason", "price_date", "close", "source"]
|
gap_rows: list[dict[str, Any]] = []
|
universe_map = base_universe_map
|
for item in failures:
|
old = universe_map[item["ticker"]]
|
if item["ticker"].endswith(".HK"):
|
gap_rows.append(
|
{"ticker": item["ticker"], "company": item["company"], "market": old["market"], "currency": old["currency"],
|
"status": "无法自动日更", "reason": "本地前复权专表首版不覆盖港股", "price_date": "", "close": "",
|
"source": "trading_xuntou.cn_stock_kline_1d_front"}
|
)
|
elif item["status"] == "PRICE_GAP":
|
gap_rows.append(
|
{"ticker": item["ticker"], "company": item["company"], "market": old["market"], "currency": old["currency"],
|
"status": "缺少完整日K", "reason": f"{PRICE_DATE}正式价格日，但前复权专表无当日日K；不回退旧价",
|
"price_date": item.get("last_price_date", ""), "close": item.get("last_price", ""),
|
"source": "trading_xuntou.cn_stock_kline_1d_front"}
|
)
|
else:
|
gap_rows.append(
|
{"ticker": item["ticker"], "company": item["company"], "market": old["market"], "currency": old["currency"],
|
"status": "估值模型缺口", "reason": "最新归母净资产不为正且TTM扣非亏损；PE/PB均无有效锚点，旧区间停止使用",
|
"price_date": PRICE_DATE, "close": old["close"], "source": "最新中报C1公告镜像+冻结V1内核"}
|
)
|
write_csv_rows(latest_root / "latest_gaps.csv", gap_rows, gap_fields)
|
labels = Counter(row["label"] for row in rows)
|
(latest_root / "latest.md").write_text(
|
"\n".join(
|
[
|
"# 股票估值每日台账最新总表", "", f"- 价格交易日：`{PRICE_DATE}`", f"- 估值更新日：`{AS_OF}`",
|
f"- 数值判定：{len(rows)}只；显式缺口：{len(gap_rows)}只。",
|
f"- 标签：偏低{labels['偏低']}、基本合理{labels['基本合理']}、偏贵{labels['偏贵']}、明显偏贵{labels['明显偏贵']}。",
|
"- 泡沫原因与量价显影将在同一发布流程的解释层补齐。", "",
|
]
|
),
|
encoding="utf-8",
|
)
|
|
audit_rows: list[dict[str, Any]] = []
|
for row in sorted(rows, key=lambda item: item["ticker"]):
|
raw = read_json(RAW_ROOT / f"{row['ticker'].replace('.', '_')}.json")
|
audit_rows.append(
|
{
|
"ticker": row["ticker"], "company": row["company"], "raw_status": raw.get("status"),
|
"latest_period": row["latest_period"], "latest_notice_date": row["latest_notice_date"],
|
"financial_update": row["financial_update"], "model_reclassified": row["model_reclassified"],
|
"consensus_count": row["consensus_count"], "excluded_forecast_count": row["excluded_forecast_count"],
|
"qa_status": row["qa_status"], "qa_warning_count": row["qa_warning_count"],
|
"finance_income_sha256": (raw.get("raw_sha256") or {}).get("income", ""),
|
"finance_balance_sha256": (raw.get("raw_sha256") or {}).get("balance", ""),
|
"finance_cashflow_sha256": (raw.get("raw_sha256") or {}).get("cashflow", ""),
|
"forecast_sha256": (raw.get("raw_sha256") or {}).get("forecast", ""),
|
"business_sha256": (raw.get("raw_sha256") or {}).get("business", ""),
|
}
|
)
|
write_csv_rows(CASE_ROOT / "data_freshness_audit.csv", audit_rows)
|
validation = fetch.mysql_rows(
|
f"""
|
SELECT
|
(SELECT COUNT(*) FROM stock_valuation.valuation_version WHERE valuation_date='{AS_OF}') AS new_versions,
|
(SELECT COUNT(*) FROM stock_valuation.valuation_version WHERE active_to IS NULL) AS active_versions,
|
(SELECT COUNT(*) FROM stock_valuation.daily_judgement WHERE trade_date='{PRICE_DATE}') AS judgements,
|
(SELECT COUNT(*) FROM stock_valuation.daily_price WHERE trade_date='{PRICE_DATE}') AS prices,
|
(SELECT COUNT(*) FROM stock_valuation.valuation_version WHERE ticker='603398.SH' AND active_to IS NULL) AS mubang_active
|
"""
|
)[0]
|
retained_gap_active = sum(item["status"] in {"UNSUPPORTED_HK", "PRICE_GAP"} for item in failures)
|
priced_without_valuation = sum(item["status"] == "VALUATION_MODEL_GAP" for item in failures)
|
expected = {
|
"new_versions": len(rows), "active_versions": len(rows) + retained_gap_active,
|
"judgements": len(rows), "prices": len(rows) + priced_without_valuation, "mubang_active": 0,
|
}
|
for key, value in expected.items():
|
if int(validation[key]) != value:
|
raise RuntimeError(f"post-publish validation failed: {key}={validation[key]}, expected={value}")
|
write_json(
|
CASE_ROOT / "publication_receipt.json",
|
{
|
"batch_id": BATCH_ID, "published_at": datetime.now().astimezone().isoformat(),
|
"database": "stock_valuation", "validation": validation,
|
"latest_csv_sha256_before_bubble_enrichment": hashlib.sha256((latest_root / "latest.csv").read_bytes()).hexdigest().upper(),
|
"latest_gaps_sha256": hashlib.sha256((latest_root / "latest_gaps.csv").read_bytes()).hexdigest().upper(),
|
"rollback_active_versions_sha256": hashlib.sha256(active_backup_path.read_bytes()).hexdigest().upper(),
|
"rollback_daily_judgement_sha256": hashlib.sha256(judgement_backup_path.read_bytes()).hexdigest().upper(),
|
},
|
)
|
print(json.dumps({"status": "PUBLISHED", "validation": validation, "gaps": gap_rows}, ensure_ascii=False, indent=2))
|
|
|
def action_revalue(workers: int, ticker: str | None, limit: int | None) -> None:
|
universe = load_universe()
|
if ticker:
|
universe = [row for row in universe if row["ticker"] == ticker]
|
if limit is not None:
|
universe = universe[:limit]
|
forced = force_reclass_tickers()
|
rows: list[dict[str, Any]] = []
|
failures: list[dict[str, Any]] = []
|
started = time.monotonic()
|
with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="valuation-v1") as pool:
|
futures = {pool.submit(build_one, item, forced): item for item in universe}
|
for index, future in enumerate(as_completed(futures), 1):
|
item = futures[future]
|
try:
|
result = future.result()
|
except Exception as exc:
|
result = {"ticker": item["ticker"], "company": item["company"], "status": "FAILED", "errors": [f"{type(exc).__name__}: {exc}"]}
|
if result["status"] == "COMPLETE":
|
rows.append(result)
|
else:
|
failures.append(result)
|
if index % 50 == 0 or index == len(universe):
|
print(f"[{index}/{len(universe)}] success={len(rows)} gaps={len(failures)} elapsed={time.monotonic()-started:.1f}s", flush=True)
|
rows.sort(key=lambda row: row["ticker"])
|
failures.sort(key=lambda row: row["ticker"])
|
if not ticker and limit is None:
|
write_outputs(rows, failures, time.monotonic() - started)
|
else:
|
print(json.dumps({"rows": rows, "failures": failures}, ensure_ascii=False, indent=2))
|
|
|
def main() -> None:
|
parser = argparse.ArgumentParser()
|
parser.add_argument("action", choices=("revalue", "publish"))
|
parser.add_argument("--workers", type=int, default=8)
|
parser.add_argument("--ticker")
|
parser.add_argument("--limit", type=int)
|
args = parser.parse_args()
|
if hasattr(sys.stdout, "reconfigure"):
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
if args.action == "revalue":
|
action_revalue(args.workers, args.ticker, args.limit)
|
else:
|
action_publish()
|
|
|
if __name__ == "__main__":
|
main()
|
