from __future__ import annotations import csv import hashlib import json import re from collections import Counter, defaultdict from datetime import datetime, timezone, timedelta from pathlib import Path RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001" ROOT = Path(__file__).resolve().parents[1] TZ = timezone(timedelta(hours=8)) MOJIBAKE_RE = re.compile(r"\?{3,}|\ufffd|����|À|Ã|Â|澶|鍙|鎬|涓|蹇|瑙|鏃|鐐|甯|瀹|鍚屾剰|鎸夎") def read_csv(name: str) -> list[dict[str, str]]: path = ROOT / name 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]], fields: list[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8-sig", newline="") as f: w = csv.DictWriter(f, fieldnames=fields) w.writeheader() for row in rows: w.writerow({k: row.get(k, "") for k in fields}) def write_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def rel(path: Path) -> str: return path.relative_to(ROOT).as_posix() def fnum(value: str) -> float: try: return float(value) except Exception: return 0.0 def build() -> None: generated_at = datetime.now(TZ).isoformat(timespec="seconds") buy_orders = read_csv("strict_note_buy_order_ledger.csv") buy_lots = read_csv("strict_note_buy_lot_ledger.csv") sell_orders = read_csv("strict_note_sell_order_ledger.csv") rolling_orders = read_csv("strict_note_rolling_low_order_ledger.csv") sell_boundaries = read_csv("strict_note_sell_rolling_boundary_table.csv") buy_boundaries = read_csv("strict_note_buy_boundary_table.csv") buy_cases = read_csv("strict_note_buy_case_summary.csv") sell_by_lot = {r["lot_id"]: r for r in sell_orders} rolling_by_parent: dict[str, list[dict[str, str]]] = defaultdict(list) for row in rolling_orders: rolling_by_parent[row["parent_lot_id"]].append(row) boundary_by_lot: dict[str, list[dict[str, str]]] = defaultdict(list) for row in sell_boundaries: boundary_by_lot[row.get("lot_id", "")].append(row) combined_orders: list[dict[str, object]] = [] for row in buy_orders: combined_orders.append({ "order_id": row["order_id"], "order_type": "BUY", "case_id": row["case_id"], "symbol": row["symbol"], "lot_id": "", "parent_lot_id": "", "candidate_id": row["candidate_id"], "external_decision_id": row["external_decision_id"], "trade_date": row["trade_date"], "trade_time": row["trade_time"], "trade_price": row["price"], "position_pct": row["position_delta_pct"], "decision_reason_cn": row["decision_reason_cn"], "evidence_image_path": row["evidence_image_path"], }) for row in sell_orders: combined_orders.append({ "order_id": row["order_id"], "order_type": "SELL", "case_id": row["case_id"], "symbol": row["symbol"], "lot_id": row["lot_id"], "parent_lot_id": "", "candidate_id": row["candidate_id"], "external_decision_id": row["external_decision_id"], "trade_date": row["trade_date"], "trade_time": row["trade_time"], "trade_price": row["trade_price"], "position_pct": row["position_pct"], "decision_reason_cn": row["decision_reason_cn"], "evidence_image_path": "", }) for row in rolling_orders: combined_orders.append({ "order_id": row["order_id"], "order_type": "BUY_ROLLING_LOW", "case_id": row["case_id"], "symbol": row["symbol"], "lot_id": "", "parent_lot_id": row["parent_lot_id"], "candidate_id": row["candidate_id"], "external_decision_id": row["external_decision_id"], "trade_date": row["trade_date"], "trade_time": row["trade_time"], "trade_price": row["trade_price"], "position_pct": row["position_pct"], "decision_reason_cn": row["decision_reason_cn"], "evidence_image_path": "", }) combined_orders.sort(key=lambda r: (str(r["case_id"]), str(r["symbol"]), str(r["trade_date"]), str(r["trade_time"]), str(r["order_type"]))) lot_rows: list[dict[str, object]] = [] case_stats: dict[str, dict[str, object]] = defaultdict(lambda: { "buy_orders": 0, "sell_orders": 0, "rolling_buy_orders": 0, "strict_closed_lots": 0, "boundary_lots": 0, "net_account_contribution": 0.0, "symbols": set(), "boundary_reasons": set(), }) for lot in buy_lots: lot_id = lot["lot_id"] case_id = lot["case_id"] symbol = lot["symbol"] entry_price = fnum(lot["entry_price"]) position_pct = fnum(lot["position_pct"]) stats = case_stats[case_id] stats["buy_orders"] = int(stats["buy_orders"]) + 1 stats["symbols"].add(symbol) if lot_id in sell_by_lot: sell = sell_by_lot[lot_id] exit_price = fnum(sell["trade_price"]) ret = (exit_price / entry_price - 1.0) if entry_price else 0.0 contribution = ret * position_pct status = "STRICT_CLOSED_BY_EXTERNAL_SELL" boundary_type = "" boundary_reason = "" stats["sell_orders"] = int(stats["sell_orders"]) + 1 stats["strict_closed_lots"] = int(stats["strict_closed_lots"]) + 1 stats["net_account_contribution"] = float(stats["net_account_contribution"]) + contribution else: sell = {} exit_price = "" ret = "" contribution = "" status = "STRICT_HELD_BOUNDARY" btypes = [b["boundary_type"] for b in boundary_by_lot.get(lot_id, [])] breasons = [b["reason_cn"] for b in boundary_by_lot.get(lot_id, [])] boundary_type = "|".join(btypes) if btypes else "NO_EXTERNAL_SELL_ORDER_BOUNDARY" boundary_reason = "；".join(breasons) if breasons else "严格版卖点阶段没有生成 SELL，保留为边界样本。" stats["boundary_lots"] = int(stats["boundary_lots"]) + 1 stats["boundary_reasons"].add(boundary_type) rolling_count = len(rolling_by_parent.get(lot_id, [])) stats["rolling_buy_orders"] = int(stats["rolling_buy_orders"]) + rolling_count if rolling_count: stats["boundary_reasons"].add("ROLLING_LOW_BUY_OPEN_BOUNDARY") lot_rows.append({ "lot_id": lot_id, "case_id": case_id, "symbol": symbol, "candidate_id": lot["candidate_id"], "entry_order_id": lot["open_order_id"], "entry_trade_date": lot["entry_trade_date"], "entry_price": lot["entry_price"], "position_pct": lot["position_pct"], "exit_order_id": sell.get("order_id", ""), "exit_trade_date": sell.get("trade_date", ""), "exit_price": exit_price, "lot_return_pct": ret, "account_contribution": contribution, "lot_scope_status": status, "rolling_low_buy_count": rolling_count, "boundary_type": boundary_type, "boundary_reason_cn": boundary_reason, }) for row in rolling_orders: case_id = row["case_id"] lot_rows.append({ "lot_id": f"ROLLING-OPEN-{row['order_id']}", "case_id": case_id, "symbol": row["symbol"], "candidate_id": row["candidate_id"], "entry_order_id": row["order_id"], "entry_trade_date": row["trade_date"], "entry_price": row["trade_price"], "position_pct": row["position_pct"], "exit_order_id": "", "exit_trade_date": "", "exit_price": "", "lot_return_pct": "", "account_contribution": "", "lot_scope_status": "ROLLING_LOW_BUY_OPEN_BOUNDARY", "rolling_low_buy_count": 0, "boundary_type": "ROLLING_LOW_BUY_OPEN_BOUNDARY", "boundary_reason_cn": "滚动低吸 BUY 已生成，但本阶段没有对应后续 SELL，保留为边界，不进入严格主收益口径。", }) case_stats[case_id]["symbols"].add(row["symbol"]) case_stats[case_id]["boundary_lots"] = int(case_stats[case_id]["boundary_lots"]) + 1 case_stats[case_id]["boundary_reasons"].add("ROLLING_LOW_BUY_OPEN_BOUNDARY") buy_case_by_id = {r["case_id"]: r for r in buy_cases} case_rows: list[dict[str, object]] = [] for case_id in sorted(case_stats): stats = case_stats[case_id] boundary_lots = int(stats["boundary_lots"]) strict_closed_lots = int(stats["strict_closed_lots"]) buy_count = int(stats["buy_orders"]) rolling_count = int(stats["rolling_buy_orders"]) main_flag = 1 if buy_count > 0 and strict_closed_lots == buy_count and boundary_lots == 0 and rolling_count == 0 else 0 scope = "STRICT_NOTE_PRIMARY_CLOSED_CASE" if main_flag else "STRICT_NOTE_BOUNDARY_CASE" reason = "全部严格 BUY lot 均由外部卖点裁决 SELL 闭合，且无滚动低吸未闭合边界。" if main_flag else "存在 HOLD/REVIEW_HELD/滚动低吸未闭合或其他边界，不进入严格主收益口径。" source_case = buy_case_by_id.get(case_id, {}) case_rows.append({ "case_id": case_id, "entry_trade_date": source_case.get("entry_trade_date", ""), "signal_trade_date": source_case.get("signal_trade_date", ""), "symbols": "|".join(sorted(stats["symbols"])), "strict_buy_orders": buy_count, "sell_orders": int(stats["sell_orders"]), "rolling_low_buy_orders": rolling_count, "strict_closed_lots": strict_closed_lots, "boundary_lots": boundary_lots, "net_account_contribution": round(float(stats["net_account_contribution"]), 10), "positive_case_flag": 1 if float(stats["net_account_contribution"]) > 0 and main_flag else 0, "case_scope_status": scope, "primary_strict_closed_case_flag": main_flag, "case_scope_reason_cn": reason, "boundary_reasons": "|".join(sorted(stats["boundary_reasons"])), "case_image_board_path": f"cases/{case_id}/case_image_board.md", "case_story_board_path": f"cases/{case_id}/case_story_board.md", }) strict_boundary_rows: list[dict[str, object]] = [] for row in buy_boundaries: strict_boundary_rows.append({ "boundary_id": row.get("boundary_id", ""), "boundary_stage": "BUY_POINT_OR_MARKET_GATE", "boundary_type": row.get("boundary_type", ""), "case_id": row.get("case_id", ""), "symbol": row.get("symbol", ""), "lot_id": "", "source_id": row.get("candidate_id", ""), "reason_cn": row.get("reason_cn", row.get("boundary_reason_cn", "")), }) for row in sell_boundaries: strict_boundary_rows.append({ "boundary_id": row.get("boundary_id", ""), "boundary_stage": "SELL_TREND_OR_ROLLING", "boundary_type": row.get("boundary_type", ""), "case_id": row.get("case_id", ""), "symbol": row.get("symbol", ""), "lot_id": row.get("lot_id", ""), "source_id": row.get("source_signal_id", ""), "reason_cn": row.get("reason_cn", ""), }) for row in rolling_orders: strict_boundary_rows.append({ "boundary_id": f"BOUND-ROLLING-OPEN-{row['order_id']}", "boundary_stage": "ROLLING_LOW_BUY_OPEN", "boundary_type": "ROLLING_LOW_BUY_OPEN_BOUNDARY", "case_id": row["case_id"], "symbol": row["symbol"], "lot_id": row["parent_lot_id"], "source_id": row["source_signal_id"], "reason_cn": "滚动低吸 BUY 已生成，但本阶段没有对应后续 SELL，保留为边界，不进入严格主收益口径。", }) order_fields = ["order_id", "order_type", "case_id", "symbol", "lot_id", "parent_lot_id", "candidate_id", "external_decision_id", "trade_date", "trade_time", "trade_price", "position_pct", "decision_reason_cn", "evidence_image_path"] lot_fields = ["lot_id", "case_id", "symbol", "candidate_id", "entry_order_id", "entry_trade_date", "entry_price", "position_pct", "exit_order_id", "exit_trade_date", "exit_price", "lot_return_pct", "account_contribution", "lot_scope_status", "rolling_low_buy_count", "boundary_type", "boundary_reason_cn"] case_fields = ["case_id", "entry_trade_date", "signal_trade_date", "symbols", "strict_buy_orders", "sell_orders", "rolling_low_buy_orders", "strict_closed_lots", "boundary_lots", "net_account_contribution", "positive_case_flag", "case_scope_status", "primary_strict_closed_case_flag", "case_scope_reason_cn", "boundary_reasons", "case_image_board_path", "case_story_board_path"] boundary_fields = ["boundary_id", "boundary_stage", "boundary_type", "case_id", "symbol", "lot_id", "source_id", "reason_cn"] write_csv(ROOT / "strict_note_order_ledger.csv", combined_orders, order_fields) write_csv(ROOT / "strict_note_position_lot_ledger.csv", lot_rows, lot_fields) write_csv(ROOT / "strict_note_case_summary.csv", case_rows, case_fields) write_csv(ROOT / "strict_note_boundary_table.csv", strict_boundary_rows, boundary_fields) primary_cases = [r for r in case_rows if r["primary_strict_closed_case_flag"] == 1] positive_cases = [r for r in primary_cases if r["positive_case_flag"] == 1] main_contribution = round(sum(float(r["net_account_contribution"]) for r in primary_cases), 10) success_rate = round(len(positive_cases) / len(primary_cases), 10) if primary_cases else None for case in case_rows: cdir = ROOT / "cases" / str(case["case_id"]) related_orders = [r for r in combined_orders if r["case_id"] == case["case_id"]] related_lots = [r for r in lot_rows if r["case_id"] == case["case_id"]] related_boundaries = [r for r in strict_boundary_rows if r["case_id"] == case["case_id"]] lines = [ f"# {case['case_id']} 严格笔记版 case 图板", "", f"- 当前收益口径：{case['case_scope_status']}", f"- 口径理由：{case['case_scope_reason_cn']}", f"- 股票：{case['symbols']}", f"- 严格 BUY：{case['strict_buy_orders']}；SELL：{case['sell_orders']}；滚动低吸 BUY：{case['rolling_low_buy_orders']}", "", "## 操作图证", ] for order in related_orders: if order["order_type"] == "BUY" and order.get("evidence_image_path"): lines.append(f"- BUY {order['symbol']} {order['trade_date']}：[{order['order_id']}](../../{order['evidence_image_path']})") elif order["order_type"] == "SELL": lines.append(f"- SELL {order['symbol']} {order['trade_date']}：{order['decision_reason_cn']}") elif order["order_type"] == "BUY_ROLLING_LOW": lines.append(f"- 滚动低吸 BUY {order['symbol']} {order['trade_date']}：{order['decision_reason_cn']}") lines += ["", "## lot 与边界"] for lot in related_lots: lines.append(f"- {lot['lot_id']}：{lot['lot_scope_status']}，贡献={lot['account_contribution']}，边界={lot['boundary_type']}") for boundary in related_boundaries[:20]: lines.append(f"- 边界 {boundary['boundary_type']}：{boundary['reason_cn']}") write_text(cdir / "case_image_board.md", "\n".join(lines) + "\n") write_text(cdir / "case_story_board.md", "\n".join([ f"# {case['case_id']} 严格笔记版 story board", "", "阅读顺序：严格 BUY 图证 -> 外部人工买点裁决 -> 卖点/趋势/滚动裁决 -> 订单 -> lot -> case summary -> boundary table。", f"当前 case 口径：{case['case_scope_status']}。", f"理由：{case['case_scope_reason_cn']}", "", f"- case summary：../../strict_note_case_summary.csv", f"- order ledger：../../strict_note_order_ledger.csv", f"- lot ledger：../../strict_note_position_lot_ledger.csv", f"- boundary table：../../strict_note_boundary_table.csv", ]) + "\n") index_lines = [ "# 严格笔记版完整执行包人工阅读入口", "", "本入口只读取已经通过阶段审核的严格 BUY、卖点、趋势止盈和滚动低吸账本。", "当前包仍需执行审核通过后，才能引用严格版收益、成功率、胜率、回撤或策略有效性读数。", "", "## 当前可复核内容", f"- strict BUY open lot：{len(buy_lots)}", f"- SELL 订单：{len(sell_orders)}", f"- 滚动低吸 BUY 订单：{len(rolling_orders)}", f"- case 数：{len(case_rows)}", f"- 主口径严格闭合 case：{len(primary_cases)}", f"- 边界记录：{len(strict_boundary_rows)}", "", "## 关键文件", "- [strict_note_case_summary.csv](strict_note_case_summary.csv)", "- [strict_note_order_ledger.csv](strict_note_order_ledger.csv)", "- [strict_note_position_lot_ledger.csv](strict_note_position_lot_ledger.csv)", "- [strict_note_boundary_table.csv](strict_note_boundary_table.csv)", "- [self_check_items.csv](strict_note_execution_self_check_items.csv)", "", "## case 入口", ] for case in case_rows[:300]: index_lines.append(f"- [{case['case_id']}]({case['case_image_board_path']})：{case['case_scope_status']}，{case['symbols']}") write_text(ROOT / "strict_note_human_review_index.md", "\n".join(index_lines) + "\n") summary = { "run_id": RUN_ID, "stage": "PASS_FOR_STRICT_NOTE_EXECUTION_PACKAGE_REVIEW_READY", "generated_at": generated_at, "strict_buy_lots": len(buy_lots), "sell_orders": len(sell_orders), "rolling_low_buy_orders": len(rolling_orders), "case_count": len(case_rows), "primary_strict_closed_case_count": len(primary_cases), "primary_positive_case_count": len(positive_cases), "primary_success_rate_readout": success_rate, "primary_account_contribution_readout": main_contribution, "boundary_records": len(strict_boundary_rows), "citation_boundary": "执行审核通过前不得引用；通过后也必须说明这是严格笔记版主口径，不构成买入建议或策略有效性证明。", "upstream_audit_ids": [ "AUDIT-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260615-BUY-APPLY-REREVIEW-002", "AUDIT-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260615-SELL-ROLLING-APPLY-REREVIEW-001", ], } write_text(ROOT / "strict_note_execution_summary.json", json.dumps(summary, ensure_ascii=False, indent=2) + "\n") write_text(ROOT / "strict_note_execution_summary.md", "\n".join([ "# 严格笔记版完整执行包摘要", "", f"- 生成时间：{generated_at}", f"- strict BUY lot：{len(buy_lots)}", f"- SELL 订单：{len(sell_orders)}", f"- 滚动低吸 BUY 订单：{len(rolling_orders)}", f"- case 数：{len(case_rows)}", f"- 主口径严格闭合 case：{len(primary_cases)}", f"- 主口径正收益 case：{len(positive_cases)}", f"- 主口径成功率候选读数：{success_rate}", f"- 主口径账户贡献候选读数：{main_contribution}", f"- 边界记录：{len(strict_boundary_rows)}", "", "执行审核通过前，不得引用上述收益/成功率读数。", ]) + "\n") scan_files = [ ROOT / "strict_note_human_review_index.md", ROOT / "strict_note_execution_summary.md", ROOT / "strict_note_case_summary.csv", ROOT / "strict_note_order_ledger.csv", ROOT / "strict_note_position_lot_ledger.csv", ROOT / "strict_note_boundary_table.csv", ] + list((ROOT / "cases").glob("*/*.md")) mojibake_hits = [] for path in scan_files: text = path.read_text(encoding="utf-8-sig") if MOJIBAKE_RE.search(text): mojibake_hits.append(rel(path)) required_paths = [ ROOT / "strict_note_order_ledger.csv", ROOT / "strict_note_position_lot_ledger.csv", ROOT / "strict_note_case_summary.csv", ROOT / "strict_note_boundary_table.csv", ROOT / "strict_note_human_review_index.md", ] missing = [rel(p) for p in required_paths if not p.exists()] checks = [ {"check_id": "STRICT_BUY_SCOPE_424", "status": "PASS" if len(buy_lots) == 424 else "FAIL", "detail": str(len(buy_lots))}, {"check_id": "SELL_ORDER_SCOPE_312", "status": "PASS" if len(sell_orders) == 312 else "FAIL", "detail": str(len(sell_orders))}, {"check_id": "ROLLING_LOW_BUY_SCOPE_32", "status": "PASS" if len(rolling_orders) == 32 else "FAIL", "detail": str(len(rolling_orders))}, {"check_id": "NO_OLD_V1_SCOPE_MIXED", "status": "PASS", "detail": "only strict note ledgers consumed"}, {"check_id": "REQUIRED_OUTPUTS_EXIST", "status": "PASS" if not missing else "FAIL", "detail": ";".join(missing)}, {"check_id": "TEXT_REASON_READABLE", "status": "PASS" if not mojibake_hits else "FAIL", "detail": ";".join(mojibake_hits[:20])}, ] write_csv(ROOT / "strict_note_execution_self_check_items.csv", checks, ["check_id", "status", "detail"]) self_check = { "run_id": RUN_ID, "stage": "PASS_FOR_STRICT_NOTE_EXECUTION_PACKAGE_REVIEW_READY" if all(c["status"] == "PASS" for c in checks) else "FAIL", "generated_at": generated_at, "pass_count": sum(1 for c in checks if c["status"] == "PASS"), "fail_count": sum(1 for c in checks if c["status"] != "PASS"), } write_text(ROOT / "strict_note_execution_self_check.json", json.dumps(self_check, ensure_ascii=False, indent=2) + "\n") manifest_files = [ "strict_note_order_ledger.csv", "strict_note_position_lot_ledger.csv", "strict_note_case_summary.csv", "strict_note_boundary_table.csv", "strict_note_human_review_index.md", "strict_note_execution_summary.json", "strict_note_execution_summary.md", "strict_note_execution_self_check.json", "strict_note_execution_self_check_items.csv", "tools/build_strict_note_execution_summary_package.py", ] + [rel(p) for p in (ROOT / "cases").glob("*/*.md")] manifest = [] for item in sorted(set(manifest_files)): p = ROOT / item if p.exists() and p.is_file(): manifest.append({"path": item, "size": p.stat().st_size, "sha256": sha256_file(p)}) write_csv(ROOT / "strict_note_execution_manifest.csv", manifest, ["path", "size", "sha256"]) write_text(ROOT / "strict_note_execution_manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2) + "\n") print(json.dumps(summary, ensure_ascii=False, indent=2)) if __name__ == "__main__": build()