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