#!/usr/bin/env python3
|
"""Enrich the stable stock-valuation ledger with valuation-bubble explanations.
|
|
The script does not change valuation ranges. It reads the current ``latest.csv``,
|
the referenced formal reports, and front-adjusted daily K-lines, then adds a
|
reproducible explanation layer. A bubble cause is an inference unless a separate
|
verified event chain exists; the output deliberately never labels it as proven
|
fund-flow causality.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import csv
|
import hashlib
|
import io
|
import json
|
import math
|
import os
|
import re
|
import statistics
|
import sys
|
from collections import Counter, defaultdict
|
from dataclasses import dataclass
|
from datetime import date, timedelta
|
from pathlib import Path
|
from typing import Iterable
|
|
import pymysql
|
|
|
SCRIPT_VERSION = "1.0"
|
BUBBLE_COLUMNS = [
|
"bubble_status",
|
"bubble_premium_pct",
|
"bubble_vs_optimistic_pct",
|
"bubble_primary_cause",
|
"bubble_secondary_cause",
|
"bubble_reason",
|
"bubble_reason_nature",
|
"bubble_confidence",
|
"bubble_evidence_basis",
|
"bubble_recheck_trigger",
|
"return_20d_pct",
|
"return_60d_pct",
|
"distance_to_ma60_pct",
|
"amount_ratio_5d_to_60d",
|
"market_sample_count",
|
]
|
|
|
THEME_RULES: list[tuple[str, tuple[str, ...]]] = [
|
(
|
"军工、商业航天或低空经济主题溢价",
|
(
|
"军工",
|
"国防",
|
"导弹",
|
"雷达",
|
"航空",
|
"航天",
|
"无人机",
|
"卫星",
|
"军品",
|
"惯性导航",
|
"火箭",
|
"低空",
|
),
|
),
|
(
|
"半导体国产替代、AI算力或先进封装预期",
|
(
|
"半导体",
|
"芯片",
|
"晶圆",
|
"光刻",
|
"刻蚀",
|
"封测",
|
"先进封装",
|
"算力",
|
"GPU",
|
"服务器",
|
"存储",
|
"EDA",
|
"光模块",
|
"PCB",
|
),
|
),
|
(
|
"机器人、自动化或智能制造成长预期",
|
(
|
"机器人",
|
"减速器",
|
"伺服",
|
"丝杠",
|
"自动化",
|
"机器视觉",
|
"智能制造",
|
"人形",
|
),
|
),
|
(
|
"创新药、医疗器械或国产替代预期",
|
(
|
"创新药",
|
"医药",
|
"生物药",
|
"临床",
|
"医疗器械",
|
"CRO",
|
"CXO",
|
"疫苗",
|
"制药",
|
"诊断",
|
),
|
),
|
(
|
"新能源、储能或电动化成长预期",
|
(
|
"新能源",
|
"锂电",
|
"电池",
|
"储能",
|
"光伏",
|
"风电",
|
"充电桩",
|
"逆变器",
|
"固态电池",
|
"电解液",
|
"正极",
|
"负极",
|
"隔膜",
|
),
|
),
|
(
|
"智能汽车、汽车电子或电动化渗透预期",
|
(
|
"汽车电子",
|
"智能驾驶",
|
"车载",
|
"汽车零部件",
|
"新能源汽车",
|
"线控",
|
"座舱",
|
),
|
),
|
(
|
"信创、网络安全或软件国产化预期",
|
(
|
"信创",
|
"网络安全",
|
"自主计算",
|
"国产软件",
|
"操作系统",
|
"数据库",
|
"工业软件",
|
"云计算",
|
"信息化",
|
),
|
),
|
(
|
"通信、光通信或卫星互联网成长预期",
|
(
|
"通信",
|
"光通信",
|
"光纤",
|
"光器件",
|
"射频",
|
"天线",
|
"卫星互联网",
|
),
|
),
|
(
|
"资源品稀缺性或周期高景气外推",
|
(
|
"黄金",
|
"白银",
|
"铜",
|
"铝",
|
"稀土",
|
"钨",
|
"锂",
|
"钴",
|
"镍",
|
"矿",
|
"有色",
|
"资源",
|
"化工",
|
"化纤",
|
"煤炭",
|
"油气",
|
),
|
),
|
(
|
"消费电子新品、AI终端或景气复苏预期",
|
(
|
"消费电子",
|
"智能终端",
|
"可穿戴",
|
"折叠屏",
|
"手机",
|
"AR",
|
"VR",
|
),
|
),
|
]
|
|
|
@dataclass(frozen=True)
|
class MarketMetrics:
|
return_20d_pct: float | None
|
return_60d_pct: float | None
|
distance_to_ma60_pct: float | None
|
amount_ratio_5d_to_60d: float | None
|
sample_count: int
|
|
|
@dataclass(frozen=True)
|
class ReportFacts:
|
business: str
|
driver: str
|
company_type: str
|
method: str
|
first_risk: str
|
negative_profit: bool
|
negative_fcf: bool
|
|
|
def parse_args() -> argparse.Namespace:
|
parser = argparse.ArgumentParser(
|
description="Add reproducible valuation-bubble explanations to latest.csv/latest.md."
|
)
|
parser.add_argument("--project-root", type=Path, default=Path.cwd())
|
parser.add_argument(
|
"--latest-csv",
|
type=Path,
|
default=Path("ana-data/result/股票估值/估值台账/latest.csv"),
|
)
|
parser.add_argument(
|
"--latest-md",
|
type=Path,
|
default=Path("ana-data/result/股票估值/估值台账/latest.md"),
|
)
|
parser.add_argument("--write", action="store_true", help="Atomically replace latest.csv/latest.md.")
|
parser.add_argument("--skip-market", action="store_true", help="Do not read front-adjusted K-lines.")
|
return parser.parse_args()
|
|
|
def resolve_under(root: Path, path: Path) -> Path:
|
root = root.resolve()
|
target = path if path.is_absolute() else root / path
|
target = target.resolve()
|
if target != root and root not in target.parents:
|
raise ValueError(f"Path escapes project root: {path}")
|
return target
|
|
|
def read_csv(path: Path) -> tuple[list[str], list[dict[str, str]]]:
|
with path.open("r", encoding="utf-8-sig", newline="") as handle:
|
reader = csv.DictReader(handle)
|
if reader.fieldnames is None:
|
raise ValueError(f"CSV has no header: {path}")
|
return list(reader.fieldnames), list(reader)
|
|
|
def clean_text(value: str, limit: int = 220) -> str:
|
value = re.sub(r"\s+", " ", value).strip(" -；;。")
|
return value[:limit]
|
|
|
def section(text: str, heading: str, next_prefix: str) -> str:
|
start = text.find(heading)
|
if start < 0:
|
return ""
|
start += len(heading)
|
end = text.find(next_prefix, start)
|
return text[start:] if end < 0 else text[start:end]
|
|
|
def first_bullet(block: str) -> str:
|
for line in block.splitlines():
|
stripped = line.strip()
|
if stripped.startswith("- "):
|
return clean_text(stripped[2:])
|
return ""
|
|
|
def all_bullets(block: str, limit: int = 3) -> list[str]:
|
values: list[str] = []
|
for line in block.splitlines():
|
stripped = line.strip()
|
if stripped.startswith("- "):
|
values.append(clean_text(stripped[2:]))
|
if len(values) >= limit:
|
break
|
return values
|
|
|
def parse_report(path: Path) -> ReportFacts:
|
text = path.read_text(encoding="utf-8-sig")
|
business_block = section(text, "## 3. 业务与利润来源", "\n## ")
|
if not business_block:
|
business_block = section(text, "## 4. 业务及利润来源", "\n## ")
|
business_bullets = all_bullets(business_block)
|
business = clean_text("；".join(business_bullets))
|
driver_match = re.search(r"(?:主要驱动是|高增长主要依赖)(.+?)(?:。|；|$)", business)
|
driver = clean_text(driver_match.group(1), 120) if driver_match else ""
|
type_match = re.search(r"公司类型：(.+)", text)
|
company_type = clean_text(type_match.group(1), 100) if type_match else ""
|
method_match = re.search(r"主模型：(.+)", text)
|
method = clean_text(method_match.group(1), 120) if method_match else ""
|
if not method:
|
model_block = section(text, "## 9. 模型选择", "\n## ")
|
model_lines = [line.strip() for line in model_block.splitlines() if line.strip()]
|
if model_lines:
|
method = clean_text(model_lines[0], 120)
|
risk_block = section(text, "### 12.1 主要风险", "\n### ")
|
if not risk_block:
|
risk_section = section(text, "## 14. 风险、上调和下调触发器", "\n## ")
|
risk_block = section(risk_section, "主要风险：", "\n上调触发器：")
|
first_risk = first_bullet(risk_block)
|
# A negative deduction-only profit warning does not mean attributable profit
|
# is negative. Use the explicit TTM attributable-profit warning only.
|
negative_profit = "QA-NEGATIVE-TTM-PROFIT" in text
|
negative_fcf = "QA-NEGATIVE-FCF" in text
|
return ReportFacts(
|
business=business,
|
driver=driver,
|
company_type=company_type,
|
method=method,
|
first_risk=first_risk,
|
negative_profit=negative_profit,
|
negative_fcf=negative_fcf,
|
)
|
|
|
def mysql_connection() -> pymysql.Connection:
|
missing = [
|
key
|
for key in ("MYSQL_HOST", "MYSQL_USER", "MYSQL_PASSWORD")
|
if not os.environ.get(key)
|
]
|
if missing:
|
raise RuntimeError(f"Missing MySQL environment variables: {', '.join(missing)}")
|
return pymysql.connect(
|
host=os.environ["MYSQL_HOST"],
|
port=int(os.environ.get("MYSQL_PORT", "3306")),
|
user=os.environ["MYSQL_USER"],
|
password=os.environ["MYSQL_PASSWORD"],
|
database="trading_xuntou",
|
charset="utf8mb4",
|
autocommit=True,
|
cursorclass=pymysql.cursors.DictCursor,
|
read_timeout=60,
|
write_timeout=60,
|
)
|
|
|
def finite_float(value: object) -> float | None:
|
if value in (None, ""):
|
return None
|
result = float(value)
|
return result if math.isfinite(result) else None
|
|
|
def market_metrics(
|
tickers: list[str], trade_date: date, skip_market: bool
|
) -> dict[str, MarketMetrics]:
|
if skip_market:
|
return {}
|
start = trade_date - timedelta(days=150)
|
placeholders = ",".join(["%s"] * len(tickers))
|
query = f"""
|
SELECT symbol, trade_date, close, amount
|
FROM cn_stock_kline_1d_front
|
WHERE symbol IN ({placeholders})
|
AND trade_date BETWEEN %s AND %s
|
AND close > 0
|
AND source = 'xtquant'
|
ORDER BY symbol, trade_date
|
"""
|
grouped: dict[str, list[tuple[date, float, float | None]]] = defaultdict(list)
|
with mysql_connection() as connection:
|
with connection.cursor() as cursor:
|
cursor.execute(query, [*tickers, start, trade_date])
|
for row in cursor.fetchall():
|
grouped[str(row["symbol"])].append(
|
(
|
row["trade_date"],
|
float(row["close"]),
|
finite_float(row["amount"]),
|
)
|
)
|
result: dict[str, MarketMetrics] = {}
|
for ticker, series in grouped.items():
|
if not series or series[-1][0] != trade_date:
|
continue
|
closes = [item[1] for item in series]
|
amounts = [item[2] for item in series]
|
current = closes[-1]
|
ret20 = current / closes[-21] - 1 if len(closes) >= 21 else None
|
ret60 = current / closes[-61] - 1 if len(closes) >= 61 else None
|
ma_window = closes[-60:]
|
ma60 = statistics.fmean(ma_window) if len(ma_window) >= 40 else None
|
distance_ma60 = current / ma60 - 1 if ma60 else None
|
valid60 = [value for value in amounts[-60:] if value and value > 0]
|
valid5 = [value for value in amounts[-5:] if value and value > 0]
|
amount_ratio = None
|
if valid60 and valid5:
|
amount_ratio = statistics.fmean(valid5) / statistics.fmean(valid60)
|
result[ticker] = MarketMetrics(
|
return_20d_pct=ret20,
|
return_60d_pct=ret60,
|
distance_to_ma60_pct=distance_ma60,
|
amount_ratio_5d_to_60d=amount_ratio,
|
sample_count=len(series),
|
)
|
return result
|
|
|
def choose_theme(facts: ReportFacts) -> str:
|
# Risk paragraphs often contain negated statements such as "no direct military
|
# evidence". They are useful as recheck triggers but must not classify the
|
# company's actual business theme.
|
haystack = " ".join((facts.business, facts.driver, facts.company_type))
|
for label, keywords in THEME_RULES:
|
if any(keyword.lower() in haystack.lower() for keyword in keywords):
|
return label
|
if any(word in haystack for word in ("重组", "并购", "资产注入", "控制权", "整合")):
|
return "并购重组、资产整合或控制权期权"
|
return "业务增长、订单兑现和利润率改善预期"
|
|
|
def pct(value: float | None) -> str:
|
return "" if value is None else f"{value * 100:.2f}"
|
|
|
def short_driver(facts: ReportFacts, theme: str) -> str:
|
if facts.driver:
|
return facts.driver
|
if facts.company_type:
|
return f"{facts.company_type}业务兑现"
|
if facts.business:
|
return facts.business[:100]
|
return theme
|
|
|
def enrich_row(
|
row: dict[str, str], facts: ReportFacts, market: MarketMetrics | None
|
) -> dict[str, str]:
|
close = float(row["close"])
|
base_high = float(row["base_high"])
|
optimistic_high = finite_float(row.get("optimistic_high"))
|
premium = close / base_high - 1
|
vs_optimistic = close / optimistic_high - 1 if optimistic_high else None
|
enriched = dict(row)
|
|
if premium <= 0:
|
enriched.update(
|
{
|
"bubble_status": "未识别估值泡沫",
|
"bubble_premium_pct": f"{premium * 100:.2f}",
|
"bubble_vs_optimistic_pct": pct(vs_optimistic),
|
"bubble_primary_cause": "不适用",
|
"bubble_secondary_cause": "不适用",
|
"bubble_reason": "当前收盘价未高于基准合理区间上沿，按本手册规则不认定估值泡沫。",
|
"bubble_reason_nature": "规则判定",
|
"bubble_confidence": "高（价格位置）",
|
"bubble_evidence_basis": "当前收盘价与正式基准合理区间比较",
|
"bubble_recheck_trigger": "价格升破基准合理区间上沿，或公司发生正式复评",
|
}
|
)
|
else:
|
theme = choose_theme(facts)
|
driver = short_driver(facts, theme)
|
primary_method = facts.method.split("，", 1)[0].strip().upper()
|
if facts.negative_profit or primary_method.startswith("PB"):
|
primary = "盈利修复或扭亏预期提前定价"
|
lead = (
|
f"当前盈利基线偏弱或常规PE适用性不足，但价格仍高于基准上沿"
|
f"{premium * 100:.1f}%，主要在交易盈利修复，并押注{driver}。"
|
)
|
else:
|
primary = theme
|
lead = (
|
f"价格高于基准合理区间上沿{premium * 100:.1f}%，主要在提前交易"
|
f"{theme}，具体押注{driver}。"
|
)
|
|
secondary: list[str] = []
|
if optimistic_high and close > optimistic_high:
|
secondary.append("超出乐观情景的叙事与情绪溢价")
|
lead += (
|
f" 当前价还高于乐观情景上沿{(close / optimistic_high - 1) * 100:.1f}%，"
|
"说明市场计入了超过现有乐观模型的额外预期。"
|
)
|
status = "泡沫-极端（超过乐观上沿）"
|
elif premium > 0.50:
|
secondary.append("乐观情景被大幅提前资本化")
|
lead += " 当前价虽未超过乐观上沿，但已经大幅提前资本化乐观情景。"
|
status = "泡沫-高"
|
elif premium > 0.15:
|
secondary.append("乐观情景提前定价")
|
lead += " 当前价位于基准与乐观上沿之间，市场已经提前支付部分乐观情景。"
|
status = "泡沫-中"
|
else:
|
secondary.append("轻度乐观预期")
|
lead += " 溢价幅度较小，更接近估值误差与轻度乐观预期的交界。"
|
status = "泡沫-轻"
|
|
momentum = False
|
if market:
|
if (
|
market.return_60d_pct is not None
|
and market.distance_to_ma60_pct is not None
|
and market.return_60d_pct >= 0.15
|
and market.distance_to_ma60_pct >= 0.08
|
):
|
momentum = True
|
secondary.append("趋势动量与交易拥挤")
|
lead += (
|
f" 近60个交易日上涨{market.return_60d_pct * 100:.1f}%，"
|
f"并高于60日均线{market.distance_to_ma60_pct * 100:.1f}%，"
|
"量价趋势可能放大估值溢价。"
|
)
|
elif (
|
market.return_20d_pct is not None
|
and market.distance_to_ma60_pct is not None
|
and market.return_20d_pct >= 0.10
|
and market.distance_to_ma60_pct >= 0.08
|
):
|
momentum = True
|
secondary.append("短期动量放大")
|
lead += (
|
f" 近20个交易日上涨{market.return_20d_pct * 100:.1f}%，"
|
f"高于60日均线{market.distance_to_ma60_pct * 100:.1f}%，"
|
"短期动量可能放大估值溢价。"
|
)
|
if market.sample_count < 61:
|
lead += (
|
f" 前复权行情样本仅{market.sample_count}个交易日，"
|
"不足以形成完整60日涨幅，长期动量不参与原因判断。"
|
)
|
count = int(row.get("consensus_count") or 0)
|
if count == 0:
|
lead += " 当前没有可用机构一致预期覆盖，原因判断更依赖估值反推与业务情景。"
|
elif count <= 2:
|
lead += f" 机构覆盖仅{count}家，预测分歧和样本偏差仍可能较大。"
|
|
evidence = "估值反推+正式报告业务/风险段落"
|
if market:
|
evidence += f"+前复权20/60日量价（样本{market.sample_count}日）"
|
confidence = "中" if facts.business and market else "中低"
|
if count == 0 or not facts.business:
|
confidence = "中低"
|
if not momentum and primary == "业务增长、订单兑现和利润率改善预期":
|
confidence = "中低"
|
recheck = facts.first_risk or f"{driver}未兑现或估值倍数回落"
|
reason_text = clean_text(lead, 620)
|
if not reason_text.endswith(("。", "！", "？")):
|
reason_text += "。"
|
enriched.update(
|
{
|
"bubble_status": status,
|
"bubble_premium_pct": f"{premium * 100:.2f}",
|
"bubble_vs_optimistic_pct": pct(vs_optimistic),
|
"bubble_primary_cause": primary,
|
"bubble_secondary_cause": "；".join(dict.fromkeys(secondary)),
|
"bubble_reason": reason_text,
|
"bubble_reason_nature": "基于估值反推、业务驱动和量价显影的推断",
|
"bubble_confidence": confidence,
|
"bubble_evidence_basis": evidence,
|
"bubble_recheck_trigger": clean_text(recheck, 180),
|
}
|
)
|
|
enriched.update(
|
{
|
"return_20d_pct": pct(market.return_20d_pct) if market else "",
|
"return_60d_pct": pct(market.return_60d_pct) if market else "",
|
"distance_to_ma60_pct": pct(market.distance_to_ma60_pct) if market else "",
|
"amount_ratio_5d_to_60d": (
|
"" if not market or market.amount_ratio_5d_to_60d is None else f"{market.amount_ratio_5d_to_60d:.4f}"
|
),
|
"market_sample_count": "" if not market else str(market.sample_count),
|
}
|
)
|
return enriched
|
|
|
def canonical_base_csv(fieldnames: list[str], rows: Iterable[dict[str, str]]) -> bytes:
|
base_fields = [field for field in fieldnames if field not in BUBBLE_COLUMNS]
|
buffer = io.StringIO(newline="")
|
writer = csv.DictWriter(
|
buffer,
|
fieldnames=base_fields,
|
extrasaction="ignore",
|
lineterminator="\n",
|
)
|
writer.writeheader()
|
writer.writerows(rows)
|
return buffer.getvalue().encode("utf-8")
|
|
|
def atomic_write_csv(path: Path, fieldnames: list[str], rows: Iterable[dict[str, str]]) -> bool:
|
temp = path.with_name(f".{path.name}.bubble.tmp")
|
try:
|
with temp.open("w", encoding="utf-8", newline="") as handle:
|
writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore", lineterminator="\n")
|
writer.writeheader()
|
writer.writerows(rows)
|
if path.exists() and temp.read_bytes() == path.read_bytes():
|
return False
|
os.replace(temp, path)
|
return True
|
finally:
|
if temp.exists():
|
temp.unlink()
|
|
|
def markdown_escape(value: object) -> str:
|
return str(value).replace("|", "\\|").replace("\n", " ")
|
|
|
def render_markdown(rows: list[dict[str, str]], source_hash: str) -> str:
|
status_counts = Counter(row["bubble_status"] for row in rows)
|
cause_counts = Counter(
|
row["bubble_primary_cause"]
|
for row in rows
|
if row["bubble_status"] != "未识别估值泡沫"
|
)
|
label_counts = Counter(row["label"] for row in rows)
|
trade_dates = sorted({row["trade_date"] for row in rows})
|
trade_date_text = trade_dates[0] if len(trade_dates) == 1 else ",".join(trade_dates)
|
bubble_count = len(rows) - status_counts.get("未识别估值泡沫", 0)
|
lines = [
|
"# 股票估值每日台账最新总表",
|
"",
|
f"- 价格交易日：`{trade_date_text}`",
|
f"- 泡沫解释版本：`{trade_date_text}/v{SCRIPT_VERSION}`",
|
f"- 记录数：`{len(rows)}`",
|
f"- 数值判定覆盖：本表共`{len(rows)}`只；无法形成当日数值判定的证券单列于同目录`latest_gaps.csv`。",
|
f"- 估值泡沫：`{bubble_count}`只；定义为当前价高于正式基准合理区间上沿。",
|
"- 原因性质：除价格位置外，泡沫原因均是基于反向估值、基础报告和量价显影的最可能解释，不是已证实的资金流因果。",
|
f"- 输入快照 SHA-256：`{source_hash}`；泡沫解释脚本：`enrich_valuation_bubbles.py v{SCRIPT_VERSION}`。",
|
"- 口径：最近完整交易日收盘价相对最近有效正式估值版本；不构成交易指令。",
|
"",
|
"## 泡沫分布",
|
"",
|
"| 泡沫状态 | 数量 |",
|
"|---|---:|",
|
]
|
for key in (
|
"未识别估值泡沫",
|
"泡沫-轻",
|
"泡沫-中",
|
"泡沫-高",
|
"泡沫-极端（超过乐观上沿）",
|
):
|
lines.append(f"| {key} | {status_counts.get(key, 0)} |")
|
lines.extend(
|
[
|
"",
|
"## 主要泡沫原因",
|
"",
|
"| 主要原因 | 数量 |",
|
"|---|---:|",
|
]
|
)
|
for cause, count in cause_counts.most_common():
|
lines.append(f"| {markdown_escape(cause)} | {count} |")
|
lines.extend(
|
[
|
"",
|
"## 全量明细",
|
"",
|
"| 代码 | 公司 | 收盘价 | 基准区间 | 原判定 | 泡沫判定 | 高于基准上沿 | 主要泡沫原因 | 原因说明 | 置信度 | 60日涨幅 | 相对MA60 | 正式报告 |",
|
"|---|---|---:|---:|---|---|---:|---|---|---|---:|---:|---|",
|
]
|
)
|
for row in rows:
|
report = f"`{row['report_path']}`"
|
base = f"{float(row['base_low']):.4f}—{float(row['base_high']):.4f}"
|
values = [
|
row["ticker"],
|
row["company"],
|
f"{float(row['close']):.4f}",
|
base,
|
row["label"],
|
row["bubble_status"],
|
f"{float(row['bubble_premium_pct']):.2f}%",
|
row["bubble_primary_cause"],
|
row["bubble_reason"],
|
row["bubble_confidence"],
|
"" if not row["return_60d_pct"] else f"{float(row['return_60d_pct']):.2f}%",
|
"" if not row["distance_to_ma60_pct"] else f"{float(row['distance_to_ma60_pct']):.2f}%",
|
report,
|
]
|
lines.append("| " + " | ".join(markdown_escape(value) for value in values) + " |")
|
lines.extend(
|
[
|
"",
|
"## 标签复核",
|
"",
|
"、".join(f"{key}{label_counts[key]}只" for key in ("偏低", "基本合理", "偏贵", "明显偏贵")),
|
"",
|
]
|
)
|
return "\n".join(lines)
|
|
|
def atomic_write_text(path: Path, content: str) -> bool:
|
temp = path.with_name(f".{path.name}.bubble.tmp")
|
try:
|
temp.write_text(content, encoding="utf-8", newline="\n")
|
if path.exists() and temp.read_bytes() == path.read_bytes():
|
return False
|
os.replace(temp, path)
|
return True
|
finally:
|
if temp.exists():
|
temp.unlink()
|
|
|
def main() -> int:
|
args = parse_args()
|
project_root = args.project_root.resolve()
|
latest_csv = resolve_under(project_root, args.latest_csv)
|
latest_md = resolve_under(project_root, args.latest_md)
|
fieldnames, rows = read_csv(latest_csv)
|
source_hash = hashlib.sha256(canonical_base_csv(fieldnames, rows)).hexdigest().upper()
|
if not rows:
|
raise RuntimeError("Current numeric judgement file is empty")
|
tickers = [row["ticker"] for row in rows]
|
if len(set(tickers)) != len(tickers):
|
raise RuntimeError("Duplicate tickers in latest.csv")
|
trade_dates = {date.fromisoformat(row["trade_date"]) for row in rows}
|
metrics: dict[str, MarketMetrics] = {}
|
for trade_date in sorted(trade_dates):
|
date_tickers = [row["ticker"] for row in rows if date.fromisoformat(row["trade_date"]) == trade_date]
|
metrics.update(market_metrics(date_tickers, trade_date, args.skip_market))
|
missing_market = sorted(set(tickers) - set(metrics)) if not args.skip_market else []
|
|
enriched: list[dict[str, str]] = []
|
missing_reports: list[str] = []
|
missing_business: list[str] = []
|
for row in rows:
|
report_path = resolve_under(project_root, Path(row["report_path"]))
|
if not report_path.is_file():
|
missing_reports.append(row["ticker"])
|
continue
|
facts = parse_report(report_path)
|
if not facts.business:
|
missing_business.append(row["ticker"])
|
enriched.append(enrich_row(row, facts, metrics.get(row["ticker"])))
|
if missing_reports:
|
raise RuntimeError(f"Missing formal reports: {missing_reports[:10]}")
|
if len(enriched) != len(rows):
|
raise RuntimeError("Enriched row count changed")
|
|
bubble_rows = [row for row in enriched if row["bubble_status"] != "未识别估值泡沫"]
|
expected_bubbles = [row for row in enriched if float(row["close"]) > float(row["base_high"])]
|
if {row["ticker"] for row in bubble_rows} != {row["ticker"] for row in expected_bubbles}:
|
raise RuntimeError("Bubble rule mismatch")
|
label_mismatch = [
|
row["ticker"]
|
for row in enriched
|
if (row["label"] in {"偏贵", "明显偏贵"})
|
!= (row["bubble_status"] != "未识别估值泡沫")
|
]
|
if label_mismatch:
|
raise RuntimeError(f"Bubble/valuation label mismatch: {label_mismatch[:10]}")
|
|
output_fields = [field for field in fieldnames if field not in BUBBLE_COLUMNS] + BUBBLE_COLUMNS
|
status_counts = Counter(row["bubble_status"] for row in enriched)
|
cause_counts = Counter(row["bubble_primary_cause"] for row in bubble_rows)
|
summary = {
|
"script_version": SCRIPT_VERSION,
|
"mode": "WRITE" if args.write else "DRY_RUN",
|
"source_csv_sha256": source_hash,
|
"row_count": len(enriched),
|
"bubble_count": len(bubble_rows),
|
"non_bubble_count": len(enriched) - len(bubble_rows),
|
"status_counts": dict(sorted(status_counts.items())),
|
"primary_cause_counts": dict(cause_counts.most_common()),
|
"market_metrics_count": len(metrics),
|
"market_under_61_count": sum(metric.sample_count < 61 for metric in metrics.values()),
|
"missing_market_count": len(missing_market),
|
"missing_market_sample": missing_market[:10],
|
"missing_business_count": len(missing_business),
|
"missing_business_sample": missing_business[:10],
|
"formal_report_count": len(enriched),
|
"label_mismatch_count": 0,
|
}
|
if args.write:
|
csv_changed = atomic_write_csv(latest_csv, output_fields, enriched)
|
md_changed = atomic_write_text(latest_md, render_markdown(enriched, source_hash))
|
summary["files_changed"] = {"latest_csv": csv_changed, "latest_md": md_changed}
|
summary["latest_csv_sha256"] = hashlib.sha256(latest_csv.read_bytes()).hexdigest().upper()
|
summary["latest_md_sha256"] = hashlib.sha256(latest_md.read_bytes()).hexdigest().upper()
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
return 0
|
|
|
if __name__ == "__main__":
|
try:
|
raise SystemExit(main())
|
except Exception as exc:
|
print(json.dumps({"status": "ERROR", "error": str(exc)}, ensure_ascii=False), file=sys.stderr)
|
raise
|
