from __future__ import annotations
|
|
import hashlib
|
import json
|
import re
|
from pathlib import Path
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
RESULT_ROOT = ROOT / "ana-data" / "result" / "股票估值"
|
BATCH_DIR = RESULT_ROOT / "20260805_batch_six_images_valuation"
|
CASE_DIR = ROOT / "ana-data" / "cases" / "股票估值" / "BATCH-STOCK-VALUATION-20260805-003"
|
|
|
def read_json(path: Path):
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
def sha256(path: Path) -> str:
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
def main() -> None:
|
batch = read_json(BATCH_DIR / "batch_results_83.json")
|
manifest = read_json(CASE_DIR / "case_manifest.json")
|
rows = batch["rows"]
|
errors: list[str] = []
|
checks = {
|
"row_count": len(rows),
|
"unique_tickers": len({row["ticker"] for row in rows}),
|
"formal_reports": 0,
|
"snapshots": 0,
|
"source_manifests": 0,
|
"calculation_results": 0,
|
"run_manifests": 0,
|
"reports_with_16_sections": 0,
|
"reports_without_placeholders": 0,
|
"qa_error_count": 0,
|
"qa_warning_count": 0,
|
"market_cap_reconciliation_max_gap": 0.0,
|
"formal_hash_matches": 0,
|
"summary_links_checked": 0,
|
}
|
hash_map = {item["ticker"]: item for item in manifest["formal_reports"]}
|
for row in rows:
|
formal = RESULT_ROOT / row["formal_path"]
|
out_dir = formal.parent
|
snapshots = list(out_dir.glob("*估值快照_20260804.json"))
|
evidence = out_dir / "source_evidence_manifest.json"
|
results_path = out_dir / "calculation" / "valuation_results.json"
|
run_manifest = out_dir / "calculation" / "run_manifest.json"
|
for path, key in ((formal, "formal_reports"), (evidence, "source_manifests"), (results_path, "calculation_results"), (run_manifest, "run_manifests")):
|
if path.exists():
|
checks[key] += 1
|
else:
|
errors.append(f"missing {path}")
|
if len(snapshots) == 1:
|
checks["snapshots"] += 1
|
else:
|
errors.append(f"snapshot count {len(snapshots)} for {row['ticker']}")
|
continue
|
text = formal.read_text(encoding="utf-8")
|
headings = {int(value) for value in re.findall(r"^## (\d+)\.", text, flags=re.MULTILINE)}
|
if headings == set(range(16)):
|
checks["reports_with_16_sections"] += 1
|
else:
|
errors.append(f"section mismatch {row['ticker']}: {sorted(headings)}")
|
if not re.search(r"\b(?:TODO|TBD|PLACEHOLDER)\b", text, flags=re.IGNORECASE):
|
checks["reports_without_placeholders"] += 1
|
else:
|
errors.append(f"placeholder in {formal}")
|
snapshot = read_json(snapshots[0])
|
results = read_json(results_path)
|
checks["qa_error_count"] += results["qa"]["error_count"]
|
checks["qa_warning_count"] += results["qa"]["warning_count"]
|
expected_cap = float(snapshot["market"]["price"]) * float(snapshot["market"]["diluted_shares"])
|
actual_cap = float(results["metrics"]["market_cap"])
|
gap = abs(expected_cap - actual_cap) / expected_cap
|
checks["market_cap_reconciliation_max_gap"] = max(checks["market_cap_reconciliation_max_gap"], gap)
|
if gap > 1e-12:
|
errors.append(f"market cap mismatch {row['ticker']}: {gap}")
|
expected_hash = hash_map.get(row["ticker"], {}).get("sha256")
|
if expected_hash == sha256(formal):
|
checks["formal_hash_matches"] += 1
|
else:
|
errors.append(f"formal hash mismatch {row['ticker']}")
|
|
summary = BATCH_DIR / "六张图片83只股票价格合理性评估批次汇总_20260805.md"
|
summary_text = summary.read_text(encoding="utf-8")
|
for relative in re.findall(r"\]\((\.\./[^)]+)\)", summary_text):
|
target = (summary.parent / relative).resolve()
|
checks["summary_links_checked"] += 1
|
if not target.exists():
|
errors.append(f"broken summary link {relative}")
|
screenshot_checks = []
|
for item in manifest["screenshots"]:
|
path = ROOT / item["path"]
|
screenshot_checks.append({
|
"path": item["path"],
|
"exists": path.exists(),
|
"hash_matches": path.exists() and sha256(path) == item["sha256"],
|
})
|
if not path.exists() or sha256(path) != item["sha256"]:
|
errors.append(f"screenshot evidence mismatch {item['path']}")
|
expected_83 = [
|
"row_count", "unique_tickers", "formal_reports", "snapshots", "source_manifests",
|
"calculation_results", "run_manifests", "reports_with_16_sections",
|
"reports_without_placeholders", "formal_hash_matches", "summary_links_checked",
|
]
|
for key in expected_83:
|
if checks[key] != 83:
|
errors.append(f"{key}={checks[key]}, expected 83")
|
if checks["qa_error_count"] != 0:
|
errors.append(f"qa_error_count={checks['qa_error_count']}")
|
acceptance = {
|
"batch_id": batch["batch_id"],
|
"as_of": batch["as_of"],
|
"status": "PASS" if not errors else "FAIL",
|
"checks": checks,
|
"screenshot_checks": screenshot_checks,
|
"errors": errors,
|
}
|
acceptance_path = BATCH_DIR / "batch_acceptance.json"
|
acceptance_path.write_text(json.dumps(acceptance, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
manifest["acceptance"] = {
|
"path": str(acceptance_path.relative_to(ROOT)).replace("\\", "/"),
|
"sha256": sha256(acceptance_path),
|
"status": acceptance["status"],
|
}
|
(CASE_DIR / "case_manifest.json").write_text(
|
json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
)
|
print(json.dumps(acceptance, ensure_ascii=False), flush=True)
|
if errors:
|
raise SystemExit(1)
|
|
|
if __name__ == "__main__":
|
main()
|
