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()
|
