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()
|
