from __future__ import annotations from typing import Any HEADINGS = [ "结论摘要", "市场和资本结构", "信源优先级", "业务及利润来源", "历史盈利", "TTM 与归一化", "机构预期", "利润质量与现金流", "模型选择", "三情景", "反向隐含条件", "交叉验证", "持有期测试", "风险与触发器", "复评清单", "数据局限和复算摘要", ] def _fmt(value: Any) -> str: if value is None: return "GAP" return str(value) def render_report( snapshot: dict[str, Any], build_report: dict[str, Any], gaps: list[dict[str, Any]], results: dict[str, Any] | None, ) -> str: market = snapshot["market"] financials = snapshot["financials"] balance = snapshot["balance_sheet"] institutions = snapshot.get("institutions", {}) analysis = snapshot.get("analysis", {}) lines = [ f"# {snapshot['meta']['company']}({snapshot['meta']['code']})价格合理性评估", "", f"数据截止:{snapshot['meta']['as_of_date']}。本报告用于研究与复算,不构成交易指令或收益承诺。", "", ] content: dict[str, list[str]] = {} if results is None: conclusion = [ "数据采集、来源回链和机械检查已完成。", "", "GAP:需要人工判断覆盖层。本终态不提供方向性价格区间或评价。", ] else: metrics = results["metrics"] base = next(item for item in results["scenarios"] if item["role"] == "base") conclusion = [ f"V1 单一计算源给出的基准情景价格范围为 {_fmt(base['price_low'])} 至 {_fmt(base['price_high'])} 元/股。", "", f"当前收盘价 {_fmt(metrics['price'])} 元/股;该结果只表达已给定情景的机械映射。", ] content[HEADINGS[0]] = conclusion content[HEADINGS[1]] = [ "| 项目 | 数值 | 日期 |", "|---|---:|---|", f"| 收盘价 | {_fmt(market['price'])} 元/股 | {market['price_timestamp']} |", f"| 摊薄股本 | {_fmt(market['diluted_shares'])} 股 | {market['shares_date']} |", f"| 平台市值 | {_fmt(market['platform_market_cap'])} 元 | {market['shares_date']} |", f"| 归母权益 | {_fmt(balance['equity'])} 元 | {balance['period_end']} |", ] content[HEADINGS[2]] = [ "核心财务以 A1 法定公告身份回链,结构化财务与行情仅承担字段取得和勾稽。", "", f"已归档原始哈希 {len(build_report.get('raw_hashes', []))} 个;as-of 检查={build_report.get('as_of_safe')}。", ] content[HEADINGS[3]] = [ *(analysis.get("profit_sources") or ["GAP:需要人工判断覆盖层。"]), ] content[HEADINGS[4]] = [ "| 期间 | 收入 | 归母净利润 | 扣非归母净利润 |", "|---|---:|---:|---:|", *[ f"| {item['period_end']} | {_fmt(item['revenue'])} | {_fmt(item['attributable_profit'])} | {_fmt(item['deduct_profit'])} |" for item in ( financials["annual"], financials["current_cumulative"], financials["prior_year_same_period"], ) ], ] if results: metrics = results["metrics"] content[HEADINGS[5]] = [ "| 指标 | V1 结果 |", "|---|---:|", f"| TTM 收入 | {_fmt(metrics['ttm_revenue'])} |", f"| TTM 归母净利润 | {_fmt(metrics['ttm_attributable_profit'])} |", f"| TTM 扣非净利润 | {_fmt(metrics['ttm_deduct_profit'])} |", f"| 归一化利润 | {_fmt(metrics['normalized_profit'])} |", ] else: content[HEADINGS[5]] = ["法定三期数据已就绪。", "", "GAP:需要人工判断覆盖层。"] forecasts = institutions.get("forecasts") or [] content[HEADINGS[6]] = [ "| 机构 | 报告日期 | 2026E 利润 |", "|---|---|---:|", *[ f"| {item['institution']} | {item['report_date']} | {_fmt((item.get('estimates', {}).get('2026') or {}).get('profit'))} |" for item in forecasts ], f"覆盖状态:{institutions.get('coverage_status', 'gap')}。", ] content[HEADINGS[7]] = [ *(analysis.get("cashflow_notes") or ["GAP:需要人工判断覆盖层。"]), "", f"年度经营现金流 {_fmt(financials['annual']['cfo'])} 元;当期累计 {_fmt(financials['current_cumulative']['cfo'])} 元。", ] content[HEADINGS[8]] = [analysis.get("primary_model") or "GAP:需要人工判断覆盖层。"] if results: content[HEADINGS[9]] = [ "| 情景 | 角色 | 价格下限 | 价格上限 |", "|---|---|---:|---:|", *[ f"| {item['name']} | {item['role']} | {_fmt(item['price_low'])} | {_fmt(item['price_high'])} |" for item in results["scenarios"] ], ] content[HEADINGS[10]] = [ "| PE | 隐含利润 |", "|---:|---:|", *[f"| {item['pe']} | {item['implied_profit']} |" for item in results["reverse_pe"]], ] metrics = results["metrics"] content[HEADINGS[11]] = [ f"PB={metrics['pb']};PS={metrics['ps']};企业价值={metrics['enterprise_value']}。", *(analysis.get("cross_checks") or []), ] holding = results["holding_period"] content[HEADINGS[12]] = [ f"{holding['years']} 年、要求回报率 {holding['required_return']} 下,所需退出价格为 {holding['required_exit_price']} 元/股。" ] else: for heading in HEADINGS[9:13]: content[heading] = ["GAP:需要人工判断覆盖层。本节不生成伪情景、伪价格或伪回报。"] risks = analysis.get("risks") or [] triggers = (analysis.get("upgrade_triggers") or []) + (analysis.get("downgrade_triggers") or []) content[HEADINGS[13]] = [ *(risks or ["GAP:需要人工判断覆盖层。"]), "", *(triggers or ["GAP:需要人工判断覆盖层。"]), ] content[HEADINGS[14]] = [ "复评时只刷新最新价格/股本、新财报、公告修订和机构预测;历史 A1 内容按哈希复用。", "", "确认发布日与数据日均不晚于新 as-of,重新运行 QA。", ] content[HEADINGS[15]] = [ f"本次共有 {len(gaps)} 个显式 gap。全部核心字段均保留 provider、source_id 与 raw hash 回链。", "", "报告数值来自 data_snapshot;存在判断覆盖时,估值数值只来自只读 V1 results。", ] for index, heading in enumerate(HEADINGS, start=1): lines.extend([f"## {index}. {heading}", "", *content[heading], ""]) return "\n".join(lines).rstrip() + "\n"