from __future__ import annotations
|
|
import hashlib
|
import importlib.util
|
import json
|
import sys
|
from pathlib import Path
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
sys.path.insert(0, str(ROOT / "dev" / "ana-dev"))
|
sys.path.insert(0, str(ROOT / "dev" / "project-dev"))
|
GENERATOR = ROOT / "ai-valuation-analyst" / "tools" / "generate_image_targets_batch_20260805.py"
|
SPEC = importlib.util.spec_from_file_location("image_targets_generator_for_repair", GENERATOR)
|
if SPEC is None or SPEC.loader is None:
|
raise RuntimeError(f"cannot load {GENERATOR}")
|
module = importlib.util.module_from_spec(SPEC)
|
sys.modules[SPEC.name] = module
|
SPEC.loader.exec_module(module)
|
|
RESULT_ROOT = ROOT / "ana-data" / "result" / "股票估值"
|
BATCH_JSON = RESULT_ROOT / "20260805_batch_six_images_valuation" / "batch_results.json"
|
|
|
def write_json(path: Path, value) -> None:
|
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
def sha256(path: Path) -> str:
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
def main() -> None:
|
batch = json.loads(BATCH_JSON.read_text(encoding="utf-8"))
|
repaired = 0
|
for row in batch["rows"]:
|
out_dir = RESULT_ROOT / Path(row["formal_path"]).parent
|
snapshot_path = next(out_dir.glob("*估值快照_20260804.json"))
|
snapshot = json.loads(snapshot_path.read_text(encoding="utf-8"))
|
old_cap = snapshot["market"].get("platform_market_cap")
|
snapshot["market"]["platform_market_cap"] = None
|
snapshot.setdefault("analysis", {})["platform_market_cap_excluded"] = old_cap
|
snapshot["analysis"]["platform_market_cap_exclusion_reason"] = (
|
"行情接口平台市值是取数时点实时值，与2026-08-04历史收盘价不在同一时点；正式市值只按历史收盘价×同口径股本复算。"
|
)
|
write_json(snapshot_path, snapshot)
|
calc_dir = out_dir / "calculation"
|
module.base.run_pipeline(snapshot_path, calc_dir, module.REGISTRY)
|
results_path = calc_dir / "valuation_results.json"
|
results = json.loads(results_path.read_text(encoding="utf-8"))
|
report = (calc_dir / "valuation_report.md").read_text(encoding="utf-8")
|
report = report.replace("价格合理性评估（流水线生成底稿）", "价格合理性评估", 1).replace(
|
"本文是研究计算底稿，不构成交易指令或收益承诺。",
|
"本文为按《股票价格合理性评估操作手册》形成的条件化研究结论，不构成交易指令或收益承诺。",
|
1,
|
)
|
formal_path = RESULT_ROOT / row["formal_path"]
|
formal_path.write_text(report, encoding="utf-8")
|
row["qa"] = results["qa"]["status"]
|
row["qa_errors"] = results["qa"]["error_count"]
|
row["qa_warnings"] = results["qa"]["warning_count"]
|
row["market_cap"] = float(results["metrics"]["market_cap"])
|
evidence_path = out_dir / "source_evidence_manifest.json"
|
evidence = json.loads(evidence_path.read_text(encoding="utf-8"))
|
note = (
|
"行情平台市值与历史收盘价不在同一时点，已从V1平台市值交叉校验输入中剔除；"
|
"原始行情哈希和平台值保留，正式市值仅按股价×股本复算。"
|
)
|
gaps = evidence.setdefault("bounded_gaps", [])
|
if note not in gaps:
|
gaps.append(note)
|
evidence["market_cap_timing_repair"] = {
|
"excluded_platform_market_cap": old_cap,
|
"formal_recomputed_market_cap": float(results["metrics"]["market_cap"]),
|
"snapshot_sha256": sha256(snapshot_path),
|
"calculation_sha256": sha256(results_path),
|
"formal_report_sha256": sha256(formal_path),
|
}
|
write_json(evidence_path, evidence)
|
repaired += 1
|
write_json(BATCH_JSON, batch)
|
print(json.dumps({
|
"repaired": repaired,
|
"qa_errors": sum(row["qa_errors"] for row in batch["rows"]),
|
"qa_warnings": sum(row["qa_warnings"] for row in batch["rows"]),
|
}, ensure_ascii=False), flush=True)
|
|
|
if __name__ == "__main__":
|
main()
|
