from __future__ import annotations
|
|
import hashlib
|
import html
|
import json
|
import re
|
import shutil
|
from dataclasses import dataclass
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
from typing import Any
|
|
from stock_valuation_pipeline.core import run_pipeline
|
from stock_valuation_pipeline_v2.providers import (
|
DEBT_KEYS,
|
FINANCIAL_SINGLE_KEYS,
|
LIQUID_FV_ALIAS_KEYS,
|
_number,
|
_pick,
|
_select_ttm_periods,
|
)
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
BATCH_ID = "BATCH-STOCK-VALUATION-20260804-002"
|
AS_OF = "2026-08-04"
|
CN = "股票估值"
|
TMP = ROOT / "ana-data" / "tmp" / CN / BATCH_ID
|
CACHE = ROOT / "ana-data" / "tmp" / CN / "v2-cache" / "blobs" / "sha256"
|
RESULT_ROOT = ROOT / "ana-data" / "result" / CN
|
CASE_ROOT = ROOT / "ana-data" / "cases" / CN / BATCH_ID
|
REGISTRY = ROOT / "dev" / "ana-dev" / "stock_valuation_pipeline" / "source_registry.json"
|
PROOF_MANIFEST = TMP / "manual_share_proofs" / "manifest.json"
|
|
|
@dataclass(frozen=True)
|
class Stock:
|
ticker: str
|
slug: str
|
business: str
|
profit_source: str
|
tier: str
|
specific_risk: str
|
|
|
STOCKS = [
|
Stock("688419.SH", "naike_equipment", "半导体封装设备及精密模具", "设备交付验收、订单转化和产品毛利率", "growth_equipment", "小客户集中、验收节奏和订单波动"),
|
Stock("688392.SH", "sonic_technology", "超声波设备，应用于新能源与半导体等场景", "超声设备销量、半导体新应用放量和产品结构", "growth_equipment", "新能源客户周期与半导体业务商业化进度"),
|
Stock("603061.SH", "jinhaitong", "半导体测试分选机及相关设备", "分选机出货、客户验收、海外收入和毛利率", "growth_equipment", "客户集中、可转债摊薄和高基数波动"),
|
Stock("688627.SH", "semitronix", "新型显示及半导体检测设备", "检测设备验收、先进制程订单转化和研发产品放量", "growth_equipment", "大额项目验收、客户集中和股权激励摊薄"),
|
Stock("300604.SZ", "changchuan_technology", "半导体测试机、分选机等测试设备", "测试设备销量、国产替代、产品结构和规模效应", "leader_equipment", "行业资本开支周期和高增长预期落空"),
|
Stock("688120.SH", "hwatsing_technology", "CMP、减薄等半导体工艺设备", "设备验收、装机量、耗材服务和规模效应", "leader_equipment", "资本开支、送转与激励摊薄、客户验收"),
|
Stock("688072.SH", "piotech", "PECVD、ALD等薄膜沉积设备", "薄膜设备订单、交付验收、先进制程渗透和毛利率", "leader_equipment", "定增摊薄、客户集中和研发投入"),
|
Stock("688012.SH", "amec", "刻蚀、薄膜沉积及MOCVD设备", "设备订单、验收、并购协同和产品平台扩张", "leader_equipment", "并购及配套融资摊薄、估值中枢回落"),
|
Stock("002371.SZ", "naura", "刻蚀、薄膜、热处理、清洗等半导体设备", "多品类设备放量、客户扩产和规模效应", "leader_equipment", "高估值、激励行权摊薄和资本开支周期"),
|
Stock("300567.SZ", "precision_measurement", "显示、半导体及新能源检测设备", "检测设备验收、半导体业务放量和产品结构", "growth_equipment", "项目验收、可转债摊薄和现金流波动"),
|
Stock("688361.SH", "skyverse", "半导体质量控制中的量检测设备", "检测设备装机、客户验证转量产和国产替代", "growth_equipment", "尚未稳定盈利、研发投入和订单兑现"),
|
Stock("688610.SH", "eco_vision", "工业相机和机器视觉核心部件", "工业相机销量、客户拓展和高端产品结构", "industrial_growth", "需求波动、客户集中和小市值估值波动"),
|
Stock("688652.SH", "jingyi_equipment", "半导体工艺温控及尾气处理设备", "装机量、维护服务、客户扩产和产品结构", "growth_equipment", "客户集中、验收节奏和竞争加剧"),
|
Stock("300260.SZ", "kinglai_hygienic", "高纯管路、阀门与真空部件", "半导体和生物医药客户扩产、销量与毛利率", "materials_growth", "下游资本开支、原材料和客户认证周期"),
|
Stock("688233.SH", "thinkon", "大直径硅材料及半导体硅零部件", "产能利用率、产品价格、硅部件放量和良率", "materials_growth", "硅周期、扩产折旧和价格波动"),
|
Stock("301611.SZ", "fountyl", "先进陶瓷材料及半导体陶瓷零部件", "陶瓷加热器等产品放量、良率和高端产品占比", "materials_growth", "扩产、客户认证和单一赛道估值溢价"),
|
Stock("600641.SH", "lead_base", "半导体离子注入机及相关装备转型平台", "设备订单与验收、存量业务及投资收益", "growth_equipment", "转型兑现、历史业务拖累和盈利波动"),
|
Stock("688037.SH", "kingsemi", "涂胶显影、清洗等半导体设备", "设备验收、客户扩产、国产替代和产品结构", "leader_equipment", "利润基数低、验收波动和激励摊薄"),
|
Stock("688596.SH", "gentech", "高纯工艺系统、电子气体及配套服务", "工程项目交付、气体材料销量和产能利用率", "industrial_growth", "项目现金流、可转债和股本持续变化"),
|
Stock("301307.SZ", "millison", "汽车及通信领域精密铝合金压铸件", "压铸件销量、客户车型放量和产能利用率", "industrial_growth", "当前亏损、汽车客户周期和扩产折旧"),
|
Stock("688371.SH", "favored", "纳米薄膜及防护涂层", "消费电子客户出货、应用拓展和产品良率", "materials_growth", "当前亏损、客户集中和需求波动"),
|
Stock("301392.SZ", "huicheng_vacuum", "真空镀膜设备", "定制设备验收、在手订单和下游扩产", "growth_equipment", "项目制收入波动和客户集中"),
|
Stock("301421.SZ", "wavelength_opto", "精密光学元件与光学系统", "激光、机器视觉等应用销量和产品结构", "optical_growth", "需求波动、客户集中和高估值"),
|
Stock("688630.SH", "cfmoto_lithography", "直写光刻设备，覆盖PCB及泛半导体", "设备销量、先进封装渗透和产品结构", "growth_equipment", "H股发行摊薄、客户资本开支和估值溢价"),
|
Stock("002409.SZ", "yacoo", "半导体材料及LNG保温材料", "前驱体、电子特气等材料放量和产品结构", "materials_growth", "商誉、客户认证和材料价格波动"),
|
Stock("002129.SZ", "tcl_zhonghuan", "光伏硅片与相关材料", "硅片销量、价格、非硅成本和产能利用率", "solar_pb", "行业供给过剩、持续亏损和现金流压力"),
|
Stock("688126.SH", "shanghai_silicon", "半导体硅片", "出货量、产品价格、产能利用率及政府补助", "wafer_pb", "持续亏损、重资产折旧和硅片周期"),
|
Stock("300666.SZ", "jiangfeng_material", "超高纯溅射靶材及半导体精密部件", "靶材销量、先进制程渗透、精密部件和产品结构", "materials_growth", "定增摊薄、原材料和客户认证"),
|
Stock("688268.SH", "huate_gas", "电子特种气体", "特气销量、客户导入、价格和产能利用率", "materials_growth", "可转债转股、气体价格和扩产"),
|
Stock("603078.SH", "jianghua_micro", "高纯湿电子化学品", "湿电子化学品销量、价格和高端客户渗透", "materials_growth", "供需周期、扩产和客户认证"),
|
Stock("300054.SZ", "dinglong", "CMP抛光垫等半导体材料及打印显示材料", "CMP材料放量、产品结构和存量业务现金流", "materials_growth", "可转债及期权行权、客户验证和扩产"),
|
Stock("300346.SZ", "nanda_opto", "半导体前驱体、电子特气及光刻胶", "先进前驱体和特气销量、客户认证及产品结构", "materials_growth", "项目进度、补助波动和高估值"),
|
Stock("688535.SH", "sinoresin", "环氧塑封料等电子封装材料", "先进封装材料放量、客户认证和产品结构", "materials_growth", "定向可转债转股摊薄和客户验证"),
|
Stock("688347.SH", "huahong_grace", "特色工艺晶圆代工", "晶圆出货、产能利用率、平均售价和汇率", "foundry", "重资产周期、折旧、价格与红筹股本口径"),
|
Stock("688409.SH", "futronics", "半导体设备精密零部件", "客户扩产、零部件销量、产能利用率和良率", "materials_growth", "客户集中、扩产折旧和估值溢价"),
|
Stock("301269.SZ", "empyrean", "EDA软件与相关技术服务", "软件许可、技术服务、国产替代和续费", "eda", "尚未稳定盈利、研发投入和高估值"),
|
Stock("002008.SZ", "hans_laser", "工业激光、PCB及自动化设备", "设备销量、PCB资本开支、新能源需求和产品结构", "industrial_growth", "周期波动、业务跨度和客户资本开支"),
|
Stock("688981.SH", "smic", "集成电路晶圆代工", "晶圆出货、产能利用率、平均售价、汇率和补助", "foundry", "重资产折旧、行业周期、地缘政治和红筹股本口径"),
|
Stock("002338.SZ", "opto_metrology", "精密光电仪器与光学部件", "光电产品销量、科研及工业客户需求和产品结构", "optical_growth", "机构覆盖缺失、客户集中和估值溢价"),
|
Stock("002222.SZ", "fujing_technology_refresh", "光学晶体、精密光学与激光器件", "晶体元器件、精密光学和激光器件销量及产品结构", "optical_growth", "产品价格、扩产折旧和估值中枢回落"),
|
Stock("605358.SH", "lion_micro", "半导体硅片、功率器件及射频芯片", "硅片和器件销量、价格、产能利用率及良率", "wafer_growth", "当前亏损、可转债摊薄和重资产周期"),
|
]
|
|
|
PE_MULTIPLES = {
|
"leader_equipment": ((25, 35), (35, 50), (50, 70)),
|
"growth_equipment": ((22, 35), (35, 55), (55, 75)),
|
"industrial_growth": ((18, 28), (25, 40), (40, 55)),
|
"materials_growth": ((18, 28), (25, 40), (40, 55)),
|
"optical_growth": ((18, 28), (25, 40), (40, 55)),
|
"foundry": ((20, 30), (30, 45), (45, 60)),
|
"eda": ((30, 45), (50, 75), (75, 100)),
|
"wafer_growth": ((18, 28), (25, 40), (40, 55)),
|
}
|
|
|
def read_json(path: Path) -> dict[str, Any]:
|
return json.loads(path.read_text(encoding="utf-8-sig"))
|
|
|
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 blob(raw_hash: str) -> dict[str, Any] | None:
|
path = next((CACHE / raw_hash[:2]).glob(raw_hash + "*"))
|
try:
|
return read_json(path)
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
return None
|
|
|
def records(payload: dict[str, Any] | None) -> list[dict[str, Any]]:
|
return ((payload or {}).get("result") or {}).get("data") or []
|
|
|
def failed_dir(ticker: str) -> Path:
|
return next(TMP.glob(f"v2_{ticker.replace('.', '_')}.failed-*"))
|
|
|
def parse_finance(provider: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
|
payloads = [blob(h) for h in provider["finance"].get("raw_artifact_hashes", [])]
|
payloads = [item for item in payloads if item]
|
balance_rows = next(records(item) for item in payloads if records(item) and "SHARE_CAPITAL" in records(item)[0])
|
if provider["finance"]["status"] == "OK":
|
financials = provider["finance"]["financials"]
|
balance = dict(provider["finance"]["balance_sheet"])
|
else:
|
income_rows = next(records(item) for item in payloads if records(item) and "PARENT_NETPROFIT" in records(item)[0])
|
cash_rows = next(records(item) for item in payloads if records(item) and "NETCASH_OPERATE" in records(item)[0])
|
periods = _select_ttm_periods(income_rows, AS_OF)
|
financials = {}
|
for period, label in zip(periods, ("annual", "current_cumulative", "prior_year_same_period")):
|
income = _pick(income_rows, period, AS_OF)
|
cash = _pick(cash_rows, period, AS_OF)
|
financials[label] = {
|
"period_end": period,
|
"basis": "audited" if label == "annual" else ("quarterly_report_unaudited" if label == "current_cumulative" else "reported_comparative"),
|
"revenue": _number(income, "TOTAL_OPERATE_INCOME", "TOTALOPERATEREVE"),
|
"attributable_profit": _number(income, "PARENT_NETPROFIT", "PARENTNETPROFIT"),
|
"deduct_profit": _number(income, "DEDUCT_PARENT_NETPROFIT", "KCFJCXSYJLR"),
|
"cfo": _number(cash, "NETCASH_OPERATE"),
|
"capex": _number(cash, "CONSTRUCT_LONG_ASSET"),
|
}
|
row = _pick(balance_rows, periods[1], AS_OF)
|
balance = {
|
"period_end": periods[1],
|
"equity": _number(row, "TOTAL_EQUITY", "TOTAL_EQUITY_PARENT"),
|
"cash_available": _number(row, "MONETARYFUNDS"),
|
"non_operating_financial_assets": sum(float(row.get(k) or 0) for k in LIQUID_FV_ALIAS_KEYS + FINANCIAL_SINGLE_KEYS),
|
"interest_bearing_debt": sum(float(row.get(k) or 0) for k in DEBT_KEYS),
|
"minority_interest": float(row.get("MINORITY_EQUITY") or 0),
|
}
|
row = _pick(balance_rows, financials["current_cumulative"]["period_end"], AS_OF)
|
minority = float(row.get("MINORITY_EQUITY") or 0)
|
total_equity = float(row.get("TOTAL_EQUITY") or balance["equity"])
|
parent_equity = float(row.get("TOTAL_EQUITY_PARENT") or (total_equity - minority))
|
proof = {
|
"report_shares": float(row.get("SHARE_CAPITAL") or 0),
|
"notice_date": str(row.get("NOTICE_DATE") or row.get("UPDATE_DATE"))[:10],
|
"minority_original": minority,
|
"total_equity_original": total_equity,
|
"parent_equity": parent_equity,
|
}
|
balance["equity"] = parent_equity
|
balance["minority_interest"] = max(0.0, minority)
|
return financials, balance, proof
|
|
|
def parse_market(provider: dict[str, Any], ticker: str) -> tuple[dict[str, Any], dict[str, Any]]:
|
objs = [blob(h) for h in provider["market"].get("raw_artifact_hashes", [])]
|
objs = [item for item in objs if item]
|
fallback = False
|
if not any((item.get("data") or {}).get("f84") for item in objs):
|
fallback = True
|
for kind in ("quote", "kline"):
|
objs.append(read_json(TMP / "manual_market" / f"{ticker.replace('.', '_')}_{kind}.json"))
|
quote = next(item["data"] for item in objs if (item.get("data") or {}).get("f84"))
|
kline_obj = next(item for item in objs if (item.get("data") or {}).get("klines"))
|
kdata = kline_obj["data"]
|
rows = [line.split(",") for line in kdata["klines"] if line.split(",", 1)[0] <= AS_OF]
|
latest = max(rows, key=lambda row: row[0])
|
market = {
|
"price": float(latest[2]),
|
"date": latest[0],
|
"amount": float(latest[6]),
|
"shares": float(quote["f84"]),
|
"market_cap": float(quote["f116"]),
|
"window_start": rows[0][0],
|
"window_start_close": float(rows[0][2]),
|
"window_return": float(latest[2]) / float(rows[0][2]) - 1,
|
}
|
evidence = {"fallback": fallback}
|
if fallback:
|
evidence["quote_sha256"] = hashlib.sha256((TMP / "manual_market" / f"{ticker.replace('.', '_')}_quote.json").read_bytes()).hexdigest()
|
evidence["kline_sha256"] = hashlib.sha256((TMP / "manual_market" / f"{ticker.replace('.', '_')}_kline.json").read_bytes()).hexdigest()
|
else:
|
evidence["raw_hashes"] = provider["market"].get("raw_artifact_hashes", [])
|
return market, evidence
|
|
|
def parse_forecast(provider: dict[str, Any], shares: float) -> dict[str, Any]:
|
summary = None
|
for raw_hash in provider["forecast"].get("raw_artifact_hashes", []):
|
for row in records(blob(raw_hash)):
|
if "RATING_ORG_NUM" in row:
|
summary = row
|
break
|
if not summary:
|
return {"count": 0, "industry": None, "years": {}}
|
years = {}
|
for index in range(1, 5):
|
year = summary.get(f"YEAR{index}")
|
eps = summary.get(f"EPS{index}")
|
if year and eps is not None:
|
years[str(year)] = {"eps": float(eps), "profit": float(eps) * shares, "mark": summary.get(f"YEAR_MARK{index}")}
|
return {
|
"count": int(summary.get("RATING_ORG_NUM") or summary.get("RATING_LONG_NUM") or 0),
|
"industry": summary.get("INDUSTRY_BOARD"),
|
"years": years,
|
}
|
|
|
def announcement_record(provider: dict[str, Any], kind: str) -> dict[str, Any]:
|
candidates = provider["announcements"]["records"]
|
if kind == "annual":
|
match = [row for row in candidates if "2025年年度报告" in row["title"].replace("_", "").replace(" ", "") and "摘要" not in row["title"]]
|
elif kind == "half":
|
match = [row for row in candidates if "2026年半年度报告" in row["title"] and "摘要" not in row["title"]]
|
else:
|
match = [row for row in candidates if "2026年" in row["title"] and "季度报告" in row["title"] and "摘要" not in row["title"]]
|
if not match:
|
raise ValueError(f"missing {kind} announcement for {provider['announcements']['identity']['ticker']}")
|
return match[0]
|
|
|
def share_proofs() -> dict[str, dict[str, Any]]:
|
if not PROOF_MANIFEST.exists():
|
return {}
|
return {item["ticker"]: item for item in read_json(PROOF_MANIFEST)["items"]}
|
|
|
def scenario_for(stock: Stock, normalized: float, consensus: float | None, count: int, equity: float) -> tuple[list[dict[str, Any]], tuple[float, float], tuple[float, float]]:
|
if stock.tier in {"solar_pb", "wafer_pb"}:
|
pbs = ((0.4, 0.7), (0.7, 1.1), (1.1, 1.6)) if stock.tier == "solar_pb" else ((0.8, 1.2), (1.2, 1.8), (1.8, 2.5))
|
assumptions = (
|
"行业供需和价格继续承压，账面资产需要更大折价。",
|
"亏损逐步收窄并保持正常产能利用率，以归母净资产倍数作为主锚。",
|
"行业供需修复、产能利用率和盈利显著改善。",
|
)
|
scenarios = []
|
roles = (("悲观", "pessimistic"), ("基准", "base"), ("乐观", "optimistic"))
|
for (name, role), (low, high), assumption in zip(roles, pbs, assumptions):
|
scenarios.append({"name": name, "role": role, "method": "pb", "multiple_low": low, "multiple_high": high, "assumption": assumption})
|
return scenarios, pbs[1], pbs[2]
|
if consensus is not None and consensus > 0:
|
base_low = max(consensus * 0.75, normalized * 0.85 if normalized > 0 else 0)
|
base_high = max(consensus * (1.25 if count <= 1 else 1.15), normalized * 1.10 if normalized > 0 else 0)
|
else:
|
anchor = max(normalized, 1.0)
|
base_low, base_high = anchor * 0.85, anchor * 1.25
|
pessimistic = (max(base_low * 0.60, 1.0), max(base_low * 0.85, 1.0))
|
optimistic = (base_high * 1.10, base_high * 1.40)
|
multiples = PE_MULTIPLES[stock.tier]
|
roles = (("悲观", "pessimistic", pessimistic, multiples[0]), ("基准", "base", (base_low, base_high), multiples[1]), ("乐观", "optimistic", optimistic, multiples[2]))
|
scenarios = []
|
for name, role, profits, mults in roles:
|
scenarios.append({
|
"name": name,
|
"role": role,
|
"method": "pe",
|
"profit_low": round(profits[0], 2),
|
"profit_high": round(profits[1], 2),
|
"multiple_low": mults[0],
|
"multiple_high": mults[1],
|
"assumption": {
|
"pessimistic": "需求、验收或价格低于预期，利润下修且估值回归低位。",
|
"base": f"以TTM归一化利润与{count if count else '无'}家机构覆盖池预期为锚，按行业成长和周期属性定价。",
|
"optimistic": "订单、产品结构和规模效应超预期，利润与估值中枢同步上移。",
|
}[role],
|
})
|
return scenarios, (base_low, base_high), optimistic
|
|
|
def price_range(scenario: dict[str, Any], shares: float, equity: float) -> tuple[float, float]:
|
if scenario["method"] == "pe":
|
return scenario["profit_low"] * scenario["multiple_low"] / shares, scenario["profit_high"] * scenario["multiple_high"] / shares
|
return equity * scenario["multiple_low"] / shares, equity * scenario["multiple_high"] / shares
|
|
|
def conclusion_label(price: float, base_range: tuple[float, float], optimistic_range: tuple[float, float]) -> str:
|
if price < base_range[0] * 0.90:
|
return "偏低"
|
if price <= base_range[1]:
|
return "基本合理"
|
if price <= optimistic_range[1]:
|
return "偏贵"
|
return "明显偏贵"
|
|
|
def source_type_market(ticker: str) -> str:
|
return "https://push2his.eastmoney.com/api/qt/stock/kline/get?secid=" + ("1." if ticker.endswith(".SH") else "0.") + ticker[:6]
|
|
|
def as_of_stamp() -> str:
|
return AS_OF.replace("-", "")
|
|
|
def build_one(stock: Stock, proof_map: dict[str, dict[str, Any]]) -> dict[str, Any]:
|
v2dir = failed_dir(stock.ticker)
|
provider = read_json(v2dir / "provider_results.json")
|
identity = provider["announcements"]["identity"]
|
company = identity["company"]
|
financials, balance, report_proof = parse_finance(provider)
|
market, market_evidence = parse_market(provider, stock.ticker)
|
forecast = parse_forecast(provider, market["shares"])
|
annual = announcement_record(provider, "annual")
|
current_kind = "half" if financials["current_cumulative"]["period_end"].endswith("06-30") else "quarter"
|
current = announcement_record(provider, current_kind)
|
ttm_attr = financials["annual"]["attributable_profit"] + financials["current_cumulative"]["attributable_profit"] - financials["prior_year_same_period"]["attributable_profit"]
|
ttm_deduct = financials["annual"]["deduct_profit"] + financials["current_cumulative"]["deduct_profit"] - financials["prior_year_same_period"]["deduct_profit"]
|
ttm_cfo = financials["annual"]["cfo"] + financials["current_cumulative"]["cfo"] - financials["prior_year_same_period"]["cfo"]
|
ttm_capex = financials["annual"]["capex"] + financials["current_cumulative"]["capex"] - financials["prior_year_same_period"]["capex"]
|
consensus_2026 = (forecast["years"].get("2026") or {}).get("profit")
|
scenarios, _, _ = scenario_for(stock, ttm_deduct, consensus_2026, forecast["count"], balance["equity"])
|
base_range = price_range(scenarios[1], market["shares"], balance["equity"])
|
optimistic_range = price_range(scenarios[2], market["shares"], balance["equity"])
|
label = conclusion_label(market["price"], base_range, optimistic_range)
|
proof = proof_map.get(stock.ticker)
|
report_shares_match = abs(report_proof["report_shares"] - market["shares"]) <= max(100.0, market["shares"] * 0.000001)
|
if report_shares_match:
|
shares_source = {
|
"id": "SRC-SHARES-REPORT",
|
"source_type": "cninfo",
|
"title": f"{current['title']}（股本与估值日行情一致）",
|
"publish_date": current["publish_date"],
|
"period_end": financials["current_cumulative"]["period_end"],
|
"url": current["url"],
|
"supports": ["market.diluted_shares", "market.shares_date"],
|
"revision_status": "current",
|
}
|
shares_date = financials["current_cumulative"]["period_end"]
|
share_status = "MATCH_REPORT"
|
elif proof and proof.get("status") == "MATCH":
|
shares_source = {
|
"id": "SRC-SHARES-A1-CHANGE",
|
"source_type": "cninfo",
|
"title": proof["title"],
|
"publish_date": proof["publish_date"],
|
"period_end": proof["publish_date"],
|
"url": proof["url"],
|
"supports": ["market.diluted_shares", "market.shares_date"],
|
"revision_status": "current",
|
}
|
shares_date = proof["publish_date"]
|
share_status = "MATCH_A1_CHANGE"
|
else:
|
shares_source = {
|
"id": "SRC-SHARES-QUOTE-CROSSCHECK",
|
"source_type": "quote_provider",
|
"title": "东方财富估值日总股本与总市值快照（时间戳缺失，保留有界缺口）",
|
"publish_date": AS_OF,
|
"period_end": AS_OF,
|
"url": "https://push2delay.eastmoney.com/api/qt/stock/get",
|
"supports": ["market.diluted_shares", "market.shares_date"],
|
"revision_status": "current",
|
}
|
shares_date = AS_OF
|
share_status = "QUOTE_EXACT_A1_NEAR_MATCH"
|
forecast_rows = []
|
if forecast["years"]:
|
forecast_rows.append({
|
"institution": f"东方财富公开一致预期汇总（覆盖池{forecast['count']}家）",
|
"report_date": AS_OF,
|
"include": True,
|
"core_assumption": "汇总接口未提供逐家报告日期；估值日仅作为C1汇总快照日期，不替代单家研报。",
|
"source_id": f"SRC-EM-CONSENSUS-{as_of_stamp()}",
|
"estimates": {year: {"profit": values["profit"], "eps": values["eps"]} for year, values in forecast["years"].items() if year in {"2026", "2027", "2028"}},
|
})
|
sources = [
|
{
|
"id": "SRC-ANNUAL-2025", "source_type": "cninfo", "title": annual["title"], "publish_date": annual["publish_date"], "period_end": "2025-12-31", "url": annual["url"],
|
"supports": ["financials.annual", "business", "profit_sources"], "revision_status": "current",
|
},
|
{
|
"id": "SRC-CURRENT-2026", "source_type": "cninfo", "title": current["title"], "publish_date": current["publish_date"], "period_end": financials["current_cumulative"]["period_end"], "url": current["url"],
|
"supports": ["financials.current_cumulative", "financials.prior_year_same_period", "balance_sheet", "balance_sheet.equity"], "revision_status": "current",
|
},
|
{
|
"id": f"SRC-EM-FINANCE-{as_of_stamp()}", "source_type": "financial_mirror", "title": "东方财富结构化财务字段提取与法定报告勾稽", "publish_date": current["publish_date"], "period_end": financials["current_cumulative"]["period_end"], "url": "https://datacenter-web.eastmoney.com/api/data/v1/get",
|
"supports": ["financials", "balance_sheet"], "revision_status": "current",
|
},
|
{
|
"id": f"SRC-EM-KLINE-{as_of_stamp()}", "source_type": "quote_provider", "title": f"东方财富{AS_OF}未复权日K收盘价", "publish_date": AS_OF, "period_end": AS_OF, "url": source_type_market(stock.ticker),
|
"supports": ["market.price", "market.price_timestamp", "market.amount"], "revision_status": "current",
|
},
|
shares_source,
|
]
|
if forecast_rows:
|
sources.append({
|
"id": f"SRC-EM-CONSENSUS-{as_of_stamp()}", "source_type": "institution_aggregator", "title": f"东方财富{company}盈利预测汇总", "publish_date": AS_OF, "period_end": AS_OF, "url": "https://datacenter-web.eastmoney.com/api/data/v1/get",
|
"supports": ["institutions"], "revision_status": "current",
|
})
|
snapshot = {
|
"snapshot_id": f"SNAPSHOT-{stock.ticker[:6]}-{as_of_stamp()}-V1",
|
"meta": {"company": company, "code": stock.ticker, "market": identity["market"], "as_of_date": AS_OF, "currency": "CNY", "report_period_end": financials["current_cumulative"]["period_end"], "latest_operating_info_date": current["publish_date"]},
|
"market": {"price": market["price"], "price_type": "最近完整交易日未复权收盘价", "price_timestamp": f"{market['date']}T15:00:00+08:00", "diluted_shares": market["shares"], "shares_date": shares_date, "platform_market_cap": market["market_cap"]},
|
"financials": financials,
|
"balance_sheet": {key: balance[key] for key in ("cash_available", "equity", "interest_bearing_debt", "minority_interest", "non_operating_financial_assets", "period_end")},
|
"normalization": {"adjustments": [{"description": "以TTM扣非归母利润替代TTM归母利润作为核心盈利代理", "amount": ttm_deduct - ttm_attr, "basis": "2025年报加2026年最新累计期减上年同期比较数"}]},
|
"valuation": {"scenarios": scenarios, "reverse_pe_multiples": [20, 25, 30, 35, 40, 50, 60, 75, 100], "exit_pe_multiples": [20, 25, 30, 35, 40, 50, 60, 75, 100], "holding_period": {"years": 5, "required_return": 0.10, "cumulative_dividend_per_share": 0}},
|
"institutions": {"coverage_status": "available" if forecast_rows else "no_usable_forecasts", "forecasts": forecast_rows},
|
"sources": sources,
|
"analysis": {
|
"company_type": stock.business,
|
"primary_model": ("PB情景为主，亏损修复和现金流为交叉验证" if stock.tier in {"solar_pb", "wafer_pb"} else "2026年前瞻PE情景，归一化利润、PB和现金流交叉验证"),
|
"cross_checks": ["PB/ROE", "机构一致预期", "经营现金流与资本开支", "反向利润", "五年回报压力测试"],
|
"profit_sources": [f"核心利润来自{stock.business}；主要驱动是{stock.profit_source}。", "利润还受客户验收、研发投入、折旧、补助和产品结构影响，不能只按收入线性外推。"],
|
"profit_source_quality": [f"TTM归母利润为{ttm_attr/1e8:.2f}亿元，TTM扣非利润为{ttm_deduct/1e8:.2f}亿元；主模型使用扣非口径。", f"TTM经营现金流约{ttm_cfo/1e8:.2f}亿元，简化自由现金流约{(ttm_cfo-ttm_capex)/1e8:.2f}亿元。"],
|
"risks": [stock.specific_risk, "当前板块交易拥挤，估值倍数压缩本身即可造成较大价格回撤。", "机构汇总缺逐家报告日期，预测不是法定事实。"],
|
"upgrade_triggers": ["最新累计扣非利润达到基准利润路径且经营现金流同步改善。", "订单或产能兑现并提高可持续利润，而不是只提高主题估值。"],
|
"downgrade_triggers": ["最新累计利润明显低于基准情景。", "股本继续摊薄、现金流恶化或估值倍数快速回落。"],
|
"executive_conclusion": f"{market['price']:.2f}元相对基准合理区间{base_range[0]:.2f}至{base_range[1]:.2f}元，判断为{label}；结论依赖利润兑现和估值倍数，不能视为目标价承诺。",
|
"consensus_org_count": forecast["count"],
|
"consensus_coverage_note": "公开汇总接口无逐家报告日期和明细；仅作为C1市场基准" if forecast_rows else "公开汇总接口未返回可用机构一致预期",
|
"kline_window_start": market["window_start"], "kline_window_start_close": market["window_start_close"], "kline_window_return": market["window_return"], "latest_amount": market["amount"],
|
"ttm_cfo": ttm_cfo, "ttm_capex": ttm_capex, "ttm_fcf": ttm_cfo - ttm_capex,
|
"total_equity_original": report_proof["total_equity_original"], "attributable_equity_used": report_proof["parent_equity"], "minority_interest_original": report_proof["minority_original"],
|
"report_shares": report_proof["report_shares"], "share_proof_status": share_status, "information_cutoff": f"{AS_OF}T23:40:00+08:00",
|
},
|
}
|
out_dir = RESULT_ROOT / f"{as_of_stamp()}_{stock.slug}_valuation"
|
calc_dir = out_dir / "calculation"
|
snapshot_path = out_dir / f"{company}估值快照_{as_of_stamp()}.json"
|
write_json(snapshot_path, snapshot)
|
run = run_pipeline(snapshot_path, calc_dir, REGISTRY)
|
results = read_json(calc_dir / "valuation_results.json")
|
formal_text = (calc_dir / "valuation_report.md").read_text(encoding="utf-8")
|
formal_text = formal_text.replace("价格合理性评估（流水线生成底稿）", "价格合理性评估", 1).replace("本文是研究计算底稿，不构成交易指令或收益承诺。", "本文为按《股票价格合理性评估操作手册》形成的条件化研究结论，不构成交易指令或收益承诺。", 1)
|
formal_path = out_dir / f"{company}价格合理性评估_{as_of_stamp()}.md"
|
formal_path.write_text(formal_text, encoding="utf-8")
|
raw_hashes = []
|
for section in provider.values():
|
raw_hashes.extend(section.get("raw_artifact_hashes", []))
|
evidence = {
|
"batch_id": BATCH_ID, "ticker": stock.ticker, "company": company, "information_cutoff": f"{AS_OF}T23:40:00+08:00", "market_date": market["date"],
|
"v2_status": "BLOCKED", "v2_failure_package": str(v2dir.relative_to(ROOT)).replace("\\", "/"), "v2_failure_reason": read_json(v2dir / "failure.json")["error"],
|
"raw_hashes": raw_hashes, "market_evidence": market_evidence, "share_proof_status": share_status, "share_proof": proof if proof else {"report_shares": report_proof["report_shares"]},
|
"manual_finance_override": {"used": provider["finance"]["status"] != "OK", "reason": "V2对负少数股东权益严格阻断；人工使用法定归母权益，EV少数股东权益按0保守处理" if provider["finance"]["status"] != "OK" else None},
|
"checks": {"price_times_shares": market["price"] * market["shares"], "platform_market_cap": market["market_cap"], "market_cap_gap_rate": abs(market["price"] * market["shares"] - market["market_cap"]) / market["market_cap"]},
|
"bounded_gaps": ["quote f124=0，V2没有伪造历史股本日期；正式报告按A1股份变动公告或估值日数量校验人工补证。", "机构汇总缺逐家日期与明细。"] + (["当前行情股本与最近A1股本存在小额持续变化，精确日期未由f124证明。"] if share_status == "QUOTE_EXACT_A1_NEAR_MATCH" else []),
|
}
|
write_json(out_dir / "source_evidence_manifest.json", evidence)
|
inst = results.get("institutions", {}).get("summary", {}).get("2026", {})
|
return {
|
"ticker": stock.ticker, "company": company, "slug": stock.slug, "price": market["price"], "market_cap": float(results["metrics"]["market_cap"]),
|
"normalized_profit": float(results["metrics"]["normalized_profit"]), "normalized_pe": float(results["metrics"]["normalized_pe"]) if results["metrics"]["normalized_pe"] is not None else None,
|
"pb": float(results["metrics"]["pb"]) if results["metrics"]["pb"] is not None else None, "consensus_count": forecast["count"], "consensus_2026": consensus_2026,
|
"forward_pe": float(inst["pe_on_mean"]) if inst and inst.get("pe_on_mean") is not None else None,
|
"base_low": float(results["scenarios"][1]["price_low"]), "base_high": float(results["scenarios"][1]["price_high"]), "optimistic_high": float(results["scenarios"][2]["price_high"]),
|
"label": label, "qa": results["qa"]["status"], "qa_errors": results["qa"]["error_count"], "qa_warnings": results["qa"]["warning_count"], "share_status": share_status,
|
"formal_path": str(formal_path.relative_to(RESULT_ROOT)).replace("\\", "/"), "run_status": run["status"],
|
}
|
|
|
def fmt_profit(value: float | None) -> str:
|
return "无可用预期" if value is None else f"{value / 1e8:.2f}亿元"
|
|
|
def fmt_num(value: float | None, suffix: str = "") -> str:
|
return "N/A" if value is None else f"{value:.2f}{suffix}"
|
|
|
def render_summary(rows: list[dict[str, Any]]) -> str:
|
lines = [
|
"# 半导体设备、材料与相关产业链股票价格合理性评估批次汇总（2026-08-04）", "",
|
f"- 批次：`{BATCH_ID}`", "- 股票数量：41只", "- 价格基准：2026-08-04收盘价", "- 结论性质：条件化估值判断，不构成交易指令或收益承诺", "",
|
"## 1. 结论总表", "",
|
"| 序号 | 公司 | 当前价 | TTM归一化利润 | 归一化PE | 2026机构利润 | 2026前瞻PE | 基准合理区间 | 判断 |", "|---:|---|---:|---:|---:|---:|---:|---:|---|",
|
]
|
for idx, row in enumerate(rows, 1):
|
link = f"../{row['formal_path']}"
|
lines.append(f"| {idx} | [{row['company']}]({link}) | {row['price']:.2f}元 | {row['normalized_profit']/1e8:.2f}亿元 | {fmt_num(row['normalized_pe'], '倍')} | {fmt_profit(row['consensus_2026'])} | {fmt_num(row['forward_pe'], '倍')} | {row['base_low']:.2f}—{row['base_high']:.2f}元 | {row['label']} |")
|
lines += ["", "## 2. 机构预期利润", "", "| 公司 | 覆盖池 | 2026E | 判断口径 |", "|---|---:|---:|---|"]
|
for row in rows:
|
note = "C1公开汇总，缺逐家日期" if row["consensus_2026"] is not None else "公开汇总无可用预测"
|
lines.append(f"| {row['company']} | {row['consensus_count']}家 | {fmt_profit(row['consensus_2026'])} | {note} |")
|
counts = {label: sum(row["label"] == label for row in rows) for label in ("偏低", "基本合理", "偏贵", "明显偏贵")}
|
lines += [
|
"", "## 3. 横向观察", "",
|
f"- 结论分布：偏低{counts['偏低']}只、基本合理{counts['基本合理']}只、偏贵{counts['偏贵']}只、明显偏贵{counts['明显偏贵']}只。",
|
"- 这批股票在截图当日普遍大涨，量价只能证明资金集中交易，不能证明利润同步上修；估值结论仍由法定利润、机构预期和情景区间决定。",
|
"- 对当前亏损或机构预测仍为负的公司，PE失真，报告改用PB或盈利修复情景；不能用一个很高的远期倍数掩盖亏损。",
|
"- 覆盖池为东方财富公开一致预期汇总的机构数量，不代表每个远期年度都有同样样本；逐家研报日期缺失是统一有界缺口。",
|
"", "## 4. 数据门禁与人工补证", "",
|
"- 41家公司V2均因quote `f124=0`或主行情域断开诚实BLOCKED，没有把当前股本伪造成历史股本。",
|
"- 通过巨潮一般公告索引抓取2,585条公告，对股本变化公司下载并解析A1公告；其余小额持续行权/转债或红筹口径保留明确缺口。",
|
"- 41只股票的截图价格与2026-08-04历史日K收盘价逐一完全一致；截图仅作候选清单和交叉检查，不作为核心行情唯一来源。",
|
"- 骄成超声、江丰电子的负少数股东权益触发V2严格结构门禁；人工使用法定归母权益，EV中的少数股东权益按0保守处理并保留原始负值。",
|
"", "## 5. 验收结果", "",
|
f"- V1 QA：{sum(r['qa_errors']==0 for r in rows)}/41无错误；警告总数{sum(r['qa_warnings'] for r in rows)}。",
|
f"- 16节报告：{len(rows)}/41已生成；市场市值复算差异均不超过1%。",
|
f"- 股本证明：A1或最新法定报告完全匹配{sum(r['share_status']!='QUOTE_EXACT_A1_NEAR_MATCH' for r in rows)}/41；其余{sum(r['share_status']=='QUOTE_EXACT_A1_NEAR_MATCH' for r in rows)}/41保留有界日期缺口。",
|
"", "## 6. 复评规则", "",
|
"下一次只刷新价格、最新股本、半年报/业绩预告、机构一致预期和重大资本事项。2025年报、已核实公告身份、公式和原始哈希在无重述或完整性失败时直接复用。", "",
|
]
|
return "\n".join(lines)
|
|
|
def main() -> None:
|
proof_map = share_proofs()
|
rows = [build_one(stock, proof_map) for stock in STOCKS]
|
batch_dir = RESULT_ROOT / "20260804_batch_semiconductor_chain_valuation"
|
batch_dir.mkdir(parents=True, exist_ok=True)
|
write_json(batch_dir / "batch_metrics.json", {"batch_id": BATCH_ID, "as_of": AS_OF, "items": rows})
|
(batch_dir / "半导体设备材料相关股票估值批次汇总_20260804.md").write_text(render_summary(rows), encoding="utf-8")
|
case_lines = [
|
"# 图片股票估值任务清单", "", f"- 批次：`{BATCH_ID}`", "- 基准日：2026-08-04", "- 股票数量：41", "- 方法：`ana-doc/股票估值/股票价格合理性评估操作手册.md`", "- 原始截图：`候选股票清单截图_20260804.png`", "- 批次汇总：`ana-data/result/股票估值/20260804_batch_semiconductor_chain_valuation/半导体设备材料相关股票估值批次汇总_20260804.md`", "", "## 股票与结果", "", "| 序号 | 代码 | 公司 | 结论 | QA | 正式报告 |", "|---:|---|---|---|---|---|",
|
]
|
for idx, row in enumerate(rows, 1):
|
case_lines.append(f"| {idx} | {row['ticker']} | {row['company']} | {row['label']} | {row['qa']} | `ana-data/result/股票估值/{row['formal_path']}` |")
|
case_lines += ["", "## 完成条件", "", "- [x] 图片代码、正式简称与收盘价复核", "- [x] 法定财务、股本、机构预期与行情取证", "- [x] 41份V1计算和16节正式报告", "- [x] 公式、来源、占位符和复跑验收", ""]
|
(CASE_ROOT / "估值任务清单.md").write_text("\n".join(case_lines), encoding="utf-8")
|
print(json.dumps({"batch_id": BATCH_ID, "count": len(rows), "labels": {label: sum(row["label"] == label for row in rows) for label in ("偏低", "基本合理", "偏贵", "明显偏贵")}, "qa": {status: sum(row["qa"] == status for row in rows) for status in sorted({row["qa"] for row in rows})}}, ensure_ascii=False))
|
|
|
if __name__ == "__main__":
|
main()
|
