from __future__ import annotations import hashlib import json import re from datetime import datetime from pathlib import Path from urllib.parse import unquote import pandas as pd RUN_ID = "RUN-ANA-WUJI-FULL-2023-2026-20260608-001" TASK_ID = "ANA-WUJI-BASELINE-2023-2026" DESIGN_ID = "DESIGN-WUJI-FULL-2023-2026-20260608" DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-FULL-2023-2026-20260608-DESIGN-001" ROOT = Path(__file__).resolve().parents[1] STAGE = "FULL_2023_2026_EXECUTION_REREVIEW_PASSED_RETURN_STAT_HELD" EXEC_REVIEW_STATUS = "全量分批执行审核通过 / 返修复审通过；RETURN_STAT_READY=false，完整结论必须按 full_return_stat_* 分层引用" EXEC_REVIEW_MESSAGE_ID = "msg_20260608134003937_c4484f0e" EXEC_HELD_AUDIT_ID = "AUDIT-ANA-WUJI-FULL-2023-2026-20260608-EXEC-001" EXEC_HELD_MESSAGE_ID = "msg_20260608135433448_d5f47560" EXEC_REREVIEW_HELD_AUDIT_ID = "AUDIT-ANA-WUJI-FULL-2023-2026-20260608-EXEC-REREVIEW-001" EXEC_REREVIEW_HELD_MESSAGE_ID = "msg_20260608142652470_d00969d7" EXEC_REREVIEW_PASS_AUDIT_ID = "AUDIT-ANA-WUJI-FULL-2023-2026-20260608-EXEC-REREVIEW-002" EXEC_REREVIEW_PASS_MESSAGE_ID = "msg_20260608145302303_8c8a3273" RETURN_SCOPE_ISSUE_ID = "ANA-ISSUE-WUJI-FULL-RETURN-STAT-SCOPE-20260608-001" RETURN_READY = False ALLOWED_UNRESOLVED = {"WINDOW_END_VALUATION_ONLY", "EXIT_DATA_GAP_HELD", "EXIT_REVIEW_HELD"} ROLE_TITLES = { "candidate_daily_100d_decision_view": "1. 选股日K图（约100个交易日）", "entry_1m_morning_review_view": "2. 买点早盘1分钟复核图", "entry_1m_late_review_view": "3. 买点尾盘1分钟复核图", "entry_1m_buy_decision_view": "4. 买入裁决1分钟图", "exit_daily_signal_review_view": "5. 卖点 / 持仓日K信号图", "exit_1m_sell_decision_view": "6. 卖出裁决1分钟图", } def now_iso() -> str: return datetime.now().astimezone().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 read_json(name: str) -> dict: path = ROOT / name return json.loads(path.read_text(encoding="utf-8")) if path.exists() else {} def read_csv(name: str) -> pd.DataFrame: path = ROOT / name return pd.read_csv(path, encoding="utf-8-sig") if path.exists() else pd.DataFrame() def write_csv(df: pd.DataFrame, path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) df.to_csv(path, index=False, encoding="utf-8-sig") def clean(value: object, default: str = "") -> str: text = str(value).strip() if text.lower() in {"", "nan", "none", "nat"}: return default return text def has_value(value: object) -> bool: return clean(value) != "" def fmt_int(value: object) -> str: if not has_value(value): return "0" return str(int(float(value))) def fmt_price(value: object) -> str: if not has_value(value): return "" return f"{float(value):.2f}" def fmt_pct(value: object) -> str: if not has_value(value): return "" return f"{float(value):.4%}" def safe_rel(from_dir: Path, root_relative_path: object) -> str: target = clean(root_relative_path) if not target: return "" return Path("../../", target).as_posix() if from_dir.name == "img" else Path(target).as_posix() def link_from_doc(doc_path: Path, root_relative_path: object) -> str: target = clean(root_relative_path) if not target: return "" try: rel = Path(target).relative_to(doc_path.parent.relative_to(ROOT)) return rel.as_posix() except ValueError: return Path(*([".."] * len(doc_path.parent.relative_to(ROOT).parts)), target).as_posix() def markdown_links(path: Path) -> list[Path]: text = path.read_text(encoding="utf-8") links: list[Path] = [] for raw in re.findall(r"\]\(([^)]+)\)", text): if "://" in raw or raw.startswith("#"): continue target = unquote(raw.split("#", 1)[0]).strip() if not target: continue links.append((path.parent / target).resolve()) return links def manifest_for_dir(base: Path) -> dict: files: list[dict] = [] for path in sorted(base.rglob("*")): if not path.is_file() or "__pycache__" in path.parts: continue rel = path.relative_to(base).as_posix() if rel == "manifest.json": continue files.append({"path": rel, "size": path.stat().st_size, "sha256": sha256_file(path)}) return { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": now_iso(), "base": base.relative_to(ROOT).as_posix() if base != ROOT else ".", "file_count": len(files), "files": files, } def role_counts(df: pd.DataFrame, column: str) -> dict: if df.empty or column not in df.columns: return {} return {str(k): int(v) for k, v in df[column].value_counts(dropna=False).to_dict().items()} def build_scope_lines( case_index: pd.DataFrame, batch_index: pd.DataFrame, selected: pd.DataFrame, orders: pd.DataFrame, lots: pd.DataFrame, case_summary: pd.DataFrame, ) -> list[str]: boundary = lots[lots.lot_status != "CLOSED_BY_AI_SELL"] if not lots.empty else pd.DataFrame() boundary_text = "；".join(f"{k} {v} 笔" for k, v in role_counts(boundary, "lot_status").items()) or "无" order_counts = role_counts(orders, "action") return [ f"- 当前 run：`{RUN_ID}`", f"- 当前阶段：`{STAGE}`", f"- 审核状态：{EXEC_REVIEW_STATUS}", f"- 设计审核：`{DESIGN_AUDIT_ID}`", f"- 范围：{len(case_index)} 个 entry_trade_date，{len(batch_index)} 个批次，每个 entry date 取前 5 个候选，共 {len(selected)} 条候选选择记录", f"- 市场闸门：打开 {int((case_index.market_gate_status == 'MKT_GATE_OPEN_PREV_DAY_UP_3000').sum())} 个 entry date，关闭 {int((case_index.market_gate_status == 'NO_TRADE_MARKET_GATE_CLOSED').sum())} 个 entry date", f"- 订单：BUY {order_counts.get('BUY', 0)}，SELL {order_counts.get('SELL', 0)}；有 BUY 的 case：{len(case_summary)}", f"- Lot：共 {len(lots)}，已闭合 {role_counts(lots, 'lot_status').get('CLOSED_BY_AI_SELL', 0)}，边界保留 {len(boundary)}（{boundary_text}）", "- `RETURN_STAT_READY=false`", "- 本入口已通过全量分批执行返修复审；完整结论仍必须按 `full_return_stat_*` 分层引用，`RETURN_STAT_READY=false` 继续保留。", ] def write_case_boards( case_index: pd.DataFrame, selected: pd.DataFrame, decisions: pd.DataFrame, lots: pd.DataFrame, case_summary: pd.DataFrame, image_manifest: pd.DataFrame, ) -> None: summary_by_case = case_summary.set_index("case_id") if not case_summary.empty else pd.DataFrame() case_scope_df = read_csv("full_return_stat_case_scope.csv") scope_by_case = case_scope_df.set_index("case_id") if not case_scope_df.empty else pd.DataFrame() for _, meta in case_index.sort_values("entry_trade_date").iterrows(): case_id = str(meta.case_id) case_dir = ROOT / "cases" / case_id case_dir.mkdir(parents=True, exist_ok=True) rows = image_manifest[image_manifest.case_id == case_id].copy() case_selected = selected[selected.case_id == case_id].copy() case_decisions = decisions[decisions.case_id == case_id].copy() case_lots = lots[lots.case_id == case_id].copy() scope_row = scope_by_case.loc[case_id] if not scope_by_case.empty and case_id in scope_by_case.index else None board_path = case_dir / "case_image_board.md" lines = [ f"# {case_id} 图片审核板", "", "用途：人工审核员按图复核本案例从选股、买入、卖点信号到卖出裁决的全链路。", "", "## 当前全量执行包状态", "", f"- 当前 run：`{RUN_ID}`", f"- 当前阶段：`{STAGE}`", f"- 审核状态：{EXEC_REVIEW_STATUS}", f"- 批次：`{meta.batch_id}`；入场日：{meta.entry_trade_date}；信号日：{meta.signal_trade_date}", f"- 市场闸门：{meta.market_gate_status}；候选数：{fmt_int(meta.candidate_count)}；入选 top5：{len(case_selected)}", "- `RETURN_STAT_READY=false`", "", ] if not summary_by_case.empty and case_id in summary_by_case.index: s = summary_by_case.loc[case_id] return_scope = clean(s.get("return_stat_scope"), "RETURN_STAT_HELD_BOUNDARY_TABLE") scope_status = clean(s.get("primary_strict_closed_case_reason")) boundary_category = clean(scope_row.get("boundary_category")) if scope_row is not None else "" boundary_reason = clean(scope_row.get("boundary_reason")) if scope_row is not None else "" lines.extend( [ "## 案例读数", "", f"- 买入 lot：{fmt_int(s.buy_lot_count)}", f"- 已闭合 lot：{fmt_int(s.closed_lot_count)}", f"- 未闭合 / 边界 lot：{fmt_int(s.unresolved_lot_count)}", f"- 闭合 lot 账户贡献合计：{fmt_pct(s.account_return_closed_lots)}（送审候选读数，不是完整 baseline 结论）", f"- 当前收益口径：`{return_scope}`", f"- 收益口径状态：`{scope_status}`", f"- 主口径标记：{fmt_int(s.get('primary_strict_closed_case_flag'))}", f"- 边界分类：{boundary_category or '无'}", f"- 边界原因：{boundary_reason or '无，当前 case 满足 PRIMARY_STRICT_CLOSED_CASE。'}", "", ] ) else: if scope_row is not None: scope_status = clean(scope_row.get("case_scope_status"), "NO_SUMMARY_BOUNDARY") return_scope = "PRIMARY_STRICT_CLOSED_CASE" if fmt_int(scope_row.get("primary_strict_closed_case_flag")) == "1" else "RETURN_STAT_HELD_BOUNDARY_TABLE" boundary_category = clean(scope_row.get("boundary_category")) boundary_reason = clean(scope_row.get("boundary_reason")) else: scope_status = "NO_SCOPE_ROW" return_scope = "RETURN_STAT_HELD_BOUNDARY_TABLE" boundary_category = "" boundary_reason = "未找到 full_return_stat_case_scope.csv 对应行。" lines.extend( [ "## 案例读数", "", "- 本案例日无 BUY，未进入主收益候选。", "- 若市场闸门关闭，应只能看到 NO_TRADE_MARKET_GATE_CLOSED 决策。", f"- 当前收益口径：`{return_scope}`", f"- 收益口径状态：`{scope_status}`", f"- 边界分类：{boundary_category or '无'}", f"- 边界原因：{boundary_reason or '无'}", "", ] ) if not case_lots.empty: lines.extend(["## 持仓状态", ""]) for _, lot in case_lots.iterrows(): exit_part = f"；卖出 {lot.exit_trade_date} {clean(lot.exit_time)} @ {fmt_price(lot.exit_price)}" if has_value(lot.exit_price) else "" contrib = f"；账户贡献 {fmt_pct(lot.account_return_contribution_pct)}" if has_value(lot.account_return_contribution_pct) else "" lines.append( f"- {lot.trade_lot_id} / {lot.symbol}：`{lot.lot_status}`；买入 {lot.entry_trade_date} {lot.entry_time} @ {fmt_price(lot.entry_price)}{exit_part}{contrib}" ) lines.append("") for role, title in ROLE_TITLES.items(): group = rows[rows.chart_role == role].copy() if group.empty: continue lines.extend([f"## {title}", ""]) for _, row in group.iterrows(): rel = link_from_doc(board_path, row.path) decision = clean(row.decision_time) note = clean(row.note) status = clean(row.status) lines.extend( [ f"### {row.symbol} {decision}".rstrip(), "", f"![{row.symbol}]({rel})", "", f"- 图状态：`{status}`", f"- 说明：{note}", "", ] ) lines.extend( [ "## 审核边界", "", "- 本板是当前全量分批执行包的图片第一入口；CSV/JSON 是反查材料。", "- 本案例读数只能按 `full_return_stat_*` 分层口径引用，不得脱离主 / 覆盖 / 辅助 lot 口径写成无边界完整 baseline 结论。", "- 非真实 SELL、数据缺口或窗口末估值 lot 必须保留边界，不得强行转 SELL。", "", ] ) board_path.write_text("\n".join(lines), encoding="utf-8") story_path = case_dir / "case_story_board.md" story = [ f"# {case_id} 一页式故事板", "", f"- 当前 run：`{RUN_ID}`", f"- 当前阶段：`{STAGE}`", f"- 批次：`{meta.batch_id}`；入场日：{meta.entry_trade_date}", f"- 市场闸门：{meta.market_gate_status}", "- 图片审核板：`case_image_board.md`", "- `RETURN_STAT_READY=false`", "", "## 收益口径", "", ] if scope_row is not None: story_scope = "PRIMARY_STRICT_CLOSED_CASE" if fmt_int(scope_row.get("primary_strict_closed_case_flag")) == "1" else "RETURN_STAT_HELD_BOUNDARY_TABLE" story.extend( [ f"- 当前收益口径：`{story_scope}`", f"- 收益口径状态：`{clean(scope_row.get('case_scope_status'))}`", f"- 边界分类：{clean(scope_row.get('boundary_category'), '无')}", f"- 边界原因：{clean(scope_row.get('boundary_reason'), '无')}", "", ] ) else: story.extend(["- 当前收益口径：`RETURN_STAT_HELD_BOUNDARY_TABLE`", "- 收益口径状态：`NO_SCOPE_ROW`", ""]) story.extend( [ "## 操作时间线", "", ] ) if case_decisions.empty: story.append("- 本案例无决策记录。") else: for _, row in case_decisions.iterrows(): image = clean(row.evidence_image_path) image_link = link_from_doc(story_path, image) if image else "" img_text = f"；图：[{Path(image_link).name}]({image_link})" if image_link else "" price = f" @ {fmt_price(row.price)}" if has_value(row.price) else "" story.append( f"- {row.decision_stage} / `{row.action_status}`：{row.symbol} {clean(row.decision_time, '无裁决时间')}{price}；{clean(row.decision_reason_cn)}{img_text}" ) story.extend(["", "## Lot 收口", ""]) if case_lots.empty: story.append("- 本案例无 BUY / 无 lot。") else: for _, lot in case_lots.iterrows(): exit_part = f"；卖出 {lot.exit_trade_date} {clean(lot.exit_time)} @ {fmt_price(lot.exit_price)}" if has_value(lot.exit_price) else "" story.append(f"- {lot.symbol}：`{lot.lot_status}`；买入 {lot.entry_trade_date} {lot.entry_time} @ {fmt_price(lot.entry_price)}{exit_part}") story.extend( [ "", "## 审核提示", "", "- 先看图片审核板，再回查账本。", "- 本页读数只能按 `full_return_stat_*` 分层口径引用，不得脱离边界样本说明转写为无边界完整 baseline 结论。", "", ] ) story_path.write_text("\n".join(story), encoding="utf-8") def write_root_and_batch_boards( case_index: pd.DataFrame, batch_index: pd.DataFrame, selected: pd.DataFrame, decisions: pd.DataFrame, orders: pd.DataFrame, lots: pd.DataFrame, case_summary: pd.DataFrame, image_manifest: pd.DataFrame, ) -> None: scope_lines = build_scope_lines(case_index, batch_index, selected, orders, lots, case_summary) summary_by_case = case_summary.set_index("case_id") if not case_summary.empty else pd.DataFrame() root_lines = [ "# 全量分批执行图片审核入口", "", *scope_lines, "", "## 批次入口", "", "| batch_id | 案例日 | 入场日期范围 | 候选 | BUY | SELL | 边界 lot | 批次图片板 | 批次故事板 |", "|---|---:|---|---:|---:|---:|---:|---|---|", ] story_lines = [ f"# {RUN_ID} 全量分批执行故事板总入口", "", "用途：给人工审核员从批次入口进入 743 个案例日；每个案例优先看图片，再回查账本。", "", "## 总体边界", "", *scope_lines, "", "## 批次总览", "", "| batch_id | 案例日 | 入场日期范围 | BUY | SELL | 边界 lot |", "|---|---:|---|---:|---:|---:|", ] for _, batch in batch_index.sort_values("batch_order").iterrows(): batch_id = str(batch.batch_id) batch_dir = ROOT / "batches" / batch_id batch_dir.mkdir(parents=True, exist_ok=True) case_ids = case_index[case_index.batch_id == batch_id].case_id.astype(str).tolist() batch_cases = case_index[case_index.case_id.isin(case_ids)].copy() batch_selected = selected[selected.case_id.isin(case_ids)].copy() batch_decisions = decisions[decisions.case_id.isin(case_ids)].copy() batch_orders = orders[orders.case_id.isin(case_ids)].copy() batch_lots = lots[lots.case_id.isin(case_ids)].copy() batch_summary = case_summary[case_summary.case_id.isin(case_ids)].copy() batch_images = image_manifest[image_manifest.case_id.isin(case_ids)].copy() boundary = batch_lots[batch_lots.lot_status != "CLOSED_BY_AI_SELL"] order_counts = role_counts(batch_orders, "action") for name, df in [ ("case_index.csv", batch_cases), ("selected_candidate_ledger.csv", batch_selected), ("decision_log.csv", batch_decisions), ("order_ledger.csv", batch_orders), ("position_lot_ledger.csv", batch_lots), ("daily_account_ledger.csv", read_csv("daily_account_ledger.csv")[read_csv("daily_account_ledger.csv").case_id.isin(case_ids)]), ("case_summary.csv", batch_summary), ("sell_decision_log.csv", read_csv("sell_decision_log.csv")[read_csv("sell_decision_log.csv").case_id.isin(case_ids)]), ("exit_resolution_log.csv", read_csv("exit_resolution_log.csv")[read_csv("exit_resolution_log.csv").case_id.isin(case_ids)]), ("image_manifest.csv", batch_images), ]: write_csv(df, batch_dir / name) batch_summary_json = { "schema_version": "1.0", "run_id": RUN_ID, "batch_id": batch_id, "stage": STAGE, "case_count": int(len(batch_cases)), "selected_candidate_rows": int(len(batch_selected)), "order_counts": role_counts(batch_orders, "action"), "lot_status_counts": role_counts(batch_lots, "lot_status"), "image_role_counts": role_counts(batch_images, "chart_role"), "return_stat_ready": False, "boundary": "batch package for audit only; full execution audit pending", } (batch_dir / "batch_summary.json").write_text(json.dumps(batch_summary_json, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") (batch_dir / "batch_summary.md").write_text( "\n".join( [ f"# {batch_id} 批次摘要", "", f"- run：`{RUN_ID}`", f"- 阶段：`{STAGE}`", f"- 案例日：{len(batch_cases)}", f"- 日期范围：{batch.entry_date_start} 至 {batch.entry_date_end}", f"- 候选：{len(batch_selected)}", f"- BUY：{order_counts.get('BUY', 0)}，SELL：{order_counts.get('SELL', 0)}", f"- 边界 lot：{len(boundary)}", "- `RETURN_STAT_READY=false`", "", ] ), encoding="utf-8", ) batch_board = [ f"# {batch_id} 批次图片审核入口", "", f"- 当前 run：`{RUN_ID}`", f"- 当前阶段：`{STAGE}`", f"- 日期范围：{batch.entry_date_start} 至 {batch.entry_date_end}", f"- 案例日：{len(batch_cases)}；候选：{len(batch_selected)}；BUY：{order_counts.get('BUY', 0)}；SELL：{order_counts.get('SELL', 0)}；边界 lot：{len(boundary)}", "- `RETURN_STAT_READY=false`", "", "| case_id | 入场日 | 市场闸门 | BUY lot | 已闭合 | 边界 lot | 图片板 | 故事板 |", "|---|---|---|---:|---:|---:|---|---|", ] for _, case in batch_cases.sort_values("entry_trade_date").iterrows(): case_id = str(case.case_id) if not summary_by_case.empty and case_id in summary_by_case.index: s = summary_by_case.loc[case_id] buy = fmt_int(s.buy_lot_count) closed = fmt_int(s.closed_lot_count) unresolved = fmt_int(s.unresolved_lot_count) else: buy = closed = unresolved = "0" batch_board.append( f"| {case_id} | {case.entry_trade_date} | {case.market_gate_status} | {buy} | {closed} | {unresolved} | [图片板](../../cases/{case_id}/case_image_board.md) | [故事板](../../cases/{case_id}/case_story_board.md) |" ) (batch_dir / "case_image_board.md").write_text("\n".join(batch_board) + "\n", encoding="utf-8") (batch_dir / "case_story_board.md").write_text("\n".join(batch_board).replace("图片审核入口", "故事板入口") + "\n", encoding="utf-8") (batch_dir / "manifest.json").write_text(json.dumps(manifest_for_dir(batch_dir), ensure_ascii=False, indent=2) + "\n", encoding="utf-8") root_lines.append( f"| {batch_id} | {len(batch_cases)} | {batch.entry_date_start} 至 {batch.entry_date_end} | {len(batch_selected)} | {order_counts.get('BUY', 0)} | {order_counts.get('SELL', 0)} | {len(boundary)} | [图片板](batches/{batch_id}/case_image_board.md) | [故事板](batches/{batch_id}/case_story_board.md) |" ) story_lines.append( f"| {batch_id} | {len(batch_cases)} | {batch.entry_date_start} 至 {batch.entry_date_end} | {order_counts.get('BUY', 0)} | {order_counts.get('SELL', 0)} | {len(boundary)} |" ) root_lines.extend( [ "", "## 图片角色统计", "", *[f"- {ROLE_TITLES.get(role, role)}：{count} 张" for role, count in role_counts(image_manifest, "chart_role").items()], "", "## Lot 状态统计", "", *[f"- {status}：{count}" for status, count in role_counts(lots, "lot_status").items()], "", ] ) story_lines.extend( [ "", "## 审核路径", "", "- 从批次入口进入案例图板。", "- 图片是人工审核第一入口；CSV/JSON 是反查证据。", "- 当前全量分批执行包已通过返修复审；完整结论仍必须按 `full_return_stat_*` 分层引用，`RETURN_STAT_READY=false` 继续保留。", "", ] ) (ROOT / "case_image_board.md").write_text("\n".join(root_lines), encoding="utf-8") (ROOT / "case_story_board.md").write_text("\n".join(story_lines), encoding="utf-8") def check(name: str, passed: bool, detail: str, rows: list[dict]) -> None: rows.append({"check_name": name, "status": "PASS" if passed else "FAIL", "detail": detail}) def to_float(value: object, default: float = 0.0) -> float: if not has_value(value): return default return float(value) def boolish(value: object) -> bool: return clean(value).lower() in {"1", "true", "yes"} def action_status_summary(df: pd.DataFrame) -> str: if df.empty or "action_status" not in df.columns: return "" counts = df.action_status.value_counts(dropna=False).to_dict() return "; ".join(f"{k}:{int(v)}" for k, v in counts.items()) def classify_case_scope(meta: pd.Series, case_lots: pd.DataFrame, entry_decisions: pd.DataFrame, summary_row: pd.Series | None) -> tuple[str, str, str]: market_status = clean(meta.get("market_gate_status")) has_buy = summary_row is not None and int(to_float(summary_row.get("buy_lot_count"))) > 0 unresolved_count = int((case_lots.lot_status != "CLOSED_BY_AI_SELL").sum()) if not case_lots.empty else 0 all_closed = has_buy and unresolved_count == 0 and bool((case_lots.lot_status == "CLOSED_BY_AI_SELL").all()) if market_status == "MKT_GATE_OPEN_PREV_DAY_UP_3000" and all_closed: return "PRIMARY_STRICT_CLOSED_CASE", "PRIMARY_STRICT_CLOSED_CASE", "市场闸门打开、有真实 BUY，且全部 lot 均为 CLOSED_BY_AI_SELL。" if market_status == "NO_TRADE_MARKET_GATE_CLOSED": return "NO_TRADE_MARKET_GATE_CLOSED", "MARKET_GATE_CLOSED", "市场闸门关闭，覆盖口径保留，不进入主收益 / 成功率口径。" if not has_buy: statuses = set(entry_decisions.action_status.astype(str).tolist()) if not entry_decisions.empty else set() if "DATA_GAP_HELD" in statuses: return "ENTRY_DATA_GAP_HELD_NO_BUY", "ENTRY_DATA_GAP_HELD", "市场闸门打开但入场分钟线或裁决数据缺口，未生成 BUY。" return "OPEN_GATE_NO_BUY_AI_REVIEWED", "NO_BUY_AI_REVIEWED", "市场闸门打开但 AI 买点复核未确认 BUY。" if unresolved_count > 0: statuses = ", ".join(sorted(case_lots[case_lots.lot_status != "CLOSED_BY_AI_SELL"].lot_status.astype(str).unique())) return "BOUNDARY_UNRESOLVED_LOT", "UNRESOLVED_LOT_BOUNDARY", f"存在未真实 SELL / 数据缺口 lot：{statuses}。" return "RETURN_STAT_HELD_OTHER", "OTHER_BOUNDARY", "未满足 PRIMARY_STRICT_CLOSED_CASE 的其他边界状态。" def build_return_stat_artifacts( case_index: pd.DataFrame, decisions: pd.DataFrame, lots: pd.DataFrame, case_summary: pd.DataFrame, ) -> pd.DataFrame: summary_by_case = {str(row.case_id): row for _, row in case_summary.iterrows()} if not case_summary.empty else {} lot_groups = {str(case_id): group.copy() for case_id, group in lots.groupby("case_id", sort=False)} if not lots.empty else {} entry_decisions = decisions[decisions.decision_stage == "ENTRY_AI_REVIEW"].copy() if not decisions.empty else pd.DataFrame() entry_groups = {str(case_id): group.copy() for case_id, group in entry_decisions.groupby("case_id", sort=False)} if not entry_decisions.empty else {} case_scope_rows: list[dict] = [] boundary_rows: list[dict] = [] primary_case_ids: set[str] = set() for _, meta in case_index.sort_values("entry_trade_date").iterrows(): case_id = str(meta.case_id) case_lots = lot_groups.get(case_id, pd.DataFrame(columns=lots.columns)) summary_row = summary_by_case.get(case_id) entry_group = entry_groups.get(case_id, pd.DataFrame(columns=entry_decisions.columns)) scope_status, boundary_category, reason = classify_case_scope(meta, case_lots, entry_group, summary_row) primary = scope_status == "PRIMARY_STRICT_CLOSED_CASE" if primary: primary_case_ids.add(case_id) buy_count = int(to_float(summary_row.get("buy_lot_count"))) if summary_row is not None else 0 closed_count = int(to_float(summary_row.get("closed_lot_count"))) if summary_row is not None else 0 unresolved_count = int(to_float(summary_row.get("unresolved_lot_count"))) if summary_row is not None else 0 case_return = to_float(summary_row.get("account_return_closed_lots")) if summary_row is not None else 0.0 unresolved_statuses = "" if not case_lots.empty: unresolved_statuses = "; ".join( f"{k}:{int(v)}" for k, v in case_lots[case_lots.lot_status != "CLOSED_BY_AI_SELL"].lot_status.value_counts(dropna=False).to_dict().items() ) case_scope_rows.append( { "run_id": RUN_ID, "case_id": case_id, "batch_id": clean(meta.get("batch_id")), "entry_trade_date": clean(meta.get("entry_trade_date")), "signal_trade_date": clean(meta.get("signal_trade_date")), "market_gate_status": clean(meta.get("market_gate_status")), "market_gate_open_flag": clean(meta.get("market_gate_open_flag")), "coverage_scope": "ALL_ENTRY_DATE_COVERAGE", "case_scope_status": scope_status, "primary_scope": "PRIMARY_STRICT_CLOSED_CASE" if primary else "", "primary_strict_closed_case_flag": int(primary), "strict_closed_lot_recalc_only_count": closed_count, "buy_lot_count": buy_count, "closed_lot_count": closed_count, "unresolved_lot_count": unresolved_count, "unresolved_lot_statuses": unresolved_statuses, "account_return_closed_lots": f"{case_return:.8f}" if summary_row is not None else "", "primary_case_success_flag": int(primary and case_return > 0), "boundary_category": "" if primary else boundary_category, "boundary_reason": "" if primary else reason, "entry_action_status_summary": action_status_summary(entry_group), "excluded_from_primary_flag": int(not primary), "source_case_summary_present": int(summary_row is not None), } ) if not primary: boundary_rows.append( { "run_id": RUN_ID, "boundary_id": f"CASE-{case_id}", "boundary_level": "CASE", "case_id": case_id, "trade_lot_id": "", "symbol": "", "entry_trade_date": clean(meta.get("entry_trade_date")), "boundary_category": boundary_category, "boundary_status": scope_status, "boundary_reason": reason, "excluded_from_primary_flag": 1, "coverage_scope": "ALL_ENTRY_DATE_COVERAGE", } ) case_scope = pd.DataFrame(case_scope_rows) primary_flag_by_case = dict(zip(case_scope.case_id, case_scope.primary_strict_closed_case_flag)) primary_reason_by_case = dict(zip(case_scope.case_id, case_scope.case_scope_status)) lot_rows: list[dict] = [] for _, lot in lots.iterrows(): case_id = str(lot.case_id) closed = clean(lot.lot_status) == "CLOSED_BY_AI_SELL" primary_case = bool(primary_flag_by_case.get(case_id, 0)) lot_scope = "STRICT_CLOSED_LOT_RECALC_ONLY" if closed else "RETURN_STAT_HELD_BOUNDARY_TABLE" boundary_category = "" if closed else clean(lot.lot_status) boundary_reason = "" if closed else "lot 未形成真实 AI SELL，必须排除出主 case 收益 / 成功率口径。" lot_rows.append( { "run_id": RUN_ID, "trade_lot_id": clean(lot.trade_lot_id), "case_id": case_id, "symbol": clean(lot.symbol), "entry_trade_date": clean(lot.entry_trade_date), "entry_time": clean(lot.entry_time), "entry_price": clean(lot.entry_price), "exit_trade_date": clean(lot.exit_trade_date), "exit_time": clean(lot.exit_time), "exit_price": clean(lot.exit_price), "lot_status": clean(lot.lot_status), "lot_scope": lot_scope, "closed_lot_recalc_included_flag": int(closed), "primary_case_included_flag": int(primary_case), "position_pct": clean(lot.position_pct), "lot_return_pct": clean(lot.lot_return_pct), "account_return_contribution_pct": clean(lot.account_return_contribution_pct), "boundary_category": boundary_category, "boundary_reason": boundary_reason, "case_scope_status": primary_reason_by_case.get(case_id, ""), } ) if not closed: boundary_rows.append( { "run_id": RUN_ID, "boundary_id": f"LOT-{clean(lot.trade_lot_id)}", "boundary_level": "LOT", "case_id": case_id, "trade_lot_id": clean(lot.trade_lot_id), "symbol": clean(lot.symbol), "entry_trade_date": clean(lot.entry_trade_date), "boundary_category": clean(lot.lot_status), "boundary_status": "LOT_NOT_REAL_SELL", "boundary_reason": boundary_reason, "excluded_from_primary_flag": 1, "coverage_scope": "STRICT_CLOSED_LOT_RECALC_ONLY_EXCLUSION", } ) lot_scope = pd.DataFrame(lot_rows) boundary_table = pd.DataFrame(boundary_rows) primary_cases = case_scope[case_scope.primary_strict_closed_case_flag == 1].copy() primary_returns = pd.to_numeric(primary_cases.account_return_closed_lots, errors="coerce").fillna(0.0) closed_lots = lot_scope[lot_scope.closed_lot_recalc_included_flag == 1].copy() closed_lot_contrib = pd.to_numeric(closed_lots.account_return_contribution_pct, errors="coerce").fillna(0.0) summary = { "schema_version": "1.0", "task_id": TASK_ID, "run_id": RUN_ID, "generated_at": now_iso(), "stage": STAGE, "execution_review_status": "EXECUTION_REREVIEW_PASSED_RETURN_STAT_HELD", "design_audit_id": DESIGN_AUDIT_ID, "held_audit_id": EXEC_HELD_AUDIT_ID, "held_message_id": EXEC_HELD_MESSAGE_ID, "rereview_held_audit_id": EXEC_REREVIEW_HELD_AUDIT_ID, "rereview_held_message_id": EXEC_REREVIEW_HELD_MESSAGE_ID, "rereview_pass_audit_id": EXEC_REREVIEW_PASS_AUDIT_ID, "rereview_pass_message_id": EXEC_REREVIEW_PASS_MESSAGE_ID, "issue_id": RETURN_SCOPE_ISSUE_ID, "return_stat_ready": False, "full_baseline_conclusion_allowed": False, "primary_scope": { "name": "PRIMARY_STRICT_CLOSED_CASE", "case_count": int(len(primary_cases)), "positive_case_count": int((primary_returns > 0).sum()), "non_positive_case_count": int((primary_returns <= 0).sum()), "candidate_success_rate_for_audit_only": float((primary_returns > 0).mean()) if len(primary_returns) else 0.0, "account_return_sum_for_audit_only": float(primary_returns.sum()) if len(primary_returns) else 0.0, "account_return_mean_for_audit_only": float(primary_returns.mean()) if len(primary_returns) else 0.0, "account_return_median_for_audit_only": float(primary_returns.median()) if len(primary_returns) else 0.0, "account_return_min_for_audit_only": float(primary_returns.min()) if len(primary_returns) else 0.0, "account_return_max_for_audit_only": float(primary_returns.max()) if len(primary_returns) else 0.0, }, "coverage_scope": { "name": "ALL_ENTRY_DATE_COVERAGE", "entry_date_count": int(len(case_scope)), "market_gate_open_entry_dates": int((case_scope.market_gate_status == "MKT_GATE_OPEN_PREV_DAY_UP_3000").sum()), "market_gate_closed_entry_dates": int((case_scope.market_gate_status == "NO_TRADE_MARKET_GATE_CLOSED").sum()), "buy_case_count": int((pd.to_numeric(case_scope.buy_lot_count, errors="coerce").fillna(0) > 0).sum()), "open_gate_no_buy_case_count": int(((case_scope.market_gate_status == "MKT_GATE_OPEN_PREV_DAY_UP_3000") & (pd.to_numeric(case_scope.buy_lot_count, errors="coerce").fillna(0) == 0)).sum()), "excluded_case_count": int((case_scope.excluded_from_primary_flag == 1).sum()), }, "lot_recalc_scope": { "name": "STRICT_CLOSED_LOT_RECALC_ONLY", "total_lot_count": int(len(lot_scope)), "closed_lot_count": int((lot_scope.closed_lot_recalc_included_flag == 1).sum()), "unresolved_lot_count": int((lot_scope.closed_lot_recalc_included_flag == 0).sum()), "positive_closed_lot_count": int((closed_lot_contrib > 0).sum()), "closed_lot_account_return_sum_for_recalc_only": float(closed_lot_contrib.sum()) if len(closed_lot_contrib) else 0.0, }, "boundary": { "case_boundary_count": int((boundary_table.boundary_level == "CASE").sum()) if not boundary_table.empty else 0, "lot_boundary_count": int((boundary_table.boundary_level == "LOT").sum()) if not boundary_table.empty else 0, "case_boundary_counts": role_counts(boundary_table[boundary_table.boundary_level == "CASE"], "boundary_category") if not boundary_table.empty else {}, "lot_boundary_counts": role_counts(boundary_table[boundary_table.boundary_level == "LOT"], "boundary_category") if not boundary_table.empty else {}, }, "timestamp_note": "entry_* generated_at values are stage artifact timestamps from long-running execution; current package status is governed by summary.json, self_check.json, full_return_stat_summary.json and manifest.json.", "boundary_statement": "Execution re-review passed for the full batch package, but RETURN_STAT_READY remains false. Readouts may only be cited with the explicit full_return_stat_* layered scopes and boundaries.", } write_csv(case_scope, ROOT / "full_return_stat_case_scope.csv") write_csv(lot_scope, ROOT / "full_return_stat_lot_scope.csv") write_csv(boundary_table, ROOT / "full_return_stat_boundary_table.csv") (ROOT / "full_return_stat_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") (ROOT / "full_return_stat_summary.md").write_text( "\n".join( [ f"# {RUN_ID} 全量收益口径分层复审通过摘要", "", f"- 阶段：`{STAGE}`", f"- 设计审核：`{DESIGN_AUDIT_ID}`", f"- 原执行审核 HELD：`{EXEC_HELD_AUDIT_ID}`", f"- 关联问题：`{RETURN_SCOPE_ISSUE_ID}`", f"- 执行复审 HELD：`{EXEC_REREVIEW_HELD_AUDIT_ID}`", f"- 执行复审通过：`{EXEC_REREVIEW_PASS_AUDIT_ID}`", "- `RETURN_STAT_READY=false`", "- 当前读数已通过执行复审，但只能按 `full_return_stat_*` 分层口径引用，不得脱离边界写成无边界完整 baseline 结论。", "", "## PRIMARY_STRICT_CLOSED_CASE", "", f"- case 数：{summary['primary_scope']['case_count']}", f"- 正收益 case：{summary['primary_scope']['positive_case_count']}", f"- 非正收益 case：{summary['primary_scope']['non_positive_case_count']}", f"- 成功率候选读数（仅供复审）：{summary['primary_scope']['candidate_success_rate_for_audit_only']:.6f}", f"- 账户贡献合计候选读数（仅供复审）：{summary['primary_scope']['account_return_sum_for_audit_only']:.8f}", "", "## ALL_ENTRY_DATE_COVERAGE", "", f"- entry date：{summary['coverage_scope']['entry_date_count']}", f"- 市场闸门打开：{summary['coverage_scope']['market_gate_open_entry_dates']}", f"- 市场闸门关闭：{summary['coverage_scope']['market_gate_closed_entry_dates']}", f"- 有 BUY case：{summary['coverage_scope']['buy_case_count']}", f"- 市场闸门打开但无 BUY case：{summary['coverage_scope']['open_gate_no_buy_case_count']}", f"- 排除出主口径 case：{summary['coverage_scope']['excluded_case_count']}", "", "## STRICT_CLOSED_LOT_RECALC_ONLY", "", f"- 总 lot：{summary['lot_recalc_scope']['total_lot_count']}", f"- 闭合 lot：{summary['lot_recalc_scope']['closed_lot_count']}", f"- 边界 lot：{summary['lot_recalc_scope']['unresolved_lot_count']}", f"- 闭合 lot 账户贡献合计（仅供复算定位）：{summary['lot_recalc_scope']['closed_lot_account_return_sum_for_recalc_only']:.8f}", "", "## 边界", "", f"- case 边界：{summary['boundary']['case_boundary_count']}", f"- lot 边界：{summary['boundary']['lot_boundary_count']}", "- 市场闸门关闭、无 BUY、非真实 SELL、数据缺口等样本不得混入主成功率 / 主收益率。", "", ] ), encoding="utf-8", ) if not case_summary.empty: case_summary = case_summary.copy() case_summary["primary_strict_closed_case_flag"] = case_summary.case_id.astype(str).map(primary_flag_by_case).fillna(0).astype(int) case_summary["primary_strict_closed_case_reason"] = case_summary.case_id.astype(str).map(primary_reason_by_case).fillna("CASE_NOT_IN_FULL_SCOPE") case_summary["return_stat_scope"] = case_summary["primary_strict_closed_case_flag"].map( {1: "PRIMARY_STRICT_CLOSED_CASE", 0: "RETURN_STAT_HELD_BOUNDARY_TABLE"} ) case_summary["return_stat_ready_global_flag"] = 0 write_csv(case_summary, ROOT / "case_summary.csv") return case_summary def run_self_check( case_index: pd.DataFrame, batch_index: pd.DataFrame, selected: pd.DataFrame, candidate_diff: pd.DataFrame, decisions: pd.DataFrame, sell_decisions: pd.DataFrame, exit_resolution: pd.DataFrame, orders: pd.DataFrame, lots: pd.DataFrame, account: pd.DataFrame, case_summary: pd.DataFrame, image_manifest: pd.DataFrame, ) -> dict: checks: list[dict] = [] action_counts = role_counts(orders, "action") lot_counts = role_counts(lots, "lot_status") buy_orders = int(action_counts.get("BUY", 0)) sell_orders = int(action_counts.get("SELL", 0)) closed = lots[lots.lot_status == "CLOSED_BY_AI_SELL"].copy() unresolved = lots[lots.lot_status != "CLOSED_BY_AI_SELL"].copy() check( "FULL_CASE_SCOPE_FROZEN", len(case_index) == 743 and case_index.entry_trade_date.nunique() == 743 and len(batch_index) == 15, f"cases={len(case_index)}, unique_dates={case_index.entry_trade_date.nunique()}, batches={len(batch_index)}", checks, ) per_case_counts = selected.groupby("case_id").size() check( "FULL_SELECTED_CANDIDATE_ROWS", len(selected) == 3715 and selected.entry_trade_date.nunique() == 743 and per_case_counts.min() == 5 and per_case_counts.max() == 5, f"selected={len(selected)}, entry_dates={selected.entry_trade_date.nunique()}, per_case_min={per_case_counts.min()}, per_case_max={per_case_counts.max()}", checks, ) check( "FULL_CANDIDATE_POOL_DIFF_PASS", not candidate_diff.empty and (candidate_diff.status == "PASS").all(), candidate_diff.to_dict("records").__repr__(), checks, ) closed_gate_cases = case_index[case_index.market_gate_status == "NO_TRADE_MARKET_GATE_CLOSED"].case_id.astype(str).tolist() closed_gate_decisions = decisions[(decisions.case_id.isin(closed_gate_cases)) & (decisions.action_status == "NO_TRADE_MARKET_GATE_CLOSED")] closed_gate_orders = orders[orders.case_id.isin(closed_gate_cases)] check( "MARKET_GATE_CLOSED_NO_TRADE", len(closed_gate_cases) == 476 and len(closed_gate_decisions) == 476 * 5 and closed_gate_orders.empty, f"closed_gate_cases={len(closed_gate_cases)}, no_trade_decisions={len(closed_gate_decisions)}, closed_gate_orders={len(closed_gate_orders)}", checks, ) entry_decisions = decisions[decisions.decision_stage == "ENTRY_AI_REVIEW"] exit_decisions = decisions[decisions.decision_stage == "EXIT_AI_REVIEW"] check( "DECISION_LOG_ENTRY_EXIT_COUNTS", len(entry_decisions) == len(selected) and len(exit_decisions) == len(sell_decisions) == len(lots), f"entry={len(entry_decisions)}, selected={len(selected)}, exit={len(exit_decisions)}, sell_decisions={len(sell_decisions)}, lots={len(lots)}", checks, ) check( "ORDERS_AND_LOTS_MATCH", buy_orders == len(lots) and sell_orders == len(closed) and len(orders) == buy_orders + sell_orders, f"orders={action_counts}; lots={len(lots)}; closed={len(closed)}", checks, ) check( "UNRESOLVED_LOTS_EXPLICIT", set(unresolved.lot_status).issubset(ALLOWED_UNRESOLVED), f"unresolved={role_counts(unresolved, 'lot_status')}", checks, ) check( "EXIT_RESOLUTION_ALL_LOTS_EXPLICIT", len(exit_resolution) == len(lots) and not exit_resolution.final_action_status.isna().any(), f"exit_resolution={len(exit_resolution)}, statuses={role_counts(exit_resolution, 'final_action_status')}", checks, ) sell_orders_df = orders[orders.action == "SELL"].copy() merged = sell_orders_df.merge(lots, left_on="source_lot_id", right_on="trade_lot_id", suffixes=("_order", "_lot")) t1_ok = True if not merged.empty: t1_ok = ( (pd.to_datetime(merged.trade_date) >= pd.to_datetime(merged.sellable_from_trade_date)).all() and (pd.to_datetime(merged.trade_date) > pd.to_datetime(merged.entry_trade_date)).all() ) check("T1_GUARD_FOR_SELL_ORDERS", bool(t1_ok), f"sell_orders={len(sell_orders_df)}", checks) lookahead_ok = True for df in [orders, decisions, sell_decisions]: if "lookahead_violation_flag" in df.columns: lookahead_ok = lookahead_ok and not df.lookahead_violation_flag.astype(str).str.lower().isin(["true", "1", "yes"]).any() check("LOOKAHEAD_FLAGS_ZERO", bool(lookahead_ok), "orders/decision/sell_decision checked", checks) lot_return_ok = True for _, lot in closed.iterrows(): entry = float(lot.entry_price) exit_price = float(lot.exit_price) position = float(lot.position_pct) lot_return = exit_price / entry - 1.0 contribution = lot_return * position lot_return_ok = lot_return_ok and abs(lot_return - float(lot.lot_return_pct)) < 1e-6 lot_return_ok = lot_return_ok and abs(contribution - float(lot.account_return_contribution_pct)) < 1e-6 check("LOT_RETURN_RECOMPUTE", bool(lot_return_ok), f"closed_lots={len(closed)}", checks) case_ok = True for _, row in case_summary.iterrows(): group = lots[lots.case_id == row.case_id] closed_count = int((group.lot_status == "CLOSED_BY_AI_SELL").sum()) unresolved_count = int((group.lot_status != "CLOSED_BY_AI_SELL").sum()) contrib = pd.to_numeric(group.account_return_contribution_pct, errors="coerce").fillna(0).sum() case_ok = case_ok and int(row.buy_lot_count) == len(group) case_ok = case_ok and int(row.closed_lot_count) == closed_count case_ok = case_ok and int(row.unresolved_lot_count) == unresolved_count case_ok = case_ok and abs(float(row.account_return_closed_lots) - float(contrib)) < 1e-6 case_ok = case_ok and str(row.strict_baseline_return_ready_flag) in {"0", "False", "false"} check("CASE_SUMMARY_RECOMPUTE", bool(case_ok), f"case_summary_rows={len(case_summary)}", checks) account_direction_ok = True account_nav_ok = True account_final_open_ok = True for case_id, group in account.groupby("case_id", sort=False): prev_cash = 1.0 prev_open = 0.0 for _, event in group.iterrows(): cash = float(event.cash_pct_after_event) open_pos = float(event.open_position_pct_after_event) nav = float(event.account_nav_after_event) account_nav_ok = account_nav_ok and abs(nav - (cash + open_pos)) < 1e-6 if event.action == "BUY": account_direction_ok = account_direction_ok and cash < prev_cash + 1e-12 and open_pos > prev_open - 1e-12 elif event.action == "SELL": account_direction_ok = account_direction_ok and cash > prev_cash - 1e-12 and open_pos < prev_open + 1e-12 and open_pos >= -1e-9 prev_cash = cash prev_open = open_pos expected_open = pd.to_numeric( lots[(lots.case_id == case_id) & (lots.lot_status != "CLOSED_BY_AI_SELL")].position_pct, errors="coerce", ).fillna(0).sum() account_final_open_ok = account_final_open_ok and abs(prev_open - expected_open) < 1e-6 check("ACCOUNT_CASH_POSITION_DIRECTION", bool(account_direction_ok), "BUY cash down/open up; SELL cash up/open down", checks) check("ACCOUNT_NAV_EQUALS_CASH_PLUS_OPEN", bool(account_nav_ok), "nav equals cash + open position after each event", checks) check("ACCOUNT_FINAL_OPEN_MATCHES_UNCLOSED_LOTS", bool(account_final_open_ok), "final open position equals unresolved lot position per case", checks) chart_rows: list[dict] = [] image_hash_ok = True for _, row in image_manifest.iterrows(): path = ROOT / str(row.path) exists = path.exists() actual = sha256_file(path) if exists else "" match = exists and actual == str(row.sha256) image_hash_ok = image_hash_ok and match chart_rows.append( { "case_id": row.case_id, "symbol": row.symbol, "chart_role": row.chart_role, "path": row.path, "exists": str(exists), "sha256_match": str(match), } ) write_csv(pd.DataFrame(chart_rows), ROOT / "chart_evidence_audit.csv") image_counts = role_counts(image_manifest, "chart_role") open_selected_count = int((selected.market_gate_status == "MKT_GATE_OPEN_PREV_DAY_UP_3000").sum()) expected_image_counts = { "candidate_daily_100d_decision_view": len(selected), "entry_1m_morning_review_view": open_selected_count, "entry_1m_late_review_view": open_selected_count, "entry_1m_buy_decision_view": buy_orders, "exit_daily_signal_review_view": len(lots), "exit_1m_sell_decision_view": sell_orders, } check("IMAGE_MANIFEST_HASH_MATCH", bool(image_hash_ok), f"images={len(image_manifest)}", checks) check( "IMAGE_ROLE_COUNTS_EXPECTED", image_counts == expected_image_counts and len(image_manifest) == sum(expected_image_counts.values()), f"actual={image_counts}; expected={expected_image_counts}", checks, ) board_paths = [ ROOT / "case_image_board.md", ROOT / "case_story_board.md", *sorted((ROOT / "batches").glob("*/case_image_board.md")), *sorted((ROOT / "batches").glob("*/case_story_board.md")), *sorted((ROOT / "cases").glob("*/case_image_board.md")), *sorted((ROOT / "cases").glob("*/case_story_board.md")), ] link_rows: list[dict] = [] links_ok = True for board in board_paths: if not board.exists(): links_ok = False link_rows.append({"board": board.relative_to(ROOT).as_posix(), "target": "", "exists": "False"}) continue for target in markdown_links(board): exists = target.exists() links_ok = links_ok and exists try: rel_target = target.relative_to(ROOT).as_posix() except ValueError: rel_target = str(target) link_rows.append({"board": board.relative_to(ROOT).as_posix(), "target": rel_target, "exists": str(exists)}) write_csv(pd.DataFrame(link_rows), ROOT / "link_evidence_audit.csv") check("MARKDOWN_LOCAL_LINKS_REACHABLE", bool(links_ok), f"boards={len(board_paths)}, links={len(link_rows)}", checks) stale_terms = [ "EXPAND_30_EXECUTION_SELF_CHECK_DONE", "7 个小样本", "30 案例日受控扩样", "扩样执行审核 HELD", "当前仍为结构试点", "当前执行审核未提交", ] status_ok = True stale_hits: list[str] = [] required_terms = [RUN_ID, STAGE, "743 个", "15 个批次", "RETURN_STAT_READY=false", "复审通过"] for board in [ROOT / "case_image_board.md", ROOT / "case_story_board.md"]: text = board.read_text(encoding="utf-8") for term in stale_terms: if term in text: status_ok = False stale_hits.append(f"{board.name}:{term}") for term in required_terms: if term not in text: status_ok = False stale_hits.append(f"{board.name}:missing:{term}") check("BOARD_STATUS_MATCHES_FULL_SUMMARY", bool(status_ok), "|".join(stale_hits) or "root boards match full-stage status", checks) batch_dirs = sorted((ROOT / "batches").glob("B*")) batch_manifest_ok = len(batch_dirs) == 15 and all((p / "manifest.json").exists() for p in batch_dirs) batch_case_total = sum(len(pd.read_csv(p / "case_index.csv", encoding="utf-8-sig")) for p in batch_dirs if (p / "case_index.csv").exists()) batch_selected_total = sum(len(pd.read_csv(p / "selected_candidate_ledger.csv", encoding="utf-8-sig")) for p in batch_dirs if (p / "selected_candidate_ledger.csv").exists()) check( "BATCH_PACKAGE_COVERAGE", batch_manifest_ok and batch_case_total == len(case_index) and batch_selected_total == len(selected), f"batch_dirs={len(batch_dirs)}, batch_case_total={batch_case_total}, batch_selected_total={batch_selected_total}", checks, ) expected_return_files = [ ROOT / "full_return_stat_case_scope.csv", ROOT / "full_return_stat_lot_scope.csv", ROOT / "full_return_stat_boundary_table.csv", ROOT / "full_return_stat_summary.md", ROOT / "full_return_stat_summary.json", ] return_files_ok = all(path.exists() and path.stat().st_size > 0 for path in expected_return_files) check( "FULL_RETURN_STAT_FILES_PRESENT", bool(return_files_ok), "; ".join(f"{path.name}:{path.exists()}:{path.stat().st_size if path.exists() else 0}" for path in expected_return_files), checks, ) case_scope = read_csv("full_return_stat_case_scope.csv") lot_scope = read_csv("full_return_stat_lot_scope.csv") boundary_table = read_csv("full_return_stat_boundary_table.csv") return_summary = read_json("full_return_stat_summary.json") case_scope_ok = ( not case_scope.empty and len(case_scope) == len(case_index) == 743 and case_scope.case_id.nunique() == 743 and (case_scope.coverage_scope == "ALL_ENTRY_DATE_COVERAGE").all() ) check( "FULL_RETURN_STAT_CASE_SCOPE_COVERAGE", bool(case_scope_ok), f"case_scope_rows={len(case_scope)}, unique_cases={case_scope.case_id.nunique() if not case_scope.empty else 0}", checks, ) primary_count = int(case_scope.primary_strict_closed_case_flag.sum()) if not case_scope.empty else 0 primary_case_ids = set(case_scope[case_scope.primary_strict_closed_case_flag == 1].case_id.astype(str)) if not case_scope.empty else set() primary_lots = lots[lots.case_id.astype(str).isin(primary_case_ids)].copy() if primary_case_ids else pd.DataFrame() primary_boundary_lots = primary_lots[primary_lots.lot_status != "CLOSED_BY_AI_SELL"] if not primary_lots.empty else pd.DataFrame() primary_summary_ok = primary_count == 234 and primary_boundary_lots.empty check( "PRIMARY_STRICT_CLOSED_CASE_SCOPE_VALID", bool(primary_summary_ok), f"primary_cases={primary_count}, primary_boundary_lots={len(primary_boundary_lots)}", checks, ) lot_scope_ok = ( not lot_scope.empty and len(lot_scope) == len(lots) and int(lot_scope.closed_lot_recalc_included_flag.sum()) == len(closed) and int((lot_scope.closed_lot_recalc_included_flag == 0).sum()) == len(unresolved) ) check( "FULL_RETURN_STAT_LOT_SCOPE_COMPLETE", bool(lot_scope_ok), f"lot_scope_rows={len(lot_scope)}, lots={len(lots)}, closed={len(closed)}, unresolved={len(unresolved)}", checks, ) case_boundary_count = int((boundary_table.boundary_level == "CASE").sum()) if not boundary_table.empty else 0 lot_boundary_count = int((boundary_table.boundary_level == "LOT").sum()) if not boundary_table.empty else 0 boundary_ok = ( not boundary_table.empty and case_boundary_count == len(case_index) - primary_count and lot_boundary_count == len(unresolved) and boundary_table.excluded_from_primary_flag.astype(str).isin(["1", "True", "true"]).all() ) check( "FULL_RETURN_STAT_BOUNDARY_TABLE_COMPLETE", bool(boundary_ok), f"case_boundaries={case_boundary_count}, expected_case_boundaries={len(case_index) - primary_count}, lot_boundaries={lot_boundary_count}, expected_lot_boundaries={len(unresolved)}", checks, ) return_summary_ok = False if return_summary: return_summary_ok = ( return_summary.get("return_stat_ready") is False and int(return_summary.get("primary_scope", {}).get("case_count", -1)) == primary_count and int(return_summary.get("coverage_scope", {}).get("entry_date_count", -1)) == len(case_index) and int(return_summary.get("lot_recalc_scope", {}).get("closed_lot_count", -1)) == len(closed) and int(return_summary.get("lot_recalc_scope", {}).get("unresolved_lot_count", -1)) == len(unresolved) ) check( "FULL_RETURN_STAT_SUMMARY_CONSISTENT", bool(return_summary_ok), f"summary_primary={return_summary.get('primary_scope', {}).get('case_count') if return_summary else ''}, primary_cases={primary_count}", checks, ) case_summary_primary_ok = ( "primary_strict_closed_case_flag" in case_summary.columns and int(case_summary.primary_strict_closed_case_flag.sum()) == primary_count and "return_stat_scope" in case_summary.columns ) check( "CASE_SUMMARY_PRIMARY_SCOPE_FLAG_PRESENT", bool(case_summary_primary_ok), f"case_summary_primary={int(case_summary.primary_strict_closed_case_flag.sum()) if 'primary_strict_closed_case_flag' in case_summary.columns else 'missing'}, scope_column={'return_stat_scope' in case_summary.columns}", checks, ) stale_scope_text = "当前收益口径：STRUCTURE_PILOT_AI_SELL_REVIEWED__EXEC_AUDIT_PENDING__NOT_RETURN_STAT_READY" case_board_scope_ok = not case_scope.empty case_board_scope_errors: list[str] = [] for _, scope in case_scope.iterrows(): case_id = str(scope.case_id) expected_scope = "PRIMARY_STRICT_CLOSED_CASE" if fmt_int(scope.get("primary_strict_closed_case_flag")) == "1" else "RETURN_STAT_HELD_BOUNDARY_TABLE" expected_status = clean(scope.get("case_scope_status")) expected_boundary = clean(scope.get("boundary_category")) for board_name in ["case_image_board.md", "case_story_board.md"]: board = ROOT / "cases" / case_id / board_name if not board.exists(): case_board_scope_ok = False if len(case_board_scope_errors) < 12: case_board_scope_errors.append(f"{case_id}/{board_name}:missing") continue text = board.read_text(encoding="utf-8") if stale_scope_text in text: case_board_scope_ok = False if len(case_board_scope_errors) < 12: case_board_scope_errors.append(f"{case_id}/{board_name}:stale_scope_text") if f"当前收益口径：`{expected_scope}`" not in text: case_board_scope_ok = False if len(case_board_scope_errors) < 12: case_board_scope_errors.append(f"{case_id}/{board_name}:scope_mismatch:{expected_scope}") if expected_status and f"收益口径状态：`{expected_status}`" not in text: case_board_scope_ok = False if len(case_board_scope_errors) < 12: case_board_scope_errors.append(f"{case_id}/{board_name}:status_mismatch:{expected_status}") if expected_scope != "PRIMARY_STRICT_CLOSED_CASE" and expected_boundary and expected_boundary not in text: case_board_scope_ok = False if len(case_board_scope_errors) < 12: case_board_scope_errors.append(f"{case_id}/{board_name}:boundary_missing:{expected_boundary}") check( "CASE_BOARD_RETURN_SCOPE_MATCHES_SCOPE_TABLE", bool(case_board_scope_ok), "case_boards=743 image + 743 story; " + (";".join(case_board_scope_errors) if case_board_scope_errors else "all match full_return_stat_case_scope.csv"), checks, ) check( "RETURN_STAT_READY_FALSE", RETURN_READY is False and (case_summary.strict_baseline_return_ready_flag.astype(str).isin(["0", "False", "false"])).all(), "full execution package remains not RETURN_STAT_READY before execution audit", checks, ) checks_df = pd.DataFrame(checks) write_csv(checks_df, ROOT / "self_check_items.csv") fail_count = int((checks_df.status == "FAIL").sum()) self_check = { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": now_iso(), "stage": STAGE, "overall_status": "PASS_FOR_FULL_2023_2026_EXECUTION_REREVIEW_READY" if fail_count == 0 else "FAIL", "check_count": int(len(checks_df)), "fail_count": fail_count, "case_count": int(len(case_index)), "batch_count": int(len(batch_index)), "selected_candidate_rows": int(len(selected)), "order_counts": action_counts, "lot_status_counts": lot_counts, "image_role_counts": image_counts, "strict_baseline_return_ready_flag": False, "boundary": "Machine self-check for full 2023-2026 batch execution package after execution re-review pass. RETURN_STAT_READY remains false and readouts require layered full_return_stat_* scope citation.", } (ROOT / "self_check.json").write_text(json.dumps(self_check, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") (ROOT / "self_check.md").write_text( "\n".join( [ "# full self_check", "", f"- run：`{RUN_ID}`", f"- 阶段：`{STAGE}`", f"- 状态：`{self_check['overall_status']}`", f"- 检查项：{self_check['check_count']}，失败：{self_check['fail_count']}", f"- 案例日：{len(case_index)}；批次：{len(batch_index)}；候选：{len(selected)}", f"- BUY：{buy_orders}；SELL：{sell_orders}；边界 lot：{len(unresolved)}", "- `RETURN_STAT_READY=false`", "", "边界：执行复审已通过，但 `RETURN_STAT_READY=false`；完整读数必须按 `full_return_stat_*` 分层口径引用。", "", ] ), encoding="utf-8", ) return self_check def write_summary_and_manifest( case_index: pd.DataFrame, batch_index: pd.DataFrame, selected: pd.DataFrame, decisions: pd.DataFrame, orders: pd.DataFrame, lots: pd.DataFrame, case_summary: pd.DataFrame, image_manifest: pd.DataFrame, self_check: dict, ) -> None: order_counts = role_counts(orders, "action") lot_counts = role_counts(lots, "lot_status") return_stat_summary = read_json("full_return_stat_summary.json") if "primary_strict_closed_case_flag" in case_summary.columns: closed_cases = case_summary[case_summary.primary_strict_closed_case_flag == 1].copy() else: closed_cases = case_summary[(case_summary.closed_lot_count > 0) & (case_summary.unresolved_lot_count == 0)].copy() audit_candidate_readouts = { "primary_strict_closed_case_candidate_count": int(len(closed_cases)), "primary_strict_closed_case_positive_count": int((closed_cases.account_return_closed_lots > 0).sum()) if not closed_cases.empty else 0, "primary_strict_closed_case_return_sum_for_audit_only": float(closed_cases.account_return_closed_lots.sum()) if not closed_cases.empty else 0.0, "closed_lot_count_for_recalc_only": int((lots.lot_status == "CLOSED_BY_AI_SELL").sum()), "closed_lot_account_return_sum_for_recalc_only": float(pd.to_numeric(lots.account_return_contribution_pct, errors="coerce").fillna(0).sum()), "not_final_conclusion": True, } summary = { "schema_version": "1.0", "task_id": TASK_ID, "run_id": RUN_ID, "design_id": DESIGN_ID, "design_audit_id": DESIGN_AUDIT_ID, "generated_at": now_iso(), "stage": STAGE, "execution_review_status": "EXECUTION_REREVIEW_PASSED_RETURN_STAT_HELD", "execution_review_original_message_id": EXEC_REVIEW_MESSAGE_ID, "execution_review_held_audit_id": EXEC_HELD_AUDIT_ID, "execution_review_held_message_id": EXEC_HELD_MESSAGE_ID, "execution_rereview_held_audit_id": EXEC_REREVIEW_HELD_AUDIT_ID, "execution_rereview_held_message_id": EXEC_REREVIEW_HELD_MESSAGE_ID, "execution_rereview_pass_audit_id": EXEC_REREVIEW_PASS_AUDIT_ID, "execution_rereview_pass_message_id": EXEC_REREVIEW_PASS_MESSAGE_ID, "issue_id": RETURN_SCOPE_ISSUE_ID, "strict_baseline_return_ready_flag": False, "scope": { "case_count": int(len(case_index)), "batch_count": int(len(batch_index)), "selected_candidate_rows": int(len(selected)), "market_gate_open_entry_dates": int((case_index.market_gate_status == "MKT_GATE_OPEN_PREV_DAY_UP_3000").sum()), "market_gate_closed_entry_dates": int((case_index.market_gate_status == "NO_TRADE_MARKET_GATE_CLOSED").sum()), "buy_case_count": int(len(case_summary)), }, "decision_counts": {f"{k[0]}::{k[1]}": int(v) for k, v in decisions.groupby(["decision_stage", "action_status"]).size().to_dict().items()}, "trade_ledger": { "order_counts": order_counts, "lot_status_counts": lot_counts, "case_summary_rows": int(len(case_summary)), }, "image_package": { "image_count": int(len(image_manifest)), "role_counts": role_counts(image_manifest, "chart_role"), }, "full_return_stat": { "case_scope": "full_return_stat_case_scope.csv", "lot_scope": "full_return_stat_lot_scope.csv", "boundary_table": "full_return_stat_boundary_table.csv", "summary": "full_return_stat_summary.json", "summary_readouts": return_stat_summary, }, "audit_candidate_readouts_not_final": audit_candidate_readouts, "self_check": { "overall_status": self_check["overall_status"], "check_count": int(self_check["check_count"]), "fail_count": int(self_check["fail_count"]), "evidence": "self_check.json", }, "boundary": ( "Full 2023-2026 batch execution package passed execution re-review after return-scope repairs. " "RETURN_STAT_READY remains false; success, return, win-rate, drawdown, and effectiveness readouts must be cited with explicit full_return_stat_* scopes and boundaries." ), } (ROOT / "summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") (ROOT / "summary.md").write_text( "\n".join( [ f"# {RUN_ID} summary", "", f"- 案例事项：`{TASK_ID}`", f"- 设计：`{DESIGN_ID}`", f"- 设计审核：`{DESIGN_AUDIT_ID}`", f"- 阶段：`{STAGE}`", "- 执行审核状态：`EXECUTION_REREVIEW_PASSED_RETURN_STAT_HELD`", f"- 原执行送审消息：`{EXEC_REVIEW_MESSAGE_ID}`", f"- 原 HELD 审计：`{EXEC_HELD_AUDIT_ID}`", f"- 执行复审 HELD：`{EXEC_REREVIEW_HELD_AUDIT_ID}`", f"- 执行复审通过：`{EXEC_REREVIEW_PASS_AUDIT_ID}`", f"- 返修问题：`{RETURN_SCOPE_ISSUE_ID}`", "- `RETURN_STAT_READY=false`", "", "## 范围", "", f"- entry_trade_date：{len(case_index)} 个", f"- 批次：{len(batch_index)} 个", f"- 选中候选：{len(selected)} 条", f"- 市场闸门打开：{summary['scope']['market_gate_open_entry_dates']} 个；关闭：{summary['scope']['market_gate_closed_entry_dates']} 个", "", "## 执行读数（送审候选，不是最终结论）", "", f"- BUY：{order_counts.get('BUY', 0)}；SELL：{order_counts.get('SELL', 0)}", f"- lot 状态：{lot_counts}", f"- 图片：{len(image_manifest)} 张", f"- 自检：{self_check['check_count']} 项，失败 {self_check['fail_count']} 项", "- 收益分层产物：`full_return_stat_case_scope.csv`、`full_return_stat_lot_scope.csv`、`full_return_stat_boundary_table.csv`、`full_return_stat_summary.json/md`", "", "## 边界", "", "- 当前包已通过全量执行返修复审，但 `RETURN_STAT_READY=false` 继续保留。", "- 完整 2023-2026 baseline 成功率、收益率、胜率、回撤或策略有效性读数必须按 `full_return_stat_*` 分层口径和边界引用。", "- 非真实 SELL、窗口末估值和数据缺口 lot 均保持边界，不强行补结论。", "", ] ), encoding="utf-8", ) manifest = manifest_for_dir(ROOT) manifest.update( { "manifest_stage": STAGE, "overall_status": self_check["overall_status"], "strict_baseline_return_ready_flag": False, "notes": "manifest.json excludes itself for stable hashing. Full execution return-scope repair passed re-review; RETURN_STAT_READY remains false.", } ) (ROOT / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def main() -> None: case_index = read_csv("case_index.csv") batch_index = read_csv("full_batch_index.csv") selected = read_csv("selected_candidate_ledger.csv") candidate_diff = read_csv("full_candidate_pool_diff.csv") decisions = read_csv("decision_log.csv") sell_decisions = read_csv("sell_decision_log.csv") exit_resolution = read_csv("exit_resolution_log.csv") orders = read_csv("order_ledger.csv") lots = read_csv("position_lot_ledger.csv") account = read_csv("daily_account_ledger.csv") case_summary = read_csv("case_summary.csv") image_manifest = read_csv("image_manifest.csv") case_summary = build_return_stat_artifacts(case_index, decisions, lots, case_summary) write_case_boards(case_index, selected, decisions, lots, case_summary, image_manifest) write_root_and_batch_boards(case_index, batch_index, selected, decisions, orders, lots, case_summary, image_manifest) self_check = run_self_check( case_index, batch_index, selected, candidate_diff, decisions, sell_decisions, exit_resolution, orders, lots, account, case_summary, image_manifest, ) write_summary_and_manifest(case_index, batch_index, selected, decisions, orders, lots, case_summary, image_manifest, self_check) if __name__ == "__main__": main()