from __future__ import annotations import csv import hashlib import json from datetime import datetime, timezone, timedelta from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[4] RUN_ID = "RUN-ANA-WUJI-V1-FINAL-CONCLUSION-20260610-001" TASK_ID = "ANA-WUJI-V1-FINAL-CONCLUSION-20260610" SOURCE_RUN_ID = "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001" SOURCE_DIR = PROJECT_ROOT / "ana-data" / "result" / SOURCE_RUN_ID OUT_DIR = PROJECT_ROOT / "ana-data" / "result" / RUN_ID SOURCE_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-EXEC-REREVIEW-003" DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-V1-FINAL-CONCLUSION-20260610-DESIGN-001" def now_iso() -> str: return datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds") 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, object]], fieldnames: list[str] | None = None) -> None: path.parent.mkdir(parents=True, exist_ok=True) if fieldnames is None: fieldnames = list(rows[0].keys()) if rows else [] with path.open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for row in rows: writer.writerow(row) def write_json(path: Path, data: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def sha256(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 file_row(path: Path, root: Path = PROJECT_ROOT) -> dict[str, object]: return { "path": path.relative_to(root).as_posix(), "size": path.stat().st_size, "sha256": sha256(path), } def source_manifest(paths: list[Path]) -> list[dict[str, object]]: rows = [] for p in paths: row = file_row(p) row["source_run_id"] = SOURCE_RUN_ID rows.append(row) return rows def copy_boundary(source_rows: list[dict[str, str]]) -> list[dict[str, object]]: rows = [] for r in source_rows: rows.append( { "boundary_id": r.get("boundary_id", ""), "boundary_level": r.get("boundary_level", ""), "case_id": r.get("case_id", ""), "source_lot_id": r.get("source_lot_id", ""), "symbol": r.get("symbol", ""), "boundary_category": r.get("boundary_category", ""), "boundary_reason": r.get("boundary_reason", ""), "source_table": "strict_boundary_table.csv", "readout_policy": "boundary_only_not_primary_return", } ) return rows def forbidden_claims(text: str) -> list[str]: claims = [ "完整 baseline 成功率为", "完整 baseline 收益率为", "无边界 baseline 成功率", "无边界 baseline 收益率", "策略有效性已证明", "策略有效性已经证明", ] return [c for c in claims if c in text] def main() -> None: OUT_DIR.mkdir(parents=True, exist_ok=True) source_paths = [ SOURCE_DIR / "summary.json", SOURCE_DIR / "summary.md", SOURCE_DIR / "strict_case_summary.csv", SOURCE_DIR / "strict_return_scope_case.csv", SOURCE_DIR / "strict_return_scope_lot.csv", SOURCE_DIR / "strict_boundary_table.csv", SOURCE_DIR / "manual_decision_external_draft.md", SOURCE_DIR / "manual_decision_external_source_ledger.csv", SOURCE_DIR / "manual_decision_ledger.csv", SOURCE_DIR / "strict_order_ledger.csv", SOURCE_DIR / "strict_position_lot_ledger.csv", SOURCE_DIR / "case_image_board.md", SOURCE_DIR / "manifest.json", ] missing_sources = [p for p in source_paths if not p.exists()] if missing_sources: raise FileNotFoundError("Missing source files: " + ", ".join(str(p) for p in missing_sources)) summary = json.loads((SOURCE_DIR / "summary.json").read_text(encoding="utf-8")) case_rows = read_csv(SOURCE_DIR / "strict_case_summary.csv") lot_rows = read_csv(SOURCE_DIR / "strict_return_scope_lot.csv") boundary_rows = read_csv(SOURCE_DIR / "strict_boundary_table.csv") order_rows = read_csv(SOURCE_DIR / "strict_order_ledger.csv") primary_cases = [r for r in case_rows if r.get("v1_primary_strict_closed_case_flag") == "1"] positive_cases = [r for r in primary_cases if r.get("v1_case_success_flag") == "1"] boundary_cases = [r for r in case_rows if r.get("v1_return_scope") != "V1_PRIMARY_STRICT_CLOSED_CASE"] closed_lots = [r for r in lot_rows if r.get("v1_lot_scope") == "V1_STRICT_CLOSED_LOT_RECALC_ONLY"] boundary_lots = [r for r in lot_rows if r.get("v1_lot_scope") != "V1_STRICT_CLOSED_LOT_RECALC_ONLY"] rolling_buy_orders = [ r for r in order_rows if r.get("action") == "BUY" and r.get("tranche_index") == "2" ] account_contribution = round(sum(float(r.get("account_return_closed_lots") or 0) for r in primary_cases), 8) success_rate = round(len(positive_cases) / len(primary_cases), 10) if primary_cases else 0 readouts = [ { "readout_id": "V1_PRIMARY_STRICT_CLOSED_CASE", "scope": "V1 主口径严格闭合 case", "case_count": len(primary_cases), "positive_case_count": len(positive_cases), "non_positive_case_count": len(primary_cases) - len(positive_cases), "success_rate": f"{success_rate:.10f}".rstrip("0").rstrip("."), "account_contribution_sum": f"{account_contribution:.8f}", "lot_count": "", "boundary_count": "", "source": "strict_case_summary.csv", "citation_boundary": "必须同时引用 V1 口径、执行复审审计 ID、v1_boundary_table.csv 和 MARKET_RISK_MINUTE_BREADTH_DATA_GAP_HELD。", }, { "readout_id": "V1_ALL_BUY_CASE_COVERAGE", "scope": "V1 有 BUY case 覆盖口径", "case_count": len(case_rows), "positive_case_count": "", "non_positive_case_count": "", "success_rate": "", "account_contribution_sum": "", "lot_count": "", "boundary_count": len(boundary_cases), "source": "strict_case_summary.csv", "citation_boundary": "只说明 V1 已买入 case 覆盖和边界，不替代主成功率分母。", }, { "readout_id": "V1_STRICT_CLOSED_LOT_RECALC_ONLY", "scope": "V1 lot 复算辅助口径", "case_count": "", "positive_case_count": "", "non_positive_case_count": "", "success_rate": "", "account_contribution_sum": "", "lot_count": len(lot_rows), "boundary_count": len(boundary_lots), "source": "strict_return_scope_lot.csv", "citation_boundary": "只用于 lot 复算和问题定位，不包装成 case 成功率。", }, { "readout_id": "V1_ROLLING_LOW_BUY", "scope": "滚动低吸 BUY 订单", "case_count": "", "positive_case_count": "", "non_positive_case_count": "", "success_rate": "", "account_contribution_sum": "", "lot_count": len(rolling_buy_orders), "boundary_count": "", "source": "strict_order_ledger.csv", "citation_boundary": "只说明滚动低吸执行数量，不单独构成策略有效性结论。", }, ] write_csv(OUT_DIR / "v1_final_readouts.csv", readouts) case_index = [] for r in case_rows: case_id = r.get("case_id", "") case_index.append( { "case_id": case_id, "v1_return_scope": r.get("v1_return_scope", ""), "primary_flag": r.get("v1_primary_strict_closed_case_flag", ""), "success_flag": r.get("v1_case_success_flag", ""), "buy_lot_count": r.get("buy_lot_count", ""), "closed_lot_count": r.get("closed_lot_count", ""), "unresolved_lot_count": r.get("unresolved_lot_count", ""), "account_return_closed_lots": r.get("account_return_closed_lots", ""), "boundary_reason": r.get("v1_boundary_reason", ""), "case_image_board": f"../{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md", "case_story_board": f"../{SOURCE_RUN_ID}/cases/{case_id}/case_story_board.md", } ) write_csv(OUT_DIR / "v1_case_readout_index.csv", case_index) v1_boundary = copy_boundary(boundary_rows) write_csv(OUT_DIR / "v1_boundary_table.csv", v1_boundary) src_manifest = source_manifest(source_paths) write_csv(OUT_DIR / "source_artifact_manifest.csv", src_manifest) top_positive = sorted(primary_cases, key=lambda r: float(r.get("account_return_closed_lots") or 0), reverse=True)[:3] top_negative = sorted(primary_cases, key=lambda r: float(r.get("account_return_closed_lots") or 0))[:3] boundary_example = boundary_cases[:1] def case_link(r: dict[str, str]) -> str: case_id = r.get("case_id", "") ret = r.get("account_return_closed_lots", "") return f"- `{case_id}`，收益贡献 `{ret}`，[图片板](../{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md)，[故事板](../{SOURCE_RUN_ID}/cases/{case_id}/case_story_board.md)" human_review = f"""# 无忌 V1 最终结论人工审核第一入口 ## 先看这里 可以引用：当前 V1 执行包已通过执行复审。允许按 `V1_PRIMARY_STRICT_CLOSED_CASE` 主口径引用 250 个严格闭合 case 的读数：99 个为正收益，成功率读数为 0.396，账户贡献合计读数为 0.12970642。 必须同时引用：V1 口径、执行复审审计 ID `{SOURCE_AUDIT_ID}`、`v1_boundary_table.csv`，以及 `MARKET_RISK_MINUTE_BREADTH_DATA_GAP_HELD`。 不能引用：不能把 V1 读数写成旧 V0 包结论，不能写成无边界完整 baseline 结论，不能作出无忌策略有效性的证明结论，不能省略市场风险分钟广度数据缺口。 当前状态： | 项目 | 内容 | |---|---| | 当前 run | `{RUN_ID}` | | 来源 run | `{SOURCE_RUN_ID}` | | 来源执行复审审计 ID | `{SOURCE_AUDIT_ID}` | | 当前设计审核审计 ID | `{DESIGN_AUDIT_ID}` | | RETURN_STAT_READY | `false` | | 执行审核状态 | `待提交执行审核` | ## V1 三层读数 1. `V1_PRIMARY_STRICT_CLOSED_CASE`：250 个严格闭合 case，正收益 99 个，成功率读数 0.396，账户贡献合计 0.12970642。 2. `V1_ALL_BUY_CASE_COVERAGE`：251 个有 BUY case，其中 1 个 case 进入边界表；该口径不替代主成功率。 3. `V1_STRICT_CLOSED_LOT_RECALC_ONLY`：735 个 lot，其中 734 个闭合、1 个边界；lot 口径只用于复算和问题定位。 ## 第一入口链接 - 来源 V1 图片总入口：[case_image_board.md](../{SOURCE_RUN_ID}/case_image_board.md) - 来源 V1 摘要：[summary.md](../{SOURCE_RUN_ID}/summary.md) - 来源 V1 手工裁决草稿：[manual_decision_external_draft.md](../{SOURCE_RUN_ID}/manual_decision_external_draft.md) - 当前 V1 结论摘要：[v1_final_conclusion_summary.md](v1_final_conclusion_summary.md) - 当前 V1 读数表：[v1_final_readouts.csv](v1_final_readouts.csv) - 当前 V1 case 索引：[v1_case_readout_index.csv](v1_case_readout_index.csv) - 当前 V1 边界表：[v1_boundary_table.csv](v1_boundary_table.csv) ## 代表性 case 主口径正收益较高样例： {chr(10).join(case_link(r) for r in top_positive)} 主口径负收益较低样例： {chr(10).join(case_link(r) for r in top_negative)} 边界样例： {chr(10).join(case_link(r) for r in boundary_example)} ## 阅读提示 1. 先看本文件的可引用 / 不可引用边界。 2. 再看 `v1_final_conclusion_summary.md` 了解整体读数。 3. 需要追溯单个案例时，从 `v1_case_readout_index.csv` 找到 `case_id`，再打开对应图片板和故事板。 4. 对卖点和滚动低吸的人工裁决来源，优先看 `manual_decision_external_draft.md` 和 `manual_decision_ledger.csv`。源单 case 图板中的图片是主要证据入口；文字理由以本包摘要、手工裁决草稿和账本字段为准。 """ (OUT_DIR / "v1_final_human_review_index.md").write_text(human_review, encoding="utf-8") summary_md = f"""# 无忌 V1 最终结论引用包摘要 ## 当前可引用状态 当前包基于已通过执行复审的 V1 结果包生成，来源审计 ID 为 `{SOURCE_AUDIT_ID}`。本包仍需执行审核；执行审核通过前，不得把 V1 读数写入正式最终案例总结。 ## V1 主口径：V1_PRIMARY_STRICT_CLOSED_CASE | 指标 | 读数 | |---|---:| | V1 主口径严格闭合 case | {len(primary_cases)} | | V1 主口径正收益 case | {len(positive_cases)} | | V1 主口径非正收益 case | {len(primary_cases) - len(positive_cases)} | | V1 主口径成功率读数 | {success_rate:.10f} | | V1 主口径账户贡献合计读数 | {account_contribution:.8f} | ## 覆盖口径 | 指标 | 读数 | |---|---:| | V1 有 BUY case | {len(case_rows)} | | V1 边界 case | {len(boundary_cases)} | | 滚动低吸 BUY 订单 | {len(rolling_buy_orders)} | ## lot 辅助口径 | 指标 | 读数 | |---|---:| | V1 lot 总数 | {len(lot_rows)} | | V1 闭合 lot | {len(closed_lots)} | | V1 边界 lot | {len(boundary_lots)} | ## 边界 `v1_boundary_table.csv` 完整保留来源边界表，共 {len(v1_boundary)} 行。必须特别保留 `MARKET_RISK_MINUTE_BREADTH_DATA_GAP_HELD`，它表示本地数据源没有开盘 10 分钟全 A 下跌家数分钟级广度，市场风险卖点不能被无边界化。 ## 禁止外推 1. 不得把 V1 读数写成旧 V0 包结论。 2. 不得写成无边界完整 baseline 成功率、收益率、胜率或回撤。 3. 不得作出无忌策略有效性的证明结论。 4. 不得省略审计 ID、V1 口径、边界表和市场风险分钟广度数据缺口。 """ (OUT_DIR / "v1_final_conclusion_summary.md").write_text(summary_md, encoding="utf-8") summary_json = { "schema_version": "1.0", "task_id": TASK_ID, "run_id": RUN_ID, "source_run_id": SOURCE_RUN_ID, "source_execution_rereview_audit_id": SOURCE_AUDIT_ID, "design_audit_id": DESIGN_AUDIT_ID, "generated_at": now_iso(), "execution_review_status": "PENDING_EXECUTION_REVIEW", "return_stat_ready": False, "readouts": { "v1_buy_case_count": len(case_rows), "v1_primary_strict_closed_cases": len(primary_cases), "v1_positive_primary_cases": len(positive_cases), "v1_primary_success_readout": success_rate, "v1_primary_account_contribution_readout": account_contribution, "v1_lot_total": len(lot_rows), "v1_closed_lots": len(closed_lots), "v1_boundary_lots": len(boundary_lots), "rolling_low_buy_orders": len(rolling_buy_orders), "boundary_rows": len(v1_boundary), }, "required_citation_context": [ "V1 scope", SOURCE_AUDIT_ID, "v1_boundary_table.csv", "MARKET_RISK_MINUTE_BREADTH_DATA_GAP_HELD", ], "forbidden": [ "old V0 package as complete note baseline", "no-boundary complete baseline conclusion", "strategy validity proven", ], } write_json(OUT_DIR / "v1_final_conclusion_summary.json", summary_json) (OUT_DIR / "README.md").write_text( f"""# 无忌 V1 最终结论引用包 本包是 V1 执行复审通过后的引用层整理，不重跑候选池、买卖裁决、人工裁决或账本。 先看：[v1_final_human_review_index.md](v1_final_human_review_index.md) 来源审计 ID：`{SOURCE_AUDIT_ID}` 设计审核 ID：`{DESIGN_AUDIT_ID}` 当前状态：待执行审核。执行审核通过前，不得把 V1 读数写入正式最终案例总结。 """, encoding="utf-8", ) generated_text = "\n".join( [ human_review, summary_md, json.dumps(summary_json, ensure_ascii=False), ] ) forbidden_hits = forbidden_claims(generated_text) checks: list[dict[str, object]] = [] def check(name: str, passed: bool, detail: str) -> None: checks.append({"item": name, "status": "PASS" if passed else "FAIL", "detail": detail}) check( "SOURCE_EXECUTION_REVIEW_PASSED", bool(summary.get("v1_execution_review_passed")) and summary.get("execution_rereview_passed_audit_id") == SOURCE_AUDIT_ID, f"audit={summary.get('execution_rereview_passed_audit_id')}", ) check("SOURCE_ARTIFACTS_HASHED", all(Path(row["path"]).exists() for row in src_manifest), f"sources={len(src_manifest)}") check( "READOUTS_MATCH_SOURCE", len(primary_cases) == int(summary["scope"]["v1_primary_strict_closed_cases"]) and len(positive_cases) == int(summary["scope"]["v1_positive_primary_cases"]) and round(float(summary["scope"]["v1_primary_success_readout"]), 10) == success_rate and round(float(summary["scope"]["v1_primary_account_contribution_readout"]), 8) == account_contribution, f"primary={len(primary_cases)}, positive={len(positive_cases)}, success={success_rate}, contribution={account_contribution}", ) check("BOUNDARY_TABLE_PRESERVED", len(v1_boundary) == len(boundary_rows), f"boundary_rows={len(v1_boundary)}") check( "MARKET_RISK_BOUNDARY_PRESENT", any(r.get("boundary_category") == "MARKET_RISK_MINUTE_BREADTH_DATA_GAP_HELD" for r in boundary_rows), "MARKET_RISK_MINUTE_BREADTH_DATA_GAP_HELD", ) check( "SCOPE_SEPARATED", len(primary_cases) + len(boundary_cases) == len(case_rows) and len(closed_lots) + len(boundary_lots) == len(lot_rows), f"cases={len(case_rows)}, primary={len(primary_cases)}, boundary={len(boundary_cases)}, lots={len(lot_rows)}", ) check( "ROLLING_LOW_BUY_READOUT_MATCHES_ORDER_LEDGER", len(rolling_buy_orders) == int(summary["scope"].get("rolling_buy_lots", len(rolling_buy_orders))), f"readout={len(rolling_buy_orders)}, source_scope={summary['scope'].get('rolling_buy_lots')}", ) link_targets = [ SOURCE_DIR / "case_image_board.md", SOURCE_DIR / "manual_decision_external_draft.md", SOURCE_DIR / "strict_order_ledger.csv", SOURCE_DIR / "strict_position_lot_ledger.csv", ] check("HUMAN_REVIEW_ENTRY_LINKS_REACHABLE", all(p.exists() for p in link_targets), f"links={len(link_targets)}") check("NO_OVERREAD_ASSERTIONS", not forbidden_hits, "forbidden_hits=" + "|".join(forbidden_hits)) write_csv(OUT_DIR / "self_check_items.csv", checks, ["item", "status", "detail"]) fail_count = sum(1 for r in checks if r["status"] != "PASS") self_check = { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": now_iso(), "status": "PASS_FOR_V1_FINAL_CONCLUSION_EXECUTION_REVIEW_READY" if fail_count == 0 else "FAIL", "pass_count": len(checks) - fail_count, "fail_count": fail_count, "items_path": "self_check_items.csv", } write_json(OUT_DIR / "self_check.json", self_check) (OUT_DIR / "self_check.md").write_text( f"""# 自检摘要 - status：`{self_check['status']}` - PASS：{self_check['pass_count']} - FAIL：{self_check['fail_count']} - 检查项：`self_check_items.csv` """, encoding="utf-8", ) # Build final package manifest last. package_files = [] for p in OUT_DIR.rglob("*"): if not p.is_file(): continue if p.name in {"manifest.csv", "manifest.json"}: continue package_files.append(file_row(p, OUT_DIR)) package_files.sort(key=lambda r: str(r["path"])) write_csv(OUT_DIR / "manifest.csv", package_files, ["path", "size", "sha256"]) write_json( OUT_DIR / "manifest.json", { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": now_iso(), "files": package_files, }, ) print( json.dumps( { "run_id": RUN_ID, "status": self_check["status"], "primary_cases": len(primary_cases), "positive_cases": len(positive_cases), "success_rate": success_rate, "manifest_files": len(package_files), }, ensure_ascii=False, ) ) if __name__ == "__main__": main()