from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
import re
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
|
|
FINAL_RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FINAL-CONCLUSION-20260615-001"
|
SOURCE_RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-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))
|
|
SOURCE_EXEC_AUDIT_ID = "AUDIT-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260615-EXEC-001"
|
MOJIBAKE_RE = re.compile(r"\?{3,}|\ufffd|����|À|Ã|Â|澶|鍙|鎬|涓|蹇|瑙|鏃|鐐|甯|瀹|鍚屾剰|鎸夎")
|
OVERREAD_RE = re.compile(r"(?<!不)是买入建议|(?<!不)构成买入建议|(?<!不)证明策略有效|策略有效性已证明|策略有效性已经证明|完整baseline成功率|完整baseline收益率|严格笔记版.*证明有效")
|
|
|
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_to_project(path: Path) -> str:
|
return path.relative_to(PROJECT_ROOT).as_posix()
|
|
|
def rel_to_root(path: Path) -> str:
|
return path.relative_to(ROOT).as_posix()
|
|
|
def main() -> None:
|
generated_at = datetime.now(TZ).isoformat(timespec="seconds")
|
source_summary = json.loads((SOURCE / "strict_note_execution_summary.json").read_text(encoding="utf-8"))
|
case_rows = read_csv(SOURCE / "strict_note_case_summary.csv")
|
boundary_rows = read_csv(SOURCE / "strict_note_boundary_table.csv")
|
|
readouts = [
|
{
|
"readout_id": "STRICT_NOTE_PRIMARY_STRICT_CLOSED_CASE",
|
"value": source_summary["primary_strict_closed_case_count"],
|
"unit": "case",
|
"meaning_cn": "严格笔记版主口径严格闭合 case 数；只包含已严格 BUY 且 lot 全部由外部卖点裁决 SELL 闭合、无未闭合滚动低吸边界的 case。",
|
"source_path": rel_to_project(SOURCE / "strict_note_execution_summary.json"),
|
},
|
{
|
"readout_id": "STRICT_NOTE_PRIMARY_POSITIVE_CASE",
|
"value": source_summary["primary_positive_case_count"],
|
"unit": "case",
|
"meaning_cn": "严格笔记版主口径中账户贡献为正的 case 数。",
|
"source_path": rel_to_project(SOURCE / "strict_note_case_summary.csv"),
|
},
|
{
|
"readout_id": "STRICT_NOTE_PRIMARY_SUCCESS_RATE",
|
"value": source_summary["primary_success_rate_readout"],
|
"unit": "ratio",
|
"meaning_cn": "严格笔记版主口径成功率读数，等于正收益主口径 case / 主口径严格闭合 case。",
|
"source_path": rel_to_project(SOURCE / "strict_note_execution_summary.json"),
|
},
|
{
|
"readout_id": "STRICT_NOTE_PRIMARY_ACCOUNT_CONTRIBUTION",
|
"value": source_summary["primary_account_contribution_readout"],
|
"unit": "account_contribution",
|
"meaning_cn": "严格笔记版主口径账户贡献合计读数。",
|
"source_path": rel_to_project(SOURCE / "strict_note_execution_summary.json"),
|
},
|
{
|
"readout_id": "STRICT_NOTE_ALL_BUY_CASE_COVERAGE",
|
"value": source_summary["case_count"],
|
"unit": "case",
|
"meaning_cn": "严格笔记版有 strict BUY 的 case 覆盖数；不替代主成功率分母。",
|
"source_path": rel_to_project(SOURCE / "strict_note_case_summary.csv"),
|
},
|
{
|
"readout_id": "STRICT_NOTE_BOUNDARY_RECORDS",
|
"value": source_summary["boundary_records"],
|
"unit": "record",
|
"meaning_cn": "严格笔记版边界记录数，包含市场闸门关闭、买点待审、分钟缺口、HOLD、REVIEW_HELD 和滚动低吸未闭合边界。",
|
"source_path": rel_to_project(SOURCE / "strict_note_boundary_table.csv"),
|
},
|
]
|
write_csv(ROOT / "strict_note_final_readouts.csv", readouts, ["readout_id", "value", "unit", "meaning_cn", "source_path"])
|
|
case_index_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", case_rows, case_index_fields)
|
write_csv(ROOT / "strict_note_boundary_table.csv", boundary_rows, list(boundary_rows[0].keys()) if boundary_rows else ["boundary_id"])
|
|
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_note_primary_strict_closed_case_count": source_summary["primary_strict_closed_case_count"],
|
"strict_note_primary_positive_case_count": source_summary["primary_positive_case_count"],
|
"strict_note_primary_success_rate_readout": source_summary["primary_success_rate_readout"],
|
"strict_note_primary_account_contribution_readout": source_summary["primary_account_contribution_readout"],
|
"strict_note_all_buy_case_coverage": source_summary["case_count"],
|
"strict_note_boundary_records": source_summary["boundary_records"],
|
"citation_policy": {
|
"allowed": "可按严格笔记版主口径、审计 ID 和边界表分层引用。",
|
"required_text": "该读数只覆盖严格笔记版已审核主口径，不构成买入建议，不证明策略有效性，不得写成无边界完整 baseline。",
|
"must_cite": [
|
SOURCE_EXEC_AUDIT_ID,
|
"strict_note_boundary_table.csv",
|
],
|
},
|
}
|
write_text(ROOT / "strict_note_final_conclusion_summary.json", json.dumps(summary, ensure_ascii=False, indent=2) + "\n")
|
|
summary_md = f"""# 严格笔记版最终引用包摘要
|
|
来源执行包：`{SOURCE_RUN_ID}`
|
|
来源执行审核 ID：`{SOURCE_EXEC_AUDIT_ID}`
|
|
## 可引用读数
|
|
在严格笔记版主口径 `STRICT_NOTE_PRIMARY_STRICT_CLOSED_CASE` 下，{summary['strict_note_primary_strict_closed_case_count']} 个严格闭合 case 中 {summary['strict_note_primary_positive_case_count']} 个为正收益，成功率读数为 {summary['strict_note_primary_success_rate_readout']}，账户贡献合计读数为 {summary['strict_note_primary_account_contribution_readout']}。
|
|
同时必须说明：严格笔记版有 strict BUY 的覆盖 case 为 {summary['strict_note_all_buy_case_coverage']} 个；边界记录为 {summary['strict_note_boundary_records']} 条，边界样本不得混入主口径。
|
|
## 必须保留的边界
|
|
1. 本包不是买入建议。
|
2. 本包不证明策略有效性。
|
3. 本包不得写成无边界完整 baseline 成功率、收益率、胜率或回撤。
|
4. 引用时必须同时写明来源执行审核 ID、严格笔记版主口径和边界表。
|
5. 不得与旧 V1 宽口径 / 代理 BUY 池读数混用。
|
"""
|
write_text(ROOT / "strict_note_final_conclusion_summary.md", summary_md)
|
|
human_index = f"""# 严格笔记版最终引用包人工第一入口
|
|
本入口给同事和审核员快速判断：严格按《笔记精简版.md》重建 BUY 池并重放 V1 卖点 / 趋势止盈 / 滚动低吸后，哪些读数可以引用，哪些不能引用。
|
|
## 可以引用
|
|
- 主口径严格闭合 case：{summary['strict_note_primary_strict_closed_case_count']}
|
- 主口径正收益 case：{summary['strict_note_primary_positive_case_count']}
|
- 主口径成功率读数：{summary['strict_note_primary_success_rate_readout']}
|
- 主口径账户贡献读数：{summary['strict_note_primary_account_contribution_readout']}
|
- 覆盖 case：{summary['strict_note_all_buy_case_coverage']}，只说明覆盖，不替代主成功率分母
|
- 边界记录：{summary['strict_note_boundary_records']}
|
|
## 必须同时引用
|
|
- 来源执行审核 ID：`{SOURCE_EXEC_AUDIT_ID}`
|
- 本最终引用包审核 ID：待审核员填写
|
- 边界表：[`strict_note_boundary_table.csv`](strict_note_boundary_table.csv)
|
- 来源执行包入口：[`strict_note_human_review_index.md`](../{SOURCE_RUN_ID}/strict_note_human_review_index.md)
|
|
## 不能引用
|
|
- 不能写成买入建议。
|
- 不能写成策略有效性证明。
|
- 不能写成无边界完整 baseline 成功率、收益率、胜率或回撤。
|
- 不能和旧 V1 宽口径 / 代理 BUY 池读数混用。
|
|
## 文件入口
|
|
- [`strict_note_final_conclusion_summary.md`](strict_note_final_conclusion_summary.md)
|
- [`strict_note_final_readouts.csv`](strict_note_final_readouts.csv)
|
- [`strict_note_case_readout_index.csv`](strict_note_case_readout_index.csv)
|
- [`strict_note_boundary_table.csv`](strict_note_boundary_table.csv)
|
- [`source_artifact_manifest.csv`](source_artifact_manifest.csv)
|
- [`self_check_items.csv`](self_check_items.csv)
|
"""
|
write_text(ROOT / "strict_note_final_human_review_index.md", human_index)
|
|
readme = f"""# {FINAL_RUN_ID}
|
|
严格笔记版最终引用包。
|
|
本包只整理已审核来源 `{SOURCE_RUN_ID}` 的读数，不重跑候选池、买点裁决、卖点裁决、滚动低吸或账户账本。
|
|
第一入口：`strict_note_final_human_review_index.md`
|
"""
|
write_text(ROOT / "README.md", readme)
|
|
source_paths = [
|
SOURCE / "strict_note_execution_summary.json",
|
SOURCE / "strict_note_case_summary.csv",
|
SOURCE / "strict_note_boundary_table.csv",
|
SOURCE / "strict_note_order_ledger.csv",
|
SOURCE / "strict_note_position_lot_ledger.csv",
|
SOURCE / "strict_note_human_review_index.md",
|
SOURCE / "strict_note_execution_self_check.json",
|
SOURCE / "strict_note_execution_manifest.csv",
|
]
|
source_manifest = []
|
for path in source_paths:
|
source_manifest.append({
|
"source_run_id": SOURCE_RUN_ID,
|
"path": rel_to_project(path),
|
"size": path.stat().st_size,
|
"sha256": sha256_file(path),
|
})
|
write_csv(ROOT / "source_artifact_manifest.csv", source_manifest, ["source_run_id", "path", "size", "sha256"])
|
|
scan_files = [
|
ROOT / "README.md",
|
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",
|
]
|
mojibake_hits = []
|
overread_hits = []
|
for path in scan_files:
|
text = path.read_text(encoding="utf-8-sig")
|
if MOJIBAKE_RE.search(text):
|
mojibake_hits.append(rel_to_root(path))
|
if OVERREAD_RE.search(text):
|
overread_hits.append(rel_to_root(path))
|
|
checks = [
|
{"check_id": "SOURCE_EXECUTION_PACKAGE_EXISTS", "status": "PASS" if SOURCE.exists() else "FAIL", "detail": rel_to_project(SOURCE)},
|
{"check_id": "READOUTS_MATCH_SOURCE_SUMMARY", "status": "PASS" if readouts[0]["value"] == source_summary["primary_strict_closed_case_count"] else "FAIL", "detail": "source summary copied"},
|
{"check_id": "BOUNDARY_TABLE_COPIED", "status": "PASS" if len(boundary_rows) == source_summary["boundary_records"] else "FAIL", "detail": str(len(boundary_rows))},
|
{"check_id": "SOURCE_ARTIFACT_HASHES_RECORDED", "status": "PASS" if len(source_manifest) == len(source_paths) else "FAIL", "detail": str(len(source_manifest))},
|
{"check_id": "NO_MOJIBAKE_IN_FINAL_PACKAGE", "status": "PASS" if not mojibake_hits else "FAIL", "detail": ";".join(mojibake_hits[:20])},
|
{"check_id": "NO_OVERREAD_CLAIMS", "status": "PASS" if not overread_hits else "FAIL", "detail": ";".join(overread_hits[:20])},
|
]
|
write_csv(ROOT / "self_check_items.csv", checks, ["check_id", "status", "detail"])
|
self_check = {
|
"run_id": FINAL_RUN_ID,
|
"stage": "PASS_FOR_STRICT_NOTE_FINAL_CONCLUSION_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 / "self_check.json", json.dumps(self_check, ensure_ascii=False, indent=2) + "\n")
|
|
manifest_items = [
|
"README.md",
|
"strict_note_final_human_review_index.md",
|
"strict_note_final_conclusion_summary.md",
|
"strict_note_final_conclusion_summary.json",
|
"strict_note_final_readouts.csv",
|
"strict_note_case_readout_index.csv",
|
"strict_note_boundary_table.csv",
|
"source_artifact_manifest.csv",
|
"self_check_items.csv",
|
"self_check.json",
|
"tools/build_strict_note_final_conclusion_package.py",
|
]
|
manifest = []
|
for item in manifest_items:
|
path = ROOT / item
|
manifest.append({"path": item, "size": path.stat().st_size, "sha256": sha256_file(path)})
|
write_csv(ROOT / "manifest.csv", manifest, ["path", "size", "sha256"])
|
write_text(ROOT / "manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2) + "\n")
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|
