from __future__ import annotations import hashlib import json import shutil from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[2] BATCH_ID = "BATCH-STOCK-VALUATION-20260805-003" AS_OF = "2026-08-04" RESULT_ROOT = ROOT / "ana-data" / "result" / "股票估值" BATCH_DIR = RESULT_ROOT / "20260805_batch_six_images_valuation" CASE_DIR = ROOT / "ana-data" / "cases" / "股票估值" / BATCH_ID INPUT_IMAGES = [ Path("C:/Users/Cai/AppData/Local/Temp/codex-clipboard-7c6d2aea-3d70-48e7-aa9d-cc3fb100b47e.png"), Path("C:/Users/Cai/AppData/Local/Temp/codex-clipboard-3b606003-d4fb-428b-aa80-5f8086bc2c11.png"), Path("C:/Users/Cai/AppData/Local/Temp/codex-clipboard-c4a11551-c1e2-48fc-971b-66cf0ade2f88.png"), Path("C:/Users/Cai/AppData/Local/Temp/codex-clipboard-e9c3ce34-3fac-449a-89e6-5cd7b1054ec7.png"), Path("C:/Users/Cai/AppData/Local/Temp/codex-clipboard-204f7277-c19e-4f86-9080-2a607c75980e.png"), Path("C:/Users/Cai/AppData/Local/Temp/codex-clipboard-18e2165b-9375-4f52-a1bd-16f43fea2cc6.png"), ] IMAGE_COUNTS = [4, 31, 5, 9, 29, 5] def read_json(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8")) def write_json(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def fmt_profit(value: float | None, currency: str) -> str: if value is None: return "无可用预测" unit = "亿港元" if currency == "HKD" else "亿元" return f"{value / 1e8:.2f}{unit}" def fmt_price(value: float, currency: str) -> str: return f"{value:.2f}{'港元' if currency == 'HKD' else '元'}" def fmt_range(row: dict[str, Any]) -> str: unit = "港元" if row["currency"] == "HKD" else "元" return f"{row['base_low']:.2f}—{row['base_high']:.2f}{unit}" def valuation_metric(row: dict[str, Any]) -> str: pe = row.get("normalized_pe") pb = row.get("pb") if pe is not None and pe > 0: return f"归一化PE {pe:.2f}倍 / PB {pb:.2f}倍" if pb is not None else f"归一化PE {pe:.2f}倍" return f"PE不适用 / PB {pb:.2f}倍" if pb is not None else "PE、PB均不适用" def render_summary(rows: list[dict[str, Any]]) -> str: counts: dict[str, int] = {} for row in rows: group = "基本合理" if row["label"].startswith("基本合理") else row["label"] counts[group] = counts.get(group, 0) + 1 no_forecast = sum(row.get("consensus_2026") is None for row in rows) share_gaps = sum(row.get("share_status") == "QUOTE_EXACT_A1_NEAR_MATCH" for row in rows) loss_count = sum(row.get("normalized_pe") is None for row in rows) warning_count = sum(row["qa_warnings"] for row in rows) lines = [ "# 六张图片83只股票价格合理性评估批次汇总（2026-08-05）", "", f"- 批次：`{BATCH_ID}`", "- 股票数：83只（上交所/深交所80只、北交所2只、港交所1只）", "- 估值基准：2026-08-04完整交易日未复权收盘价；财务和机构信息截止估值日", "- 计算：V1统一内核复算市值、TTM、归一化利润、PE/PB/PS、三情景、反向利润和五年回报压力测试", "- 结论性质：条件化研究判断，不构成交易指令或收益承诺", "", "## 1. 结论摘要", "", f"- 结论分布：偏低{counts.get('偏低', 0)}只、基本合理{counts.get('基本合理', 0)}只、偏贵{counts.get('偏贵', 0)}只、明显偏贵{counts.get('明显偏贵', 0)}只。", f"- {loss_count}只股票的TTM归一化利润不为正，PE失真或不适用，正式报告改用PB或盈利修复情景。", f"- {len(rows)-no_forecast}只有公开2026年机构盈利预测，{no_forecast}只没有可用机构预测；空缺没有用公司预告或自有情景冒充。", f"- 83份报告QA错误为0，保留{warning_count}条业务警告；其中内地股票{share_gaps}只存在行情股本与最近法定股本的小额持续变化日期缺口。", "- 截图中的股票多为机器人、半导体、锂电和军工主题活跃标的；估值结论必须服从利润与现金流，不能用当日涨幅代替价值证明。", "", "## 2. 83只逐股结果", "", "| 序号 | 图片 | 代码 | 公司 | 当前价 | 核心估值指标 | 2026机构利润 | 基准合理区间 | 判断 | 正式报告 |", "|---:|---:|---|---|---:|---|---:|---:|---|---|", ] for idx, row in enumerate(rows, 1): profit = fmt_profit(row.get("consensus_2026"), row["currency"]) if row["currency"] == "HKD" and row.get("consensus_median_2026") is not None: profit += f"；中位{fmt_profit(row['consensus_median_2026'], 'HKD')}" link = f"../{row['formal_path']}" lines.append( f"| {idx} | {row['image_no']}-{row['image_index']} | {row['ticker']} | {row['company']} | " f"{fmt_price(row['price'], row['currency'])} | {valuation_metric(row)} | {profit}（{row['consensus_count']}家） | " f"{fmt_range(row)} | {row['label']} | [查看]({link}) |" ) undervalued = [row for row in rows if row["label"] == "偏低"] lines += [ "", "## 3. 横向阅读提示", "", "### 3.1 当前落在偏低区间的公司", "", ] for row in undervalued: lines.append( f"- {row['company']}（{row['ticker']}）：当前{fmt_price(row['price'], row['currency'])}，" f"基准区间{fmt_range(row)}；仍须逐份阅读盈利兑现和行业周期条件，不能把‘偏低’直接解释为交易建议。" ) lines += [ "", "### 3.2 基本合理与高估值公司", "", "- 基本合理只表示当前价格落在基准情景区间内，不表示风险低；优必选尤其依赖收入高增长、亏损收窄和现金消耗受控。", "- 偏贵或明显偏贵主要来自当前价格显著超过基准利润×合理倍数，或亏损公司仍享有很高PB/PS；报告中的乐观区间不是目标价承诺。", "- 当机构预测分歧大、覆盖缺失或股本存在持续行权/转债变化时，应优先读逐股报告的第5节、第11节和第12节。", "", "## 4. 优必选港股口径", "", "- 2026-08-04收盘价86.95港元；港交所7月月报证明总股本为5.03401373亿股。", "- 2025年归母亏损7.03亿元人民币；7家机构2026年预测范围为亏损4.02亿元至盈利3.49亿元人民币，逐家数字已在正式报告列出。", "- 按2026-08-03历史汇率0.8612人民币/港元统一换算后，当前约5.22倍PB、18.84倍2025年PS；PB基准区间66.60—99.90港元，判断为基本合理但高风险、高弹性。", "", "## 5. 数据、信源和缺口", "", "- 内地82只：公告、法定财务、机构汇总和历史K线由登记公开provider并发获取；V2因行情`f124=0`无法证明历史股本时诚实BLOCKED，正式报告用最新法定报告、公告或有界近似补证。", "- 优必选：财务和股本使用港交所法定披露，价格使用历史日K，机构预测使用ETNet逐家公开明细；港股没有A股式扣非归母字段，因此以归母亏损作保守代理。", "- 行情平台实时市值与8月4日历史收盘价不在同一时点，已从正式QA输入剔除；原始值与哈希保留，正式市值统一按估值日股价×股本复算。", "- 机构预测是估值日市场基准，不是法定事实；报告将机构利润、公司已实现利润和自有估值情景分开列示。", "", "## 6. 验收结果", "", f"- 正式逐股报告：{len(rows)}/{len(rows)}；每份均含第0—15节。", f"- V1 QA：错误0；警告{warning_count}，均保留在逐股第11节。", "- 批次机器表、逐股快照、来源manifest、计算结果和运行manifest均已归档；截图原件及哈希保存在批次案例目录。", "- 本批次只做估值研究，不输出买入、卖出、持有、仓位或止损指令。", "", ] return "\n".join(lines) def render_case_record(rows: list[dict[str, Any]], screenshots: list[dict[str, Any]]) -> str: lines = [ "# 六张图片83只股票估值任务清单", "", f"- 批次：`{BATCH_ID}`", "- 用户范围：六张图片中的全部股票", "- 识别结果：83只，去重后仍为83只", "- 估值日：2026-08-04", "- 任务状态：已完成", "- 普通验收边界：由当前用户验收；未建立独立审核或管理授权链", "", "## 1. 原始图片", "", "| 图片 | 股票数 | 归档路径 | SHA-256 |", "|---:|---:|---|---|", ] for item in screenshots: lines.append(f"| {item['image_no']} | {item['stock_count']} | `{item['path']}` | `{item['sha256']}` |") lines += [ "", "## 2. 识别和结果入口", "", "| 序号 | 图片位置 | 代码 | 公司 | 结论 | 正式报告 |", "|---:|---:|---|---|---|---|", ] for idx, row in enumerate(rows, 1): rel = "../../../result/股票估值/" + row["formal_path"] lines.append(f"| {idx} | {row['image_no']}-{row['image_index']} | {row['ticker']} | {row['company']} | {row['label']} | [报告]({rel}) |") lines += [ "", "## 3. 数据和计算路径", "", "- 内地82只使用V2登记公开信源并发取数，V1作为唯一公式内核；V2失败包和内容寻址缓存保留在角色私有临时目录或`ana-data/tmp/`，不冒充正式结果。", "- 优必选使用港交所法定披露、历史日K、公开机构明细和估值日前最近可核验汇率，金额统一换算为港元进入V1。", "- 正式报告、快照、来源证据manifest和计算三件套位于`ana-data/result/股票估值/20260805_*_valuation/`。", "- 批次总表：`../../../result/股票估值/20260805_batch_six_images_valuation/六张图片83只股票价格合理性评估批次汇总_20260805.md`。", "", "## 4. 已知缺口", "", "- 29只内地股票的行情股本与最近法定报告股本存在小额持续行权、转债或其他变化，股本总量用于当前估值，但精确变动日期保留为有界缺口。", "- 13只股票没有可用机构2026年盈利预测；报告明确写为无覆盖，不用公司预告或自有情景替代。", "- 港股没有A股式扣非利润；优必选使用归母亏损作保守代理。", "- 结论为条件化研究判断，不构成交易指令或收益承诺。", "", ] return "\n".join(lines) def main() -> None: batch = read_json(BATCH_DIR / "batch_results.json") rows = batch["rows"] hk = read_json(RESULT_ROOT / "20260805_ubtech_robotics_valuation" / "summary_row.json") for row in rows: row["currency"] = "CNY" rows.insert(5, hk) cursor = 0 for image_no, count in enumerate(IMAGE_COUNTS, 1): for image_index in range(1, count + 1): rows[cursor]["image_no"] = image_no rows[cursor]["image_index"] = image_index cursor += 1 if cursor != 83 or len(rows) != 83: raise RuntimeError(f"image/row count mismatch: cursor={cursor}, rows={len(rows)}") if len({row["ticker"] for row in rows}) != 83: raise RuntimeError("ticker duplicate detected") screenshot_dir = CASE_DIR / "raw" / "screenshots" screenshot_dir.mkdir(parents=True, exist_ok=True) screenshots = [] for index, (source, count) in enumerate(zip(INPUT_IMAGES, IMAGE_COUNTS), 1): if not source.exists(): raise FileNotFoundError(source) destination = screenshot_dir / f"input_{index:02d}.png" shutil.copy2(source, destination) screenshots.append({ "image_no": index, "stock_count": count, "path": str(destination.relative_to(ROOT)).replace("\\", "/"), "sha256": sha256(destination), "bytes": destination.stat().st_size, }) formal_hashes = [] for row in rows: formal = RESULT_ROOT / row["formal_path"] if not formal.exists(): raise FileNotFoundError(formal) formal_hashes.append({ "ticker": row["ticker"], "company": row["company"], "path": str(formal.relative_to(ROOT)).replace("\\", "/"), "sha256": sha256(formal), }) write_json(BATCH_DIR / "batch_results_83.json", { "batch_id": BATCH_ID, "as_of": AS_OF, "row_count": len(rows), "rows": rows, "failures": [], }) summary_path = BATCH_DIR / "六张图片83只股票价格合理性评估批次汇总_20260805.md" summary_path.write_text(render_summary(rows), encoding="utf-8") case_path = CASE_DIR / "估值任务清单.md" case_path.parent.mkdir(parents=True, exist_ok=True) case_path.write_text(render_case_record(rows, screenshots), encoding="utf-8") manifest = { "batch_id": BATCH_ID, "as_of": AS_OF, "target_count": 83, "unique_ticker_count": 83, "screenshots": screenshots, "formal_reports": formal_hashes, "batch_summary": { "path": str(summary_path.relative_to(ROOT)).replace("\\", "/"), "sha256": sha256(summary_path), }, "qa": { "error_count": sum(row["qa_errors"] for row in rows), "warning_count": sum(row["qa_warnings"] for row in rows), "formal_report_count": len(formal_hashes), }, } write_json(CASE_DIR / "case_manifest.json", manifest) print(json.dumps({ "rows": len(rows), "qa_errors": manifest["qa"]["error_count"], "qa_warnings": manifest["qa"]["warning_count"], "summary": str(summary_path.relative_to(ROOT)).replace("\\", "/"), }, ensure_ascii=False), flush=True) if __name__ == "__main__": main()