Cai
2026-08-22 2042980bbf75b0eb72b725048536a878ca028ab2
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
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"