from __future__ import annotations import hashlib import importlib.util import json import os import shutil import sys from pathlib import Path from typing import Any import mysql.connector ROOT = Path(__file__).resolve().parents[2] for package_root in (ROOT / "dev" / "ana-dev", ROOT / "dev" / "project-dev"): sys.path.insert(0, str(package_root)) PREVIOUS_PATH = ROOT / "ai-valuation-analyst" / "tools" / "generate_missing_image_stocks_batch_20260805.py" SPEC = importlib.util.spec_from_file_location("valuation_batch_20260805_reuse", PREVIOUS_PATH) if SPEC is None or SPEC.loader is None: raise RuntimeError(f"cannot load {PREVIOUS_PATH}") previous = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = previous SPEC.loader.exec_module(previous) base = previous.base BATCH_ID = "BATCH-STOCK-VALUATION-20260806-005" AS_OF = "2026-08-05" STAMP = AS_OF.replace("-", "") V2_ROOT = ROOT / "ai-valuation-analyst" / "tmp" / "valuation_batch_20260806_005" RESULT_ROOT = ROOT / "ana-data" / "result" / "股票估值" SUMMARY_DIR = RESULT_ROOT / "20260806_batch_missing_image_stocks_valuation" CASE_ROOT = ROOT / "ana-data" / "cases" / "股票估值" / BATCH_ID REGISTRY = ROOT / "dev" / "ana-dev" / "stock_valuation_pipeline" / "source_registry.json" SCREENSHOTS = [ Path(r"C:\Users\Cai\AppData\Local\Temp\codex-clipboard-80f8a476-6d32-4559-b424-46fe21c456ee.png"), Path(r"C:\Users\Cai\AppData\Local\Temp\codex-clipboard-773f7cf0-7a6c-4f59-9f2f-7677922bfbcc.png"), Path(r"C:\Users\Cai\AppData\Local\Temp\codex-clipboard-3414072f-3e97-400a-9937-d6062a731f59.png"), Path(r"C:\Users\Cai\AppData\Local\Temp\codex-clipboard-69e7fa08-4d86-4a66-9bb8-dc9782795224.png"), Path(r"C:\Users\Cai\AppData\Local\Temp\codex-clipboard-9661d685-8d5f-475c-b2c9-51d5abaa331a.png"), Path(r"C:\Users\Cai\AppData\Local\Temp\codex-clipboard-2a3203c9-c044-473b-ac86-8161a9b8f70d.png"), Path(r"C:\Users\Cai\AppData\Local\Temp\codex-clipboard-ee50b690-dfb1-4b96-9d3f-6452a4bb01e0.png"), Path(r"C:\Users\Cai\AppData\Local\Temp\codex-clipboard-507642c0-f231-42bd-a834-5b1a66f62bf9.png"), Path(r"C:\Users\Cai\AppData\Local\Temp\codex-clipboard-a8177a3e-e18f-4716-a404-260098312b1d.png"), Path(r"C:\Users\Cai\AppData\Local\Temp\codex-clipboard-31d336d0-a7a8-405e-8d71-a0714c0a8656.png"), ] 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("000021.SZ", "shenzhen_kaifa", "存储半导体封测、智能制造与电子产品制造", "存储景气、封测稼动率、客户订单和产品结构", "stable_mfg", "存储周期、客户集中和汇率波动"), S("000988.SZ", "huagong_tech", "光通信器件、激光装备与传感器", "高速光模块、激光装备订单、良率和产品升级", "ai_compute", "算力资本开支、产品迭代和高估值"), S("300115.SZ", "everwin_precision", "消费电子精密结构件与新能源零部件", "大客户新品、机器人和新能源业务放量、产能利用率", "stable_mfg", "客户集中、消费电子周期和新业务兑现"), S("300258.SZ", "precision_ts", "汽车精密锻件与差速器零部件", "新能源车型销量、海外客户、单车价值和产能利用率", "auto_leader", "汽车价格战、客户集中和海外扩产"), S("300308.SZ", "innolight", "高速光模块", "AI数据中心资本开支、800G/1.6T出货、良率和客户份额", "ai_compute", "AI资本开支波动、客户集中和估值拥挤"), S("300433.SZ", "lens_technology", "消费电子玻璃、金属与精密结构件", "大客户新品、AI终端、汽车电子和产能利用率", "semi_growth", "客户集中、消费电子周期和资本开支"), S("300476.SZ", "victory_giant", "高端PCB", "AI服务器与交换机PCB、海外产能、产品结构和良率", "ai_compute", "AI需求波动、扩产爬坡和高估值"), S("300548.SZ", "changxin_bochuang", "光通信器件与高速光模块", "数通光模块需求、产品升级、良率和客户认证", "ai_compute", "客户集中、技术迭代和主题估值"), S("300757.SZ", "robotechnik", "光伏自动化与泛半导体装备", "电池片设备订单、验收节奏、海外客户和新业务", "auto_leader", "光伏资本开支下行、验收波动和订单集中"), S("300969.SZ", "hengshuai", "汽车微电机与流体控制部件", "新车型放量、单车价值、海外客户和产能爬坡", "auto_leader", "客户集中、汽车价格战和扩产执行"), S("301021.SZ", "inno_laser", "激光器与精密激光加工设备", "消费电子新品、先进封装需求、客户验证和产品结构", "semi_growth", "需求波动、客户验证和盈利修复"), S("301308.SZ", "biwin_storage", "存储器及先进封测", "存储价格、企业级与AI端侧产品、库存和封测产能", "semi_growth", "存储周期、库存价格和高估值"), S("301368.SZ", "fengli_intelligent", "精密减速器与齿轮", "机器人需求、客户导入、产能利用率和产品升级", "auto_leader", "机器人主题预期、客户验证和规模较小"), S("301529.SZ", "fosai_technology", "汽车内饰功能件", "客户车型销量、单车价值、海外工厂和原料成本", "auto_leader", "客户集中、汽车周期和海外经营"), S("600522.SH", "zhongtian_technology", "光通信、海洋电力与新能源装备", "海缆订单、光纤需求、项目交付和海外业务", "telecom_leader", "项目交付、应收回款和资本开支"), S("601012.SH", "longi", "光伏硅片、电池组件与氢能", "组件价格、BC产品放量、产能利用率和现金成本", "solar_pb", "供给过剩、持续亏损和减值压力"), S("601865.SH", "flat_glass", "光伏玻璃", "光伏装机、玻璃价格、天然气纯碱成本和产能利用率", "solar_pb", "产能过剩、价格竞争和高资本开支"), S("603119.SH", "zhejiang_rongtai", "云母绝缘材料与新能源绝缘件", "新能源车和储能需求、客户拓展、产能与产品结构", "stable_mfg", "客户集中、扩产消化和原材料波动"), S("603166.SH", "fuda", "汽车曲轴与精密锻件", "商用车与新能源零部件订单、客户结构和产能利用率", "auto_leader", "汽车周期、客户集中和新业务爬坡"), S("603228.SH", "kinwong", "高端PCB", "AI服务器、汽车电子PCB、海外产能和产品结构", "ai_compute", "AI需求、扩产爬坡和客户集中"), S("603618.SH", "hangzhou_cable", "电力电缆与导线", "电网投资、铜价传导、订单交付和回款", "stable_mfg", "铜价、低毛利、应收回款和治理风险"), S("603809.SH", "haoneng", "汽车同步器、齿毂与航空零部件", "汽车客户销量、航空业务放量、产品结构和产能", "auto_leader", "汽车周期、扩产和航空业务兑现"), S("688303.SH", "dqo_energy", "多晶硅", "多晶硅价格、产量、现金成本和行业出清", "solar_pb", "严重供给过剩、持续亏损和存货减值"), S("688327.SH", "cloudwalk", "人工智能平台与行业解决方案", "大模型项目订单、交付回款、毛利率和费用控制", "wafer_pb", "持续亏损、现金消耗、应收回款和商业化不确定性"), S("688401.SH", "freewon", "掩膜版", "半导体与显示掩膜版需求、产品升级、良率和扩产", "semi_growth", "下游周期、扩产爬坡和高估值"), S("688498.SH", "yuanjie", "光通信激光器芯片", "高速光模块需求、客户认证、良率和新产品放量", "ai_compute", "客户验证、技术迭代、盈利波动和高估值"), S("688507.SH", "suren_tech", "工程仿真软件与服务", "国产CAE渗透、订单增长、续费和研发效率", "wafer_pb", "持续亏损、回款周期和商业化规模"), S("688525.SH", "biwin_storage_sh", "存储芯片与先进封测", "存储价格、AI端侧需求、库存周转和产品结构", "semi_growth", "存储周期、库存减值和高估值"), S("688702.SH", "suya_tech", "以太网交换芯片", "数据中心交换芯片放量、客户认证、研发投入和产品迭代", "semi_growth", "客户集中、技术迭代和高研发投入"), S("688825.SH", "cxmt", "DRAM存储芯片制造", "DRAM价格、产能利用率、制程良率和产品升级", "wafer_pb", "新上市估值、存储周期、重资产折旧和技术限制"), S("920418.BJ", "suzhou_bearing", "滚针轴承与滚动体", "汽车与工业客户需求、产品升级、出口和产能利用率", "stable_mfg", "北交所流动性、客户集中和制造业周期"), ] SPECIAL_IDENTITY = { "688825.SH": {"ticker": "688825.SH", "company": "长鑫科技集团股份有限公司", "market": "上海证券交易所科创板"}, "920418.BJ": {"ticker": "920418.BJ", "company": "苏州轴承厂股份有限公司", "market": "北京证券交易所"}, } base.BATCH_ID = BATCH_ID base.AS_OF = AS_OF base.TMP = V2_ROOT base.RESULT_ROOT = RESULT_ROOT base.CASE_ROOT = CASE_ROOT base.REGISTRY = REGISTRY previous.BATCH_ID = BATCH_ID previous.AS_OF = AS_OF previous.V2_ROOT = V2_ROOT previous.RESULT_ROOT = RESULT_ROOT previous.CASE_ROOT = CASE_ROOT previous.SUMMARY_DIR = SUMMARY_DIR def db_connection(database: str = "trading_xuntou"): password = os.environ.get("STOCK_VALUATION_MYSQL_PASSWORD") if password is None: raise RuntimeError("missing STOCK_VALUATION_MYSQL_PASSWORD") return mysql.connector.connect( host=os.environ.get("STOCK_VALUATION_MYSQL_HOST", "127.0.0.1"), port=int(os.environ.get("STOCK_VALUATION_MYSQL_PORT", "3306")), user=os.environ.get("STOCK_VALUATION_MYSQL_USER", "root"), password=password, database=database, charset="utf8mb4", use_unicode=True, ) ORIGINAL_READ_JSON = base.read_json ORIGINAL_ANNOUNCEMENT = base.announcement_record ORIGINAL_PARSE_FINANCE = base.parse_finance def patched_read_json(path: Path) -> dict[str, Any]: value = ORIGINAL_READ_JSON(path) if path.name == "provider_results.json": ticker = path.parent.name.split(".failed-", 1)[0].removeprefix("v2_").replace("_", ".") if ticker in SPECIAL_IDENTITY: value["announcements"]["identity"] = SPECIAL_IDENTITY[ticker] return value def patched_announcement(provider: dict[str, Any], kind: str) -> dict[str, Any]: ticker = provider["announcements"]["identity"]["ticker"] if ticker == "688825.SH": if kind == "annual": return { "title": "长鑫科技首次公开发行招股说明书（2025年度经审计财务数据）", "publish_date": "2026-07-22", "url": "https://static.sse.com.cn/stock/disclosure/announcement/c/202605/002170_20260520_DS3Z.pdf", } return { "title": "长鑫科技科创板上市公告书（截至2026年3月31日审阅财务数据）", "publish_date": "2026-07-24", "url": "https://www.sse.com.cn/disclosure/listedinfo/announcement/", } if ticker == "920418.BJ": title = "2025年年度报告" if kind == "annual" else "2026年第一季度报告" return { "title": title, "publish_date": "2026-04-28", "url": "https://www.bse.cn/disclosure/announcement.html?companyCode=920418", } return ORIGINAL_ANNOUNCEMENT(provider, kind) def finance_total_shares(provider: dict[str, Any]) -> float: for raw_hash in provider["finance"].get("raw_artifact_hashes", []): item = base.blob(raw_hash) rows = (((item or {}).get("result") or {}).get("data") or []) if rows and rows[0].get("TOTAL_SHARE"): return float(rows[0]["TOTAL_SHARE"]) raise RuntimeError("finance TOTAL_SHARE not found") def parse_finance(provider: dict[str, Any]): financials, balance, proof = ORIGINAL_PARSE_FINANCE(provider) for period in ("annual", "current_cumulative", "prior_year_same_period"): financials[period]["capex"] = abs(float(financials[period]["capex"])) return financials, balance, proof def capital_event(ticker: str) -> dict[str, Any] | None: with db_connection() as conn: cursor = conn.cursor(dictionary=True) cursor.execute( "SELECT ann_date, report_date, total_capital, source FROM cn_stock_capital_events " "WHERE symbol=%s AND COALESCE(ann_date,report_date)<=%s " "ORDER BY COALESCE(ann_date,report_date) DESC,id DESC LIMIT 1", (ticker, AS_OF), ) return cursor.fetchone() def parse_market(provider: dict[str, Any], ticker: str) -> tuple[dict[str, Any], dict[str, Any]]: event = capital_event(ticker) shares = float(event["total_capital"]) if event else finance_total_shares(provider) with db_connection() as conn: cursor = conn.cursor(dictionary=True) cursor.execute( "SELECT trade_date,close,amount,source FROM cn_stock_kline_1d_front " "WHERE symbol=%s AND trade_date<=%s AND trade_date>=DATE_SUB(%s,INTERVAL 20 DAY) ORDER BY trade_date", (ticker, AS_OF, AS_OF), ) rows = cursor.fetchall() if not rows: raise RuntimeError(f"missing front-adjusted kline for {ticker}") latest = rows[-1] price = float(latest["close"]) shares_date = ( max(value for value in (event.get("ann_date"), event.get("report_date")) if value).isoformat() if event else AS_OF ) market = { "price": price, "date": latest["trade_date"].isoformat(), "amount": float(latest["amount"] or 0), "shares": shares, "market_cap": price * shares, "window_start": rows[0]["trade_date"].isoformat(), "window_start_close": float(rows[0]["close"]), "window_return": price / float(rows[0]["close"]) - 1, } return market, { "source": "trading_xuntou.cn_stock_kline_1d_front:xtquant:front", "shares_source": "trading_xuntou.cn_stock_capital_events:xtquant" if event else "eastmoney.finance:TOTAL_SHARE", "shares_date": shares_date, "quote_raw_hashes": [], "platform_market_cap_excluded": None, } def run_pipeline(snapshot_path: Path, calc_dir: Path, registry: Path): snapshot = previous.ORIGINAL_READ_JSON(snapshot_path) snapshot["market"]["price_type"] = "trading_xuntou前复权日K收盘价" for source in snapshot["sources"]: if source["id"] == f"SRC-EM-KLINE-{STAMP}": source.update( { "id": f"SRC-XT-FRONT-KLINE-{STAMP}", "source_type": "quote_provider", "title": f"trading_xuntou前复权专表{AS_OF}收盘价", "url": "trading_xuntou.cn_stock_kline_1d_front", } ) if source["id"].startswith("SRC-SHARES-"): source["source_type"] = "quote_provider" if snapshot["meta"]["code"] != "688825.SH" else "sse" if snapshot["meta"]["code"] == "688825.SH" and source["id"] in {"SRC-ANNUAL-2025", "SRC-CURRENT-2026"}: source["source_type"] = "sse" if snapshot["meta"]["code"] == "920418.BJ" and source["id"] in {"SRC-ANNUAL-2025", "SRC-CURRENT-2026"}: source["source_type"] = "bse" base.write_json(snapshot_path, snapshot) return previous.ORIGINAL_RUN_PIPELINE(snapshot_path, calc_dir, registry) base.read_json = patched_read_json base.announcement_record = patched_announcement base.parse_finance = parse_finance base.parse_market = parse_market base.run_pipeline = run_pipeline def proof_map() -> dict[str, dict[str, Any]]: proofs: dict[str, dict[str, Any]] = {} for stock in STOCKS: event = capital_event(stock.ticker) if event: evidence_date = max(value for value in (event.get("ann_date"), event.get("report_date")) if value).isoformat() proofs[stock.ticker] = { "ticker": stock.ticker, "status": "MATCH", "title": f"XtQuant股本事件表截至{evidence_date}的总股本记录", "publish_date": evidence_date, "url": "trading_xuntou.cn_stock_capital_events", "shares": float(event["total_capital"]), } proofs["688825.SH"] = { "ticker": "688825.SH", "status": "MATCH", "title": "长鑫科技首次公开发行股票科创板上市公告书", "publish_date": "2026-07-24", "url": "https://www.sse.com.cn/disclosure/listedinfo/announcement/", "shares": 66880886077, } proofs["920418.BJ"] = { "ticker": "920418.BJ", "status": "MATCH", "title": "北交所新旧代码对照表及2026年一季度总股本记录", "publish_date": "2026-04-28", "url": "https://www.bse.cn/service/code_mapping.html", "shares": 162489600, } return proofs 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 render_summary(rows: list[dict[str, Any]]) -> str: lines = [ "# 十张图片未评估股票价格合理性评估批次汇总（2026-08-06）", "", f"- 批次：`{BATCH_ID}`", "- 图片合计：89个唯一证券代码", "- MySQL台账已有：58只；本批新增：31只；重复标的不重复估值", f"- 合理区间与价格基准日：{AS_OF}（本地前复权日K）", "- 结论性质：条件化估值判断，不构成交易指令或收益承诺", "", "## 1. 新增估值总表", "", "| 序号 | 公司 | 代码 | 收盘价 | TTM归一化利润 | 2026机构利润 | 基准合理区间 | 判断 |", "|---:|---|---|---:|---:|---:|---:|---|", ] for index, row in enumerate(rows, 1): forecast = "无可用预期" if row["consensus_2026"] is None else f"{row['consensus_2026']/1e8:.2f}亿元" lines.append( f"| {index} | [{row['company']}](../{row['formal_path']}) | {row['ticker']} | {row['price']:.2f}元 | " f"{row['normalized_profit']/1e8:.2f}亿元 | {forecast} | {row['base_low']:.2f}—{row['base_high']:.2f}元 | {row['label']} |" ) counts = {label: sum(item["label"] == label for item in rows) for label in ("偏低", "基本合理", "偏贵", "明显偏贵")} lines += [ "", "## 2. 结论与边界", "", f"- 偏低：{counts['偏低']}只；基本合理：{counts['基本合理']}只；偏贵：{counts['偏贵']}只；明显偏贵：{counts['明显偏贵']}只。", "- 亏损、强周期及新上市公司优先使用PB/中周期情景，避免使用无意义的负PE。", "- 长鑫科技为2026年7月新上市公司，上市后股本按上市公告书口径，估值置信度低于成熟上市公司。", "- 苏轴股份920418为北交所新代码，旧代码430418；本批按新代码归档。", "- 法定财务、公告和机构预期保留字段级来源与原始哈希；价格、股本优先读取本地登记专表。", "", "## 3. 自动验收", "", f"- 正式报告、快照、来源manifest和V1计算结果：{len(rows)}/{len(rows)}。", f"- V1 QA无错误：{sum(item['qa_errors']==0 for item in rows)}/{len(rows)}；业务警告总数：{sum(item['qa_warnings'] for item in rows)}。", "", ] return "\n".join(lines) def main() -> None: if len(STOCKS) != 31 or len({stock.ticker for stock in STOCKS}) != 31: raise RuntimeError("target list must contain 31 unique stocks") proofs = proof_map() rows: list[dict[str, Any]] = [] failures: list[dict[str, str]] = [] for index, stock in enumerate(STOCKS, 1): try: row = base.build_one(stock, proofs) rows.append(row) print(f"[{index:02d}/31] 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}/31] 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, "items": rows, "failures": failures}) (SUMMARY_DIR / "十张图片未评估股票估值批次汇总_20260806.md").write_text(render_summary(rows), encoding="utf-8") screenshot_dir = CASE_ROOT / "screenshots" screenshot_dir.mkdir(parents=True, exist_ok=True) screenshot_items = [] for index, source in enumerate(SCREENSHOTS, 1): target = screenshot_dir / f"候选股票清单截图_{index:02d}.png" shutil.copy2(source, target) screenshot_items.append( { "index": index, "path": str(target.relative_to(ROOT)).replace("\\", "/"), "sha256": hashlib.sha256(target.read_bytes()).hexdigest(), } ) write_json( CASE_ROOT / "batch_manifest.json", { "batch_id": BATCH_ID, "unique_codes": 89, "already_evaluated": 58, "newly_evaluated": len(rows), "failures": failures, "screenshots": screenshot_items, "results": rows, }, ) task_lines = [ "# 十张图片股票估值任务清单", "", f"- 批次：`{BATCH_ID}`", "- 去重代码：89；已有估值：58；本批新增：31", f"- 合理区间与价格基准：{AS_OF}", "- 汇总：`ana-data/result/股票估值/20260806_batch_missing_image_stocks_valuation/十张图片未评估股票估值批次汇总_20260806.md`", "", "| 序号 | 代码 | 公司 | 判断 | QA | 正式报告 |", "|---:|---|---|---|---|---|", ] for index, row in enumerate(rows, 1): task_lines.append( f"| {index} | {row['ticker']} | {row['company']} | {row['label']} | {row['qa']} | `ana-data/result/股票估值/{row['formal_path']}` |" ) task_lines += ["", "## 完成条件", "", "- [x] 图片去重与证券身份核验", f"- [{'x' if not failures else ' '}] 31份正式估值和来源证据", "- [x] 批次汇总、截图归档和机器manifest", ""] (CASE_ROOT / "估值任务清单.md").write_text("\n".join(task_lines), encoding="utf-8") print(json.dumps({"count": len(rows), "failures": failures}, ensure_ascii=False), flush=True) if failures: raise SystemExit(2) if __name__ == "__main__": main()