from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
import os
|
import tempfile
|
from collections import Counter
|
from pathlib import Path
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
COVERAGE_PATH = PROJECT_ROOT / "ana-data/result/股票估值/行业调研标的估值覆盖/全部行业调研标的估值覆盖清单.csv"
|
LATEST_PATH = PROJECT_ROOT / "ana-data/result/股票估值/估值台账/latest.csv"
|
OUTPUT_DIR = PROJECT_ROOT / "ana-data/result/股票估值/行业调研标的估值覆盖"
|
RESULT_ROOT = PROJECT_ROOT / "ana-data/result/股票估值"
|
OUTPUT_CSV = OUTPUT_DIR / "半导体估值排序.csv"
|
OUTPUT_MD = OUTPUT_DIR / "半导体估值排序.md"
|
OUTPUT_MANIFEST = OUTPUT_DIR / "全部行业调研标的估值覆盖manifest.json"
|
|
FIELDS = [
|
"代码",
|
"公司",
|
"所属行业",
|
"收盘价",
|
"基准区间",
|
"估值位置偏离",
|
"原判定",
|
"泡沫判定",
|
"高于基准上沿",
|
"主要泡沫原因",
|
"原因说明",
|
"置信度",
|
"60日涨幅",
|
"相对MA60",
|
"正式报告",
|
]
|
def read_csv(path: Path) -> list[dict[str, str]]:
|
with path.open("r", encoding="utf-8-sig", newline="") as handle:
|
return list(csv.DictReader(handle))
|
|
|
def sha256(path: Path) -> str:
|
return hashlib.sha256(path.read_bytes()).hexdigest().upper()
|
|
|
def as_float(value: str) -> float | None:
|
try:
|
return float(value)
|
except (TypeError, ValueError):
|
return None
|
|
|
def signed_pct(value: str) -> str:
|
number = as_float(value)
|
return "—" if number is None else f"{number:+.2f}%"
|
|
|
def valuation_position(row: dict[str, str]) -> tuple[int, float, str] | None:
|
close = as_float(row.get("close", ""))
|
low = as_float(row.get("base_low", ""))
|
high = as_float(row.get("base_high", ""))
|
if close is None or low is None or high is None or low <= 0 or high <= low:
|
return None
|
if close < low:
|
discount = (low - close) / low * 100.0
|
return 0, discount, f"低于下沿 {discount:.2f}%"
|
if close <= high:
|
midpoint = (low + high) / 2.0
|
deviation = (close - midpoint) / midpoint * 100.0
|
if deviation < -0.005:
|
display = f"低于中枢 {abs(deviation):.2f}%"
|
elif deviation > 0.005:
|
display = f"高于中枢 {deviation:.2f}%"
|
else:
|
display = "位于中枢 0.00%"
|
return 1, deviation, display
|
premium = (close - high) / high * 100.0
|
return 2, premium, f"高于上沿 {premium:.2f}%"
|
|
|
def display_industries(value: str) -> str:
|
return value.replace(";", "、")
|
|
|
def resolve_report_path(source: dict[str, str]) -> str:
|
if source.get("report_path"):
|
return source["report_path"]
|
code = source["ticker"].split(".", 1)[0]
|
candidates = sorted(RESULT_ROOT.glob(f"*_{code}_valuation/*价格合理性评估*.md"))
|
if len(candidates) != 1:
|
raise RuntimeError(f"Expected one formal report for {source['ticker']}, got {len(candidates)}")
|
return candidates[0].relative_to(PROJECT_ROOT).as_posix()
|
|
|
def escape_markdown(value: str) -> str:
|
return value.replace("\\", "\\\\").replace("|", "\\|").replace("\n", " ")
|
|
|
def atomic_write_text(path: Path, text: str) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
|
try:
|
with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle:
|
handle.write(text)
|
os.replace(temp_name, path)
|
except BaseException:
|
try:
|
os.unlink(temp_name)
|
except FileNotFoundError:
|
pass
|
raise
|
|
|
def atomic_write_csv(path: Path, rows: list[dict[str, str]]) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
|
try:
|
with os.fdopen(fd, "w", encoding="utf-8", newline="") as handle:
|
writer = csv.DictWriter(handle, fieldnames=FIELDS, lineterminator="\n")
|
writer.writeheader()
|
writer.writerows(rows)
|
os.replace(temp_name, path)
|
except BaseException:
|
try:
|
os.unlink(temp_name)
|
except FileNotFoundError:
|
pass
|
raise
|
|
|
def main() -> int:
|
coverage = [row for row in read_csv(COVERAGE_PATH) if "半导体" in row["industries"].split(";")]
|
latest = {row["ticker"]: row for row in read_csv(LATEST_PATH)}
|
ranked: list[tuple[tuple[int, float, str], dict[str, str]]] = []
|
|
for source in coverage:
|
current = latest.get(source["ticker"])
|
report_path = resolve_report_path(source)
|
if current is None:
|
output = {
|
"代码": source["ticker"],
|
"公司": source["company"],
|
"所属行业": display_industries(source["industries"]),
|
"收盘价": "—",
|
"基准区间": "—",
|
"估值位置偏离": "—",
|
"原判定": "特殊结论",
|
"泡沫判定": "不适用",
|
"高于基准上沿": "—",
|
"主要泡沫原因": "常规PE/PB失效",
|
"原因说明": source["coverage_status"],
|
"置信度": "—",
|
"60日涨幅": "—",
|
"相对MA60": "—",
|
"正式报告": report_path,
|
}
|
ranked.append(((4, float("inf"), source["ticker"]), output))
|
continue
|
|
close = as_float(current["close"])
|
low = as_float(current["base_low"])
|
high = as_float(current["base_high"])
|
premium = as_float(current["bubble_premium_pct"])
|
position = valuation_position(current)
|
output = {
|
"代码": source["ticker"],
|
"公司": source["company"],
|
"所属行业": display_industries(source["industries"]),
|
"收盘价": "—" if close is None else f"{close:.2f}",
|
"基准区间": "—" if low is None or high is None else f"{low:.2f}—{high:.2f}",
|
"估值位置偏离": "—" if position is None else position[2],
|
"原判定": current["label"],
|
"泡沫判定": current["bubble_status"],
|
"高于基准上沿": "—" if premium is None else f"{max(0.0, premium):.2f}%",
|
"主要泡沫原因": current["bubble_primary_cause"],
|
"原因说明": current["bubble_reason"],
|
"置信度": current["bubble_confidence"],
|
"60日涨幅": signed_pct(current["return_60d_pct"]),
|
"相对MA60": signed_pct(current["distance_to_ma60_pct"]),
|
"正式报告": report_path,
|
}
|
if position is None:
|
key = (3, float("inf"), source["ticker"])
|
else:
|
zone, metric, _ = position
|
key = (zone, -metric if zone == 0 else metric, source["ticker"])
|
ranked.append((key, output))
|
|
ranked.sort(key=lambda item: item[0])
|
rows = [item[1] for item in ranked]
|
if len(rows) != 179 or len({row["代码"] for row in rows}) != 179:
|
raise RuntimeError(f"Expected 179 unique semiconductor securities, got {len(rows)}")
|
|
atomic_write_csv(OUTPUT_CSV, rows)
|
label_counts = Counter(row["原判定"] for row in rows)
|
trade_dates = sorted({latest[row["代码"]]["trade_date"] for row in rows if row["代码"] in latest})
|
lines = [
|
"# 半导体股票估值排序",
|
"",
|
f"- 价格日：{'、'.join(trade_dates)}；半导体研究证券共{len(rows)}只。",
|
"- 排序：低于下沿 → 基准区间内 → 高于上沿 → 特殊结论。低于下沿按折价从大到小；区间内按相对中枢偏离从低到高；高于上沿按溢价从小到大。",
|
f"- 分布：偏低{label_counts['偏低']}只、基本合理{label_counts['基本合理']}只、偏贵{label_counts['偏贵']}只、明显偏贵{label_counts['明显偏贵']}只、特殊结论{label_counts['特殊结论']}只。",
|
"- “估值位置偏离”分段计算：低于下沿用 `(下沿-收盘价)/下沿`；区间内用 `(收盘价-区间中枢)/区间中枢`；高于上沿用 `(收盘价-上沿)/上沿`。",
|
"- 该排序只比较当前价格在既定合理区间中的位置，不代表经营风险已经消除，也不把不同公司的估值区间置信度视为完全相同。",
|
"- “高于基准上沿”为0表示没有超过基准合理区间上沿；泡沫原因除价格位置外属于最可能解释，不是已经证实的资金因果。",
|
f"- 输入：`全部行业调研标的估值覆盖清单.csv` SHA-256 `{sha256(COVERAGE_PATH)}`；`../估值台账/latest.csv` SHA-256 `{sha256(LATEST_PATH)}`。",
|
"- 合理区间是条件化估值，不是目标价、交易指令或收益承诺。",
|
"",
|
"| " + " | ".join(FIELDS) + " |",
|
"|" + "|".join(["---"] * len(FIELDS)) + "|",
|
]
|
for row in rows:
|
values = []
|
for field in FIELDS:
|
value = row[field]
|
if field == "正式报告" and value:
|
value = f"[报告](../../../../{value})"
|
values.append(escape_markdown(value))
|
lines.append("| " + " | ".join(values) + " |")
|
lines.append("")
|
atomic_write_text(OUTPUT_MD, "\n".join(lines))
|
|
manifest = json.loads(OUTPUT_MANIFEST.read_text(encoding="utf-8"))
|
manifest["semiconductor_ranking_trade_date"] = "、".join(trade_dates)
|
manifest["semiconductor_ranking_md_sha256"] = sha256(OUTPUT_MD).lower()
|
manifest["semiconductor_ranking_csv_sha256"] = sha256(OUTPUT_CSV).lower()
|
atomic_write_text(
|
OUTPUT_MANIFEST,
|
json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
|
)
|
|
print(json.dumps({
|
"rows": len(rows),
|
"label_counts": dict(label_counts),
|
"trade_dates": trade_dates,
|
"csv": str(OUTPUT_CSV.relative_to(PROJECT_ROOT)),
|
"csv_sha256": sha256(OUTPUT_CSV),
|
"markdown": str(OUTPUT_MD.relative_to(PROJECT_ROOT)),
|
"markdown_sha256": sha256(OUTPUT_MD),
|
"manifest": str(OUTPUT_MANIFEST.relative_to(PROJECT_ROOT)),
|
"manifest_sha256": sha256(OUTPUT_MANIFEST),
|
}, ensure_ascii=False))
|
return 0
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
