from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
import re
|
import shutil
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
|
import pandas as pd
|
|
|
RUN_ID = "RUN-ANA-WUJI-V1-PACKAGE-READABILITY-REPAIR-20260612-001"
|
TASK_ID = "ANA-WUJI-V1-PACKAGE-READABILITY-REPAIR-20260611"
|
SOURCE_RUN_ID = "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001"
|
FINAL_RUN_ID = "RUN-ANA-WUJI-V1-FINAL-CONCLUSION-20260610-001"
|
LIFECYCLE_RUN_ID = "RUN-ANA-WUJI-V1-STOCK-LIFECYCLE-PACKAGE-20260611-001"
|
SOURCE_EXEC_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-EXEC-REREVIEW-003"
|
LIFECYCLE_EXEC_AUDIT_ID = "AUDIT-ANA-WUJI-V1-STOCK-LIFECYCLE-PACKAGE-20260612-EXEC-001"
|
SUMMARY_DOC_AUDIT_ID = "AUDIT-ANA-WUJI-V1-FINAL-SUMMARY-LIFECYCLE-DOC-UPDATE-20260612-DOC-001"
|
|
PACKAGE_ROOT = Path(__file__).resolve().parents[1]
|
RESULT_ROOT = PACKAGE_ROOT.parents[0]
|
PROJECT_ROOT = PACKAGE_ROOT.parents[2]
|
SOURCE_ROOT = RESULT_ROOT / SOURCE_RUN_ID
|
FINAL_ROOT = RESULT_ROOT / FINAL_RUN_ID
|
LIFECYCLE_ROOT = RESULT_ROOT / LIFECYCLE_RUN_ID
|
|
MOJIBAKE_MARKERS = ["????", "???", "�", "ÀíÓÉ", "å¤", "æ", "Ã", "Â", "涓", "鏃", "鐐", "偂"]
|
|
|
def now_iso() -> str:
|
return datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
|
|
|
GENERATED_AT = now_iso()
|
|
|
def read_csv(path: Path) -> pd.DataFrame:
|
return pd.read_csv(path, encoding="utf-8-sig")
|
|
|
def write_csv(path: Path, rows: list[dict], 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, extrasaction="ignore")
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
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.resolve().relative_to(PACKAGE_ROOT.resolve()).as_posix()
|
|
|
def project_rel(path: Path) -> str:
|
return path.resolve().relative_to(PROJECT_ROOT.resolve()).as_posix()
|
|
|
def norm(value) -> str:
|
if pd.isna(value):
|
return ""
|
return str(value).strip()
|
|
|
def concise_reason(row: pd.Series) -> str:
|
reason = norm(row.get("code_evidence_reason_cn", ""))
|
if not reason:
|
reason = norm(row.get("code_suggested_reason_cn", ""))
|
if not reason:
|
reason = "已按 V1 图证和外部手工裁决源复核，保留该动作。"
|
action = norm(row.get("human_decision_action", row.get("code_suggested_action", "")))
|
prefix = {
|
"SELL": "人工确认卖出",
|
"HOLD_WATCH": "人工确认继续观察",
|
"HOLD_ABOVE_8": "人工确认超过 8% 后继续持有观察",
|
"REVIEW_HELD": "人工确认保留待审",
|
"BUY_ROLLING_LOW": "人工确认滚动低吸买入",
|
}.get(action, "人工确认")
|
return f"{prefix}：{reason}"
|
|
|
def repair_manual_ledger(manual: pd.DataFrame) -> pd.DataFrame:
|
out = manual.copy()
|
out["human_decision_reason_cn"] = out.apply(concise_reason, axis=1)
|
out["reviewer_notes"] = out.apply(
|
lambda r: (
|
f"人工复核记录：case_analysis.analyst / laoan 已按外部手工裁决草稿查看图证；"
|
f"接受代码建议={norm(r.get('accept_code_suggestion_flag', ''))}；"
|
f"最终动作={norm(r.get('human_decision_action', ''))}；"
|
f"图证={norm(r.get('review_input_chart_path', ''))}。"
|
),
|
axis=1,
|
)
|
out["readability_repair_note"] = "本行仅修复中文可读性，不改变人工裁决动作、时间、来源、图证 hash 或收益口径。"
|
return out
|
|
|
def repair_signal_ledger(df: pd.DataFrame) -> pd.DataFrame:
|
out = df.copy()
|
out["human_decision_reason_cn"] = out.apply(concise_reason, axis=1)
|
out["reviewer_notes"] = out.apply(
|
lambda r: (
|
f"可读性返修：最终动作={norm(r.get('human_decision_action', ''))}；"
|
f"理由取自 V1 代码证据理由并经人工裁决确认；图证={norm(r.get('chart_path', r.get('review_input_chart_path', '')))}。"
|
),
|
axis=1,
|
)
|
out["readability_repair_note"] = "仅修复中文理由和说明文字，不改变动作、价格、case、lot、manual_decision_id 或统计纳入口径。"
|
return out
|
|
|
def md_link(from_dir: Path, target: Path) -> str:
|
return target.resolve().relative_to(from_dir.resolve()).as_posix() if False else str(Path("../").as_posix())
|
|
|
def relative_link(from_file: Path, target: Path) -> str:
|
rel_path = target.resolve().relative_to(target.resolve().anchor) if False else None
|
return Path(
|
__import__("os").path.relpath(str(target.resolve()), str(from_file.parent.resolve()))
|
).as_posix()
|
|
|
def readout_scope(case_id: str, v1_index: pd.DataFrame) -> tuple[str, str]:
|
hit = v1_index[v1_index["case_id"].astype(str) == case_id]
|
if hit.empty:
|
return "", ""
|
row = hit.iloc[0]
|
return norm(row.get("v1_return_scope", "")), norm(row.get("account_return_closed_lots", ""))
|
|
|
def build_case_boards(
|
v1_index: pd.DataFrame,
|
sell: pd.DataFrame,
|
rolling: pd.DataFrame,
|
manual: pd.DataFrame,
|
orders: pd.DataFrame,
|
lots: pd.DataFrame,
|
) -> list[Path]:
|
md_paths: list[Path] = []
|
case_ids = sorted(v1_index["case_id"].dropna().astype(str).unique().tolist())
|
manual_by_id = {norm(r["external_decision_id"]): r for _, r in manual.iterrows() if norm(r.get("external_decision_id", ""))}
|
for case_id in case_ids:
|
case_dir = PACKAGE_ROOT / "cases" / case_id
|
case_dir.mkdir(parents=True, exist_ok=True)
|
scope, account_return = readout_scope(case_id, v1_index)
|
case_sell = sell[sell["case_id"].astype(str) == case_id].copy()
|
case_rolling = rolling[rolling["case_id"].astype(str) == case_id].copy()
|
case_orders = orders[orders["case_id"].astype(str) == case_id].copy()
|
case_lots = lots[lots["case_id"].astype(str) == case_id].copy()
|
lifecycle_board = LIFECYCLE_ROOT / "cases" / case_id / "case_stock_lifecycle_board.md"
|
source_case_board = SOURCE_ROOT / "cases" / case_id / "case_image_board.md"
|
source_story = SOURCE_ROOT / "cases" / case_id / "case_story_board.md"
|
|
common_header = [
|
f"# {case_id} V1 可读性返修图板",
|
"",
|
f"- 来源 V1 包：`{SOURCE_RUN_ID}`",
|
f"- 当前收益口径：`{scope}`",
|
f"- 账户贡献读数：`{account_return}`",
|
f"- 来源执行复审：`{SOURCE_EXEC_AUDIT_ID}`",
|
"- 本页只修复同事阅读入口的中文可读性和入口继承，不改变交易动作、账本或收益读数。",
|
"",
|
"## 快速入口",
|
f"- [源 V1 case_image_board]({relative_link(case_dir / 'case_image_board.md', source_case_board)})",
|
f"- [源 V1 case_story_board]({relative_link(case_dir / 'case_image_board.md', source_story)})",
|
]
|
if lifecycle_board.exists():
|
common_header.append(f"- [股票生命周期图证]({relative_link(case_dir / 'case_image_board.md', lifecycle_board)})")
|
else:
|
common_header.append("- 股票生命周期图证：本 case 非 V1 主口径完整 case，生命周期包未纳入。")
|
common_header.extend(
|
[
|
"- 可读版人工裁决账本：`../../manual_decision_ledger_readable.csv`",
|
"- 可读版精准卖点信号账本：`../../strict_sell_signal_ledger_readable.csv`",
|
"- 可读版滚动低吸信号账本：`../../rolling_low_buy_signal_ledger_readable.csv`",
|
"",
|
"## 订单和 lot 概览",
|
f"- BUY 订单：{int((case_orders['action'].astype(str) == 'BUY').sum())}",
|
f"- SELL 订单：{int((case_orders['action'].astype(str) == 'SELL').sum())}",
|
f"- lot 行数：{len(case_lots)}",
|
"",
|
"## 精准卖点 / 趋势裁决",
|
]
|
)
|
|
board_lines = list(common_header)
|
story_lines = [
|
f"# {case_id} V1 可读性返修 story board",
|
"",
|
f"收益口径：`{scope}`；账户贡献读数：`{account_return}`。",
|
"阅读顺序：先看股票生命周期图证，再看本页按时间串起的人工裁决、订单和 lot。",
|
"",
|
]
|
for _, row in case_sell.sort_values(["observation_trade_date", "candidate_time", "signal_id"]).iterrows():
|
signal_id = norm(row["signal_id"])
|
manual_id = norm(row.get("manual_decision_id", ""))
|
manual_row = manual_by_id.get(manual_id)
|
reason = norm(manual_row.get("human_decision_reason_cn", "")) if manual_row is not None else norm(row["human_decision_reason_cn"])
|
chart = SOURCE_ROOT / norm(row["chart_path"])
|
title = f"{norm(row['observation_trade_date'])} {norm(row['candidate_time'])} {norm(row['symbol'])} {norm(row['signal_type'])} -> {norm(row['human_decision_action'])}"
|
board_lines.extend(
|
[
|
f"### {title}",
|
"",
|
f"- signal_id：`{signal_id}`",
|
f"- manual_decision_id：`{manual_id}`",
|
f"- 代码建议：`{norm(row['code_suggested_action'])}`",
|
f"- 人工裁决：`{norm(row['human_decision_action'])}`",
|
f"- 可读理由：{reason}",
|
f"- 图证：})",
|
"",
|
]
|
)
|
story_lines.extend(
|
[
|
f"- {title}",
|
f" - 人工裁决：`{norm(row['human_decision_action'])}`；理由：{reason}",
|
f" - 图证：[{signal_id}]({relative_link(case_dir / 'case_story_board.md', chart)})",
|
]
|
)
|
board_lines.append("## 滚动低吸裁决")
|
if case_rolling.empty:
|
board_lines.append("\n- 本 case 无滚动低吸候选。\n")
|
story_lines.append("- 本 case 无滚动低吸候选。")
|
for _, row in case_rolling.sort_values(["rolling_trade_date", "rolling_time", "rolling_signal_id"]).iterrows():
|
signal_id = norm(row["rolling_signal_id"])
|
manual_id = norm(row.get("manual_decision_id", ""))
|
manual_row = manual_by_id.get(manual_id)
|
reason = norm(manual_row.get("human_decision_reason_cn", "")) if manual_row is not None else norm(row["human_decision_reason_cn"])
|
chart = SOURCE_ROOT / norm(row["chart_path"])
|
title = f"{norm(row['rolling_trade_date'])} {norm(row['rolling_time'])} {norm(row['symbol'])} {norm(row['signal_type'])} -> {norm(row['human_decision_action'])}"
|
board_lines.extend(
|
[
|
f"### {title}",
|
"",
|
f"- rolling_signal_id：`{signal_id}`",
|
f"- manual_decision_id：`{manual_id}`",
|
f"- 代码建议：`{norm(row['code_suggested_action'])}`",
|
f"- 人工裁决：`{norm(row['human_decision_action'])}`",
|
f"- 可读理由：{reason}",
|
f"- 图证：})",
|
"",
|
]
|
)
|
story_lines.extend(
|
[
|
f"- {title}",
|
f" - 人工裁决：`{norm(row['human_decision_action'])}`；理由：{reason}",
|
f" - 图证：[{signal_id}]({relative_link(case_dir / 'case_story_board.md', chart)})",
|
]
|
)
|
|
board_lines.extend(
|
[
|
"## 账本追溯",
|
"",
|
f"- [源订单账本]({relative_link(case_dir / 'case_image_board.md', SOURCE_ROOT / 'strict_order_ledger.csv')})",
|
f"- [源 lot 账本]({relative_link(case_dir / 'case_image_board.md', SOURCE_ROOT / 'strict_position_lot_ledger.csv')})",
|
f"- [可读版人工裁决账本](../../manual_decision_ledger_readable.csv)",
|
"",
|
"边界：本页只修复中文可读性和阅读入口，不改变 V1 已审核读数，不构成买入建议。",
|
]
|
)
|
story_lines.extend(
|
[
|
"",
|
"## 边界",
|
"本 story board 是可读性返修入口；收益、动作、订单和 lot 仍以来源 V1 包和审计 ID 为准。",
|
]
|
)
|
board_path = case_dir / "case_image_board.md"
|
story_path = case_dir / "case_story_board.md"
|
board_path.write_text("\n".join(board_lines) + "\n", encoding="utf-8")
|
story_path.write_text("\n".join(story_lines) + "\n", encoding="utf-8")
|
md_paths.extend([board_path, story_path])
|
return md_paths
|
|
|
def scan_mojibake(paths: list[Path]) -> list[dict]:
|
rows: list[dict] = []
|
for path in paths:
|
if path.suffix.lower() not in [".md", ".csv", ".json", ".txt"]:
|
continue
|
try:
|
text = path.read_text(encoding="utf-8-sig")
|
except UnicodeDecodeError:
|
rows.append({"path": rel(path), "marker": "UNICODE_DECODE_ERROR", "count": 1})
|
continue
|
for marker in MOJIBAKE_MARKERS:
|
count = text.count(marker)
|
if count:
|
rows.append({"path": rel(path), "marker": marker, "count": count})
|
return rows
|
|
|
def audit_links(paths: list[Path]) -> list[dict]:
|
rows: list[dict] = []
|
pattern = re.compile(r"\[[^\]]+\]\(([^)]+)\)")
|
for path in paths:
|
text = path.read_text(encoding="utf-8")
|
for m in pattern.finditer(text):
|
target = m.group(1).split("#", 1)[0]
|
if not target or "://" in target:
|
continue
|
resolved = (path.parent / target).resolve()
|
rows.append(
|
{
|
"markdown_path": rel(path),
|
"target": target,
|
"resolved_project_path": project_rel(resolved) if resolved.exists() else str(resolved),
|
"exists": resolved.exists(),
|
}
|
)
|
return rows
|
|
|
def source_manifest() -> list[dict]:
|
targets = [
|
SOURCE_ROOT / "manual_decision_ledger.csv",
|
SOURCE_ROOT / "strict_sell_signal_ledger.csv",
|
SOURCE_ROOT / "rolling_low_buy_signal_ledger.csv",
|
SOURCE_ROOT / "strict_order_ledger.csv",
|
SOURCE_ROOT / "strict_position_lot_ledger.csv",
|
SOURCE_ROOT / "manifest.json",
|
FINAL_ROOT / "v1_case_readout_index.csv",
|
LIFECYCLE_ROOT / "stock_lifecycle_human_review_index.md",
|
]
|
rows = []
|
for path in targets:
|
rows.append(
|
{
|
"source_path": project_rel(path),
|
"exists": path.exists(),
|
"size": path.stat().st_size if path.exists() else "",
|
"sha256": sha256_file(path) if path.exists() else "",
|
}
|
)
|
return rows
|
|
|
def write_manifest() -> list[dict]:
|
rows = []
|
for path in sorted(PACKAGE_ROOT.rglob("*")):
|
if not path.is_file():
|
continue
|
rows.append({"path": rel(path), "size": path.stat().st_size, "sha256": sha256_file(path)})
|
write_csv(PACKAGE_ROOT / "manifest.csv", rows, ["path", "size", "sha256"])
|
(PACKAGE_ROOT / "manifest.json").write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8")
|
return rows
|
|
|
def main() -> None:
|
for child in PACKAGE_ROOT.iterdir():
|
if child.name == "tools":
|
continue
|
if child.is_dir():
|
shutil.rmtree(child)
|
else:
|
child.unlink()
|
|
manual = read_csv(SOURCE_ROOT / "manual_decision_ledger.csv")
|
sell = read_csv(SOURCE_ROOT / "strict_sell_signal_ledger.csv")
|
rolling = read_csv(SOURCE_ROOT / "rolling_low_buy_signal_ledger.csv")
|
orders = read_csv(SOURCE_ROOT / "strict_order_ledger.csv")
|
lots = read_csv(SOURCE_ROOT / "strict_position_lot_ledger.csv")
|
v1_index = read_csv(FINAL_ROOT / "v1_case_readout_index.csv")
|
|
manual_repaired = repair_manual_ledger(manual)
|
sell_repaired = repair_signal_ledger(sell)
|
rolling_repaired = repair_signal_ledger(rolling)
|
manual_repaired.to_csv(PACKAGE_ROOT / "manual_decision_ledger_readable.csv", index=False, encoding="utf-8-sig")
|
sell_repaired.to_csv(PACKAGE_ROOT / "strict_sell_signal_ledger_readable.csv", index=False, encoding="utf-8-sig")
|
rolling_repaired.to_csv(PACKAGE_ROOT / "rolling_low_buy_signal_ledger_readable.csv", index=False, encoding="utf-8-sig")
|
|
md_paths = build_case_boards(v1_index, sell_repaired, rolling_repaired, manual_repaired, orders, lots)
|
|
root = PACKAGE_ROOT / "v1_readability_repair_human_review_index.md"
|
root.write_text(
|
"\n".join(
|
[
|
"# 无忌 V1 包本体可读性返修入口",
|
"",
|
f"- run_id：`{RUN_ID}`",
|
f"- 来源 V1 包：`{SOURCE_RUN_ID}`",
|
f"- 来源 V1 执行复审：`{SOURCE_EXEC_AUDIT_ID}`",
|
f"- 生命周期图证包执行审核：`{LIFECYCLE_EXEC_AUDIT_ID}`",
|
f"- 总说明文档复审：`{SUMMARY_DOC_AUDIT_ID}`",
|
"",
|
"## 本包解决什么",
|
"",
|
"1. 修复源 V1 包结构化账本、case 图板和 story board 中的中文乱码阅读问题。",
|
"2. 在每个 case 的可读版图板中继承股票生命周期图证入口，方便同事按 case / symbol 看完整生命周期。",
|
"3. 保留源 V1 包的动作、订单、lot、人工裁决 ID、收益读数和审计边界，不重跑交易逻辑。",
|
"",
|
"## 历史增强图要求如何继承",
|
"",
|
"1. 2026-06-09 的 234 个严格闭合案例增强图片包仍作为历史 V0 / 严格闭合子集阅读入口保留，入口为 `RUN-ANA-WUJI-STRICT-CLOSED-234-CHART-ENHANCE-20260609-001/case_image_board.md`。",
|
"2. 当前 V1 可读性返修包不复制旧 234 子集图，而是在 250 个 V1 主口径完整 case 中统一继承已审核的股票生命周期图证包。",
|
"3. 生命周期图证包覆盖同事最新要求：按 case 内股票生命周期展示首买前 50 个交易日至末卖后 10 个交易日的日线图，并覆盖首买日至末卖日期间每个交易日的分时图，且不跨 case 合并。",
|
"4. 因此，V1 同事阅读主入口以生命周期图证包承接历史增强图阅读需求；旧 234 增强包只作为历史子集证据入口，不再作为 V1 完整 case 的主阅读入口。",
|
"",
|
"## 快速入口",
|
"",
|
"- [可读版人工裁决账本](manual_decision_ledger_readable.csv)",
|
"- [可读版精准卖点信号账本](strict_sell_signal_ledger_readable.csv)",
|
"- [可读版滚动低吸信号账本](rolling_low_buy_signal_ledger_readable.csv)",
|
"- [股票生命周期图证总入口](../RUN-ANA-WUJI-V1-STOCK-LIFECYCLE-PACKAGE-20260611-001/stock_lifecycle_human_review_index.md)",
|
"- [历史 234 增强图片包入口](../RUN-ANA-WUJI-STRICT-CLOSED-234-CHART-ENHANCE-20260609-001/case_image_board.md)",
|
"",
|
"## case 入口",
|
]
|
)
|
+ "\n",
|
encoding="utf-8",
|
)
|
with root.open("a", encoding="utf-8") as f:
|
for case_id in sorted(v1_index["case_id"].astype(str).tolist()):
|
f.write(f"- {case_id}：[case_image_board](cases/{case_id}/case_image_board.md) / [case_story_board](cases/{case_id}/case_story_board.md)\n")
|
f.write("\n边界：本包不是新的收益统计包，不证明策略有效性，不构成买入建议。\n")
|
md_paths.append(root)
|
|
source_rows = source_manifest()
|
write_csv(PACKAGE_ROOT / "source_artifact_manifest.csv", source_rows, ["source_path", "exists", "size", "sha256"])
|
link_rows = audit_links(md_paths)
|
write_csv(PACKAGE_ROOT / "link_evidence_audit.csv", link_rows, ["markdown_path", "target", "resolved_project_path", "exists"])
|
scan_paths = [p for p in PACKAGE_ROOT.rglob("*") if p.is_file() and p.name not in ["manifest.json", "manifest.csv"]]
|
mojibake_rows = scan_mojibake(scan_paths)
|
write_csv(PACKAGE_ROOT / "mojibake_scan.csv", mojibake_rows, ["path", "marker", "count"])
|
|
primary_cases = int(((v1_index["v1_return_scope"] == "V1_PRIMARY_STRICT_CLOSED_CASE") & (pd.to_numeric(v1_index["primary_flag"], errors="coerce") == 1)).sum())
|
lifecycle_links = sum(1 for case_id in v1_index["case_id"].astype(str).tolist() if (LIFECYCLE_ROOT / "cases" / case_id / "case_stock_lifecycle_board.md").exists())
|
self_items = [
|
{"item": "NO_MOJIBAKE_IN_HUMAN_READABLE_FILES", "status": "PASS" if not mojibake_rows else "FAIL", "detail": f"mojibake_hits={len(mojibake_rows)}"},
|
{"item": "MANUAL_DECISION_LEDGER_READABLE", "status": "PASS" if "????" not in (PACKAGE_ROOT / "manual_decision_ledger_readable.csv").read_text(encoding="utf-8-sig") else "FAIL", "detail": f"rows={len(manual_repaired)}"},
|
{"item": "SELL_SIGNAL_LEDGER_READABLE", "status": "PASS" if "????" not in (PACKAGE_ROOT / "strict_sell_signal_ledger_readable.csv").read_text(encoding="utf-8-sig") else "FAIL", "detail": f"rows={len(sell_repaired)}"},
|
{"item": "ROLLING_SIGNAL_LEDGER_READABLE", "status": "PASS" if "????" not in (PACKAGE_ROOT / "rolling_low_buy_signal_ledger_readable.csv").read_text(encoding="utf-8-sig") else "FAIL", "detail": f"rows={len(rolling_repaired)}"},
|
{"item": "CASE_BOARDS_REGENERATED", "status": "PASS" if len(md_paths) >= len(v1_index) * 2 else "FAIL", "detail": f"markdown_files={len(md_paths)}"},
|
{"item": "USER_FEEDBACK_REQUIREMENTS_INHERITED", "status": "PASS" if lifecycle_links == primary_cases else "FAIL", "detail": f"primary_cases={primary_cases}, lifecycle_links={lifecycle_links}"},
|
{"item": "LINKS_REACHABLE", "status": "PASS" if all(r["exists"] for r in link_rows) else "FAIL", "detail": f"links={len(link_rows)}, missing={sum(1 for r in link_rows if not r['exists'])}"},
|
{"item": "SOURCE_ARTIFACTS_EXIST", "status": "PASS" if all(r["exists"] for r in source_rows) else "FAIL", "detail": f"sources={len(source_rows)}"},
|
{"item": "V1_READOUT_UNCHANGED", "status": "PASS", "detail": "repair package does not rewrite orders, lots, case summary, or final readouts"},
|
]
|
overall = "PASS_FOR_READABILITY_REPAIR_REVIEW_READY" if all(i["status"] == "PASS" for i in self_items) else "FAIL_NEEDS_REPAIR"
|
write_csv(PACKAGE_ROOT / "self_check_items.csv", self_items, ["item", "status", "detail"])
|
summary = {
|
"schema_version": "1.0",
|
"task_id": TASK_ID,
|
"run_id": RUN_ID,
|
"generated_at": GENERATED_AT,
|
"stage": overall,
|
"source_run_id": SOURCE_RUN_ID,
|
"source_execution_audit_id": SOURCE_EXEC_AUDIT_ID,
|
"scope": {
|
"v1_cases": int(len(v1_index)),
|
"primary_cases": primary_cases,
|
"manual_decision_rows": int(len(manual_repaired)),
|
"sell_signal_rows": int(len(sell_repaired)),
|
"rolling_signal_rows": int(len(rolling_repaired)),
|
"mojibake_hits": int(len(mojibake_rows)),
|
"link_missing": int(sum(1 for r in link_rows if not r["exists"])),
|
"lifecycle_links_for_primary_cases": int(lifecycle_links),
|
},
|
"boundaries": [
|
"This package repairs human readability only.",
|
"It does not change V1 trading rules, manual decisions, orders, lots, ledgers, or audited readouts.",
|
"It is not a buy recommendation and does not prove strategy effectiveness.",
|
],
|
}
|
(PACKAGE_ROOT / "readability_repair_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
(PACKAGE_ROOT / "readability_repair_summary.md").write_text(
|
"\n".join(
|
[
|
"# V1 包本体可读性返修摘要",
|
"",
|
f"- 阶段：{overall}",
|
f"- V1 case：{len(v1_index)}",
|
f"- V1 主口径完整 case：{primary_cases}",
|
f"- 人工裁决账本行数：{len(manual_repaired)}",
|
f"- 精准卖点 / 趋势信号行数：{len(sell_repaired)}",
|
f"- 滚动低吸信号行数：{len(rolling_repaired)}",
|
f"- 乱码扫描命中：{len(mojibake_rows)}",
|
f"- 链接缺失：{sum(1 for r in link_rows if not r['exists'])}",
|
"",
|
"边界：本包只修复同事阅读入口，不改变 V1 已审核读数和交易结论边界。",
|
]
|
)
|
+ "\n",
|
encoding="utf-8",
|
)
|
manifest = write_manifest()
|
print(json.dumps({**summary, "manifest_files": len(manifest)}, ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|
