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"
|