| | |
| | | RESULT_ROOT = PROJECT_ROOT / "ana-data/result/股票估值" |
| | | OUTPUT_CSV = OUTPUT_DIR / "半导体估值排序.csv" |
| | | OUTPUT_MD = OUTPUT_DIR / "半导体估值排序.md" |
| | | OUTPUT_MANIFEST = OUTPUT_DIR / "全部行业调研标的估值覆盖manifest.json" |
| | | |
| | | FIELDS = [ |
| | | "代码", |
| | |
| | | "所属行业", |
| | | "收盘价", |
| | | "基准区间", |
| | | "估值位置偏离", |
| | | "原判定", |
| | | "泡沫判定", |
| | | "高于基准上沿", |
| | |
| | | "相对MA60", |
| | | "正式报告", |
| | | ] |
| | | LABEL_RANK = {"偏低": 0, "基本合理": 1, "偏贵": 2, "明显偏贵": 3} |
| | | |
| | | |
| | | 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)) |
| | |
| | | return "—" if number is None else f"{number:+.2f}%" |
| | | |
| | | |
| | | def interval_position(row: dict[str, str]) -> float: |
| | | 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 high <= low: |
| | | return float("inf") |
| | | return (close - low) / (high - low) |
| | | 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: |
| | |
| | | "所属行业": display_industries(source["industries"]), |
| | | "收盘价": "—", |
| | | "基准区间": "—", |
| | | "估值位置偏离": "—", |
| | | "原判定": "特殊结论", |
| | | "泡沫判定": "不适用", |
| | | "高于基准上沿": "—", |
| | |
| | | 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}%", |
| | |
| | | "相对MA60": signed_pct(current["distance_to_ma60_pct"]), |
| | | "正式报告": report_path, |
| | | } |
| | | key = (LABEL_RANK.get(current["label"], 4), interval_position(current), source["ticker"]) |
| | | 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]) |
| | |
| | | "# 半导体股票估值排序", |
| | | "", |
| | | 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)}`。", |
| | | "- 合理区间是条件化估值,不是目标价、交易指令或收益承诺。", |
| | |
| | | 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), |
| | |
| | | "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 |
| | | |