from __future__ import annotations import importlib.util import json import shutil import sys from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[2] BASE_PATH = ROOT / "ai-valuation-analyst" / "tools" / "generate_semiconductor_batch_20260804.py" SPEC = importlib.util.spec_from_file_location("valuation_batch_base", BASE_PATH) if SPEC is None or SPEC.loader is None: raise RuntimeError(f"cannot load {BASE_PATH}") base = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = base SPEC.loader.exec_module(base) BATCH_ID = "BATCH-STOCK-VALUATION-20260805-003" AS_OF = "2026-08-04" RUN_DATE = "20260805" CN = "股票估值" V2_ROOT = ROOT / "ai-valuation-analyst" / "tmp" / "valuation_batch_20260805_003" CACHE = ROOT / "ai-valuation-analyst" / "tmp" / "v2_cache" / "blobs" / "sha256" STAGING = ROOT / "ai-valuation-analyst" / "tmp" / "valuation_batch_20260805_003_generated" RESULT_ROOT = ROOT / "ana-data" / "result" / CN CASE_ROOT = ROOT / "ana-data" / "cases" / CN / BATCH_ID SUMMARY_DIR = RESULT_ROOT / "20260805_batch_six_images_valuation" REGISTRY = ROOT / "dev" / "ana-dev" / "stock_valuation_pipeline" / "source_registry.json" def S(ticker: str, slug: str, business: str, driver: str, tier: str, risk: str): return base.Stock(ticker, slug, business, driver, tier, risk) STOCKS = [ S("603650.SH", "rachem", "光刻胶及电子化学品材料", "半导体材料销量、客户认证、产品结构和原料价差", "materials_growth", "客户验证、原材料价格及半导体景气波动"), S("300576.SZ", "rongda_photosensitive", "光刻胶及配套感光化学品", "PCB与半导体光刻胶销量、价格和高端产品占比", "materials_growth", "客户认证、竞争加剧及估值溢价"), S("601133.SH", "baicheng_system", "洁净室与高科技厂房系统集成", "在手订单转化、项目验收、毛利率和回款", "stable_mfg", "工程项目验收、应收账款和下游资本开支"), S("688502.SH", "magnity_optics", "高端精密光学器件和光学系统", "半导体、生命科学和AR/VR客户放量及良率", "high_growth", "客户集中、扩产消化和高估值"), S("605488.SH", "fulai_new_material", "功能性涂布复合材料及电子材料", "标签与电子材料销量、产品结构和原料价差", "materials_growth", "新业务爬坡、原料价格和需求波动"), S("688716.SH", "china_peek", "PEEK等高性能聚合物材料", "PEEK销量、机器人等新场景渗透和毛利率", "high_growth", "需求兑现、扩产和单一材料估值溢价"), S("601689.SH", "top_group", "汽车底盘系统及轻量化零部件", "核心客户车型放量、单车价值量和产能利用率", "auto_leader", "客户集中、汽车价格战和人形机器人预期"), S("000700.SZ", "molding_technology", "汽车内外饰件与保险杠", "客户车型销量、海外产能和产品结构", "stable_mfg", "汽车周期、海外经营和低毛利"), S("301413.SZ", "ampron", "压力和温度传感器", "汽车与储能传感器出货、客户导入和产品结构", "high_growth", "客户认证、产能爬坡和估值波动"), S("603179.SH", "xinquan_automotive", "汽车内外饰系统", "新能源车型放量、海外扩产和单车价值量", "auto_leader", "客户集中、海外扩产和汽车价格战"), S("603730.SH", "daimei", "汽车遮阳系统及内饰件", "海外客户出货、单车配套价值和成本控制", "stable_mfg", "海外需求、贸易政策和客户集中"), S("002979.SZ", "leadshine", "运动控制器、伺服系统和步进系统", "工业自动化需求、伺服渗透和机器人业务", "industrial_growth", "制造业资本开支和机器人预期兑现"), S("002850.SZ", "kedali", "动力电池精密结构件", "电池客户出货、海外产能和单位盈利", "battery_leader", "客户集中、扩产折旧和电池价格压力"), S("300580.SZ", "best", "精密零部件、涡轮增压及新能源部件", "汽车零部件销量、丝杠等新产品放量和毛利率", "industrial_growth", "新业务兑现和汽车客户周期"), S("603009.SH", "beite_technology", "底盘零部件与精密主轴等汽车部件", "汽车部件销量、新产品量产和客户车型放量", "industrial_growth", "低毛利、客户周期和新业务预期"), S("600580.SH", "wolong_electric", "工业电机及驱动系统", "高效电机销量、产品结构、海外业务和原料成本", "stable_mfg", "宏观制造业周期和机器人主题溢价"), S("003021.SZ", "zhaowei", "微型传动系统", "汽车电子、医疗和机器人传动产品放量", "high_growth", "客户导入、研发投入和高估值"), S("301596.SZ", "rendi", "精密传动零部件和滚珠丝杠", "滚珠丝杠及精密零部件销量、产能利用率", "high_growth", "机器人需求兑现和小市值波动"), S("603667.SH", "cugroup", "轴承、丝杠及汽车零部件", "轴承与丝杠销量、客户拓展和产品结构", "industrial_growth", "制造业周期、扩产和机器人预期"), S("601100.SH", "hengli_hydraulics", "液压元件和精密传动部件", "工程机械景气、泵阀销量、海外收入和新产品", "auto_leader", "工程机械周期和新业务估值溢价"), S("688711.SH", "macmic", "功率半导体芯片与模块", "IGBT/模块销量、产品价格和产能利用率", "materials_growth", "功率半导体周期、价格竞争和扩产"), S("300660.SZ", "jiangsu_leili", "微特电机及驱动系统", "家电、汽车与机器人电机销量和产品结构", "industrial_growth", "传统需求波动和机器人业务兑现"), S("002896.SZ", "zhongda_leader", "减速器及精密传动设备", "减速器销量、自动化需求和机器人产品放量", "high_growth", "机器人主题估值、竞争和产能爬坡"), S("300124.SZ", "inovance", "工业自动化和新能源汽车电驱", "变频伺服、电驱出货、市场份额和规模效应", "auto_leader", "工业周期、竞争加剧和海外扩张"), S("301550.SZ", "silin", "汽车轴承和精密传动部件", "汽车轴承销量、客户拓展和机器人新业务", "industrial_growth", "客户集中和新业务商业化"), S("002472.SZ", "shuanghuan", "汽车齿轮与精密传动部件", "新能源汽车齿轮、机器人减速器和产能利用率", "auto_leader", "汽车价格战、扩产和机器人预期"), S("688017.SH", "leader_harmonic", "谐波减速器及精密传动", "工业机器人需求、市场份额和人形机器人放量", "high_growth", "需求兑现、竞争和高估值"), S("603662.SH", "keli_sensing", "称重传感器与工业物联网", "传感器销量、工业控制方案和并购协同", "industrial_growth", "并购整合、传统称重需求和机器人预期"), S("002050.SZ", "sanhuahua", "制冷控制元件及新能源汽车热管理", "热管理单车价值、客户车型放量和全球份额", "auto_leader", "客户周期、海外贸易和机器人预期"), S("688322.SH", "orbbec", "3D视觉传感器及解决方案", "机器人与消费电子客户放量、产品结构和规模效应", "high_growth", "持续盈利能力、客户导入和高估值"), S("688160.SH", "kinco", "工业自动化控制与驱动产品", "伺服、人机界面和机器人客户销量", "industrial_growth", "制造业需求和机器人主题溢价"), S("002747.SZ", "estun", "工业机器人和自动化系统", "机器人销量、国产替代、集成业务质量和费用率", "high_growth", "盈利质量、商誉、价格竞争和高估值"), S("688165.SH", "efort", "工业机器人整机与系统集成", "机器人销量、业务结构和规模降本", "generic_pb", "持续亏损、商誉和价格竞争"), S("300607.SZ", "topstar", "工业机器人、注塑机及自动化装备", "设备销量、订单验收和产品结构", "industrial_growth", "制造业周期和机器人预期"), S("688392.SH", "sonic_technology_refresh", "超声波设备及半导体、新能源应用", "设备交付、客户验收和半导体新产品放量", "growth_equipment", "客户集中、验收节奏和高估值"), S("920522.BJ", "nake_nuor", "锂电池辊压及相关装备", "设备订单、交付验收、海外客户和毛利率", "growth_equipment", "客户资本开支、北交所流动性和验收波动"), S("688005.SH", "ronbay", "锂电池正极材料", "正极材料销量、金属价格传导、海外产能和单位盈利", "battery_cycle", "行业供给、金属价格、应收与海外扩产"), S("920185.BJ", "btr", "锂电池负极与正极材料", "材料销量、价格、产能利用率和原料成本", "battery_cycle", "行业供给、客户集中和北交所流动性"), S("002709.SZ", "capchem", "电解液及锂电化学品", "电解液销量、价格、添加剂自供和海外业务", "battery_cycle", "价格竞争、原料波动和扩产"), S("002594.SZ", "byd", "新能源汽车、动力电池和电子业务", "汽车销量、单车盈利、产品结构和海外扩张", "battery_leader", "汽车价格战、海外贸易和资本开支"), S("301217.SZ", "tongguan_foil", "锂电铜箔和电子电路铜箔", "铜箔销量、加工费、良率和产能利用率", "battery_cycle", "供给过剩、加工费和扩产折旧"), S("002812.SZ", "semeda", "锂电池隔膜", "隔膜销量、单价、海外客户和单位成本", "battery_cycle", "价格竞争、产能利用率和应收风险"), S("300390.SZ", "canmax", "锂电材料与锂资源", "锂盐销量与价格、矿端成本和材料业务", "battery_cycle", "锂价周期、资源项目和存货减值"), S("688063.SH", "pylontech", "储能电池系统", "海外储能出货、库存去化、产品价格和渠道恢复", "battery_cycle", "海外库存、贸易政策和需求波动"), S("300750.SZ", "catl", "动力与储能电池系统", "电池出货、单位盈利、技术平台和海外市场", "battery_leader", "价格竞争、海外贸易和资本开支"), S("001283.SZ", "highpower", "消费和动力锂电池", "电池出货、客户新品、越南产能和单位盈利", "battery_cycle", "客户集中、海外扩产和消费电子周期"), S("002460.SZ", "ganfeng_lithium", "锂资源、锂盐和电池", "锂价、资源产量、冶炼成本和电池业务", "battery_cycle", "锂价下行、海外资源和减值"), S("300450.SZ", "lead_intelligent", "锂电及智能制造装备", "设备订单、验收、海外扩张和毛利率", "growth_equipment", "电池资本开支、验收和应收账款"), S("688270.SH", "zhenlei", "军用射频芯片和微系统", "型号放量、军品订单交付和产品结构", "defense_growth", "ST风险、军品验收和客户集中"), S("001270.SZ", "chengchang_refresh", "相控阵雷达射频芯片", "军品订单、批产交付和新型号放量", "defense_growth", "ST风险、军品订单和收入确认波动"), S("002265.SZ", "construction_industry_refresh", "特种装备和汽车零部件", "特种装备交付、汽车业务销量和产品结构", "defense_stable", "军品采购节奏和民品盈利波动"), S("688084.SH", "jingpin_refresh", "军用机器人和光电侦察装备", "军品项目验收、批产订单和新产品交付", "defense_growth", "订单集中、验收波动和小市值估值"), S("603712.SH", "seven_twelve_refresh", "军用无线通信和专网通信", "军品通信装备交付、订单恢复和产品结构", "defense_growth", "订单延期、应收账款和验收节奏"), S("000519.SZ", "north_arrows_refresh", "特种装备及超硬材料", "军品订单、金刚石产品销量和价格", "defense_stable", "军品采购和培育钻石价格周期"), S("688543.SH", "guoke_military_refresh", "导弹火箭固体发动机动力与弹药装备", "军品批产、型号进展和订单交付", "defense_growth", "客户集中、验收节奏和高估值"), S("300200.SZ", "gaomeng", "复合聚氨酯胶粘剂", "包装、汽车和新能源胶粘剂销量及原料价差", "stable_mfg", "原料价格、需求波动和并购整合"), S("600399.SH", "fushun_special_steel", "高温合金和特种钢", "高端特钢销量、产品结构、原料价格和产能利用率", "defense_stable", "原料周期、需求波动和历史治理风险"), S("300034.SZ", "cisri_gaona", "高温合金及航空发动机材料", "航空航天订单、产能释放和产品结构", "defense_growth", "军品验收、扩产和原料价格"), S("002179.SZ", "avic_optronics_refresh", "高可靠连接器和互连系统", "军工、新能源与通信连接器销量和产品结构", "defense_stable", "军品需求、汽车竞争和估值中枢"), S("000733.SZ", "zhenhua_technology", "军用电子元器件", "元器件订单、产品升级和军品需求恢复", "defense_stable", "军品采购节奏、价格和库存"), S("688002.SH", "raytron_refresh", "红外探测器、热成像与微波器件", "红外芯片与整机销量、海外拓展和产品结构", "defense_growth", "军民品需求、竞争和高估值"), S("600698.SH", "hunan_ty", "军用车辆与汽车零部件", "军品交付和增压器等汽车零部件销量", "defense_stable", "盈利基数低、军品订单和治理变化"), S("601606.SH", "great_wall_military_refresh", "迫击炮、弹药与智能装备", "军品订单、资产整合和批产交付", "defense_growth", "军品验收、重组预期和高估值"), S("002025.SZ", "aerospace_electric", "高端连接器、继电器和微特电机", "军工与通信订单、产品结构和规模效应", "defense_stable", "军品采购节奏和应收账款"), S("000768.SZ", "avic_xian", "大中型军民用飞机整机", "飞机交付数量、产品结构和生产效率", "defense_stable", "交付节奏、成本控制和大额合同波动"), S("688122.SH", "western_superconducting", "高端钛合金、高温合金与超导材料", "航空材料销量、产品结构和产能利用率", "defense_growth", "军品需求、扩产和原料价格"), S("002985.SZ", "beimo", "航空制动系统和机轮刹车", "军机配套、民航维修和新型号放量", "defense_growth", "军品订单、应收账款和客户集中"), S("600879.SH", "aerospace_electronics", "航天电子、无人系统和军用组件", "航天型号任务、军品订单和资产整合", "defense_stable", "项目验收、低毛利和应收账款"), S("600562.SH", "glarun", "雷达系统及军用电子", "雷达装备交付、订单恢复和产品结构", "defense_growth", "军品采购、项目验收和资产整合"), S("002414.SZ", "guide_infrared_refresh", "红外探测器、热成像与装备系统", "军民红外产品销量、型号批产和规模效应", "defense_growth", "军品订单、产品降价和高估值"), S("600760.SH", "avic_shenyang", "战斗机整机制造", "军机交付、产品结构和生产效率", "defense_stable", "军品交付节奏和估值中枢"), S("002413.SZ", "radar_technology", "雷达电子与卫星通信", "军品项目验收、雷达及通信产品订单", "defense_growth", "盈利波动、军品验收和资产减值"), S("002149.SZ", "western_materials", "稀有金属材料及制品", "钛材、层状复合材料和军工需求", "defense_stable", "原料价格、扩产和需求波动"), S("600893.SH", "aero_engine", "航空发动机整机", "发动机交付、维修保障和生产效率", "defense_stable", "交付节奏、成本控制和大额合同"), S("300008.SZ", "bestway_marine", "船海工程与防务装备", "船舶订单、项目交付和产能利用率", "defense_growth", "项目履约、周期波动和应收账款"), S("688297.SH", "avic_uav", "大型固定翼无人机系统", "无人机系统交付、出口订单和产品结构", "defense_growth", "订单集中、交付节奏和高估值"), S("002389.SZ", "casc_rainbow", "无人机与新材料", "无人机订单交付、出口和膜业务改善", "defense_growth", "项目验收、出口政策和民品拖累"), S("002373.SZ", "transinfo", "智慧交通与物联网", "交通数字化项目、ETC和云服务收入", "transport_tech", "政府项目回款、商誉和盈利质量"), S("301091.SZ", "shenzhen_urban_transport", "城市交通规划、数字平台和工程咨询", "交通数字化项目、咨询订单和数据服务", "transport_tech", "政府项目回款、订单波动和高估值"), S("300098.SZ", "gosuncn", "车联网、公共安全与通信物联网", "车联网项目、模组销量和费用控制", "transport_tech", "持续盈利能力、应收账款和重组预期"), S("002869.SZ", "genvict", "ETC与车路协同设备", "ETC更新、车路云项目和海外业务", "transport_tech", "政策项目节奏、应收账款和主题估值"), S("300552.SZ", "wanji", "智能交通、激光雷达和车路协同", "ETC、车路云项目和激光雷达出货", "transport_tech", "持续盈利能力、项目回款和主题估值"), ] base.BATCH_ID = BATCH_ID base.AS_OF = AS_OF base.TMP = V2_ROOT base.CACHE = CACHE base.RESULT_ROOT = STAGING base.CASE_ROOT = CASE_ROOT base.REGISTRY = REGISTRY base.PROOF_MANIFEST = V2_ROOT / "manual_share_proofs" / "manifest.json" base.PE_MULTIPLES.update( { "stable_mfg": ((12, 18), (18, 28), (28, 38)), "auto_leader": ((15, 22), (22, 32), (32, 45)), "high_growth": ((25, 40), (40, 60), (60, 85)), "battery_leader": ((15, 22), (22, 32), (32, 45)), "battery_cycle": ((10, 16), (15, 25), (25, 35)), "defense_stable": ((18, 25), (25, 35), (35, 48)), "defense_growth": ((22, 32), (32, 48), (48, 65)), "transport_tech": ((18, 28), (28, 42), (42, 60)), "generic_pb": ((1, 1), (1, 1), (1, 1)), } ) ORIGINAL_SCENARIO = base.scenario_for def scenario_for(stock, normalized: float, consensus: float | None, count: int, equity: float): if stock.tier == "generic_pb" or (normalized <= 0 and (consensus is None or consensus <= 0)): bands = ((0.8, 1.2), (1.2, 2.0), (2.0, 3.0)) assumptions = ( "盈利尚未稳定，按归母净资产折价并要求现金消耗受控。", "业务逐步接近盈亏平衡，按归母净资产的中性倍数定价。", "新业务放量并形成可持续盈利，给予成长型资产溢价。", ) scenarios = [] for (name, role), (low, high), assumption in zip( (("悲观", "pessimistic"), ("基准", "base"), ("乐观", "optimistic")), bands, assumptions ): scenarios.append( { "name": name, "role": role, "method": "pb", "multiple_low": low, "multiple_high": high, "assumption": assumption, } ) return scenarios, bands[1], bands[2] return ORIGINAL_SCENARIO(stock, normalized, consensus, count, equity) base.scenario_for = scenario_for 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 refresh_pb_narrative(out_dir: Path) -> None: snapshots = list(out_dir.glob("*估值快照_20260804.json")) if len(snapshots) != 1: raise RuntimeError(f"snapshot count invalid: {out_dir}") snapshot_path = snapshots[0] snapshot = json.loads(snapshot_path.read_text(encoding="utf-8")) excluded_cap = snapshot["market"].get("platform_market_cap") snapshot["market"]["platform_market_cap"] = None snapshot["analysis"]["platform_market_cap_excluded"] = excluded_cap snapshot["analysis"]["platform_market_cap_exclusion_reason"] = ( "行情接口平台市值是取数时点实时值，与2026-08-04历史收盘价不在同一时点；正式市值只按历史收盘价×同口径股本复算。" ) if snapshot["valuation"]["scenarios"][1]["method"] == "pb": snapshot["analysis"]["primary_model"] = "PB情景为主，盈利修复、经营现金流和机构预测为交叉验证" write_json(snapshot_path, snapshot) calc_dir = out_dir / "calculation" base.run_pipeline(snapshot_path, calc_dir, REGISTRY) report = (calc_dir / "valuation_report.md").read_text(encoding="utf-8") report = report.replace("价格合理性评估（流水线生成底稿）", "价格合理性评估", 1).replace( "本文是研究计算底稿，不构成交易指令或收益承诺。", "本文为按《股票价格合理性评估操作手册》形成的条件化研究结论，不构成交易指令或收益承诺。", 1, ) formal = next(out_dir.glob("*价格合理性评估_20260804.md")) formal.write_text(report, encoding="utf-8") def move_one(row: dict[str, Any]) -> dict[str, Any]: src_dir = STAGING / Path(row["formal_path"]).parent dest_dir = RESULT_ROOT / f"{RUN_DATE}_{row['slug']}_valuation" if dest_dir.exists(): raise FileExistsError(dest_dir) refresh_pb_narrative(src_dir) shutil.move(str(src_dir), str(dest_dir)) formal_name = Path(row["formal_path"]).name row["formal_path"] = f"{dest_dir.name}/{formal_name}" results = json.loads((dest_dir / "calculation" / "valuation_results.json").read_text(encoding="utf-8")) row["base_low"] = float(results["scenarios"][1]["price_low"]) row["base_high"] = float(results["scenarios"][1]["price_high"]) row["optimistic_high"] = float(results["scenarios"][2]["price_high"]) row["qa"] = results["qa"]["status"] row["qa_errors"] = results["qa"]["error_count"] row["qa_warnings"] = results["qa"]["warning_count"] evidence_path = dest_dir / "source_evidence_manifest.json" evidence = json.loads(evidence_path.read_text(encoding="utf-8")) evidence.setdefault("bounded_gaps", []).append( "行情平台市值与历史收盘价不在同一时点，已从V1平台市值交叉校验输入中剔除；原始行情哈希和平台值保留，正式市值仅按股价×股本复算。" ) write_json(evidence_path, evidence) return row def fmt_profit(value: float | None) -> str: return "无可用预期" if value is None else f"{value / 1e8:.2f}亿元" def fmt_multiple(value: float | None) -> str: return "N/A" if value is None else f"{value:.2f}倍" def render_summary(rows: list[dict[str, Any]]) -> str: counts: dict[str, int] = {} for row in rows: counts[row["label"]] = counts.get(row["label"], 0) + 1 lines = [ "# 六张图片股票价格合理性评估批次汇总（内地82只，2026-08-05）", "", f"- 批次：`{BATCH_ID}`", "- 价格基准：2026-08-04完整交易日未复权收盘价", "- 范围：六张图片中的82只内地股票；优必选（09880.HK）在同批次单独按港股口径评估", "- 方法：V2并发取数与证据冻结，V1作为唯一公式和16节报告生成内核", "- 结论性质：条件化估值判断，不构成交易指令或收益承诺", "", "## 1. 结论分布", "", "；".join(f"{key}{value}只" for key, value in sorted(counts.items())) + "。", "", "## 2. 逐股结论总表", "", "| 序号 | 公司 | 代码 | 收盘价 | TTM归一化利润 | 归一化PE | 2026机构利润 | 前瞻PE | 基准合理区间 | 判断 |", "|---:|---|---|---:|---:|---:|---:|---:|---:|---|", ] for idx, row in enumerate(rows, 1): link = "../" + row["formal_path"] lines.append( f"| {idx} | [{row['company']}]({link}) | {row['ticker']} | {row['price']:.2f}元 | " f"{row['normalized_profit']/1e8:.2f}亿元 | {fmt_multiple(row['normalized_pe'])} | " f"{fmt_profit(row['consensus_2026'])} | {fmt_multiple(row['forward_pe'])} | " f"{row['base_low']:.2f}—{row['base_high']:.2f}元 | {row['label']} |" ) lines += [ "", "## 3. 机构预期利润", "", "| 公司 | 覆盖池 | 2026E归母利润 | 使用说明 |", "|---|---:|---:|---|", ] for row in rows: note = "C1公开一致预期汇总，缺逐家报告日期" if row["consensus_2026"] is not None else "公开汇总未返回可用预测" lines.append(f"| {row['company']} | {row['consensus_count']}家 | {fmt_profit(row['consensus_2026'])} | {note} |") lines += [ "", "## 4. 信源与质量说明", "", "- A1：巨潮资讯法定年报、最新季报/半年报，支持财务事实、公司身份和利润来源。", "- B1/C1：东方财富完整交易日K线、结构化财务镜像和公开一致预期汇总；财务镜像与A1报告期勾稽。", "- trading_xuntou 本地行情最大日期为2026-02-24，未用于2026-08-04当前价，只作为证券身份和历史股本交叉核对。", "- V2因行情接口f124无法证明历史股本日期而诚实阻断；正式报告仅在最新法定报告股本与行情总股本一致时视为匹配，其余保留明确日期缺口。", "- 7家公司负少数股东权益触发V2严格结构阻断；正式报告保留原始值，归母权益直接取法定口径，EV中的少数股东权益按0保守处理。", "- 北交所两家公司因V2首期未接北交所公告索引，在单独补证后纳入最终总表。", "", "## 5. 如何使用", "", "优先查看结论表筛选明显偏贵或进入合理区间的公司，再进入单股16节报告核对利润来源、机构预期、情景假设、反向利润和风险触发器。合理区间是条件区间，不是目标价。", "", ] return "\n".join(lines) def main() -> None: if len(STOCKS) != 82 or len({s.ticker for s in STOCKS}) != 82: raise RuntimeError("target list must contain 82 unique mainland stocks") if STAGING.exists(): shutil.rmtree(STAGING) STAGING.mkdir(parents=True, exist_ok=True) rows = [] failures = [] for index, stock in enumerate(STOCKS, 1): try: row = base.build_one(stock, {}) rows.append(move_one(row)) print(f"[{index:02d}/82] OK {stock.ticker} {row['company']}", flush=True) except Exception as exc: failures.append({"ticker": stock.ticker, "error": f"{type(exc).__name__}: {exc}"}) print(f"[{index:02d}/82] FAIL {stock.ticker} {failures[-1]['error']}", flush=True) SUMMARY_DIR.mkdir(parents=True, exist_ok=True) write_json(SUMMARY_DIR / "batch_results.json", {"batch_id": BATCH_ID, "as_of": AS_OF, "rows": rows, "failures": failures}) (SUMMARY_DIR / "六张图片82只内地股票价格合理性评估批次汇总_20260805.md").write_text(render_summary(rows), encoding="utf-8") print(json.dumps({"success": len(rows), "failures": failures}, ensure_ascii=False), flush=True) if failures: raise SystemExit(2) BJ_MANUAL = { "920522.BJ": { "company": "纳科诺尔", "annual_date": "2026-04-13", "current_date": "2026-04-28", }, "920185.BJ": { "company": "贝特瑞", "annual_date": "2026-04-24", "current_date": "2026-04-29", }, } def patch_bj_snapshot_sources(out_dir: Path, ticker: str) -> None: snapshot_path = next(out_dir.glob("*估值快照_20260804.json")) snapshot = json.loads(snapshot_path.read_text(encoding="utf-8")) snapshot["meta"]["market"] = "北京证券交易所" for source in snapshot["sources"]: if source["id"] in {"SRC-ANNUAL-2025", "SRC-CURRENT-2026", "SRC-SHARES-REPORT"}: source["source_type"] = "bse" write_json(snapshot_path, snapshot) calc_dir = out_dir / "calculation" base.run_pipeline(snapshot_path, calc_dir, REGISTRY) report = (calc_dir / "valuation_report.md").read_text(encoding="utf-8") report = report.replace("价格合理性评估（流水线生成底稿）", "价格合理性评估", 1).replace( "本文是研究计算底稿，不构成交易指令或收益承诺。", "本文为按《股票价格合理性评估操作手册》形成的条件化研究结论，不构成交易指令或收益承诺。", 1, ) next(out_dir.glob("*价格合理性评估_20260804.md")).write_text(report, encoding="utf-8") def supplement_bj() -> None: original_read_json = base.read_json def patched_read_json(path: Path): value = original_read_json(path) if path.name != "provider_results.json": return value ticker = path.parent.name.split(".failed-", 1)[0].removeprefix("v2_").replace("_", ".") manual = BJ_MANUAL.get(ticker) if not manual: return value locator = f"https://www.bse.cn/disclosure/announcement.html?companyCode={ticker[:6]}" value["announcements"]["identity"] = { "company": manual["company"], "market": "北京证券交易所", "org_id": ticker[:6], "ticker": ticker, } value["announcements"]["records"] = [ { "announcement_id": f"BSE-{ticker[:6]}-2025-ANNUAL", "publish_date": manual["annual_date"], "source_id": f"BSE-{ticker[:6]}-2025-ANNUAL", "title": "2025年年度报告", "url": locator, }, { "announcement_id": f"BSE-{ticker[:6]}-2026-Q1", "publish_date": manual["current_date"], "source_id": f"BSE-{ticker[:6]}-2026-Q1", "title": "2026年第一季度报告", "url": locator, }, ] return value base.read_json = patched_read_json existing = json.loads((SUMMARY_DIR / "batch_results.json").read_text(encoding="utf-8")) rows = existing["rows"] for stock in (item for item in STOCKS if item.ticker in BJ_MANUAL): row = base.build_one(stock, {}) src_dir = STAGING / Path(row["formal_path"]).parent patch_bj_snapshot_sources(src_dir, stock.ticker) rows.append(move_one(row)) dest_dir = RESULT_ROOT / f"{RUN_DATE}_{row['slug']}_valuation" evidence_path = dest_dir / "source_evidence_manifest.json" evidence = json.loads(evidence_path.read_text(encoding="utf-8")) evidence["bounded_gaps"].append("V2首期未接北交所公告索引；正式报告以北交所公告查询页和法定披露日期补证，结构化财务原始哈希继续保留。") write_json(evidence_path, evidence) print(f"BJ OK {stock.ticker} {row['company']}", flush=True) order = {stock.ticker: index for index, stock in enumerate(STOCKS)} rows.sort(key=lambda item: order[item["ticker"]]) write_json(SUMMARY_DIR / "batch_results.json", {"batch_id": BATCH_ID, "as_of": AS_OF, "rows": rows, "failures": []}) (SUMMARY_DIR / "六张图片82只内地股票价格合理性评估批次汇总_20260805.md").write_text(render_summary(rows), encoding="utf-8") print(json.dumps({"success": len(rows), "failures": []}, ensure_ascii=False), flush=True) if __name__ == "__main__": if len(sys.argv) > 1 and sys.argv[1] == "--supplement-bj": supplement_bj() else: main()