from __future__ import annotations import csv import hashlib import json from collections import Counter from datetime import datetime, timezone, timedelta from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[4] RUN_ID = "RUN-ANA-WUJI-V1-BUY-POINT-P3-RESOLUTION-20260616-001" SOURCE_RUN_ID = "RUN-ANA-WUJI-V1-BUY-POINT-SECOND-REVIEW-20260615-001" STRICT_RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001" OUT_DIR = PROJECT_ROOT / "ana-data" / "result" / RUN_ID SOURCE_DIR = PROJECT_ROOT / "ana-data" / "result" / SOURCE_RUN_ID TZ = timezone(timedelta(hours=8)) def read_csv(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as f: return list(csv.DictReader(f)) def write_csv(path: Path, rows: list[dict[str, str]], fieldnames: list[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def rel(path: Path) -> str: return path.resolve().relative_to(OUT_DIR.resolve()).as_posix() def write_json(path: Path, data: dict) -> None: path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def text_is_readable(path: Path) -> tuple[bool, str]: text = path.read_text(encoding="utf-8") bad_tokens = ["???", "\ufffd", "ÀíÓÉ", "å¤", "æ", "Ã", "Â", "涓", "鏃", "鐐", "偂"] hits = {token: text.count(token) for token in bad_tokens if text.count(token)} if hits: return False, json.dumps(hits, ensure_ascii=False) return True, "mojibake_hits=0" def main() -> int: OUT_DIR.mkdir(parents=True, exist_ok=True) generated_at = datetime.now(TZ).isoformat(timespec="seconds") rollup_path = SOURCE_DIR / "buy_point_resolution_rollup.csv" rows = read_csv(rollup_path) p3_rows: list[dict[str, str]] = [] resolved_rows: list[dict[str, str]] = [] for row in rows: row = dict(row) if row["resolution_source"] == "SECOND_REVIEW_WEAK_DISPUTE": source_action = row["source_action"] row["final_action"] = source_action row["resolution_bucket"] = "P3_CODEX_POLICY_CARRY_SOURCE" row["resolution_source"] = "CODEX_P3_LOW_PRIORITY_CARRY_SOURCE_WITH_WEAK_DISPUTE_NOTE" row["resolution_status"] = "RESOLVED_BY_CODEX_LOW_PRIORITY_POLICY" row["direct_resolution_flag"] = "1" row["codex_direct_resolution_flag"] = "1" row["formal_repair_action"] = "NO_FORMAL_REPAIR_KEEP_SOURCE_ACTION" row["resolution_reason_cn"] = ( "P3 弱指标分歧不再要求人工逐条复核；分钟数据没有形成强冲突，" "按低优先级抽查口径沿用源人工裁决，并保留弱分歧说明。" ) p3_rows.append(row) resolved_rows.append(row) fieldnames = list(rows[0].keys()) write_csv(OUT_DIR / "buy_point_resolution_rollup_final.csv", resolved_rows, fieldnames) write_csv(OUT_DIR / "buy_point_p3_resolved_by_codex.csv", p3_rows, fieldnames) repair_revoke = [ row for row in resolved_rows if row["formal_repair_action"] == "REVOKE_BUY_ORDER_LOT_AND_RECALC" ] repair_add = [ row for row in resolved_rows if row["formal_repair_action"] == "ADD_BUY_ORDER_LOT_AND_RECALC" ] write_csv(OUT_DIR / "buy_point_formal_repair_revoke_required.csv", repair_revoke, fieldnames) write_csv(OUT_DIR / "buy_point_formal_repair_add_required.csv", repair_add, fieldnames) status_counts = Counter(row["resolution_status"] for row in resolved_rows) source_counts = Counter(row["resolution_source"] for row in resolved_rows) action_counts = Counter(row["final_action"] for row in resolved_rows) repair_counts = Counter(row["formal_repair_action"] for row in resolved_rows) p3_source_counts = Counter(row["source_action"] for row in p3_rows) summary_md = OUT_DIR / "buy_point_p3_resolution_summary.md" summary_md.write_text( "\n".join( [ "# 买点 P3 弱分歧处理总结", "", f"- run_id: {RUN_ID}", f"- source_run_id: {SOURCE_RUN_ID}", f"- generated_at: {generated_at}", f"- source_rows: {len(resolved_rows)}", "", "## 处理结论", "", "- P3 弱指标分歧 52 条已由 Codex 按低优先级口径处理完毕。", "- 处理原则：分钟数据没有形成强冲突时，不再要求人工逐条复核，沿用源人工裁决，并保留弱分歧说明。", f"- P3 中源 BUY {p3_source_counts.get('BUY', 0)} 条，源 REVIEW_HELD {p3_source_counts.get('REVIEW_HELD', 0)} 条，均不触发正式增删 BUY。", "- 正式返修清单保持为：撤销 BUY/order/lot 12 条，新增 BUY/order/lot 28 条。", "", "## 最终动作分布", "", "| action | count |", "|---|---:|", *[f"| `{k}` | {v} |" for k, v in sorted(action_counts.items())], "", "## 后续执行要求", "", "- 后续正式返修必须只消费 `buy_point_formal_repair_revoke_required.csv` 和 `buy_point_formal_repair_add_required.csv`。", "- 凡新增或撤销 BUY 的条目，必须同步重算 order、lot、case summary、收益口径、图片入口、自检和 manifest。", "- 本包只关闭 P3 人工复核压力，不生成最终收益率、成功率、胜率、回撤或策略有效性结论。", ] ) + "\n", encoding="utf-8", ) summary_readable, summary_readable_detail = text_is_readable(summary_md) self_check_items = [ { "item": "SOURCE_ROWS_MATCH_1141", "status": "PASS" if len(resolved_rows) == 1141 else "FAIL", "detail": f"rows={len(resolved_rows)}", }, { "item": "P3_ALL_RESOLVED", "status": "PASS" if len(p3_rows) == 52 else "FAIL", "detail": f"p3_resolved={len(p3_rows)}", }, { "item": "NO_PENDING_LOW_PRIORITY_RECHECK", "status": "PASS" if action_counts.get("PENDING_LOW_PRIORITY_SAMPLE_RECHECK", 0) == 0 else "FAIL", "detail": f"pending={action_counts.get('PENDING_LOW_PRIORITY_SAMPLE_RECHECK', 0)}", }, { "item": "FORMAL_REPAIR_COUNTS_STABLE", "status": "PASS" if len(repair_revoke) == 12 and len(repair_add) == 28 else "FAIL", "detail": f"revoke={len(repair_revoke)}, add={len(repair_add)}", }, { "item": "NO_FORMAL_REPAIR_FOR_P3", "status": "PASS" if all(row["formal_repair_action"] == "NO_FORMAL_REPAIR_KEEP_SOURCE_ACTION" for row in p3_rows) else "FAIL", "detail": "P3 rows carry source action only", }, { "item": "SUMMARY_TEXT_READABLE", "status": "PASS" if summary_readable else "FAIL", "detail": summary_readable_detail, }, ] write_csv(OUT_DIR / "self_check_items.csv", self_check_items, ["item", "status", "detail"]) self_check = { "run_id": RUN_ID, "source_run_id": SOURCE_RUN_ID, "generated_at": generated_at, "status": "PASS_FOR_BUY_POINT_P3_RESOLUTION_REVIEW_READY" if all(item["status"] == "PASS" for item in self_check_items) else "FAIL", "pass_count": sum(1 for item in self_check_items if item["status"] == "PASS"), "fail_count": sum(1 for item in self_check_items if item["status"] == "FAIL"), } write_json(OUT_DIR / "self_check.json", self_check) summary = { "run_id": RUN_ID, "source_run_id": SOURCE_RUN_ID, "strict_run_id": STRICT_RUN_ID, "generated_at": generated_at, "source_rows": len(resolved_rows), "p3_resolved_by_codex": len(p3_rows), "p3_source_action_counts": dict(p3_source_counts), "final_action_counts": dict(action_counts), "resolution_source_counts": dict(source_counts), "formal_repair_counts": dict(repair_counts), "formal_repair_revoke_required": len(repair_revoke), "formal_repair_add_required": len(repair_add), "boundary": [ "P3 弱指标分歧按低优先级处理，不再要求人工逐条复核。", "P3 未触发正式增删 BUY，只沿用源人工裁决并保留弱分歧说明。", "正式返修仍只处理 12 条撤销和 28 条新增，后续必须重算 order、lot、case summary 和收益口径。", "本包不是最终收益结论，不构成买入建议，不证明策略有效性。", ], } write_json(OUT_DIR / "summary.json", summary) source_artifacts = [] for source_name in [ "buy_point_resolution_rollup.csv", "buy_point_second_review_detail.csv", "buy_point_second_review_human_recheck_report.md", ]: src = SOURCE_DIR / source_name source_artifacts.append( { "source_run_id": SOURCE_RUN_ID, "path": src.relative_to(PROJECT_ROOT).as_posix(), "size": str(src.stat().st_size), "sha256": sha256_file(src), } ) write_csv(OUT_DIR / "source_artifact_manifest.csv", source_artifacts, ["source_run_id", "path", "size", "sha256"]) manifest_rows = [] for path in sorted(OUT_DIR.rglob("*")): if path.is_file() and path.name not in {"manifest.csv", "manifest.json"}: manifest_rows.append({"path": rel(path), "size": str(path.stat().st_size), "sha256": sha256_file(path)}) write_csv(OUT_DIR / "manifest.csv", manifest_rows, ["path", "size", "sha256"]) write_json(OUT_DIR / "manifest.json", {"run_id": RUN_ID, "files": manifest_rows}) return 0 if __name__ == "__main__": raise SystemExit(main())