from __future__ import annotations import csv import hashlib import json from datetime import datetime, timedelta, timezone from pathlib import Path import pandas as pd RUN_ID = "RUN-ANA-WUJI-V1-BUY-POINT-SECOND-REVIEW-20260615-001" SOURCE_RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001" TZ = timezone(timedelta(hours=8)) ROOT = Path(__file__).resolve().parents[1] DETAIL_PATH = ROOT / "buy_point_second_review_detail.csv" P1_LEDGER_PATH = ROOT / "p1_step_review_packets" / "p1_step_review_resolution_ledger.csv" P1_SUMMARY_PATH = ROOT / "p1_step_review_packets" / "p1_step_review_resolution_summary.md" def now_iso() -> str: return datetime.now(TZ).isoformat(timespec="seconds") 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 write_csv(df: pd.DataFrame, path: Path) -> None: df.to_csv(path, index=False, encoding="utf-8-sig") def write_text(path: Path, text: str) -> None: path.write_text(text, encoding="utf-8-sig") def safe_text(value: object) -> str: if value is None: return "" text = str(value) if text.lower() == "nan": return "" return text def p1_lookup() -> dict[str, dict[str, str]]: if not P1_LEDGER_PATH.exists(): return {} p1 = pd.read_csv(P1_LEDGER_PATH, dtype=str, encoding="utf-8-sig").fillna("") return {row["candidate_id"]: row.to_dict() for _, row in p1.iterrows()} def build_rollup() -> tuple[pd.DataFrame, pd.DataFrame]: detail = pd.read_csv(DETAIL_PATH, dtype=str, encoding="utf-8-sig").fillna("") p1_by_candidate = p1_lookup() rows: list[dict[str, str]] = [] for _, row in detail.iterrows(): candidate_id = row["candidate_id"] source_action = row["human_decision_action"] second_status = row["second_review_status"] priority = row.get("human_recheck_priority", "") final_action = source_action resolution_bucket = "SOURCE_MANUAL_CARRIED_NO_DATA_RECHECK" resolution_source = "SOURCE_MANUAL_DECISION_NO_LOCAL_DATA_RECHECK" resolution_status = "NOT_INDEPENDENTLY_CONFIRMED" resolution_reason = row.get("second_review_reason_cn", "") formal_repair_action = "NO_REPAIR_DECISION_RECORDED" direct_resolution_flag = "0" codex_direct_resolution_flag = "0" if candidate_id in p1_by_candidate: p1 = p1_by_candidate[candidate_id] final_action = p1["final_action"] resolution_bucket = "USER_P1_FINAL" resolution_source = "USER_MANUAL_P1_CONFIRMATION" resolution_status = "FINAL_CONFIRMED" resolution_reason = p1["resolution_reason_cn"] direct_resolution_flag = "1" if source_action == "BUY" and final_action == "REVIEW_HELD": formal_repair_action = "REVOKE_BUY_ORDER_LOT_AND_RECALC" else: formal_repair_action = "KEEP_SOURCE_DECISION_NO_REPAIR" elif second_status == "AGREE": resolution_bucket = "CODEX_DATA_DIRECT_AGREE" resolution_source = "CODEX_DATA_SECOND_REVIEW_AGREE" resolution_status = "DATA_DIRECT_CONFIRMED" resolution_reason = row.get("second_review_reason_cn", "") formal_repair_action = "KEEP_SOURCE_DECISION_NO_REPAIR" direct_resolution_flag = "1" codex_direct_resolution_flag = "1" elif priority.startswith("P2_"): final_action = "PENDING_HUMAN_RECHECK" resolution_bucket = "P2_PENDING_HUMAN_RECHECK" resolution_source = "SECOND_REVIEW_DISPUTE" resolution_status = "PENDING_RECHECK" formal_repair_action = "WAIT_FOR_HUMAN_RECHECK" elif priority.startswith("P3_"): final_action = "PENDING_LOW_PRIORITY_SAMPLE_RECHECK" resolution_bucket = "P3_PENDING_SAMPLE_RECHECK" resolution_source = "SECOND_REVIEW_WEAK_DISPUTE" resolution_status = "PENDING_LOW_PRIORITY_RECHECK" formal_repair_action = "WAIT_FOR_SAMPLE_RECHECK" rows.append( { "case_id": row["case_id"], "candidate_id": candidate_id, "symbol": row["symbol"], "entry_trade_date": row["entry_trade_date"], "source_action": source_action, "final_action": final_action, "resolution_bucket": resolution_bucket, "resolution_source": resolution_source, "resolution_status": resolution_status, "direct_resolution_flag": direct_resolution_flag, "codex_direct_resolution_flag": codex_direct_resolution_flag, "formal_repair_action": formal_repair_action, "second_review_status": second_status, "second_review_issue_code": row["second_review_issue_code"], "human_recheck_priority": priority, "local_minute_status": row["local_minute_status"], "minute_rows": row["minute_rows"], "close_ret_pct": row["close_ret_pct"], "above_open_ratio": row["above_open_ratio"], "early_max_ret_pct": row["early_max_ret_pct"], "early_max_ret_time": row["early_max_ret_time"], "tail_max_ret_pct": row["tail_max_ret_pct"], "tail_max_ret_time": row["tail_max_ret_time"], "source_human_reason_cn": row["human_decision_reason_cn"], "resolution_reason_cn": resolution_reason, "source_review_chart_abs_path": row["source_review_chart_abs_path"], } ) rollup = pd.DataFrame(rows) p1 = pd.DataFrame(p1_by_candidate.values()) return rollup, p1 def table_counts(series: pd.Series) -> str: counts = series.value_counts(dropna=False).sort_index() lines = ["| item | count |", "|---|---:|"] for name, count in counts.items(): lines.append(f"| `{safe_text(name)}` | {int(count)} |") return "\n".join(lines) def render_summary(rollup: pd.DataFrame, p1: pd.DataFrame, generated_at: str) -> str: confirmed = rollup[rollup["direct_resolution_flag"].eq("1")] codex = rollup[rollup["codex_direct_resolution_flag"].eq("1")] repair = rollup[rollup["formal_repair_action"].eq("REVOKE_BUY_ORDER_LOT_AND_RECALC")] source_buy_confirmed = confirmed[confirmed["final_action"].eq("BUY")] source_held_confirmed = confirmed[confirmed["final_action"].eq("REVIEW_HELD")] lines = [ "# 买点裁决总览（含 Codex 数据直裁）", "", f"- run_id: {RUN_ID}", f"- source_run_id: {SOURCE_RUN_ID}", f"- updated_at: {generated_at}", f"- source_rows: {len(rollup)}", "", "## 当前结论", "", f"- 已形成明确裁决：{len(confirmed)} 条", f"- 其中用户 P1 最终裁决：{len(p1)} 条", f"- 其中 Codex 数据二审直接同意源裁决：{len(codex)} 条", f"- 已确认 BUY：{len(source_buy_confirmed)} 条", f"- 已确认 REVIEW_HELD：{len(source_held_confirmed)} 条", f"- 需要正式返修撤销 BUY/order/lot 并重算：{len(repair)} 条", f"- 仍需 P2/P3 人工复核或抽查：{len(rollup[rollup['resolution_status'].str.contains('PENDING', na=False)])} 条", f"- 无本地分钟数据覆盖、仅沿用源人工裁决且未二审确认：{len(rollup[rollup['resolution_bucket'].eq('SOURCE_MANUAL_CARRIED_NO_DATA_RECHECK')])} 条", "", "## 裁决来源分布", "", table_counts(rollup["resolution_bucket"]), "", "## 最终动作分布", "", table_counts(rollup["final_action"]), "", "## 输出文件", "", "- `buy_point_resolution_rollup.csv`: 1141 条总台账，合并用户 P1 裁决、Codex 数据直裁、待复核和无数据覆盖状态。", "- `buy_point_codex_direct_resolution.csv`: 388 条 Codex 数据二审直接同意源裁决的明细。", "- `buy_point_formal_repair_required.csv`: 12 条需要正式返修撤销 BUY/order/lot 并重算的明细。", "", "## 口径边界", "", "- `CODEX_DATA_DIRECT_AGREE` 表示本机分钟数据复算与源人工裁决一致，因此列入“数据二审直接同意源裁决”；它不是新的买卖规则，也不替代后续正式结果包返修。", "- `SOURCE_MANUAL_CARRIED_NO_DATA_RECHECK` 表示本机分钟数据没有覆盖，当前没有形成二审证据，不把它算作 Codex 直接裁决。", "- 正式结果包返修时，不能只改人工裁决表；凡撤销 BUY 的条目，都需要同步重算 order、lot、case 汇总和收益口径。", ] return "\n".join(lines) + "\n" def render_p1_summary(p1: pd.DataFrame, generated_at: str) -> str: if p1.empty: return "# P1 买点逐条人工复核裁决汇总\n\n- 当前没有 P1 裁决记录。\n" keep = p1[p1["final_action"].eq("BUY")] revoke = p1[p1["final_action"].eq("REVIEW_HELD")] lines = [ "# P1 买点逐条人工复核裁决汇总", "", f"- updated_at: {generated_at}", f"- resolved_count: {len(p1)}", f"- 维持 BUY: {len(keep)}", f"- 改为 REVIEW_HELD: {len(revoke)}", "", "| # | case | candidate | symbol | date | original | final | reason |", "|---:|---|---|---|---|---|---|---|", ] ordered = p1.copy() ordered["packet_order_num"] = pd.to_numeric(ordered["packet_order"], errors="coerce") ordered = ordered.sort_values("packet_order_num") for _, row in ordered.iterrows(): lines.append( "| {packet_order} | {case_id} | {candidate_id} | {symbol} | {date} | {original} | {final} | {reason} |".format( packet_order=row["packet_order"], case_id=row["case_id"], candidate_id=row["candidate_id"], symbol=row["symbol"], date=row["entry_trade_date"], original=row["original_action"], final=row["final_action"], reason=row["resolution_reason_cn"], ) ) lines.extend( [ "", "## 待处理影响", "", "- 维持 BUY 的条目保留最新结果包中的 BUY/order/lot，不进入买点撤销重算清单。", "- 改为 REVIEW_HELD 的条目需要在正式返修时撤销对应 BUY/order/lot；如果已经有后续 SELL，也要一并重算 lot、case 汇总和收益口径。", "- `WUJI-STRICT-20250317` 同时有 `920570.BJ` 和 `603206.SH` 两条买点被判定不适合作为买点，正式返修时应作为同一个 case 合并处理。", ] ) return "\n".join(lines) + "\n" def update_manifest(generated_at: str, extra_paths: list[Path]) -> None: manifest_paths = [ ROOT / "buy_point_second_review_action_status_counts.csv", ROOT / "buy_point_second_review_detail.csv", ROOT / "buy_point_second_review_human_recheck_report.md", ROOT / "buy_point_second_review_human_top80.csv", ROOT / "buy_point_second_review_recheck_list.csv", ROOT / "buy_point_second_review_strong_recheck.csv", ROOT / "buy_point_second_review_summary.json", ROOT / "buy_point_second_review_summary.md", ROOT / "tools" / "review_buy_point_decisions.py", ROOT / "tools" / "build_p1_step_review_packets.py", ROOT / "tools" / "build_buy_point_resolution_rollup.py", *extra_paths, ] files = [] for path in manifest_paths: if not path.exists(): continue rel = path.relative_to(ROOT).as_posix() files.append({"path": rel, "size": path.stat().st_size, "sha256": sha256_file(path)}) with (ROOT / "manifest.csv").open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=["path", "size", "sha256"]) writer.writeheader() writer.writerows(files) manifest = { "run_id": RUN_ID, "updated_at": generated_at, "file_count": len(files), "files": files, } (ROOT / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") def main() -> None: generated_at = now_iso() rollup, p1 = build_rollup() rollup_path = ROOT / "buy_point_resolution_rollup.csv" codex_path = ROOT / "buy_point_codex_direct_resolution.csv" repair_path = ROOT / "buy_point_formal_repair_required.csv" summary_path = ROOT / "buy_point_resolution_summary.md" write_csv(rollup, rollup_path) write_csv(rollup[rollup["codex_direct_resolution_flag"].eq("1")], codex_path) write_csv(rollup[rollup["formal_repair_action"].eq("REVOKE_BUY_ORDER_LOT_AND_RECALC")], repair_path) write_text(summary_path, render_summary(rollup, p1, generated_at)) write_text(P1_SUMMARY_PATH, render_p1_summary(p1, generated_at)) update_manifest(generated_at, [rollup_path, codex_path, repair_path, summary_path]) if __name__ == "__main__": main()