from __future__ import annotations
|
|
import hashlib
|
import json
|
import math
|
import os
|
import tempfile
|
from dataclasses import dataclass
|
from datetime import date, datetime
|
from decimal import Decimal, InvalidOperation
|
from pathlib import Path
|
from statistics import median
|
from typing import Any, Iterable
|
from urllib.parse import urlparse
|
|
from . import __version__
|
|
|
ZERO = Decimal("0")
|
ONE = Decimal("1")
|
YI = Decimal("100000000")
|
|
|
class InputError(ValueError):
|
def __init__(self, errors: list[str]):
|
self.errors = errors
|
super().__init__("; ".join(errors))
|
|
|
@dataclass(frozen=True)
|
class QAIssue:
|
issue_id: str
|
severity: str
|
field: str
|
message: str
|
action: str
|
recalc_scope: str
|
|
def as_dict(self) -> dict[str, str]:
|
return {
|
"issue_id": self.issue_id,
|
"severity": self.severity,
|
"field": self.field,
|
"message": self.message,
|
"action": self.action,
|
"recalc_scope": self.recalc_scope,
|
}
|
|
|
def load_snapshot(path: Path) -> dict[str, Any]:
|
with path.open("r", encoding="utf-8") as handle:
|
payload = json.load(handle, parse_float=Decimal, parse_int=Decimal)
|
if not isinstance(payload, dict):
|
raise InputError(["输入根节点必须是 JSON 对象"])
|
return payload
|
|
|
def _load_json(path: Path) -> dict[str, Any]:
|
with path.open("r", encoding="utf-8") as handle:
|
payload = json.load(handle, parse_float=Decimal, parse_int=Decimal)
|
if not isinstance(payload, dict):
|
raise InputError([f"{path} 的根节点必须是 JSON 对象"])
|
return payload
|
|
|
def _required(mapping: dict[str, Any], key: str, path: str, errors: list[str]) -> Any:
|
if key not in mapping or mapping[key] is None or mapping[key] == "":
|
errors.append(f"缺少必填字段：{path}.{key}")
|
return None
|
return mapping[key]
|
|
|
def _decimal(value: Any, field: str, errors: list[str], *, allow_none: bool = False) -> Decimal | None:
|
if value is None and allow_none:
|
return None
|
try:
|
result = value if isinstance(value, Decimal) else Decimal(str(value))
|
except (InvalidOperation, ValueError, TypeError):
|
errors.append(f"字段 {field} 必须是数字")
|
return None
|
if not result.is_finite():
|
errors.append(f"字段 {field} 必须是有限数字")
|
return None
|
return result
|
|
|
def _date(value: Any, field: str, errors: list[str]) -> date | None:
|
if not isinstance(value, str):
|
errors.append(f"字段 {field} 必须是 YYYY-MM-DD 日期")
|
return None
|
try:
|
return date.fromisoformat(value[:10])
|
except ValueError:
|
errors.append(f"字段 {field} 不是有效日期：{value}")
|
return None
|
|
|
def _get_decimal(mapping: dict[str, Any], key: str, path: str, errors: list[str], *, optional: bool = False) -> Decimal | None:
|
if optional and (key not in mapping or mapping[key] is None):
|
return None
|
value = _required(mapping, key, path, errors)
|
return _decimal(value, f"{path}.{key}", errors, allow_none=optional)
|
|
|
def validate_snapshot(snapshot: dict[str, Any], registry: dict[str, Any]) -> None:
|
errors: list[str] = []
|
as_of: date | None = None
|
for group in ("meta", "market", "financials", "balance_sheet", "valuation", "institutions", "sources"):
|
if group not in snapshot:
|
errors.append(f"缺少必填分组：{group}")
|
|
if errors:
|
raise InputError(errors)
|
|
meta = snapshot["meta"]
|
market = snapshot["market"]
|
financials = snapshot["financials"]
|
balance = snapshot["balance_sheet"]
|
valuation = snapshot["valuation"]
|
institutions = snapshot["institutions"]
|
|
if not isinstance(meta, dict) or not isinstance(market, dict):
|
errors.append("meta 和 market 必须是对象")
|
else:
|
for key in ("company", "code", "market", "as_of_date", "currency", "report_period_end", "latest_operating_info_date"):
|
_required(meta, key, "meta", errors)
|
as_of = _date(meta.get("as_of_date"), "meta.as_of_date", errors)
|
report_end = _date(meta.get("report_period_end"), "meta.report_period_end", errors)
|
latest_info = _date(meta.get("latest_operating_info_date"), "meta.latest_operating_info_date", errors)
|
if as_of and as_of > date.today():
|
errors.append("meta.as_of_date 不得晚于当前日期")
|
if as_of and report_end and report_end > as_of:
|
errors.append("meta.report_period_end 不得晚于估值日")
|
if as_of and latest_info and latest_info > as_of:
|
errors.append("meta.latest_operating_info_date 不得晚于估值日")
|
|
price = _get_decimal(market, "price", "market", errors)
|
shares = _get_decimal(market, "diluted_shares", "market", errors)
|
price_timestamp = _required(market, "price_timestamp", "market", errors)
|
price_date = _date(price_timestamp, "market.price_timestamp", errors) if price_timestamp else None
|
shares_date = _date(_required(market, "shares_date", "market", errors), "market.shares_date", errors)
|
if as_of and price_date and price_date > as_of:
|
errors.append("market.price_timestamp 不得晚于估值日")
|
if as_of and shares_date and shares_date > as_of:
|
errors.append("market.shares_date 不得晚于估值日")
|
if price is not None and price <= ZERO:
|
errors.append("market.price 必须大于 0")
|
if shares is not None and shares <= ZERO:
|
errors.append("market.diluted_shares 必须大于 0")
|
_get_decimal(market, "platform_market_cap", "market", errors, optional=True)
|
|
if not isinstance(financials, dict):
|
errors.append("financials 必须是对象")
|
else:
|
for period_name in ("annual", "current_cumulative", "prior_year_same_period"):
|
period = financials.get(period_name)
|
if not isinstance(period, dict):
|
errors.append(f"financials.{period_name} 必须是对象")
|
continue
|
period_end = _date(
|
_required(period, "period_end", f"financials.{period_name}", errors),
|
f"financials.{period_name}.period_end",
|
errors,
|
)
|
if as_of and period_end and period_end > as_of:
|
errors.append(f"financials.{period_name}.period_end 不得晚于估值日")
|
for metric in ("revenue", "attributable_profit", "deduct_profit"):
|
_get_decimal(period, metric, f"financials.{period_name}", errors)
|
_get_decimal(period, "cfo", f"financials.{period_name}", errors, optional=True)
|
capex = _get_decimal(period, "capex", f"financials.{period_name}", errors, optional=True)
|
if capex is not None and capex < ZERO:
|
errors.append(f"financials.{period_name}.capex 采用现金流出正数口径，不得小于 0")
|
|
if not isinstance(balance, dict):
|
errors.append("balance_sheet 必须是对象")
|
else:
|
balance_end = _date(_required(balance, "period_end", "balance_sheet", errors), "balance_sheet.period_end", errors)
|
if as_of and balance_end and balance_end > as_of:
|
errors.append("balance_sheet.period_end 不得晚于估值日")
|
for key in ("equity", "cash_available", "non_operating_financial_assets", "interest_bearing_debt", "minority_interest"):
|
value = _get_decimal(balance, key, "balance_sheet", errors)
|
if value is not None and key != "equity" and value < ZERO:
|
errors.append(f"balance_sheet.{key} 不得小于 0")
|
|
if not isinstance(valuation, dict):
|
errors.append("valuation 必须是对象")
|
else:
|
scenarios = valuation.get("scenarios")
|
if not isinstance(scenarios, list) or not scenarios:
|
errors.append("valuation.scenarios 必须是非空数组")
|
else:
|
roles: list[str] = []
|
for index, scenario in enumerate(scenarios):
|
p = f"valuation.scenarios[{index}]"
|
if not isinstance(scenario, dict):
|
errors.append(f"{p} 必须是对象")
|
continue
|
method = _required(scenario, "method", p, errors)
|
role = _required(scenario, "role", p, errors)
|
_required(scenario, "name", p, errors)
|
if role:
|
roles.append(str(role))
|
if method == "pe":
|
values: dict[str, Decimal | None] = {}
|
for key in ("profit_low", "profit_high", "multiple_low", "multiple_high"):
|
values[key] = _get_decimal(scenario, key, p, errors)
|
if all(values[key] is not None for key in values):
|
if values["profit_low"] <= ZERO or values["profit_high"] <= ZERO:
|
errors.append(f"{p} 的 PE 情景利润必须大于 0；亏损情景请改用 PB")
|
if values["multiple_low"] <= ZERO or values["multiple_high"] <= ZERO:
|
errors.append(f"{p} 的 PE 倍数必须大于 0")
|
if values["profit_low"] > values["profit_high"] or values["multiple_low"] > values["multiple_high"]:
|
errors.append(f"{p} 的低值不得大于高值")
|
elif method == "pb":
|
values = {}
|
for key in ("multiple_low", "multiple_high"):
|
values[key] = _get_decimal(scenario, key, p, errors)
|
if all(values[key] is not None for key in values):
|
if values["multiple_low"] <= ZERO or values["multiple_high"] <= ZERO:
|
errors.append(f"{p} 的 PB 倍数必须大于 0")
|
if values["multiple_low"] > values["multiple_high"]:
|
errors.append(f"{p} 的低值不得大于高值")
|
else:
|
errors.append(f"{p}.method 只支持 pe 或 pb")
|
if roles.count("base") != 1:
|
errors.append("valuation.scenarios 必须且只能包含一个 role=base 的基准情景")
|
if roles.count("optimistic") > 1 or roles.count("pessimistic") > 1:
|
errors.append("乐观或悲观情景角色不得重复")
|
reverse_pes = valuation.get("reverse_pe_multiples")
|
if not isinstance(reverse_pes, list) or not reverse_pes:
|
errors.append("valuation.reverse_pe_multiples 必须是非空数组")
|
else:
|
for index, value in enumerate(reverse_pes):
|
multiple = _decimal(value, f"valuation.reverse_pe_multiples[{index}]", errors)
|
if multiple is not None and multiple <= ZERO:
|
errors.append("反向 PE 倍数必须大于 0")
|
holding = valuation.get("holding_period")
|
if not isinstance(holding, dict):
|
errors.append("valuation.holding_period 必须是对象")
|
else:
|
years = _decimal(holding.get("years"), "valuation.holding_period.years", errors)
|
required_return = _decimal(holding.get("required_return"), "valuation.holding_period.required_return", errors)
|
_decimal(holding.get("cumulative_dividend_per_share", 0), "valuation.holding_period.cumulative_dividend_per_share", errors)
|
if years is not None and (years <= ZERO or years != years.to_integral_value()):
|
errors.append("valuation.holding_period.years 必须是正整数")
|
if required_return is not None and required_return <= Decimal("-1"):
|
errors.append("valuation.holding_period.required_return 必须大于 -100%")
|
|
if not isinstance(institutions, dict):
|
errors.append("institutions 必须是对象")
|
else:
|
coverage = institutions.get("coverage_status")
|
if coverage not in ("available", "no_usable_forecasts"):
|
errors.append("institutions.coverage_status 只支持 available 或 no_usable_forecasts")
|
forecasts = institutions.get("forecasts")
|
if not isinstance(forecasts, list):
|
errors.append("institutions.forecasts 必须是数组；无预测时使用空数组")
|
elif coverage == "available" and not forecasts:
|
errors.append("coverage_status=available 时 forecasts 不得为空")
|
elif coverage == "no_usable_forecasts" and forecasts:
|
errors.append("coverage_status=no_usable_forecasts 时 forecasts 必须为空")
|
elif forecasts:
|
for index, forecast in enumerate(forecasts):
|
p = f"institutions.forecasts[{index}]"
|
if not isinstance(forecast, dict):
|
errors.append(f"{p} 必须是对象")
|
continue
|
_required(forecast, "institution", p, errors)
|
report_date = _date(_required(forecast, "report_date", p, errors), f"{p}.report_date", errors)
|
if as_of and report_date and report_date > as_of:
|
errors.append(f"{p}.report_date 不得晚于估值日")
|
estimates = _required(forecast, "estimates", p, errors)
|
if not isinstance(estimates, dict) or not estimates:
|
errors.append(f"{p}.estimates 必须是非空年度预测对象")
|
continue
|
for year, estimate in estimates.items():
|
ep = f"{p}.estimates.{year}"
|
if not str(year).isdigit() or len(str(year)) != 4:
|
errors.append(f"{ep} 的年度键必须是四位年份")
|
if not isinstance(estimate, dict):
|
errors.append(f"{ep} 必须是对象")
|
continue
|
_get_decimal(estimate, "profit", ep, errors)
|
_get_decimal(estimate, "eps", ep, errors)
|
|
normalization = snapshot.get("normalization", {"adjustments": []})
|
if not isinstance(normalization, dict) or not isinstance(normalization.get("adjustments", []), list):
|
errors.append("normalization.adjustments 必须是数组")
|
else:
|
for index, adjustment in enumerate(normalization.get("adjustments", [])):
|
p = f"normalization.adjustments[{index}]"
|
if not isinstance(adjustment, dict):
|
errors.append(f"{p} 必须是对象")
|
continue
|
_required(adjustment, "description", p, errors)
|
_get_decimal(adjustment, "amount", p, errors)
|
|
sources = snapshot.get("sources")
|
source_types = registry.get("source_types", {})
|
if not isinstance(sources, list) or not sources:
|
errors.append("sources 必须是非空数组")
|
else:
|
ids: set[str] = set()
|
for index, source in enumerate(sources):
|
p = f"sources[{index}]"
|
if not isinstance(source, dict):
|
errors.append(f"{p} 必须是对象")
|
continue
|
source_id = _required(source, "id", p, errors)
|
source_type = _required(source, "source_type", p, errors)
|
_required(source, "title", p, errors)
|
publish_date = _date(_required(source, "publish_date", p, errors), f"{p}.publish_date", errors)
|
if as_of and publish_date and publish_date > as_of:
|
# The QA layer also emits the affected-field repair scope when the date is valid.
|
pass
|
supports = _required(source, "supports", p, errors)
|
if source_id in ids:
|
errors.append(f"重复 source id：{source_id}")
|
ids.add(str(source_id))
|
if source_type not in source_types:
|
errors.append(f"未知 source_type：{source_type}")
|
if not isinstance(supports, list) or not supports:
|
errors.append(f"{p}.supports 必须是非空数组")
|
source_ids = ids
|
if isinstance(institutions, dict) and isinstance(institutions.get("forecasts"), list):
|
for index, forecast in enumerate(institutions["forecasts"]):
|
if isinstance(forecast, dict) and forecast.get("source_id") not in (None, "") and forecast["source_id"] not in source_ids:
|
errors.append(f"institutions.forecasts[{index}].source_id 未在 sources 中登记")
|
|
if errors:
|
raise InputError(errors)
|
|
|
def _ttm(annual: dict[str, Any], current: dict[str, Any], prior: dict[str, Any], metric: str) -> Decimal:
|
return Decimal(annual[metric]) + Decimal(current[metric]) - Decimal(prior[metric])
|
|
|
def _safe_ratio(numerator: Decimal, denominator: Decimal) -> Decimal | None:
|
if denominator == ZERO:
|
return None
|
return numerator / denominator
|
|
|
def _cagr(start: Decimal, end_plus_dividend: Decimal, years: int) -> Decimal | None:
|
if start <= ZERO or end_plus_dividend <= ZERO or years <= 0:
|
return None
|
return Decimal(str(math.pow(float(end_plus_dividend / start), 1.0 / years) - 1.0))
|
|
|
def _forecast_statuses(snapshot: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]]]:
|
meta = snapshot["meta"]
|
latest_info = date.fromisoformat(meta["latest_operating_info_date"])
|
current_year = str(date.fromisoformat(meta["as_of_date"]).year)
|
current_profit = Decimal(snapshot["financials"]["current_cumulative"]["attributable_profit"])
|
detail: list[dict[str, Any]] = []
|
usable_by_year: dict[str, list[dict[str, Decimal]]] = {}
|
|
for item in snapshot["institutions"]["forecasts"]:
|
statuses: list[str] = []
|
report_date = date.fromisoformat(item["report_date"])
|
if report_date < latest_info:
|
statuses.append("STALE_BEFORE_LATEST_DISCLOSURE")
|
estimate = item.get("estimates", {}).get(current_year)
|
if estimate and current_profit > ZERO and Decimal(estimate["profit"]) < current_profit:
|
statuses.append("STALE_BELOW_REPORTED_CUMULATIVE")
|
include_requested = bool(item.get("include", True))
|
included = include_requested and not statuses
|
row = {
|
"institution": item["institution"],
|
"report_date": item["report_date"],
|
"estimates": item.get("estimates", {}),
|
"core_assumption": item.get("core_assumption", "未提供"),
|
"source_id": item.get("source_id"),
|
"statuses": statuses or ["CURRENT"],
|
"included": included,
|
}
|
detail.append(row)
|
if included:
|
for year, values in item.get("estimates", {}).items():
|
usable_by_year.setdefault(str(year), []).append(
|
{"profit": Decimal(values["profit"]), "eps": Decimal(values["eps"])}
|
)
|
|
summary: dict[str, dict[str, Any]] = {}
|
shares = Decimal(snapshot["market"]["diluted_shares"])
|
market_cap = Decimal(snapshot["market"]["price"]) * shares
|
for year, estimates in sorted(usable_by_year.items()):
|
profits = [row["profit"] for row in estimates]
|
eps_values = [row["eps"] for row in estimates]
|
mean_profit = sum(profits, ZERO) / Decimal(len(profits))
|
mean_eps = sum(eps_values, ZERO) / Decimal(len(eps_values))
|
median_profit = Decimal(str(median(profits)))
|
summary[year] = {
|
"institution_count": len(profits),
|
"profit_min": min(profits),
|
"profit_mean": mean_profit,
|
"profit_median": median_profit,
|
"profit_max": max(profits),
|
"eps_mean": mean_eps,
|
"pe_on_mean": market_cap / mean_profit if mean_profit > ZERO else None,
|
"eps_profit_reconciliation_gap": abs(mean_eps * shares - mean_profit) / abs(mean_profit) if mean_profit else None,
|
}
|
return detail, summary
|
|
|
def _source_issues(snapshot: dict[str, Any], registry: dict[str, Any]) -> list[QAIssue]:
|
issues: list[QAIssue] = []
|
source_types = registry["source_types"]
|
supported: set[str] = set()
|
core_supported: set[str] = set()
|
as_of = date.fromisoformat(snapshot["meta"]["as_of_date"])
|
|
for source in snapshot["sources"]:
|
config = source_types[source["source_type"]]
|
supported.update(str(item) for item in source["supports"])
|
if config.get("core_numeric"):
|
core_supported.update(str(item) for item in source["supports"])
|
publish = date.fromisoformat(source["publish_date"])
|
if publish > as_of:
|
issues.append(QAIssue(
|
"QA-SOURCE-FUTURE-DATE", "ERROR", source["id"],
|
"来源发布日期晚于估值日，存在后视偏差。", "改用估值日当时可得版本。", "受该来源支持的字段及下游估值"
|
))
|
url = source.get("url")
|
allowed_domains = config.get("allowed_domains", [])
|
if url and allowed_domains:
|
hostname = (urlparse(url).hostname or "").lower()
|
if not any(hostname == domain or hostname.endswith("." + domain) for domain in allowed_domains):
|
issues.append(QAIssue(
|
"QA-SOURCE-DOMAIN-MISMATCH", "WARNING", source["id"],
|
f"URL 域名 {hostname or '未知'} 与注册的 {source['source_type']} 不匹配。",
|
"核实来源类型或改用注册域名原文。", "来源登记"
|
))
|
if source.get("revision_status") == "superseded":
|
issues.append(QAIssue(
|
"QA-SOURCE-SUPERSEDED", "ERROR", source["id"],
|
"旧版本已被修订文件取代，不得作为主计算输入。", "切换到 superseded_by 指定的新版本。", "受影响字段及全部下游计算"
|
))
|
|
required_core = {
|
"market.price",
|
"market.diluted_shares",
|
"financials.annual",
|
"financials.current_cumulative",
|
"financials.prior_year_same_period",
|
"balance_sheet.equity",
|
"institutions",
|
}
|
for field in sorted(required_core - core_supported):
|
severity = "WARNING" if field == "institutions" else "ERROR"
|
issues.append(QAIssue(
|
"QA-CORE-SOURCE-MISSING", severity, field,
|
"核心字段没有由注册表中允许承载核心数字的来源支持。",
|
"补充法定披露、行情接口或机构预测汇总页来源。", field
|
))
|
return issues
|
|
|
def compute_valuation(snapshot: dict[str, Any], registry: dict[str, Any]) -> dict[str, Any]:
|
validate_snapshot(snapshot, registry)
|
market = snapshot["market"]
|
financials = snapshot["financials"]
|
annual = financials["annual"]
|
current = financials["current_cumulative"]
|
prior = financials["prior_year_same_period"]
|
balance = snapshot["balance_sheet"]
|
valuation = snapshot["valuation"]
|
|
price = Decimal(market["price"])
|
shares = Decimal(market["diluted_shares"])
|
market_cap = price * shares
|
ttm_revenue = _ttm(annual, current, prior, "revenue")
|
ttm_profit = _ttm(annual, current, prior, "attributable_profit")
|
ttm_deduct_profit = _ttm(annual, current, prior, "deduct_profit")
|
|
adjustments = snapshot.get("normalization", {}).get("adjustments", [])
|
normalized_profit = ttm_profit + sum((Decimal(item["amount"]) for item in adjustments), ZERO)
|
|
equity = Decimal(balance["equity"])
|
debt = Decimal(balance["interest_bearing_debt"])
|
minority = Decimal(balance["minority_interest"])
|
cash_available = Decimal(balance["cash_available"])
|
non_operating_assets = Decimal(balance["non_operating_financial_assets"])
|
enterprise_value = market_cap + debt + minority - cash_available - non_operating_assets
|
book_value_per_share = equity / shares if equity > ZERO else None
|
pb = market_cap / equity if equity > ZERO else None
|
|
annual_cfo = Decimal(annual["cfo"]) if annual.get("cfo") is not None else None
|
annual_capex = Decimal(annual["capex"]) if annual.get("capex") is not None else None
|
annual_fcf = annual_cfo - annual_capex if annual_cfo is not None and annual_capex is not None else None
|
current_cfo = Decimal(current["cfo"]) if current.get("cfo") is not None else None
|
current_capex = Decimal(current["capex"]) if current.get("capex") is not None else None
|
current_fcf = current_cfo - current_capex if current_cfo is not None and current_capex is not None else None
|
|
scenarios: list[dict[str, Any]] = []
|
for item in valuation["scenarios"]:
|
method = item["method"]
|
if method == "pe":
|
value_low = Decimal(item["profit_low"]) * Decimal(item["multiple_low"])
|
value_high = Decimal(item["profit_high"]) * Decimal(item["multiple_high"])
|
else:
|
value_low = equity * Decimal(item["multiple_low"])
|
value_high = equity * Decimal(item["multiple_high"])
|
price_low = value_low / shares
|
price_high = value_high / shares
|
scenarios.append({
|
"name": item["name"],
|
"role": item["role"],
|
"method": method,
|
"assumption": item.get("assumption", "未提供"),
|
"equity_value_low": value_low,
|
"equity_value_high": value_high,
|
"price_low": price_low,
|
"price_high": price_high,
|
"upside_to_low": price_low / price - ONE,
|
"upside_to_high": price_high / price - ONE,
|
})
|
|
holding = valuation["holding_period"]
|
years = int(Decimal(holding["years"]))
|
required_return = Decimal(holding["required_return"])
|
dividends = Decimal(holding.get("cumulative_dividend_per_share", 0))
|
required_exit_price = price * ((ONE + required_return) ** years) - dividends
|
for item in scenarios:
|
item["cagr_low"] = _cagr(price, item["price_low"] + dividends, years)
|
item["cagr_high"] = _cagr(price, item["price_high"] + dividends, years)
|
|
reverse_pe: list[dict[str, Any]] = []
|
for value in valuation["reverse_pe_multiples"]:
|
multiple = Decimal(value)
|
implied_profit = market_cap / multiple
|
reverse_pe.append({
|
"pe": multiple,
|
"implied_eps": price / multiple,
|
"implied_profit": implied_profit,
|
"implied_margin": implied_profit / ttm_revenue if ttm_revenue > ZERO else None,
|
})
|
|
required_profit: list[dict[str, Any]] = []
|
for value in valuation.get("exit_pe_multiples", valuation["reverse_pe_multiples"]):
|
multiple = Decimal(value)
|
required_profit.append({
|
"exit_pe": multiple,
|
"required_profit": required_exit_price * shares / multiple,
|
})
|
|
institution_detail, institution_summary = _forecast_statuses(snapshot)
|
issues = _source_issues(snapshot, registry)
|
|
platform_cap = market.get("platform_market_cap")
|
if platform_cap is not None:
|
gap = abs(Decimal(platform_cap) - market_cap) / market_cap
|
if gap > Decimal("0.01"):
|
issues.append(QAIssue(
|
"QA-MARKET-CAP-RECONCILIATION", "ERROR", "market.platform_market_cap",
|
f"平台市值与股价×摊薄股本差异 {gap:.2%}，超过 1%。",
|
"检查股本日期、价格单位、证券类别、复权和接口缩放。", "市值、PE、PB、EV、全部情景"
|
))
|
|
if ttm_profit <= ZERO:
|
issues.append(QAIssue(
|
"QA-NEGATIVE-TTM-PROFIT", "WARNING", "metrics.ttm_attributable_profit",
|
"TTM 归母利润不为正，报告 PE 不适用。", "使用正常化情景、PB 或反向估值。", "PE 模块"
|
))
|
if ttm_deduct_profit <= ZERO:
|
issues.append(QAIssue(
|
"QA-NEGATIVE-TTM-DEDUCT", "WARNING", "metrics.ttm_deduct_profit",
|
"TTM 扣非利润不为正，扣非 PE 不适用。", "检查主营盈利恢复条件。", "扣非 PE 模块"
|
))
|
if annual_fcf is not None and annual_fcf < ZERO:
|
issues.append(QAIssue(
|
"QA-NEGATIVE-FCF", "WARNING", "metrics.annual_fcf",
|
"最近完整年度简化自由现金流为负。", "拆解营运资本、资本开支与回款持续性。", "DCF 与现金流交叉验证"
|
))
|
for row in institution_detail:
|
if row["statuses"] != ["CURRENT"]:
|
issues.append(QAIssue(
|
"QA-INSTITUTION-STALE", "WARNING", f"institutions.{row['institution']}",
|
"机构预测已早于最新经营信息或低于已实现累计利润。",
|
"保留作公告前预期对照，但不纳入当前一致预期。", "机构预期与 Forward PE"
|
))
|
if snapshot["institutions"]["coverage_status"] == "no_usable_forecasts":
|
issues.append(QAIssue(
|
"QA-INSTITUTION-NONE", "INFO", "institutions",
|
"估值日没有可用机构盈利预测，已明确记录为空。",
|
"不得以公司业绩预告或自有情景冒充机构预期。", "机构预期模块"
|
))
|
|
for year, summary in institution_summary.items():
|
gap = summary["eps_profit_reconciliation_gap"]
|
if gap is not None and gap > Decimal("0.03"):
|
issues.append(QAIssue(
|
"QA-INSTITUTION-EPS-GAP", "WARNING", f"institutions.{year}",
|
f"机构平均 EPS×股本与平均利润差异 {gap:.2%}，超过 3%。",
|
"检查摊薄股本、机构口径和单位。", "机构预测汇总"
|
))
|
|
base = next(item for item in scenarios if item["role"] == "base")
|
optimistic = next((item for item in scenarios if item["role"] == "optimistic"), None)
|
if optimistic and price > optimistic["price_high"]:
|
price_position = "ABOVE_OPTIMISTIC_RANGE"
|
price_label = "当前价格高于乐观情景上限"
|
elif price > base["price_high"]:
|
price_position = "ABOVE_BASE_RANGE"
|
price_label = "当前价格高于基准合理区间"
|
elif price >= base["price_low"]:
|
price_position = "WITHIN_BASE_RANGE"
|
price_label = "当前价格处于基准合理区间"
|
else:
|
price_position = "BELOW_BASE_RANGE"
|
price_label = "当前价格低于基准合理区间"
|
|
error_count = sum(issue.severity == "ERROR" for issue in issues)
|
warning_count = sum(issue.severity == "WARNING" for issue in issues)
|
qa_status = "BLOCKED_INPUT_ERRORS" if error_count else ("PASS_WITH_WARNINGS" if warning_count else "PASS")
|
|
return {
|
"engine_version": __version__,
|
"meta": snapshot["meta"],
|
"metrics": {
|
"price": price,
|
"diluted_shares": shares,
|
"market_cap": market_cap,
|
"enterprise_value": enterprise_value,
|
"ttm_revenue": ttm_revenue,
|
"ttm_attributable_profit": ttm_profit,
|
"ttm_deduct_profit": ttm_deduct_profit,
|
"normalized_profit": normalized_profit,
|
"ttm_eps": ttm_profit / shares,
|
"normalized_eps": normalized_profit / shares,
|
"reported_pe": market_cap / ttm_profit if ttm_profit > ZERO else None,
|
"deduct_pe": market_cap / ttm_deduct_profit if ttm_deduct_profit > ZERO else None,
|
"normalized_pe": market_cap / normalized_profit if normalized_profit > ZERO else None,
|
"book_value_per_share": book_value_per_share,
|
"pb": pb,
|
"ps": market_cap / ttm_revenue if ttm_revenue > ZERO else None,
|
"annual_fcf": annual_fcf,
|
"current_fcf": current_fcf,
|
"annual_cash_conversion": annual_cfo / Decimal(annual["attributable_profit"]) if annual_cfo is not None and Decimal(annual["attributable_profit"]) > ZERO else None,
|
},
|
"normalization_adjustments": adjustments,
|
"scenarios": scenarios,
|
"reverse_pe": reverse_pe,
|
"holding_period": {
|
"years": years,
|
"required_return": required_return,
|
"cumulative_dividend_per_share": dividends,
|
"required_exit_price": required_exit_price,
|
"required_profit_by_exit_pe": required_profit,
|
},
|
"institutions": {
|
"coverage_status": snapshot["institutions"]["coverage_status"],
|
"summary": institution_summary,
|
"detail": institution_detail,
|
},
|
"conclusion": {
|
"price_position": price_position,
|
"price_label": price_label,
|
"base_midpoint": (base["price_low"] + base["price_high"]) / Decimal("2"),
|
"safety_margin_to_base_midpoint": ((base["price_low"] + base["price_high"]) / Decimal("2") - price) / ((base["price_low"] + base["price_high"]) / Decimal("2")),
|
},
|
"qa": {
|
"status": qa_status,
|
"error_count": error_count,
|
"warning_count": warning_count,
|
"info_count": sum(issue.severity == "INFO" for issue in issues),
|
"issues": [issue.as_dict() for issue in issues],
|
},
|
}
|
|
|
def _fmt_amount(value: Decimal | None, digits: int = 2) -> str:
|
if value is None:
|
return "N/A"
|
return f"{value / YI:.{digits}f} 亿元"
|
|
|
def _fmt_num(value: Decimal | None, digits: int = 2, suffix: str = "") -> str:
|
if value is None:
|
return "N/A"
|
return f"{value:.{digits}f}{suffix}"
|
|
|
def _fmt_pct(value: Decimal | None, digits: int = 1) -> str:
|
if value is None:
|
return "N/A"
|
return f"{value * 100:.{digits}f}%"
|
|
|
def _md(value: Any) -> str:
|
return str(value).replace("|", "\\|").replace("\n", " ")
|
|
|
def _bullet_lines(items: Iterable[Any], empty: str = "无") -> list[str]:
|
values = [str(item) for item in items if str(item).strip()]
|
return [f"- {_md(item)}" for item in values] if values else [f"- {empty}"]
|
|
|
def render_report(snapshot: dict[str, Any], results: dict[str, Any], registry: dict[str, Any]) -> str:
|
meta = results["meta"]
|
metrics = results["metrics"]
|
analysis = snapshot.get("analysis", {})
|
lines: list[str] = [
|
f"# {_md(meta['company'])}（{_md(meta['code'])}）价格合理性评估（流水线生成底稿）",
|
"",
|
f"> 估值基准日：{meta['as_of_date']} ",
|
f"> 价格时点：{snapshot['market']['price_timestamp']} ",
|
f"> 最新经营信息日：{meta['latest_operating_info_date']} ",
|
f"> 货币：{meta['currency']}；金额计算统一使用元，表格按亿元展示。 ",
|
"> 本文是研究计算底稿，不构成交易指令或收益承诺。",
|
"",
|
"## 0. 结论摘要",
|
"",
|
f"**机械计算结论：{results['conclusion']['price_label']}。**",
|
"",
|
"| 指标 | 结果 |",
|
"|---|---:|",
|
f"| 当前价格 | {_fmt_num(metrics['price'], 2, ' 元/股')} |",
|
f"| 总市值 | {_fmt_amount(metrics['market_cap'])} |",
|
f"| 企业价值 EV | {_fmt_amount(metrics['enterprise_value'])} |",
|
f"| TTM 归母净利润 | {_fmt_amount(metrics['ttm_attributable_profit'], 4)} |",
|
f"| TTM 扣非净利润 | {_fmt_amount(metrics['ttm_deduct_profit'], 4)} |",
|
f"| TTM 收入 | {_fmt_amount(metrics['ttm_revenue'], 4)} |",
|
f"| 报告 PE | {_fmt_num(metrics['reported_pe'], 2, ' 倍')} |",
|
f"| PB | {_fmt_num(metrics['pb'], 2, ' 倍')} |",
|
f"| PS | {_fmt_num(metrics['ps'], 2, ' 倍')} |",
|
f"| 自动质检 | {results['qa']['status']}（错误 {results['qa']['error_count']} / 警告 {results['qa']['warning_count']}） |",
|
"",
|
]
|
if analysis.get("executive_conclusion"):
|
lines.extend([f"> **人工判断：** {_md(analysis['executive_conclusion'])}", ""])
|
|
lines.extend([
|
"## 1. 市场与资本结构快照",
|
"",
|
"| 项目 | 数值 | 公式或口径 |",
|
"|---|---:|---|",
|
f"| 股价 | {_fmt_num(metrics['price'], 2, ' 元')} | {snapshot['market']['price_type']} |",
|
f"| 完全摊薄股本 | {_fmt_num(metrics['diluted_shares'] / YI, 6, ' 亿股')} | 股本日：{snapshot['market']['shares_date']} |",
|
f"| 复算总市值 | {_fmt_amount(metrics['market_cap'], 4)} | 股价 × 完全摊薄股本 |",
|
f"| 归母净资产 | {_fmt_amount(Decimal(snapshot['balance_sheet']['equity']), 4)} | 截至 {snapshot['balance_sheet']['period_end']} |",
|
f"| 每股净资产 | {_fmt_num(metrics['book_value_per_share'], 4, ' 元')} | 净资产 ÷ 摊薄股本 |",
|
f"| 企业价值 | {_fmt_amount(metrics['enterprise_value'], 4)} | 市值 + 有息负债 + 少数股东权益 - 可用现金 - 非经营金融资产 |",
|
"",
|
"## 2. 信息来源与固定优先级",
|
"",
|
f"本次使用信源注册表版本 `{registry['registry_version']}`。来源身份和优先级沿用固定注册表；本次只检查公司、报告期、版本、修订状态和数据口径。",
|
"",
|
"| ID | 等级 | 来源 | 发布日 | 支持字段 | 修订状态 |",
|
"|---|---|---|---|---|---|",
|
])
|
source_types = registry["source_types"]
|
for source in snapshot["sources"]:
|
config = source_types[source["source_type"]]
|
title = f"[{_md(source['title'])}]({source['url']})" if source.get("url") else _md(source["title"])
|
lines.append(
|
f"| {_md(source['id'])} | {config['tier']} | {title} | {source['publish_date']} | {_md(', '.join(source['supports']))} | {_md(source.get('revision_status', 'current'))} |"
|
)
|
lines.extend([
|
"",
|
"停止搜索规则：取得法定披露或合规效率型来源、日期与口径明确、无修订、数字可勾稽且无数量级异常后，不再为同一数字寻找更多转载来源。",
|
"",
|
"## 3. 业务与利润来源",
|
"",
|
])
|
lines.extend(_bullet_lines(analysis.get("profit_sources", []), "快照未填写人工利润来源说明"))
|
lines.extend(["", "### 3.1 利润来源可靠性", ""])
|
lines.extend(_bullet_lines(analysis.get("profit_source_quality", []), "需由分析员结合分部、合同与回款补充"))
|
|
lines.extend([
|
"",
|
"## 4. TTM 与归一化利润",
|
"",
|
"```text",
|
"TTM = 最近完整年度 + 本期累计/预告 - 上年同期累计",
|
f"TTM 收入 = {_fmt_amount(Decimal(snapshot['financials']['annual']['revenue']), 4)} + {_fmt_amount(Decimal(snapshot['financials']['current_cumulative']['revenue']), 4)} - {_fmt_amount(Decimal(snapshot['financials']['prior_year_same_period']['revenue']), 4)} = {_fmt_amount(metrics['ttm_revenue'], 4)}",
|
f"TTM 归母 = {_fmt_amount(Decimal(snapshot['financials']['annual']['attributable_profit']), 4)} + {_fmt_amount(Decimal(snapshot['financials']['current_cumulative']['attributable_profit']), 4)} - {_fmt_amount(Decimal(snapshot['financials']['prior_year_same_period']['attributable_profit']), 4)} = {_fmt_amount(metrics['ttm_attributable_profit'], 4)}",
|
f"TTM 扣非 = {_fmt_amount(Decimal(snapshot['financials']['annual']['deduct_profit']), 4)} + {_fmt_amount(Decimal(snapshot['financials']['current_cumulative']['deduct_profit']), 4)} - {_fmt_amount(Decimal(snapshot['financials']['prior_year_same_period']['deduct_profit']), 4)} = {_fmt_amount(metrics['ttm_deduct_profit'], 4)}",
|
"```",
|
"",
|
"| 归一化调整 | 对可持续利润的带符号影响 | 依据 |",
|
"|---|---:|---|",
|
])
|
adjustments = results["normalization_adjustments"]
|
if adjustments:
|
for item in adjustments:
|
lines.append(f"| {_md(item['description'])} | {_fmt_amount(Decimal(item['amount']), 4)} | {_md(item.get('basis', '未提供'))} |")
|
else:
|
lines.append("| 无显式调整 | 0.0000 亿元 | TTM 利润只作报告口径，不等同于情景利润 |")
|
lines.extend([
|
"",
|
f"归一化结果为 **{_fmt_amount(metrics['normalized_profit'], 4)}**；只有调整项证据充分时才可用作主估值利润。",
|
"",
|
"## 5. 机构盈利预期",
|
"",
|
f"覆盖状态：`{results['institutions']['coverage_status']}`。早于最新经营信息日、或低于已实现累计利润的预测自动排除出当前一致预期。",
|
"",
|
"| 年度 | 机构数 | 利润最小值 | 利润均值 | 利润中位数 | 利润最大值 | EPS 均值 | 对应 PE |",
|
"|---|---:|---:|---:|---:|---:|---:|---:|",
|
])
|
if results["institutions"]["summary"]:
|
for year, row in results["institutions"]["summary"].items():
|
lines.append(
|
f"| {year} | {row['institution_count']} | {_fmt_amount(row['profit_min'])} | {_fmt_amount(row['profit_mean'])} | {_fmt_amount(row['profit_median'])} | {_fmt_amount(row['profit_max'])} | {_fmt_num(row['eps_mean'], 3)} | {_fmt_num(row['pe_on_mean'], 2, ' 倍')} |"
|
)
|
else:
|
lines.append("| 无可用预测 | 0 | N/A | N/A | N/A | N/A | N/A | N/A |")
|
lines.extend([
|
"",
|
"| 机构 | 研报日期 | 预测 | 核心假设 | 状态 | 纳入一致预期 |",
|
"|---|---|---|---|---|---|",
|
])
|
if results["institutions"]["detail"]:
|
for row in results["institutions"]["detail"]:
|
estimates = "; ".join(
|
f"{year}: 利润 {_fmt_amount(Decimal(value['profit']))}, EPS {_fmt_num(Decimal(value['eps']), 3)}"
|
for year, value in row["estimates"].items()
|
)
|
lines.append(
|
f"| {_md(row['institution'])} | {row['report_date']} | {_md(estimates)} | {_md(row['core_assumption'])} | {_md(', '.join(row['statuses']))} | {'是' if row['included'] else '否'} |"
|
)
|
else:
|
lines.append("| 无 | — | — | — | NO_USABLE_FORECASTS | 否 |")
|
|
lines.extend([
|
"",
|
"## 6. 利润质量、现金流与资产负债",
|
"",
|
"| 指标 | 结果 | 判断入口 |",
|
"|---|---:|---|",
|
f"| 最近年度 CFO | {_fmt_amount(Decimal(snapshot['financials']['annual']['cfo']) if snapshot['financials']['annual'].get('cfo') is not None else None, 4)} | 与利润、营运资本对照 |",
|
f"| 最近年度资本开支 | {_fmt_amount(Decimal(snapshot['financials']['annual']['capex']) if snapshot['financials']['annual'].get('capex') is not None else None, 4)} | 现金流出正数口径 |",
|
f"| 最近年度简化 FCF | {_fmt_amount(metrics['annual_fcf'], 4)} | CFO - 资本开支 |",
|
f"| 本期累计简化 FCF | {_fmt_amount(metrics['current_fcf'], 4)} | 仅在 CFO 与资本开支均可得时计算 |",
|
f"| 可用现金 | {_fmt_amount(Decimal(snapshot['balance_sheet']['cash_available']), 4)} | 已排除受限或经营最低现金（由输入方确认） |",
|
f"| 有息负债 | {_fmt_amount(Decimal(snapshot['balance_sheet']['interest_bearing_debt']), 4)} | 最新时点 |",
|
"",
|
])
|
lines.extend(_bullet_lines(analysis.get("cashflow_notes", []), "需结合应收、存货、合同资产与回款补充人工解释"))
|
|
lines.extend([
|
"",
|
"## 7. 模型选择",
|
"",
|
f"- 公司类型：{_md(analysis.get('company_type', '未填写'))}",
|
f"- 主模型：{_md(analysis.get('primary_model', '按情景表配置'))}",
|
f"- 交叉验证：{_md(', '.join(analysis.get('cross_checks', [])) or '至少补充一种独立交叉模型')}",
|
"",
|
"## 8. 悲观、基准、乐观情景估值",
|
"",
|
"| 情景 | 方法 | 核心假设 | 股权价值区间 | 每股价值区间 | 相对当前价 | 五年年化区间 |",
|
"|---|---|---|---:|---:|---:|---:|",
|
])
|
for row in results["scenarios"]:
|
lines.append(
|
f"| {_md(row['name'])} | {row['method'].upper()} | {_md(row['assumption'])} | {_fmt_amount(row['equity_value_low'])}～{_fmt_amount(row['equity_value_high'])} | {_fmt_num(row['price_low'], 2)}～{_fmt_num(row['price_high'], 2)} 元 | {_fmt_pct(row['upside_to_low'])}～{_fmt_pct(row['upside_to_high'])} | {_fmt_pct(row['cagr_low'])}～{_fmt_pct(row['cagr_high'])} |"
|
)
|
|
lines.extend([
|
"",
|
"## 9. 当前价格反向隐含利润",
|
"",
|
"| 假设 PE | 隐含 EPS | 隐含归母利润 | 占 TTM 收入 |",
|
"|---:|---:|---:|---:|",
|
])
|
for row in results["reverse_pe"]:
|
lines.append(
|
f"| {_fmt_num(row['pe'], 0, ' 倍')} | {_fmt_num(row['implied_eps'], 3, ' 元')} | {_fmt_amount(row['implied_profit'])} | {_fmt_pct(row['implied_margin'])} |"
|
)
|
|
hp = results["holding_period"]
|
lines.extend([
|
"",
|
"## 10. 持有期回报压力测试",
|
"",
|
f"若要求 {hp['years']} 年年化回报 {_fmt_pct(hp['required_return'])}，并假设累计每股分红 {_fmt_num(hp['cumulative_dividend_per_share'], 2)} 元，则所需退出价格为 **{_fmt_num(hp['required_exit_price'], 2)} 元**。",
|
"",
|
"| 退出 PE | 届时所需归母利润 |",
|
"|---:|---:|",
|
])
|
for row in hp["required_profit_by_exit_pe"]:
|
lines.append(f"| {_fmt_num(row['exit_pe'], 0, ' 倍')} | {_fmt_amount(row['required_profit'])} |")
|
|
lines.extend([
|
"",
|
"## 11. 自动质检与异常驱动复核",
|
"",
|
f"总体状态：`{results['qa']['status']}`。ERROR 会阻止把底稿当作正式结论；WARNING 只要求局部补查，不应重跑全部流程。",
|
"",
|
"| 异常 ID | 严重度 | 数据项 | 现象 | 需要补什么 | 修复后重算范围 |",
|
"|---|---|---|---|---|---|",
|
])
|
if results["qa"]["issues"]:
|
for issue in results["qa"]["issues"]:
|
lines.append(
|
f"| {issue['issue_id']} | {issue['severity']} | {_md(issue['field'])} | {_md(issue['message'])} | {_md(issue['action'])} | {_md(issue['recalc_scope'])} |"
|
)
|
else:
|
lines.append("| — | PASS | — | 未发现自动异常 | — | — |")
|
|
lines.extend(["", "## 12. 风险、上调与下调触发器", "", "### 12.1 主要风险", ""])
|
lines.extend(_bullet_lines(analysis.get("risks", []), "快照未填写公司特有风险"))
|
lines.extend(["", "### 12.2 上调触发器", ""])
|
lines.extend(_bullet_lines(analysis.get("upgrade_triggers", []), "快照未填写"))
|
lines.extend(["", "### 12.3 下调或失效触发器", ""])
|
lines.extend(_bullet_lines(analysis.get("downgrade_triggers", []), "快照未填写"))
|
|
lines.extend([
|
"",
|
"## 13. 复评增量清单",
|
"",
|
"下次复评默认只更新：价格与市值、最新股本、最新财报/预告及修正、机构预测、现金债务、重大合同和公司行动。历史年报原文及已验证公式在没有重述触发器时直接复用。",
|
"",
|
"## 14. 复算摘要",
|
"",
|
"```text",
|
f"总市值 = {metrics['price']} × {metrics['diluted_shares']} = {metrics['market_cap']} 元",
|
f"TTM 收入 = 年度 + 本期累计 - 上年同期 = {metrics['ttm_revenue']} 元",
|
f"TTM 归母 = 年度 + 本期累计 - 上年同期 = {metrics['ttm_attributable_profit']} 元",
|
f"TTM 扣非 = 年度 + 本期累计 - 上年同期 = {metrics['ttm_deduct_profit']} 元",
|
f"PB = 市值 ÷ 归母净资产 = {metrics['pb'] if metrics['pb'] is not None else 'N/A'}",
|
f"PS = 市值 ÷ TTM 收入 = {metrics['ps'] if metrics['ps'] is not None else 'N/A'}",
|
"```",
|
"",
|
"## 15. 结论边界",
|
"",
|
"- 自动引擎冻结算术、口径、机构时效和异常检查，不替代对利润持续性、订单真实性、竞争格局和治理风险的人工判断。",
|
"- 机构预测、公司预告和本报告情景已分开标记，不把预测写成事实。",
|
"- 情景价值不是目标价承诺；若自动质检存在 ERROR，必须先修复输入再形成正式判断。",
|
"- 未披露的资产注入、并购或政策收益不得计入基础价值。",
|
"",
|
"---",
|
"",
|
f"生成器版本：`{results['engine_version']}`；输入快照：`{snapshot.get('snapshot_id', '未命名')}`。",
|
])
|
return "\n".join(lines) + "\n"
|
|
|
def _jsonable(value: Any) -> Any:
|
if isinstance(value, Decimal):
|
return format(value, "f")
|
if isinstance(value, dict):
|
return {str(key): _jsonable(item) for key, item in value.items()}
|
if isinstance(value, list):
|
return [_jsonable(item) for item in value]
|
return value
|
|
|
def _canonical_bytes(value: Any) -> bytes:
|
return json.dumps(_jsonable(value), ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
|
def _sha256_bytes(value: bytes) -> str:
|
return hashlib.sha256(value).hexdigest()
|
|
|
def _sha256_file(path: Path) -> str:
|
digest = hashlib.sha256()
|
with path.open("rb") as handle:
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
digest.update(chunk)
|
return digest.hexdigest()
|
|
|
def _engine_fingerprint() -> str:
|
package_dir = Path(__file__).resolve().parent
|
digest = hashlib.sha256()
|
for name in ("__init__.py", "__main__.py", "cli.py", "core.py", "snapshot.schema.json"):
|
path = package_dir / name
|
digest.update(name.encode("utf-8"))
|
digest.update(b"\0")
|
digest.update(path.read_bytes())
|
digest.update(b"\0")
|
return digest.hexdigest()
|
|
|
def _atomic_write(path: Path, payload: bytes) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
descriptor, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
|
try:
|
with os.fdopen(descriptor, "wb") as handle:
|
handle.write(payload)
|
handle.flush()
|
os.fsync(handle.fileno())
|
os.replace(temp_name, path)
|
finally:
|
if os.path.exists(temp_name):
|
os.unlink(temp_name)
|
|
|
def run_pipeline(
|
snapshot_path: Path,
|
output_dir: Path,
|
source_registry_path: Path,
|
*,
|
force: bool = False,
|
) -> dict[str, Any]:
|
snapshot_path = Path(snapshot_path)
|
output_dir = Path(output_dir)
|
source_registry_path = Path(source_registry_path)
|
snapshot = load_snapshot(snapshot_path)
|
registry = _load_json(source_registry_path)
|
engine_fingerprint = _engine_fingerprint()
|
fingerprint = _sha256_bytes(_canonical_bytes({
|
"engine_version": __version__,
|
"engine_fingerprint": engine_fingerprint,
|
"snapshot": snapshot,
|
"source_registry": registry,
|
}))
|
|
report_path = output_dir / "valuation_report.md"
|
results_path = output_dir / "valuation_results.json"
|
manifest_path = output_dir / "run_manifest.json"
|
if not force and manifest_path.exists() and report_path.exists() and results_path.exists():
|
try:
|
prior = _load_json(manifest_path)
|
artifacts = prior.get("artifacts", {})
|
if (
|
prior.get("status") == "COMPLETE"
|
and prior.get("fingerprint") == fingerprint
|
and artifacts.get("valuation_report.md", {}).get("sha256") == _sha256_file(report_path)
|
and artifacts.get("valuation_results.json", {}).get("sha256") == _sha256_file(results_path)
|
):
|
return {
|
"status": "REUSED",
|
"fingerprint": fingerprint,
|
"output_dir": str(output_dir.resolve()),
|
"report": str(report_path.resolve()),
|
"results": str(results_path.resolve()),
|
"manifest": str(manifest_path.resolve()),
|
}
|
except (OSError, json.JSONDecodeError, InputError):
|
pass
|
|
results = compute_valuation(snapshot, registry)
|
report = render_report(snapshot, results, registry)
|
results_payload = json.dumps(_jsonable(results), ensure_ascii=False, indent=2, sort_keys=True).encode("utf-8") + b"\n"
|
report_payload = report.encode("utf-8")
|
_atomic_write(results_path, results_payload)
|
_atomic_write(report_path, report_payload)
|
manifest = {
|
"manifest_version": "1.0.0",
|
"status": "COMPLETE",
|
"fingerprint": fingerprint,
|
"engine_version": __version__,
|
"engine_fingerprint": engine_fingerprint,
|
"snapshot_path": str(snapshot_path.resolve()),
|
"source_registry_path": str(source_registry_path.resolve()),
|
"qa_status": results["qa"]["status"],
|
"artifacts": {
|
"valuation_report.md": {"sha256": _sha256_bytes(report_payload), "bytes": len(report_payload)},
|
"valuation_results.json": {"sha256": _sha256_bytes(results_payload), "bytes": len(results_payload)},
|
},
|
}
|
manifest_payload = json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True).encode("utf-8") + b"\n"
|
_atomic_write(manifest_path, manifest_payload)
|
return {
|
"status": "GENERATED",
|
"fingerprint": fingerprint,
|
"qa_status": results["qa"]["status"],
|
"output_dir": str(output_dir.resolve()),
|
"report": str(report_path.resolve()),
|
"results": str(results_path.resolve()),
|
"manifest": str(manifest_path.resolve()),
|
}
|
