from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
import os
|
from collections import Counter
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
from typing import Any
|
|
|
TASK_ID = "ANA-WUJI-BASELINE-2023-2026"
|
DESIGN_ID = "DESIGN-WUJI-FINAL-CONCLUSION-20260608"
|
RUN_ID = "RUN-ANA-WUJI-FINAL-CONCLUSION-20260608-001"
|
SOURCE_RUN_ID = "RUN-ANA-WUJI-FULL-2023-2026-20260608-001"
|
DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-FINAL-CONCLUSION-20260608-DESIGN-001"
|
SOURCE_EXEC_AUDIT_ID = "AUDIT-ANA-WUJI-FULL-2023-2026-20260608-EXEC-REREVIEW-002"
|
DESIGN_REVIEW_MESSAGE_ID = "msg_20260608152148950_4b5694b6"
|
EXECUTION_REVIEW_REQUEST_MESSAGE_ID = "msg_20260608153645898_fe3781a7"
|
EXECUTION_AUDIT_ID = "AUDIT-ANA-WUJI-FINAL-CONCLUSION-20260608-EXEC-001"
|
EXECUTION_REVIEW_RESULT_MESSAGE_ID = "msg_20260608154423790_54bb73d5"
|
FINAL_EXECUTION_REVIEW_STATUS = "EXECUTION_REVIEW_PASSED_LAYERED_CITATION_ALLOWED_RETURN_STAT_HELD"
|
STAGE = "FINAL_CONCLUSION_EXECUTION_REVIEW_PASSED_LAYERED_CITATION_ALLOWED_RETURN_STAT_HELD"
|
OVERALL_STATUS = "PASS_FOR_FINAL_CONCLUSION_EXECUTION_REVIEW_PASSED_LAYERED_CITATION_ALLOWED"
|
EXPECTED_SOURCE_STAGE = "FULL_2023_2026_EXECUTION_REREVIEW_PASSED_RETURN_STAT_HELD"
|
|
SCRIPT_PATH = Path(__file__).resolve()
|
TOOLS_DIR = SCRIPT_PATH.parent
|
RUN_DIR = TOOLS_DIR.parent
|
PROJECT_ROOT = RUN_DIR.parents[2]
|
SOURCE_RUN_DIR = PROJECT_ROOT / "ana-data" / "result" / SOURCE_RUN_ID
|
|
|
def now_iso() -> str:
|
tz = timezone(timedelta(hours=8))
|
return datetime.now(tz).replace(microsecond=0).isoformat()
|
|
|
def read_json(path: Path) -> dict[str, Any]:
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
def write_json(path: Path, data: dict[str, Any]) -> None:
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
def read_csv(path: Path) -> list[dict[str, str]]:
|
with path.open("r", encoding="utf-8-sig", newline="") as f:
|
rows = []
|
for row in csv.DictReader(f):
|
rows.append({(key or "").strip(): (value or "") for key, value in row.items()})
|
return rows
|
|
|
def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str]) -> None:
|
with path.open("w", encoding="utf-8", newline="") as f:
|
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
writer.writeheader()
|
for row in rows:
|
writer.writerow({key: row.get(key, "") for key in fieldnames})
|
|
|
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_link(target: Path, base: Path = RUN_DIR) -> str:
|
return Path(os.path.relpath(target, base)).as_posix()
|
|
|
def rel_project(path: Path) -> str:
|
return Path(os.path.relpath(path, PROJECT_ROOT)).as_posix()
|
|
|
def float_value(value: str) -> float:
|
if value is None or value == "":
|
return 0.0
|
return float(value)
|
|
|
def fmt_rate(value: Any) -> str:
|
return f"{float(value):.10f}"
|
|
|
def fmt_money(value: Any) -> str:
|
return f"{float(value):.8f}"
|
|
|
def count_where(rows: list[dict[str, str]], key: str, value: str) -> int:
|
return sum(1 for row in rows if row.get(key) == value)
|
|
|
def first_where(rows: list[dict[str, str]], key: str, value: str) -> dict[str, str] | None:
|
return next((row for row in rows if row.get(key) == value), None)
|
|
|
def source_artifact(path: Path, role: str) -> dict[str, Any]:
|
exists = path.exists()
|
return {
|
"artifact_role": role,
|
"source_path": rel_project(path),
|
"exists": str(exists),
|
"size": path.stat().st_size if exists else "",
|
"sha256": sha256_file(path) if exists else "",
|
"final_package_reference": rel_link(path),
|
}
|
|
|
def add_check(items: list[dict[str, Any]], check_id: str, passed: bool, detail: str) -> None:
|
items.append(
|
{
|
"check_id": check_id,
|
"status": "PASS" if passed else "FAIL",
|
"detail": detail,
|
}
|
)
|
|
|
def metric_row(
|
scope: str,
|
metric_key: str,
|
metric_name_cn: str,
|
value: Any,
|
source_file: str,
|
source_field: str,
|
citation_boundary: str,
|
citation_status: str = "EXECUTION_REVIEW_PASSED_LAYERED_CITATION_ONLY_RETURN_STAT_HELD",
|
) -> dict[str, Any]:
|
return {
|
"scope": scope,
|
"metric_key": metric_key,
|
"metric_name_cn": metric_name_cn,
|
"value": value,
|
"citation_status": citation_status,
|
"source_file": source_file,
|
"source_field": source_field,
|
"citation_boundary": citation_boundary,
|
}
|
|
|
def build() -> None:
|
RUN_DIR.mkdir(parents=True, exist_ok=True)
|
generated_at = now_iso()
|
|
source_summary = read_json(SOURCE_RUN_DIR / "summary.json")
|
source_self_check = read_json(SOURCE_RUN_DIR / "self_check.json")
|
source_manifest = read_json(SOURCE_RUN_DIR / "manifest.json")
|
return_summary = read_json(SOURCE_RUN_DIR / "full_return_stat_summary.json")
|
case_scope = read_csv(SOURCE_RUN_DIR / "full_return_stat_case_scope.csv")
|
lot_scope = read_csv(SOURCE_RUN_DIR / "full_return_stat_lot_scope.csv")
|
boundary_rows = read_csv(SOURCE_RUN_DIR / "full_return_stat_boundary_table.csv")
|
batch_index = read_csv(SOURCE_RUN_DIR / "full_batch_index.csv")
|
|
primary = return_summary["primary_scope"]
|
coverage = return_summary["coverage_scope"]
|
lot_recalc = return_summary["lot_recalc_scope"]
|
boundary = return_summary["boundary"]
|
|
case_scope_count = len(case_scope)
|
primary_case_count = count_where(case_scope, "primary_strict_closed_case_flag", "1")
|
positive_case_count = sum(
|
1
|
for row in case_scope
|
if row.get("primary_strict_closed_case_flag") == "1"
|
and float_value(row.get("account_return_closed_lots", "")) > 0
|
)
|
lot_count = len(lot_scope)
|
closed_lot_count = count_where(lot_scope, "lot_scope", "STRICT_CLOSED_LOT_RECALC_ONLY")
|
boundary_lot_count = count_where(lot_scope, "lot_scope", "RETURN_STAT_HELD_BOUNDARY_TABLE")
|
positive_closed_lots = sum(
|
1
|
for row in lot_scope
|
if row.get("lot_scope") == "STRICT_CLOSED_LOT_RECALC_ONLY"
|
and float_value(row.get("account_return_contribution_pct", "")) > 0
|
)
|
case_boundary_count = count_where(boundary_rows, "boundary_level", "CASE")
|
lot_boundary_count = count_where(boundary_rows, "boundary_level", "LOT")
|
|
source_files = [
|
("source summary", SOURCE_RUN_DIR / "summary.json"),
|
("source summary markdown", SOURCE_RUN_DIR / "summary.md"),
|
("source self check", SOURCE_RUN_DIR / "self_check.json"),
|
("source manifest", SOURCE_RUN_DIR / "manifest.json"),
|
("return stat summary", SOURCE_RUN_DIR / "full_return_stat_summary.json"),
|
("return stat summary markdown", SOURCE_RUN_DIR / "full_return_stat_summary.md"),
|
("case scope", SOURCE_RUN_DIR / "full_return_stat_case_scope.csv"),
|
("lot scope", SOURCE_RUN_DIR / "full_return_stat_lot_scope.csv"),
|
("boundary table", SOURCE_RUN_DIR / "full_return_stat_boundary_table.csv"),
|
("root image board", SOURCE_RUN_DIR / "case_image_board.md"),
|
("root story board", SOURCE_RUN_DIR / "case_story_board.md"),
|
("batch index", SOURCE_RUN_DIR / "full_batch_index.csv"),
|
]
|
|
primary_case = first_where(case_scope, "primary_strict_closed_case_flag", "1")
|
market_closed_case = first_where(case_scope, "boundary_category", "MARKET_GATE_CLOSED")
|
unresolved_case = first_where(case_scope, "boundary_category", "UNRESOLVED_LOT_BOUNDARY")
|
entry_gap_case = first_where(case_scope, "boundary_category", "ENTRY_DATA_GAP_HELD")
|
exit_gap_lot = first_where(lot_scope, "lot_status", "EXIT_DATA_GAP_HELD")
|
exit_gap_case = None
|
if exit_gap_lot:
|
exit_gap_case = next((row for row in case_scope if row.get("case_id") == exit_gap_lot["case_id"]), None)
|
|
sample_rows = [
|
("主口径正收益样例", primary_case),
|
("市场闸门关闭样例", market_closed_case),
|
("未解决 lot 边界样例", unresolved_case),
|
("入场数据缺口样例", entry_gap_case),
|
("退出数据缺口样例", exit_gap_case),
|
]
|
sample_links: list[dict[str, str]] = []
|
for label, row in sample_rows:
|
if not row:
|
continue
|
case_id = row["case_id"]
|
batch_id = row.get("batch_id", "")
|
board = SOURCE_RUN_DIR / "cases" / case_id / "case_image_board.md"
|
story = SOURCE_RUN_DIR / "cases" / case_id / "case_story_board.md"
|
source_files.append((label + " case image board", board))
|
source_files.append((label + " case story board", story))
|
sample_links.append(
|
{
|
"label": label,
|
"case_id": case_id,
|
"batch_id": batch_id,
|
"entry_trade_date": row.get("entry_trade_date", ""),
|
"case_scope_status": row.get("case_scope_status", ""),
|
"return_stat_scope": row.get("primary_scope") or row.get("case_scope_status", ""),
|
"boundary_category": row.get("boundary_category", ""),
|
"board_link": rel_link(board),
|
"story_link": rel_link(story),
|
}
|
)
|
|
for batch in batch_index:
|
source_files.append((f"{batch['batch_id']} batch image board", SOURCE_RUN_DIR / batch["batch_dir"] / "case_image_board.md"))
|
|
source_artifact_rows = [source_artifact(path, role) for role, path in source_files]
|
write_csv(
|
RUN_DIR / "source_artifact_manifest.csv",
|
source_artifact_rows,
|
["artifact_role", "source_path", "exists", "size", "sha256", "final_package_reference"],
|
)
|
|
readout_rows = [
|
metric_row("PRIMARY_STRICT_CLOSED_CASE", "primary_case_count", "主口径 case 数", primary["case_count"], "full_return_stat_summary.json", "primary_scope.case_count", "只覆盖严格闭合 case。"),
|
metric_row("PRIMARY_STRICT_CLOSED_CASE", "positive_case_count", "主口径正收益 case 数", primary["positive_case_count"], "full_return_stat_summary.json", "primary_scope.positive_case_count", "成功定义为主口径 case 账户贡献大于 0。"),
|
metric_row("PRIMARY_STRICT_CLOSED_CASE", "non_positive_case_count", "主口径非正收益 case 数", primary["non_positive_case_count"], "full_return_stat_summary.json", "primary_scope.non_positive_case_count", "与正收益 case 合计等于主口径 case。"),
|
metric_row("PRIMARY_STRICT_CLOSED_CASE", "candidate_success_rate", "主口径严格闭合 case 成功率候选读数", fmt_rate(primary["candidate_success_rate_for_audit_only"]), "full_return_stat_summary.json", "primary_scope.candidate_success_rate_for_audit_only", "不得脱离主口径边界写成完整 baseline 成功率。"),
|
metric_row("PRIMARY_STRICT_CLOSED_CASE", "account_return_sum", "主口径账户贡献合计候选读数", fmt_money(primary["account_return_sum_for_audit_only"]), "full_return_stat_summary.json", "primary_scope.account_return_sum_for_audit_only", "不得脱离主口径边界写成完整 baseline 收益率。"),
|
metric_row("PRIMARY_STRICT_CLOSED_CASE", "account_return_mean", "主口径 case 平均账户贡献候选读数", fmt_money(primary["account_return_mean_for_audit_only"]), "full_return_stat_summary.json", "primary_scope.account_return_mean_for_audit_only", "只在主口径样本内解释。"),
|
metric_row("PRIMARY_STRICT_CLOSED_CASE", "account_return_median", "主口径 case 账户贡献中位数候选读数", fmt_money(primary["account_return_median_for_audit_only"]), "full_return_stat_summary.json", "primary_scope.account_return_median_for_audit_only", "只在主口径样本内解释。"),
|
metric_row("ALL_ENTRY_DATE_COVERAGE", "entry_date_count", "覆盖 entry date 数", coverage["entry_date_count"], "full_return_stat_summary.json", "coverage_scope.entry_date_count", "覆盖口径只说明样本覆盖，不替代主成功率分母。"),
|
metric_row("ALL_ENTRY_DATE_COVERAGE", "market_gate_open_entry_dates", "市场闸门打开 entry date", coverage["market_gate_open_entry_dates"], "full_return_stat_summary.json", "coverage_scope.market_gate_open_entry_dates", "用于覆盖说明。"),
|
metric_row("ALL_ENTRY_DATE_COVERAGE", "market_gate_closed_entry_dates", "市场闸门关闭 entry date", coverage["market_gate_closed_entry_dates"], "full_return_stat_summary.json", "coverage_scope.market_gate_closed_entry_dates", "市场闸门关闭样本不进入主收益 / 成功率口径。"),
|
metric_row("ALL_ENTRY_DATE_COVERAGE", "buy_case_count", "有 BUY case 数", coverage["buy_case_count"], "full_return_stat_summary.json", "coverage_scope.buy_case_count", "不等于主口径 case 数。"),
|
metric_row("ALL_ENTRY_DATE_COVERAGE", "excluded_case_count", "排除出主口径 case 数", coverage["excluded_case_count"], "full_return_stat_summary.json", "coverage_scope.excluded_case_count", "排除样本进入覆盖口径或边界表。"),
|
metric_row("STRICT_CLOSED_LOT_RECALC_ONLY", "total_lot_count", "lot 总数", lot_recalc["total_lot_count"], "full_return_stat_summary.json", "lot_recalc_scope.total_lot_count", "lot 口径只用于复算和问题定位。"),
|
metric_row("STRICT_CLOSED_LOT_RECALC_ONLY", "closed_lot_count", "闭合 lot 数", lot_recalc["closed_lot_count"], "full_return_stat_summary.json", "lot_recalc_scope.closed_lot_count", "不得包装成 case 成功率。"),
|
metric_row("STRICT_CLOSED_LOT_RECALC_ONLY", "unresolved_lot_count", "未解决 / 边界 lot 数", lot_recalc["unresolved_lot_count"], "full_return_stat_summary.json", "lot_recalc_scope.unresolved_lot_count", "必须排除出主 case 收益 / 成功率口径。"),
|
metric_row("STRICT_CLOSED_LOT_RECALC_ONLY", "positive_closed_lot_count", "正收益闭合 lot 数", lot_recalc["positive_closed_lot_count"], "full_return_stat_summary.json", "lot_recalc_scope.positive_closed_lot_count", "只作为 lot 复算读数。"),
|
metric_row("STRICT_CLOSED_LOT_RECALC_ONLY", "closed_lot_account_return_sum", "闭合 lot 账户贡献合计", fmt_money(lot_recalc["closed_lot_account_return_sum_for_recalc_only"]), "full_return_stat_summary.json", "lot_recalc_scope.closed_lot_account_return_sum_for_recalc_only", "只作为 lot 复算读数。"),
|
metric_row("RETURN_STAT_HELD_BOUNDARY_TABLE", "case_boundary_count", "case 边界数", boundary["case_boundary_count"], "full_return_stat_summary.json", "boundary.case_boundary_count", "边界样本不得混入主口径。"),
|
metric_row("RETURN_STAT_HELD_BOUNDARY_TABLE", "lot_boundary_count", "lot 边界数", boundary["lot_boundary_count"], "full_return_stat_summary.json", "boundary.lot_boundary_count", "边界 lot 不得强行转真实 SELL。"),
|
]
|
for key, value in boundary["case_boundary_counts"].items():
|
readout_rows.append(metric_row("RETURN_STAT_HELD_BOUNDARY_TABLE", f"case_boundary_{key}", f"case 边界：{key}", value, "full_return_stat_summary.json", f"boundary.case_boundary_counts.{key}", "边界样本不得混入主口径。"))
|
for key, value in boundary["lot_boundary_counts"].items():
|
readout_rows.append(metric_row("RETURN_STAT_HELD_BOUNDARY_TABLE", f"lot_boundary_{key}", f"lot 边界：{key}", value, "full_return_stat_summary.json", f"boundary.lot_boundary_counts.{key}", "边界 lot 不得强行转真实 SELL。"))
|
|
write_csv(
|
RUN_DIR / "final_conclusion_readouts.csv",
|
readout_rows,
|
["scope", "metric_key", "metric_name_cn", "value", "citation_status", "source_file", "source_field", "citation_boundary"],
|
)
|
|
final_boundary_rows = []
|
for row in boundary_rows:
|
final_boundary_rows.append(
|
{
|
**row,
|
"final_conclusion_scope": "RETURN_STAT_HELD_BOUNDARY_TABLE",
|
"final_policy": "保留为边界；不得混入 PRIMARY_STRICT_CLOSED_CASE，不得强行补结论。",
|
"source_file": "full_return_stat_boundary_table.csv",
|
}
|
)
|
boundary_fieldnames = list(boundary_rows[0].keys()) + ["final_conclusion_scope", "final_policy", "source_file"]
|
write_csv(RUN_DIR / "final_boundary_table.csv", final_boundary_rows, boundary_fieldnames)
|
|
config = {
|
"schema_version": "1.0",
|
"task_id": TASK_ID,
|
"design_id": DESIGN_ID,
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"stage": STAGE,
|
"source_run_id": SOURCE_RUN_ID,
|
"source_run_path": rel_project(SOURCE_RUN_DIR),
|
"design_audit_id": DESIGN_AUDIT_ID,
|
"source_execution_rereview_audit_id": SOURCE_EXEC_AUDIT_ID,
|
"design_review_message_id": DESIGN_REVIEW_MESSAGE_ID,
|
"execution_review_request_message_id": EXECUTION_REVIEW_REQUEST_MESSAGE_ID,
|
"execution_audit_id": EXECUTION_AUDIT_ID,
|
"execution_review_result_message_id": EXECUTION_REVIEW_RESULT_MESSAGE_ID,
|
"return_stat_ready": False,
|
"final_execution_review_status": FINAL_EXECUTION_REVIEW_STATUS,
|
"allowed_action": "Cite current readouts only with explicit layered scopes and boundaries.",
|
"forbidden_actions": [
|
"Do not rerun candidate pool, buy/sell decisions, or account ledgers.",
|
"Do not cite readouts as unbounded full baseline success, return, win-rate, drawdown, or effectiveness.",
|
"Do not set RETURN_STAT_READY=true; reviewer approved layered citation only.",
|
],
|
}
|
write_json(RUN_DIR / "final_conclusion_config.json", config)
|
|
config_md = f"""# 最终结论引用包配置
|
|
| 项目 | 内容 |
|
|---|---|
|
| 任务 | `{TASK_ID}` |
|
| 设计 ID | `{DESIGN_ID}` |
|
| run_id | `{RUN_ID}` |
|
| 当前阶段 | `{STAGE}` |
|
| 设计审核审计 ID | `{DESIGN_AUDIT_ID}` |
|
| 来源全量 run | `{SOURCE_RUN_ID}` |
|
| 来源执行复审审计 ID | `{SOURCE_EXEC_AUDIT_ID}` |
|
| 来源结果包 | `{rel_link(SOURCE_RUN_DIR)}` |
|
| 执行审核请求消息 | `{EXECUTION_REVIEW_REQUEST_MESSAGE_ID}` |
|
| 执行审核结果消息 | `{EXECUTION_REVIEW_RESULT_MESSAGE_ID}` |
|
| 执行审核审计 ID | `{EXECUTION_AUDIT_ID}` |
|
| RETURN_STAT_READY | `false` |
|
| 执行审核状态 | `通过；仅允许分层引用；RETURN_STAT_READY 继续为 false` |
|
|
本包只读取已通过复审的全量分批结果包，不重跑候选池、买卖裁决或账本。执行审核已通过，允许把当前读数写入面向同事的最终案例总结，但必须采用分层引用和降读文本；不得脱离 `PRIMARY_STRICT_CLOSED_CASE`、`ALL_ENTRY_DATE_COVERAGE`、`STRICT_CLOSED_LOT_RECALC_ONLY` 与边界样本说明写成完整 baseline 结论。
|
"""
|
(RUN_DIR / "final_conclusion_config.md").write_text(config_md, encoding="utf-8")
|
|
primary_success = fmt_rate(primary["candidate_success_rate_for_audit_only"])
|
primary_return_sum = fmt_money(primary["account_return_sum_for_audit_only"])
|
primary_mean = fmt_money(primary["account_return_mean_for_audit_only"])
|
lot_return_sum = fmt_money(lot_recalc["closed_lot_account_return_sum_for_recalc_only"])
|
allowed_citation_text = [
|
f"在 `PRIMARY_STRICT_CLOSED_CASE` 主口径下，{primary['case_count']} 个严格闭合 case 中 {primary['positive_case_count']} 个为正收益，主口径成功率读数为 {primary_success}，主口径账户贡献合计为 {primary_return_sum}。",
|
f"全样本覆盖为 {coverage['entry_date_count']} 个 entry date，其中市场闸门打开 {coverage['market_gate_open_entry_dates']}、关闭 {coverage['market_gate_closed_entry_dates']}；覆盖口径只说明样本覆盖，不替代主成功率。",
|
f"辅助 lot 口径中 {lot_recalc['total_lot_count']} 个 lot，{lot_recalc['closed_lot_count']} 个闭合、{lot_recalc['unresolved_lot_count']} 个边界；lot 口径只用于复算和问题定位，不包装成 case 成功率。",
|
f"{boundary['case_boundary_count']} 个 case 边界和 {boundary['lot_boundary_count']} 个 lot 边界均不得混入主口径。",
|
]
|
|
summary_data = {
|
"schema_version": "1.0",
|
"task_id": TASK_ID,
|
"design_id": DESIGN_ID,
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"stage": STAGE,
|
"design_audit_id": DESIGN_AUDIT_ID,
|
"source_run_id": SOURCE_RUN_ID,
|
"source_stage": source_summary["stage"],
|
"source_execution_rereview_audit_id": SOURCE_EXEC_AUDIT_ID,
|
"source_execution_rereview_pass_message_id": source_summary.get("execution_rereview_pass_message_id"),
|
"execution_review_request_message_id": EXECUTION_REVIEW_REQUEST_MESSAGE_ID,
|
"execution_audit_id": EXECUTION_AUDIT_ID,
|
"execution_review_result_message_id": EXECUTION_REVIEW_RESULT_MESSAGE_ID,
|
"return_stat_ready": False,
|
"final_execution_review_status": FINAL_EXECUTION_REVIEW_STATUS,
|
"citation_state": "Execution review passed. Cite only with explicit layered scopes and boundaries; RETURN_STAT_READY remains false.",
|
"allowed_citation_text": allowed_citation_text,
|
"primary_scope": primary,
|
"coverage_scope": coverage,
|
"lot_recalc_scope": lot_recalc,
|
"boundary": boundary,
|
"source_self_check": {
|
"overall_status": source_self_check.get("overall_status"),
|
"check_count": source_self_check.get("check_count"),
|
"fail_count": source_self_check.get("fail_count"),
|
},
|
"source_manifest": {
|
"file_count": source_manifest.get("file_count"),
|
"generated_at": source_manifest.get("generated_at"),
|
},
|
"sample_links": sample_links,
|
"boundary_statement": "主口径只覆盖 234 个严格闭合 case；覆盖口径覆盖 743 个 entry date；辅助 lot 口径只用于 lot 复算；509 个 case 边界和 18 个 lot 边界不得混入主口径。",
|
}
|
write_json(RUN_DIR / "final_conclusion_summary.json", summary_data)
|
|
sample_md = "\n".join(
|
f"- {row['label']}：`{row['case_id']}`，entry `{row['entry_trade_date']}`，[图片板]({row['board_link']})，[故事板]({row['story_link']})"
|
for row in sample_links
|
)
|
|
summary_md = f"""# 无忌交易系统最终结论引用包摘要
|
|
## 当前可引用状态
|
|
当前包执行审核已通过，审计 ID 为 `{EXECUTION_AUDIT_ID}`。审核员允许把当前读数写入面向同事的最终案例总结，但必须采用分层引用 / 降读文本；`RETURN_STAT_READY=false` 继续保留，不得把下列读数脱离分层边界写成完整 baseline 成功率、收益率、胜率、回撤或策略有效性结论。
|
|
允许引用文本边界：
|
|
1. {allowed_citation_text[0]}
|
2. {allowed_citation_text[1]}
|
3. {allowed_citation_text[2]}
|
4. {allowed_citation_text[3]}
|
|
## 三层口径读数
|
|
### 主口径：PRIMARY_STRICT_CLOSED_CASE
|
|
主口径只覆盖市场闸门打开、有真实 BUY，且所有 lot 都由真实 `CLOSED_BY_AI_SELL` 闭合的 case。
|
|
| 指标 | 读数 |
|
|---|---:|
|
| 主口径 case | {primary['case_count']} |
|
| 正收益 case | {primary['positive_case_count']} |
|
| 非正收益 case | {primary['non_positive_case_count']} |
|
| 主口径严格闭合 case 成功率候选读数 | {primary_success} |
|
| 主口径账户贡献合计候选读数 | {primary_return_sum} |
|
| 主口径 case 平均账户贡献候选读数 | {primary_mean} |
|
|
### 覆盖口径：ALL_ENTRY_DATE_COVERAGE
|
|
覆盖口径用于说明全样本覆盖、市场闸门和无交易 / 边界状态，不替代主成功率分母。
|
|
| 指标 | 读数 |
|
|---|---:|
|
| entry date | {coverage['entry_date_count']} |
|
| 市场闸门打开 entry date | {coverage['market_gate_open_entry_dates']} |
|
| 市场闸门关闭 entry date | {coverage['market_gate_closed_entry_dates']} |
|
| 有 BUY case | {coverage['buy_case_count']} |
|
| 市场闸门打开但无 BUY case | {coverage['open_gate_no_buy_case_count']} |
|
| 排除出主口径 case | {coverage['excluded_case_count']} |
|
|
### 辅助 lot 口径：STRICT_CLOSED_LOT_RECALC_ONLY
|
|
辅助 lot 口径只用于 lot 复算和问题定位，不得包装成 case 成功率。
|
|
| 指标 | 读数 |
|
|---|---:|
|
| lot 总数 | {lot_recalc['total_lot_count']} |
|
| 闭合 lot | {lot_recalc['closed_lot_count']} |
|
| 未解决 / 边界 lot | {lot_recalc['unresolved_lot_count']} |
|
| 正收益闭合 lot | {lot_recalc['positive_closed_lot_count']} |
|
| 闭合 lot 账户贡献合计 | {lot_return_sum} |
|
|
## 边界样本
|
|
| 边界 | 读数 |
|
|---|---:|
|
| case 边界合计 | {boundary['case_boundary_count']} |
|
| lot 边界合计 | {boundary['lot_boundary_count']} |
|
| MARKET_GATE_CLOSED | {boundary['case_boundary_counts'].get('MARKET_GATE_CLOSED', 0)} |
|
| UNRESOLVED_LOT_BOUNDARY | {boundary['case_boundary_counts'].get('UNRESOLVED_LOT_BOUNDARY', 0)} |
|
| ENTRY_DATA_GAP_HELD | {boundary['case_boundary_counts'].get('ENTRY_DATA_GAP_HELD', 0)} |
|
| NO_BUY_AI_REVIEWED | {boundary['case_boundary_counts'].get('NO_BUY_AI_REVIEWED', 0)} |
|
| WINDOW_END_VALUATION_ONLY | {boundary['lot_boundary_counts'].get('WINDOW_END_VALUATION_ONLY', 0)} |
|
| EXIT_DATA_GAP_HELD | {boundary['lot_boundary_counts'].get('EXIT_DATA_GAP_HELD', 0)} |
|
|
## 人工审核样例入口
|
|
{sample_md}
|
|
## 来源
|
|
- 来源全量包：[case_image_board.md]({rel_link(SOURCE_RUN_DIR / 'case_image_board.md')})
|
- 来源分层摘要：[full_return_stat_summary.md]({rel_link(SOURCE_RUN_DIR / 'full_return_stat_summary.md')})
|
- 来源 case scope：[full_return_stat_case_scope.csv]({rel_link(SOURCE_RUN_DIR / 'full_return_stat_case_scope.csv')})
|
- 来源 lot scope：[full_return_stat_lot_scope.csv]({rel_link(SOURCE_RUN_DIR / 'full_return_stat_lot_scope.csv')})
|
- 来源边界表：[full_return_stat_boundary_table.csv]({rel_link(SOURCE_RUN_DIR / 'full_return_stat_boundary_table.csv')})
|
"""
|
(RUN_DIR / "final_conclusion_summary.md").write_text(summary_md, encoding="utf-8")
|
|
batch_links = "\n".join(
|
f"- `{batch['batch_id']}`：{batch['entry_date_start']} 至 {batch['entry_date_end']}，[图片板]({rel_link(SOURCE_RUN_DIR / batch['batch_dir'] / 'case_image_board.md')})"
|
for batch in batch_index
|
)
|
human_index_md = f"""# 最终结论人工审核第一入口
|
|
## 先看这里
|
|
可以审核：当前包是否把已通过复审的全量分层读数，正确转换成最终结论引用材料。
|
|
可以引用：执行审核已通过，允许按 `PRIMARY_STRICT_CLOSED_CASE`、`ALL_ENTRY_DATE_COVERAGE`、`STRICT_CLOSED_LOT_RECALC_ONLY` 三层口径写入面向同事的最终案例总结。
|
|
不能引用：不能把 `PRIMARY_STRICT_CLOSED_CASE` 的 {primary_success} 脱离主口径边界写成完整 baseline 成功率，不能把 {primary_return_sum} 脱离主口径边界写成完整 baseline 收益率，不能声称已经证明策略有效性，不能把案例事项标记为最终完成，不能把 `RETURN_STAT_READY` 改为 true。
|
|
当前状态：
|
|
| 项目 | 内容 |
|
|---|---|
|
| 当前 run | `{RUN_ID}` |
|
| 来源 run | `{SOURCE_RUN_ID}` |
|
| 来源执行复审审计 ID | `{SOURCE_EXEC_AUDIT_ID}` |
|
| 当前设计审核审计 ID | `{DESIGN_AUDIT_ID}` |
|
| 执行审核请求消息 | `{EXECUTION_REVIEW_REQUEST_MESSAGE_ID}` |
|
| 执行审核结果消息 | `{EXECUTION_REVIEW_RESULT_MESSAGE_ID}` |
|
| 执行审核审计 ID | `{EXECUTION_AUDIT_ID}` |
|
| RETURN_STAT_READY | `false` |
|
| 执行审核状态 | `通过；仅允许分层引用；RETURN_STAT_READY 继续为 false` |
|
|
## 三层口径
|
|
1. `PRIMARY_STRICT_CLOSED_CASE`：234 个严格闭合 case，正收益 113 个，成功率候选读数 {primary_success}，账户贡献合计候选读数 {primary_return_sum}。
|
2. `ALL_ENTRY_DATE_COVERAGE`：覆盖 743 个 entry date，其中市场闸门打开 267 个、关闭 476 个；该口径不替代主成功率。
|
3. `STRICT_CLOSED_LOT_RECALC_ONLY`：710 个 lot，其中 692 个闭合、18 个边界；该口径只用于 lot 复算。
|
|
## 第一入口链接
|
|
- 来源全量图片总入口：[case_image_board.md]({rel_link(SOURCE_RUN_DIR / 'case_image_board.md')})
|
- 来源全量故事板：[case_story_board.md]({rel_link(SOURCE_RUN_DIR / 'case_story_board.md')})
|
- 来源分层摘要：[full_return_stat_summary.md]({rel_link(SOURCE_RUN_DIR / 'full_return_stat_summary.md')})
|
- 当前最终摘要：[final_conclusion_summary.md](final_conclusion_summary.md)
|
- 当前读数表：[final_conclusion_readouts.csv](final_conclusion_readouts.csv)
|
- 当前边界表：[final_boundary_table.csv](final_boundary_table.csv)
|
|
## 代表性 case
|
|
{sample_md}
|
|
## 批次入口
|
|
{batch_links}
|
"""
|
(RUN_DIR / "final_human_review_index.md").write_text(human_index_md, encoding="utf-8")
|
|
colleague_summary_data = {
|
"schema_version": "1.0",
|
"task_id": TASK_ID,
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"execution_audit_id": EXECUTION_AUDIT_ID,
|
"execution_review_result_message_id": EXECUTION_REVIEW_RESULT_MESSAGE_ID,
|
"return_stat_ready": False,
|
"final_execution_review_status": FINAL_EXECUTION_REVIEW_STATUS,
|
"allowed_citation_text": allowed_citation_text,
|
"primary_scope": {
|
"case_count": primary["case_count"],
|
"positive_case_count": primary["positive_case_count"],
|
"success_rate_readout": primary_success,
|
"account_return_sum_readout": primary_return_sum,
|
},
|
"coverage_scope": {
|
"entry_date_count": coverage["entry_date_count"],
|
"market_gate_open_entry_dates": coverage["market_gate_open_entry_dates"],
|
"market_gate_closed_entry_dates": coverage["market_gate_closed_entry_dates"],
|
},
|
"lot_recalc_scope": {
|
"total_lot_count": lot_recalc["total_lot_count"],
|
"closed_lot_count": lot_recalc["closed_lot_count"],
|
"boundary_lot_count": lot_recalc["unresolved_lot_count"],
|
},
|
"boundary": {
|
"case_boundary_count": boundary["case_boundary_count"],
|
"lot_boundary_count": boundary["lot_boundary_count"],
|
},
|
}
|
write_json(RUN_DIR / "final_case_summary_for_colleagues.json", colleague_summary_data)
|
|
colleague_summary_md = f"""# 无忌交易系统最终案例总结（分层引用版）
|
|
## 一句话结论
|
|
本案例已经完成最终结论引用包执行审核，审计 ID 为 `{EXECUTION_AUDIT_ID}`。当前允许把读数按分层口径写给同事看，但 `RETURN_STAT_READY=false` 继续保留，本总结不能被解读为无忌 baseline 已经得到无保留的收益 / 成功率 / 胜率 / 回撤或策略有效性结论。
|
|
## 可以引用
|
|
1. {allowed_citation_text[0]}
|
2. {allowed_citation_text[1]}
|
3. {allowed_citation_text[2]}
|
4. {allowed_citation_text[3]}
|
|
## 不可以引用
|
|
1. 不能把 `{primary_success}` 写成脱离 `PRIMARY_STRICT_CLOSED_CASE` 的完整 baseline 成功率。
|
2. 不能把 `{primary_return_sum}` 写成脱离 `PRIMARY_STRICT_CLOSED_CASE` 的完整 baseline 收益率。
|
3. 不能把 `ALL_ENTRY_DATE_COVERAGE` 或 `STRICT_CLOSED_LOT_RECALC_ONLY` 包装成主成功率或主收益率。
|
4. 不能声称无忌 baseline 策略有效性已经被证明。
|
5. 不能把 `RETURN_STAT_READY` 改为 true，也不能把案例事项标记为最终完成。
|
|
## 给人工复核的入口
|
|
- 最终人工入口：[final_human_review_index.md](final_human_review_index.md)
|
- 最终结论摘要：[final_conclusion_summary.md](final_conclusion_summary.md)
|
- 分层读数表：[final_conclusion_readouts.csv](final_conclusion_readouts.csv)
|
- 边界样本表：[final_boundary_table.csv](final_boundary_table.csv)
|
- 来源全量图片入口：[case_image_board.md]({rel_link(SOURCE_RUN_DIR / 'case_image_board.md')})
|
|
## 审计链
|
|
- 最终结论引用设计审核：`{DESIGN_AUDIT_ID}`
|
- 全量执行返修复审：`{SOURCE_EXEC_AUDIT_ID}`
|
- 最终结论引用包执行审核：`{EXECUTION_AUDIT_ID}`
|
"""
|
(RUN_DIR / "final_case_summary_for_colleagues.md").write_text(colleague_summary_md, encoding="utf-8")
|
|
banned_phrases = [
|
"无边界完整收益率",
|
"无边界完整成功率",
|
"策略有效性已证实",
|
]
|
report_text = "\n".join(
|
[
|
summary_md,
|
human_index_md,
|
colleague_summary_md,
|
config_md,
|
json.dumps(summary_data, ensure_ascii=False),
|
json.dumps(colleague_summary_data, ensure_ascii=False),
|
]
|
)
|
|
check_items: list[dict[str, Any]] = []
|
add_check(
|
check_items,
|
"SOURCE_STAGE_MATCHES_EXPECTED",
|
source_summary.get("stage") == EXPECTED_SOURCE_STAGE,
|
f"source_stage={source_summary.get('stage')}",
|
)
|
add_check(
|
check_items,
|
"SOURCE_EXEC_REREVIEW_AUDIT_MATCHES",
|
source_summary.get("execution_rereview_pass_audit_id") == SOURCE_EXEC_AUDIT_ID,
|
f"source_audit={source_summary.get('execution_rereview_pass_audit_id')}",
|
)
|
add_check(
|
check_items,
|
"SOURCE_SELF_CHECK_PASS",
|
int(source_self_check.get("fail_count", -1)) == 0,
|
f"source_self_check={source_self_check.get('overall_status')}; fail_count={source_self_check.get('fail_count')}",
|
)
|
add_check(
|
check_items,
|
"SOURCE_RETURN_STAT_READY_FALSE",
|
source_summary.get("strict_baseline_return_ready_flag") is False and return_summary.get("return_stat_ready") is False,
|
f"summary_ready={source_summary.get('strict_baseline_return_ready_flag')}; return_summary_ready={return_summary.get('return_stat_ready')}",
|
)
|
add_check(
|
check_items,
|
"SUMMARY_CASE_SCOPE_COUNTS_MATCH",
|
primary_case_count == int(primary["case_count"]) and positive_case_count == int(primary["positive_case_count"]) and case_scope_count == int(coverage["entry_date_count"]),
|
f"case_scope={case_scope_count}; primary={primary_case_count}; positive={positive_case_count}",
|
)
|
add_check(
|
check_items,
|
"SUMMARY_LOT_SCOPE_COUNTS_MATCH",
|
lot_count == int(lot_recalc["total_lot_count"]) and closed_lot_count == int(lot_recalc["closed_lot_count"]) and boundary_lot_count == int(lot_recalc["unresolved_lot_count"]) and positive_closed_lots == int(lot_recalc["positive_closed_lot_count"]),
|
f"lots={lot_count}; closed={closed_lot_count}; boundary={boundary_lot_count}; positive_closed={positive_closed_lots}",
|
)
|
add_check(
|
check_items,
|
"SUMMARY_BOUNDARY_COUNTS_MATCH",
|
case_boundary_count == int(boundary["case_boundary_count"]) and lot_boundary_count == int(boundary["lot_boundary_count"]),
|
f"case_boundary={case_boundary_count}; lot_boundary={lot_boundary_count}",
|
)
|
add_check(
|
check_items,
|
"SOURCE_LINKS_REACHABLE",
|
all(row["exists"] == "True" for row in source_artifact_rows),
|
f"source_artifacts={len(source_artifact_rows)}; missing={sum(1 for row in source_artifact_rows if row['exists'] != 'True')}",
|
)
|
add_check(
|
check_items,
|
"FINAL_REPORT_BANNED_PHRASES_ABSENT",
|
not any(phrase in report_text for phrase in banned_phrases),
|
"checked phrases: unbounded success/return and strategy-effectiveness assertions",
|
)
|
add_check(
|
check_items,
|
"FINAL_REPORT_RETURN_STAT_READY_FALSE_PRESENT",
|
"RETURN_STAT_READY=false" in report_text or '"return_stat_ready": false' in report_text,
|
"final package explicitly keeps RETURN_STAT_READY=false",
|
)
|
add_check(
|
check_items,
|
"EXECUTION_AUDIT_PASS_STATUS_PRESENT",
|
EXECUTION_AUDIT_ID in report_text and FINAL_EXECUTION_REVIEW_STATUS in json.dumps(summary_data, ensure_ascii=False),
|
f"execution_audit_id={EXECUTION_AUDIT_ID}; status={FINAL_EXECUTION_REVIEW_STATUS}",
|
)
|
add_check(
|
check_items,
|
"COLLEAGUE_SUMMARY_BOUNDARY_PRESENT",
|
"509 个 case 边界和 18 个 lot 边界均不得混入主口径" in colleague_summary_md,
|
"colleague summary keeps required boundary statement",
|
)
|
required_outputs = [
|
"final_conclusion_config.md",
|
"final_conclusion_config.json",
|
"final_conclusion_summary.md",
|
"final_conclusion_summary.json",
|
"final_conclusion_readouts.csv",
|
"final_boundary_table.csv",
|
"final_human_review_index.md",
|
"final_case_summary_for_colleagues.md",
|
"final_case_summary_for_colleagues.json",
|
"source_artifact_manifest.csv",
|
]
|
add_check(
|
check_items,
|
"MINIMUM_OUTPUT_FILES_EXIST",
|
all((RUN_DIR / name).exists() for name in required_outputs),
|
f"required_outputs={len(required_outputs)}",
|
)
|
|
fail_count = sum(1 for row in check_items if row["status"] != "PASS")
|
self_check = {
|
"schema_version": "1.0",
|
"task_id": TASK_ID,
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"stage": STAGE,
|
"overall_status": OVERALL_STATUS if fail_count == 0 else "FAIL_FOR_FINAL_CONCLUSION_EXECUTION_REVIEW",
|
"check_count": len(check_items),
|
"fail_count": fail_count,
|
"source_run_id": SOURCE_RUN_ID,
|
"source_manifest_file_count": source_manifest.get("file_count"),
|
"return_stat_ready": False,
|
"boundary": "Execution review passed for layered citation only; RETURN_STAT_READY remains false.",
|
}
|
write_json(RUN_DIR / "self_check.json", self_check)
|
write_csv(RUN_DIR / "self_check_items.csv", check_items, ["check_id", "status", "detail"])
|
self_check_md = "# 最终结论引用包自检\n\n"
|
self_check_md += f"- 当前阶段：`{STAGE}`\n"
|
self_check_md += f"- 总体状态：`{self_check['overall_status']}`\n"
|
self_check_md += f"- 检查项：{len(check_items)}\n"
|
self_check_md += f"- FAIL：{fail_count}\n\n"
|
self_check_md += "| 检查项 | 状态 | 说明 |\n|---|---|---|\n"
|
for row in check_items:
|
self_check_md += f"| `{row['check_id']}` | {row['status']} | {row['detail']} |\n"
|
(RUN_DIR / "self_check.md").write_text(self_check_md, encoding="utf-8")
|
|
manifest_files = []
|
for path in sorted(RUN_DIR.rglob("*")):
|
if path.is_dir() or path.name == "manifest.json" or "__pycache__" in path.parts or path.suffix == ".pyc":
|
continue
|
rel = Path(os.path.relpath(path, RUN_DIR)).as_posix()
|
manifest_files.append(
|
{
|
"path": rel,
|
"size": path.stat().st_size,
|
"sha256": sha256_file(path),
|
}
|
)
|
manifest = {
|
"schema_version": "1.0",
|
"task_id": TASK_ID,
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"base": ".",
|
"manifest_stage": STAGE,
|
"overall_status": self_check["overall_status"],
|
"file_count": len(manifest_files),
|
"strict_baseline_return_ready_flag": False,
|
"files": manifest_files,
|
}
|
write_json(RUN_DIR / "manifest.json", manifest)
|
|
add_check(
|
check_items,
|
"MANIFEST_COVERAGE_COMPLETE",
|
len(manifest_files) >= len(required_outputs) + 4,
|
f"manifest_files={len(manifest_files)}",
|
)
|
fail_count = sum(1 for row in check_items if row["status"] != "PASS")
|
self_check["check_count"] = len(check_items)
|
self_check["fail_count"] = fail_count
|
self_check["overall_status"] = OVERALL_STATUS if fail_count == 0 else "FAIL_FOR_FINAL_CONCLUSION_EXECUTION_REVIEW"
|
write_json(RUN_DIR / "self_check.json", self_check)
|
write_csv(RUN_DIR / "self_check_items.csv", check_items, ["check_id", "status", "detail"])
|
self_check_md = "# 最终结论引用包自检\n\n"
|
self_check_md += f"- 当前阶段：`{STAGE}`\n"
|
self_check_md += f"- 总体状态：`{self_check['overall_status']}`\n"
|
self_check_md += f"- 检查项：{len(check_items)}\n"
|
self_check_md += f"- FAIL：{fail_count}\n\n"
|
self_check_md += "| 检查项 | 状态 | 说明 |\n|---|---|---|\n"
|
for row in check_items:
|
self_check_md += f"| `{row['check_id']}` | {row['status']} | {row['detail']} |\n"
|
(RUN_DIR / "self_check.md").write_text(self_check_md, encoding="utf-8")
|
|
manifest_files = []
|
for path in sorted(RUN_DIR.rglob("*")):
|
if path.is_dir() or path.name == "manifest.json" or "__pycache__" in path.parts or path.suffix == ".pyc":
|
continue
|
rel = Path(os.path.relpath(path, RUN_DIR)).as_posix()
|
manifest_files.append(
|
{
|
"path": rel,
|
"size": path.stat().st_size,
|
"sha256": sha256_file(path),
|
}
|
)
|
manifest["overall_status"] = self_check["overall_status"]
|
manifest["file_count"] = len(manifest_files)
|
manifest["files"] = manifest_files
|
write_json(RUN_DIR / "manifest.json", manifest)
|
|
|
if __name__ == "__main__":
|
build()
|
