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-STRICT-CLOSED-234-REVIEW-GUIDE-20260608-001" TASK_ID = "ANA-WUJI-STRICT-CLOSED-234-REVIEW-GUIDE-20260608" SOURCE_RUN_ID = "RUN-ANA-WUJI-FULL-2023-2026-20260608-001" SOURCE_AUDIT_ID = "AUDIT-ANA-WUJI-FULL-2023-2026-20260608-EXEC-REREVIEW-002" FINAL_AUDIT_ID = "AUDIT-ANA-WUJI-FINAL-CONCLUSION-20260608-EXEC-001" PACKAGE_DIR = PROJECT_ROOT / "ana-data" / "result" / RUN_ID SOURCE_DIR = PROJECT_ROOT / "ana-data" / "result" / SOURCE_RUN_ID TZ = timezone(timedelta(hours=8)) def rel(path: Path) -> str: return path.resolve().relative_to(PROJECT_ROOT).as_posix() 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: with path.open("w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, lineterminator="\n") writer.writeheader() for row in rows: writer.writerow({k: row.get(k, "") for k in fieldnames}) 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 read_json(path: Path) -> dict: with path.open("r", encoding="utf-8") as f: return json.load(f) def write_json(path: Path, data: dict | list) -> None: path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def md_link(path: Path, label: str | None = None) -> str: target = rel(path) return f"[{label or target}](../../../{target})" def build() -> None: generated_at = datetime.now(TZ).isoformat(timespec="seconds") PACKAGE_DIR.mkdir(parents=True, exist_ok=True) case_scope = read_csv(SOURCE_DIR / "full_return_stat_case_scope.csv") case_summary = {r["case_id"]: r for r in read_csv(SOURCE_DIR / "case_summary.csv")} selected = read_csv(SOURCE_DIR / "full_selected_candidate_ledger.csv") lots = read_csv(SOURCE_DIR / "full_return_stat_lot_scope.csv") orders = read_csv(SOURCE_DIR / "order_ledger.csv") summary = read_json(SOURCE_DIR / "full_return_stat_summary.json") selected_by_case: dict[str, list[dict[str, str]]] = {} for row in selected: selected_by_case.setdefault(row["case_id"], []).append(row) lot_counts: dict[str, dict[str, int]] = {} for row in lots: case_id = row["case_id"] item = lot_counts.setdefault(case_id, {"lots": 0, "closed": 0, "boundary": 0}) item["lots"] += 1 if row.get("lot_scope") == "STRICT_CLOSED_LOT_RECALC_ONLY": item["closed"] += 1 else: item["boundary"] += 1 order_counts: dict[str, dict[str, int]] = {} for row in orders: case_id = row["case_id"] item = order_counts.setdefault(case_id, {"orders": 0, "buy": 0, "sell": 0}) item["orders"] += 1 action = row.get("action", "") if action == "BUY": item["buy"] += 1 elif action == "SELL": item["sell"] += 1 strict_cases = [ row for row in case_scope if row.get("case_scope_status") == "PRIMARY_STRICT_CLOSED_CASE" and row.get("primary_strict_closed_case_flag") == "1" ] strict_cases.sort(key=lambda r: (r["entry_trade_date"], r["case_id"])) index_rows: list[dict[str, object]] = [] file_rows: list[dict[str, object]] = [] for row in strict_cases: case_id = row["case_id"] case_dir = SOURCE_DIR / "cases" / case_id cs = case_summary.get(case_id, {}) candidates = selected_by_case.get(case_id, []) symbols = ";".join(c.get("symbol", "") for c in candidates[:5]) candidate_ids = ";".join(c.get("candidate_id", "") for c in candidates[:5]) lc = lot_counts.get(case_id, {"lots": 0, "closed": 0, "boundary": 0}) oc = order_counts.get(case_id, {"orders": 0, "buy": 0, "sell": 0}) account_return = row.get("account_return_closed_lots") or cs.get("account_return_closed_lots", "") success_flag = row.get("primary_case_success_flag", "") index_rows.append( { "case_id": case_id, "batch_id": row.get("batch_id", ""), "entry_trade_date": row.get("entry_trade_date", ""), "signal_trade_date": row.get("signal_trade_date", ""), "market_gate_status": row.get("market_gate_status", ""), "primary_scope": row.get("case_scope_status", ""), "success_flag": success_flag, "account_return_closed_lots": account_return, "buy_lot_count": row.get("buy_lot_count", ""), "closed_lot_count": row.get("closed_lot_count", ""), "order_buy_count": oc["buy"], "order_sell_count": oc["sell"], "top5_symbols": symbols, "top5_candidate_ids": candidate_ids, "case_image_board": rel(case_dir / "case_image_board.md"), "case_story_board": rel(case_dir / "case_story_board.md"), "case_candidate_ledger": rel(case_dir / "candidate_ledger.csv"), } ) file_rows.append( { "case_id": case_id, "case_dir": rel(case_dir), "case_image_board": rel(case_dir / "case_image_board.md"), "case_story_board": rel(case_dir / "case_story_board.md"), "case_candidate_ledger": rel(case_dir / "candidate_ledger.csv"), "case_image_manifest": rel(case_dir / "image_manifest.csv"), "source_case_scope": rel(SOURCE_DIR / "full_return_stat_case_scope.csv"), "source_lot_scope": rel(SOURCE_DIR / "full_return_stat_lot_scope.csv"), "source_case_summary": rel(SOURCE_DIR / "case_summary.csv"), "source_decision_log": rel(SOURCE_DIR / "decision_log.csv"), "source_order_ledger": rel(SOURCE_DIR / "order_ledger.csv"), "source_position_lot_ledger": rel(SOURCE_DIR / "position_lot_ledger.csv"), "source_daily_account_ledger": rel(SOURCE_DIR / "daily_account_ledger.csv"), } ) readme_path = PACKAGE_DIR / "README.md" navigation_path = PACKAGE_DIR / "strict_closed_234_navigation.md" index_path = PACKAGE_DIR / "strict_closed_234_case_index.csv" file_map_path = PACKAGE_DIR / "strict_closed_234_file_map.csv" self_check_path = PACKAGE_DIR / "self_check.json" self_check_items_path = PACKAGE_DIR / "self_check_items.csv" manifest_path = PACKAGE_DIR / "manifest.json" write_csv( index_path, index_rows, [ "case_id", "batch_id", "entry_trade_date", "signal_trade_date", "market_gate_status", "primary_scope", "success_flag", "account_return_closed_lots", "buy_lot_count", "closed_lot_count", "order_buy_count", "order_sell_count", "top5_symbols", "top5_candidate_ids", "case_image_board", "case_story_board", "case_candidate_ledger", ], ) write_csv( file_map_path, file_rows, [ "case_id", "case_dir", "case_image_board", "case_story_board", "case_candidate_ledger", "case_image_manifest", "source_case_scope", "source_lot_scope", "source_case_summary", "source_decision_log", "source_order_ledger", "source_position_lot_ledger", "source_daily_account_ledger", ], ) primary = summary["primary_scope"] coverage = summary["coverage_scope"] lot_scope = summary["lot_recalc_scope"] boundary = summary["boundary"] readme = f"""# 234 个严格闭合案例阅读包 本包只负责把已审核通过的无忌全量执行包中 `PRIMARY_STRICT_CLOSED_CASE` 的 234 个案例串起来，方便同事拿到 git 仓库后按图、按表、按账本复核。它不新增交易结论，不重跑候选池、买卖裁决或账户账本。 ## 当前可引用边界 - 来源 run：`{SOURCE_RUN_ID}` - 来源执行复审：`{SOURCE_AUDIT_ID}` - 最终结论引用审核：`{FINAL_AUDIT_ID}` - `RETURN_STAT_READY=false` - 主口径：`PRIMARY_STRICT_CLOSED_CASE`，{primary["case_count"]} 个严格闭合 case，正收益 {primary["positive_case_count"]} 个，成功率读数 {primary["candidate_success_rate_for_audit_only"]:.10f}，账户贡献合计 {primary["account_return_sum_for_audit_only"]:.8f} - 覆盖口径：`ALL_ENTRY_DATE_COVERAGE`，{coverage["entry_date_count"]} 个 entry date，市场闸门打开 {coverage["market_gate_open_entry_dates"]}，关闭 {coverage["market_gate_closed_entry_dates"]} - 辅助 lot 口径：`STRICT_CLOSED_LOT_RECALC_ONLY`，{lot_scope["total_lot_count"]} 个 lot，闭合 {lot_scope["closed_lot_count"]}，边界 {lot_scope["unresolved_lot_count"]} - 边界：{boundary["case_boundary_count"]} 个 case 边界和 {boundary["lot_boundary_count"]} 个 lot 边界不得混入主口径 ## 先看哪些文件 1. `ana-doc/wuji/无忌交易系统234个严格闭合案例阅读说明.md`：给同事看的总说明，解释怎么看、每类文件干什么。 2. `strict_closed_234_case_index.csv`：234 个主口径案例索引，一行一个 case，含日期、批次、收益读数、top5 股票和图片入口。 3. `strict_closed_234_file_map.csv`：逐 case 文件地图，说明这个 case 的图片板、故事板、候选账本、全局账本在哪里。 4. `ana-data/result/{SOURCE_RUN_ID}/case_image_board.md`：全量包根图片入口，可跳到所有 case。 5. `ana-data/result/{SOURCE_RUN_ID}/full_return_stat_case_scope.csv`：正式判断某个 case 是否属于主口径的 scope 表。 ## 单个案例怎么读 1. 在 `strict_closed_234_case_index.csv` 选一个 `case_id`。 2. 打开该行的 `case_image_board`。先看日 K 选股图，再看 1 分钟买点图，再看卖点裁决图。 3. 打开同目录 `case_story_board.md`，按文字串起选股、买入、卖出、lot 状态和边界。 4. 回到全局表核对：`full_return_stat_case_scope.csv` 看主口径，`full_return_stat_lot_scope.csv` 看 lot，`order_ledger.csv` 看订单，`position_lot_ledger.csv` 看每笔 lot，`daily_account_ledger.csv` 看账户资金流。 5. 只把 `PRIMARY_STRICT_CLOSED_CASE` 行用于主口径读数；不要把市场闸门关闭、无 BUY、未真实 SELL 或数据缺口样本混进来。 """ readme_path.write_text(readme, encoding="utf-8") navigation = f"""# 234 个严格闭合案例导航 生成时间：{generated_at} ## 读图顺序 每个 case 的 `case_image_board.md` 都是人工审核第一入口： 1. 选股日 K 图：验证信号日以前的形态、放量、长上影、近期涨停记忆和候选排序。 2. 买点 1 分钟图：验证入场日早盘是否触发买点，且不读取未来数据。 3. 卖点 1 分钟图：验证 T+1 之后的卖点确认和卖出原因。 4. 案例读数区：确认当前收益口径为 `PRIMARY_STRICT_CLOSED_CASE`，边界分类为无。 ## 表格串联方式 | 你要核对什么 | 看哪个文件 | 用什么键串起来 | |---|---|---| | 234 个主口径案例名单 | `strict_closed_234_case_index.csv` | `case_id` | | 某个 case 是否真的进主口径 | `../{SOURCE_RUN_ID}/full_return_stat_case_scope.csv` | `case_id` | | 入选前 5 候选股票 | `../{SOURCE_RUN_ID}/full_selected_candidate_ledger.csv` | `case_id` / `candidate_id` | | 买入 / 卖出裁决 | `../{SOURCE_RUN_ID}/decision_log.csv` | `case_id` / `candidate_id` | | 真实订单事件 | `../{SOURCE_RUN_ID}/order_ledger.csv` | `case_id` / `order_id` | | 每笔仓位 lot | `../{SOURCE_RUN_ID}/position_lot_ledger.csv` | `case_id` / `trade_lot_id` | | lot 口径归属 | `../{SOURCE_RUN_ID}/full_return_stat_lot_scope.csv` | `case_id` / `trade_lot_id` | | 账户现金和持仓流 | `../{SOURCE_RUN_ID}/daily_account_ledger.csv` | `case_id` / 日期 | | case 汇总收益 | `../{SOURCE_RUN_ID}/case_summary.csv` | `case_id` | | 被排除边界 | `../{SOURCE_RUN_ID}/full_return_stat_boundary_table.csv` | `case_id` / `trade_lot_id` | ## 不要这样读 - 不要把 234 个主口径案例说成 743 个 entry date 的全体表现。 - 不要把辅助 lot 口径包装成 case 成功率。 - 不要忽略 `RETURN_STAT_READY=false`。 - 不要只看 CSV 数字而跳过图片入口；无忌专项流程要求图片是人工审核第一入口。 """ navigation_path.write_text(navigation, encoding="utf-8") checks = [ { "check_id": "SOURCE_RUN_PRESENT", "status": "PASS" if SOURCE_DIR.exists() else "FAIL", "detail": rel(SOURCE_DIR), }, { "check_id": "STRICT_CLOSED_CASE_COUNT_234", "status": "PASS" if len(strict_cases) == 234 else "FAIL", "detail": f"strict_closed_cases={len(strict_cases)}", }, { "check_id": "ALL_STRICT_CASES_HAVE_IMAGE_BOARD", "status": "PASS" if all((SOURCE_DIR / "cases" / r["case_id"] / "case_image_board.md").exists() for r in strict_cases) else "FAIL", "detail": "case_image_board.md exists for every strict closed case", }, { "check_id": "ALL_STRICT_CASES_HAVE_STORY_BOARD", "status": "PASS" if all((SOURCE_DIR / "cases" / r["case_id"] / "case_story_board.md").exists() for r in strict_cases) else "FAIL", "detail": "case_story_board.md exists for every strict closed case", }, { "check_id": "SOURCE_SUMMARY_RETURN_STAT_HELD", "status": "PASS" if summary.get("return_stat_ready") is False and summary.get("full_baseline_conclusion_allowed") is False else "FAIL", "detail": "return_stat_ready=false; full_baseline_conclusion_allowed=false", }, ] write_csv(self_check_items_path, checks, ["check_id", "status", "detail"]) self_check = { "schema_version": "1.0", "task_id": TASK_ID, "run_id": RUN_ID, "source_run_id": SOURCE_RUN_ID, "generated_at": generated_at, "status": "PASS_FOR_STRICT_CLOSED_234_REVIEW_GUIDE_READY" if all(c["status"] == "PASS" for c in checks) else "FAIL", "pass_count": sum(1 for c in checks if c["status"] == "PASS"), "fail_count": sum(1 for c in checks if c["status"] == "FAIL"), "strict_closed_case_count": len(strict_cases), "return_stat_ready": False, "notes": [ "本包是阅读说明和索引包，不新增收益结论。", "所有主口径读数必须回到来源 full_return_stat_* 产物和审计边界。", ], } write_json(self_check_path, self_check) manifest_files = [ readme_path, navigation_path, index_path, file_map_path, self_check_path, self_check_items_path, Path(__file__), ] manifest = { "schema_version": "1.0", "task_id": TASK_ID, "run_id": RUN_ID, "source_run_id": SOURCE_RUN_ID, "generated_at": generated_at, "stage": "STRICT_CLOSED_234_REVIEW_GUIDE_READY", "strict_closed_case_count": len(strict_cases), "return_stat_ready": False, "files": [ { "path": rel(path), "size": path.stat().st_size, "sha256": sha256(path), } for path in manifest_files ], } write_json(manifest_path, manifest) print(json.dumps(self_check, ensure_ascii=False, indent=2)) if __name__ == "__main__": build()