from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
import re
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
|
|
FINAL_RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FINAL-CONCLUSION-AFTER-BUY-REPAIR-20260616-001"
|
SOURCE_RUN_ID = "RUN-ANA-WUJI-V1-SELL-ROLLING-REPLAY-AFTER-BUY-REPAIR-20260616-001"
|
SOURCE_EXEC_AUDIT_ID = "AUDIT-ANA-WUJI-V1-SELL-ROLLING-REPLAY-AFTER-BUY-REPAIR-20260616-EXEC-001"
|
ROOT = Path(__file__).resolve().parents[1]
|
PROJECT_ROOT = ROOT.parents[2]
|
SOURCE = PROJECT_ROOT / "ana-data" / "result" / SOURCE_RUN_ID
|
TZ = timezone(timedelta(hours=8))
|
MOJIBAKE_RE = re.compile(r"\?{3,}|\ufffd|锟斤拷|脌|脙|脗|鎵|涓|鐗|寰|瀹|绗|璁|圭")
|
OVERREAD_RE = re.compile(r"策略有效性已证明|无边界完整baseline|完整baseline成功率|完整baseline收益率")
|
|
|
def now_iso() -> str:
|
return datetime.now(TZ).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]], fields: list[str]) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
with path.open("w", encoding="utf-8-sig", newline="") as f:
|
writer = csv.DictWriter(f, fieldnames=fields)
|
writer.writeheader()
|
for row in rows:
|
writer.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_project(path: Path) -> str:
|
return path.relative_to(PROJECT_ROOT).as_posix()
|
|
|
def rel_root(path: Path) -> str:
|
return path.relative_to(ROOT).as_posix()
|
|
|
def main() -> None:
|
generated_at = now_iso()
|
summary = json.loads((SOURCE / "strict_note_execution_summary.json").read_text(encoding="utf-8"))
|
cases = read_csv(SOURCE / "strict_note_case_summary.csv")
|
boundaries = read_csv(SOURCE / "strict_note_boundary_table.csv")
|
|
readouts = [
|
("STRICT_NOTE_AFTER_BUY_REPAIR_BUY_LOT", summary["strict_buy_lots"], "lot", "正式买点返修后进入严格卖点/滚动重放的 strict BUY lot 数"),
|
("STRICT_NOTE_AFTER_BUY_REPAIR_SELL_ORDER", summary["sell_orders"], "order", "严格卖点外部裁决生成的 SELL 订单数"),
|
("STRICT_NOTE_AFTER_BUY_REPAIR_ROLLING_LOW_BUY", summary["rolling_low_buy_orders"], "order", "滚动低吸 BUY 订单数，只说明执行数量，不单独构成策略结论"),
|
("STRICT_NOTE_AFTER_BUY_REPAIR_CASE_COVERAGE", summary["case_count"], "case", "有 strict BUY 的 case 覆盖数，不替代主成功率分母"),
|
("STRICT_NOTE_AFTER_BUY_REPAIR_PRIMARY_CLOSED_CASE", summary["primary_strict_closed_case_count"], "case", "主口径严格闭合 case 数"),
|
("STRICT_NOTE_AFTER_BUY_REPAIR_PRIMARY_POSITIVE_CASE", summary["primary_positive_case_count"], "case", "主口径正收益 case 数"),
|
("STRICT_NOTE_AFTER_BUY_REPAIR_PRIMARY_SUCCESS_RATE", summary["primary_success_rate_readout"], "ratio", "主口径成功率读数"),
|
("STRICT_NOTE_AFTER_BUY_REPAIR_PRIMARY_ACCOUNT_CONTRIBUTION", summary["primary_account_contribution_readout"], "account_contribution", "主口径账户贡献合计读数"),
|
("STRICT_NOTE_AFTER_BUY_REPAIR_BOUNDARY_RECORDS", summary["boundary_records"], "record", "边界记录数，边界样本不得混入主口径"),
|
]
|
readout_rows = [
|
{
|
"readout_id": rid,
|
"value": value,
|
"unit": unit,
|
"meaning_cn": meaning,
|
"source_path": rel_project(SOURCE / "strict_note_execution_summary.json"),
|
}
|
for rid, value, unit, meaning in readouts
|
]
|
write_csv(ROOT / "strict_note_final_readouts.csv", readout_rows, ["readout_id", "value", "unit", "meaning_cn", "source_path"])
|
|
case_fields = [
|
"case_id",
|
"entry_trade_date",
|
"signal_trade_date",
|
"symbols",
|
"case_scope_status",
|
"primary_strict_closed_case_flag",
|
"net_account_contribution",
|
"positive_case_flag",
|
"strict_buy_orders",
|
"sell_orders",
|
"rolling_low_buy_orders",
|
"boundary_lots",
|
"case_image_board_path",
|
"case_story_board_path",
|
]
|
write_csv(ROOT / "strict_note_case_readout_index.csv", cases, case_fields)
|
write_csv(ROOT / "strict_note_boundary_table.csv", boundaries, list(boundaries[0].keys()) if boundaries else ["boundary_id"])
|
|
final_summary = {
|
"run_id": FINAL_RUN_ID,
|
"source_run_id": SOURCE_RUN_ID,
|
"generated_at": generated_at,
|
"source_execution_audit_id": SOURCE_EXEC_AUDIT_ID,
|
"strict_buy_lots": summary["strict_buy_lots"],
|
"sell_orders": summary["sell_orders"],
|
"rolling_low_buy_orders": summary["rolling_low_buy_orders"],
|
"case_count": summary["case_count"],
|
"primary_strict_closed_case_count": summary["primary_strict_closed_case_count"],
|
"primary_positive_case_count": summary["primary_positive_case_count"],
|
"primary_success_rate_readout": summary["primary_success_rate_readout"],
|
"primary_account_contribution_readout": summary["primary_account_contribution_readout"],
|
"boundary_records": summary["boundary_records"],
|
"citation_policy": {
|
"required": [
|
SOURCE_EXEC_AUDIT_ID,
|
"strict_note_boundary_table.csv",
|
"严格笔记版主口径",
|
"不构成买入建议",
|
"不证明策略有效性",
|
"不得与旧 V1 宽口径或旧 424 lot 读数混用",
|
],
|
"forbidden": [
|
"无边界完整 baseline 成功率",
|
"无边界完整 baseline 收益率",
|
"策略有效性已证明",
|
"买入建议",
|
],
|
},
|
}
|
write_text(ROOT / "strict_note_final_conclusion_summary.json", json.dumps(final_summary, ensure_ascii=False, indent=2) + "\n")
|
|
summary_md = f"""# 严格笔记版最终引用包摘要
|
|
来源执行包：`{SOURCE_RUN_ID}`
|
|
来源执行审核 ID：`{SOURCE_EXEC_AUDIT_ID}`
|
|
## 可以引用什么
|
|
在严格笔记版主口径 `STRICT_NOTE_AFTER_BUY_REPAIR_PRIMARY_CLOSED_CASE` 下，{summary['primary_strict_closed_case_count']} 个主口径严格闭合 case 中，{summary['primary_positive_case_count']} 个为正收益，成功率读数为 {summary['primary_success_rate_readout']}，账户贡献合计读数为 {summary['primary_account_contribution_readout']}。
|
|
覆盖读数：有 strict BUY 的 case 为 {summary['case_count']} 个，strict BUY lot 为 {summary['strict_buy_lots']} 个，SELL 订单为 {summary['sell_orders']} 个，滚动低吸 BUY 订单为 {summary['rolling_low_buy_orders']} 个，边界记录为 {summary['boundary_records']} 条。
|
|
## 必须同时说明什么
|
|
- 必须引用来源执行审核 ID：`{SOURCE_EXEC_AUDIT_ID}`。
|
- 必须说明这是严格笔记版主口径，不是旧 V1 宽口径 / 代理 BUY 池读数。
|
- 必须保留 `strict_note_boundary_table.csv`，边界样本不得混入主成功率或主收益读数。
|
- 不构成买入建议，不证明策略有效性，不得写成无边界完整 baseline。
|
- 不得与旧 424 lot、旧 V1 宽口径 BUY 或未过审 BUY / SELL / rolling 口径混用。
|
"""
|
write_text(ROOT / "strict_note_final_conclusion_summary.md", summary_md)
|
|
index_md = f"""# 严格笔记版最终人工阅读入口
|
|
## 先看结论边界
|
|
可以引用：{summary['primary_strict_closed_case_count']} 个主口径严格闭合 case，{summary['primary_positive_case_count']} 个正收益，成功率读数 {summary['primary_success_rate_readout']}，账户贡献读数 {summary['primary_account_contribution_readout']}。
|
|
必须同时引用：`{SOURCE_EXEC_AUDIT_ID}`、严格笔记版主口径、边界表、非买入建议、非策略有效性证明。
|
|
不能引用：无边界完整 baseline 成功率 / 收益率，不能和旧 V1 宽口径或旧 424 lot 读数混用。
|
|
## 文件入口
|
|
- 来源执行包入口：[{SOURCE_RUN_ID}/strict_note_human_review_index.md](../{SOURCE_RUN_ID}/strict_note_human_review_index.md)
|
- 最终读数表：[strict_note_final_readouts.csv](strict_note_final_readouts.csv)
|
- case 索引：[strict_note_case_readout_index.csv](strict_note_case_readout_index.csv)
|
- 边界表：[strict_note_boundary_table.csv](strict_note_boundary_table.csv)
|
- 来源 manifest：[source_artifact_manifest.csv](source_artifact_manifest.csv)
|
"""
|
write_text(ROOT / "strict_note_final_human_review_index.md", index_md)
|
|
source_paths = [
|
SOURCE / "strict_note_execution_summary.json",
|
SOURCE / "strict_note_case_summary.csv",
|
SOURCE / "strict_note_boundary_table.csv",
|
SOURCE / "strict_note_human_review_index.md",
|
SOURCE / "strict_note_execution_self_check.json",
|
SOURCE / "strict_note_execution_manifest.csv",
|
]
|
source_manifest = [
|
{"path": rel_project(path), "size": path.stat().st_size, "sha256": sha256_file(path)}
|
for path in source_paths
|
]
|
write_csv(ROOT / "source_artifact_manifest.csv", source_manifest, ["path", "size", "sha256"])
|
|
text_paths = [
|
ROOT / "strict_note_final_human_review_index.md",
|
ROOT / "strict_note_final_conclusion_summary.md",
|
ROOT / "strict_note_final_conclusion_summary.json",
|
ROOT / "strict_note_final_readouts.csv",
|
ROOT / "strict_note_case_readout_index.csv",
|
ROOT / "strict_note_boundary_table.csv",
|
]
|
overread_paths = [
|
ROOT / "strict_note_final_human_review_index.md",
|
ROOT / "strict_note_final_conclusion_summary.md",
|
ROOT / "strict_note_final_readouts.csv",
|
ROOT / "strict_note_case_readout_index.csv",
|
ROOT / "strict_note_boundary_table.csv",
|
]
|
mojibake_hits = []
|
overread_hits = []
|
for path in text_paths:
|
text = path.read_text(encoding="utf-8")
|
if MOJIBAKE_RE.search(text):
|
mojibake_hits.append(rel_root(path))
|
if path in overread_paths and OVERREAD_RE.search(text):
|
overread_hits.append(rel_root(path))
|
checks = [
|
{"check_id": "SOURCE_SUMMARY_EXISTS", "status": "PASS" if (SOURCE / "strict_note_execution_summary.json").exists() else "FAIL", "detail": rel_project(SOURCE / "strict_note_execution_summary.json")},
|
{"check_id": "READOUTS_MATCH_SOURCE", "status": "PASS" if summary["primary_strict_closed_case_count"] == 122 and summary["primary_positive_case_count"] == 47 else "FAIL", "detail": f"primary={summary['primary_strict_closed_case_count']}, positive={summary['primary_positive_case_count']}"},
|
{"check_id": "BOUNDARY_TABLE_COPIED", "status": "PASS" if len(boundaries) == summary["boundary_records"] else "FAIL", "detail": f"boundary={len(boundaries)}"},
|
{"check_id": "SOURCE_ARTIFACT_HASH_READY", "status": "PASS" if len(source_manifest) == len(source_paths) else "FAIL", "detail": str(len(source_manifest))},
|
{"check_id": "TEXT_READABLE_NO_MOJIBAKE", "status": "PASS" if not mojibake_hits else "FAIL", "detail": ",".join(mojibake_hits)},
|
{"check_id": "NO_FORBIDDEN_OVERREAD", "status": "PASS" if not overread_hits else "FAIL", "detail": ",".join(overread_hits)},
|
]
|
write_csv(ROOT / "self_check_items.csv", checks, ["check_id", "status", "detail"])
|
stage = "PASS_FOR_STRICT_NOTE_FINAL_CONCLUSION_REVIEW_READY" if all(c["status"] == "PASS" for c in checks) else "FAIL"
|
write_text(ROOT / "self_check.json", json.dumps({
|
"run_id": FINAL_RUN_ID,
|
"generated_at": generated_at,
|
"stage": stage,
|
"pass_count": sum(1 for c in checks if c["status"] == "PASS"),
|
"fail_count": sum(1 for c in checks if c["status"] == "FAIL"),
|
}, ensure_ascii=False, indent=2) + "\n")
|
|
manifest_paths = text_paths + [
|
ROOT / "source_artifact_manifest.csv",
|
ROOT / "self_check.json",
|
ROOT / "self_check_items.csv",
|
ROOT / "tools" / "build_strict_note_final_after_buy_repair.py",
|
]
|
manifest = [
|
{"path": rel_root(path), "size": path.stat().st_size, "sha256": sha256_file(path)}
|
for path in manifest_paths
|
]
|
write_csv(ROOT / "manifest.csv", manifest, ["path", "size", "sha256"])
|
write_text(ROOT / "manifest.json", json.dumps({"run_id": FINAL_RUN_ID, "generated_at": generated_at, "file_count": len(manifest), "files": manifest}, ensure_ascii=False, indent=2) + "\n")
|
write_text(ROOT / "README.md", "# 严格笔记版最终引用包\n\n请先打开 `strict_note_final_human_review_index.md`。本包只整理已审核执行汇总读数，不重跑交易。\n")
|
print(json.dumps({"run_id": FINAL_RUN_ID, "stage": stage, "primary": summary["primary_strict_closed_case_count"], "positive": summary["primary_positive_case_count"]}, ensure_ascii=False))
|
|
|
if __name__ == "__main__":
|
main()
|
