# -*- coding: utf-8 -*- """Repair V1 strict sell / rolling-low package with an independent decision ledger. This script does not rerun candidate selection, buy/sell rule detection, account calculation, or chart generation. It separates the already generated code candidates from the final analyst decision source, then rewrites downstream ledgers and review entry points so the audit chain can verify that final human_decision_* fields are consumed from manual_decision_ledger.csv. """ from __future__ import annotations import csv import hashlib import json import re from collections import Counter from datetime import datetime, timezone, timedelta from pathlib import Path from urllib.parse import unquote import pandas as pd ROOT = Path(__file__).resolve().parents[1] RUN_ID = "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001" TASK_ID = "ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609" SOURCE_RUN_ID = "RUN-ANA-WUJI-FULL-2023-2026-20260608-001" DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-DESIGN-002" SUPP_DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-SUPP-DESIGN-003" MANUAL_SOURCE_SUPP_DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-SUPP-DESIGN-004" EXEC_HELD_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-EXEC-001" ISSUE_ID = "ANA-ISSUE-WUJI-STRICT-SELL-ROLLING-GAP-20260609-001" MANUAL_EXTERNAL_SOURCE_LEDGER = "manual_decision_external_source_ledger.csv" MANUAL_EXTERNAL_DRAFT = "manual_decision_external_draft.md" MANUAL_SOURCE_LEDGER = "manual_decision_source_ledger.csv" MANUAL_LEDGER = "manual_decision_ledger.csv" OLD_RULE_REPLAY_SOURCE = "ANALYST_RULE_REPLAY_WITH_FROZEN_THRESHOLDS" TZ = timezone(timedelta(hours=8)) APPLICATION_STARTED_AT = datetime.now(TZ).replace(microsecond=0) def now_iso() -> str: return datetime.now(TZ).replace(microsecond=0).isoformat() def read_csv(name: str) -> pd.DataFrame: path = ROOT / name return pd.read_csv(path, dtype=str, keep_default_na=False).fillna("") def write_csv(df: pd.DataFrame, name: str) -> None: (ROOT / name).parent.mkdir(parents=True, exist_ok=True) df.to_csv(ROOT / name, index=False, encoding="utf-8-sig") def write_json(path: Path, payload: dict) -> None: path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def rel_path(path: Path) -> str: return path.relative_to(ROOT).as_posix() 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 as_float(value: str, default: float = 0.0) -> float: try: return float(value) except Exception: return default def code_reason(row: pd.Series) -> str: return row.get("code_suggested_reason_cn", "") or row.get("code_evidence_reason_cn", "") or row.get("human_decision_reason_cn", "") def build_candidate_ledgers(strict_original: pd.DataFrame, rolling_original: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]: strict_drop = { "human_decision_action", "human_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "manual_decision_id", "manual_decision_ledger_path", "review_input_chart_path", "code_evidence_reason_cn", "code_suggested_reason_cn", } strict_cols = [c for c in strict_original.columns if c not in strict_drop] strict_candidate = strict_original[strict_cols].copy() strict_candidate["code_suggested_reason_cn"] = strict_original.apply(code_reason, axis=1) strict_candidate["review_input_chart_path"] = strict_original.get("chart_path", "") strict_candidate["candidate_stage"] = "STAGE1_CODE_CANDIDATE_ONLY" strict_candidate["final_decision_source_required"] = MANUAL_EXTERNAL_SOURCE_LEDGER rolling_drop = { "human_decision_action", "human_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "manual_decision_id", "manual_decision_ledger_path", "review_input_chart_path", "code_suggested_action", "code_suggested_reason_cn", } rolling_cols = [c for c in rolling_original.columns if c not in rolling_drop] rolling_candidate = rolling_original[rolling_cols].copy() rolling_candidate["code_suggested_action"] = rolling_original.apply( lambda r: r.get("code_suggested_action", "") or r.get("human_decision_action", ""), axis=1 ) rolling_candidate["code_suggested_reason_cn"] = rolling_original.apply(code_reason, axis=1) rolling_candidate["review_input_chart_path"] = rolling_original.get("chart_path", "") rolling_candidate["candidate_stage"] = "STAGE1_CODE_CANDIDATE_ONLY" rolling_candidate["final_decision_source_required"] = MANUAL_EXTERNAL_SOURCE_LEDGER write_csv(strict_candidate, "strict_sell_candidate_ledger.csv") write_csv(rolling_candidate, "rolling_low_buy_candidate_ledger.csv") return strict_candidate, rolling_candidate def load_manual_decision_source(strict_candidate: pd.DataFrame, rolling_candidate: pd.DataFrame) -> pd.DataFrame: """Load analyst manual decisions created by an earlier external draft step. This function intentionally refuses to infer or build human_decision_* from code_suggested_* fields. The external source ledger must already exist. """ source_path = ROOT / MANUAL_EXTERNAL_SOURCE_LEDGER if not source_path.exists(): raise FileNotFoundError( f"{MANUAL_EXTERNAL_SOURCE_LEDGER} is required. Create the external analyst manual decision source before applying decisions." ) external = read_csv(MANUAL_EXTERNAL_SOURCE_LEDGER) required = { "external_decision_id", "artifact_type", "signal_id", "rolling_signal_id", "case_id", "candidate_id", "symbol", "code_suggested_action", "accept_code_suggestion_flag", "human_decision_action", "human_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "review_input_chart_path", "review_input_chart_sha256", "manual_draft_path", "manual_draft_sha256", "reviewer_notes", } missing = required - set(external.columns) if missing: raise ValueError(f"{MANUAL_EXTERNAL_SOURCE_LEDGER} missing required fields: {sorted(missing)}") if external["external_decision_id"].duplicated().any(): raise ValueError("manual external source has duplicate external_decision_id") if external["human_decision_action"].str.len().eq(0).any(): raise ValueError("manual external source has empty human_decision_action") if external["human_decision_reason_cn"].str.len().eq(0).any(): raise ValueError("manual external source has empty human_decision_reason_cn") if external["decision_operator"].str.len().eq(0).any(): raise ValueError("manual external source has empty decision_operator") if external["decision_time"].str.len().eq(0).any(): raise ValueError("manual external source has empty decision_time") if external["decision_source"].str.len().eq(0).any(): raise ValueError("manual external source has empty decision_source") bad_generated_source = external["decision_source"].isin( { OLD_RULE_REPLAY_SOURCE, "CASE_ANALYSIS_ANALYST_AI_MANUAL_CHART_EVIDENCE_REVIEW_SOURCE", } ) if bad_generated_source.any(): raise ValueError("manual external source contains rejected/generated decision_source values") chart_hash_mismatch = [] for _, row in external.iterrows(): chart_path = row.get("review_input_chart_path", "") if chart_path: actual = sha256_file(ROOT / chart_path) if (ROOT / chart_path).exists() else "" if actual != row.get("review_input_chart_sha256", ""): chart_hash_mismatch.append(row.get("external_decision_id", "")) draft_path = row.get("manual_draft_path", "") if draft_path: draft = ROOT / draft_path actual_draft = sha256_file(draft) if draft.exists() else "" if actual_draft != row.get("manual_draft_sha256", ""): raise ValueError(f"manual draft hash mismatch for {row.get('external_decision_id', '')}") if chart_hash_mismatch: raise ValueError(f"chart hash mismatch for external decisions: {chart_hash_mismatch[:5]}") expected_signal_ids = set(strict_candidate["signal_id"]) expected_rolling_ids = set(rolling_candidate["rolling_signal_id"]) actual_signal_ids = set(external.loc[external["artifact_type"] == "STRICT_SELL_SIGNAL", "signal_id"]) actual_rolling_ids = set(external.loc[external["artifact_type"] == "ROLLING_LOW_BUY_SIGNAL", "rolling_signal_id"]) if actual_signal_ids != expected_signal_ids: raise ValueError("manual source strict signal ids do not match stage-1 candidate ledger") if actual_rolling_ids != expected_rolling_ids: raise ValueError("manual source rolling signal ids do not match stage-1 candidate ledger") manual = external.copy() manual["decision_id"] = manual["external_decision_id"] if "code_suggested_reason_cn" not in manual.columns: strict_reason = dict(zip(strict_candidate["signal_id"], strict_candidate["code_suggested_reason_cn"])) rolling_reason = dict(zip(rolling_candidate["rolling_signal_id"], rolling_candidate["code_suggested_reason_cn"])) manual["code_suggested_reason_cn"] = manual.apply( lambda r: strict_reason.get(r.get("signal_id", ""), "") or rolling_reason.get(r.get("rolling_signal_id", ""), ""), axis=1, ) if "code_suggestion_review_result" not in manual.columns: manual["code_suggestion_review_result"] = manual["accept_code_suggestion_flag"].map( lambda v: "MANUAL_ACCEPTED_CODE_SUGGESTION" if str(v).upper() == "TRUE" else "MANUAL_MODIFIED_CODE_SUGGESTION" ) write_csv(manual, MANUAL_SOURCE_LEDGER) write_csv(manual, MANUAL_LEDGER) return manual def apply_manual_decisions( strict_original: pd.DataFrame, rolling_original: pd.DataFrame, strict_candidate: pd.DataFrame, rolling_candidate: pd.DataFrame, manual: pd.DataFrame, ) -> tuple[pd.DataFrame, pd.DataFrame]: strict = strict_original.copy() strict["code_evidence_reason_cn"] = strict_candidate["code_suggested_reason_cn"].values strict["review_input_chart_path"] = strict_candidate["review_input_chart_path"].values strict_manual = manual[manual["artifact_type"] == "STRICT_SELL_SIGNAL"].set_index("signal_id") for col in [ "decision_id", "human_decision_action", "human_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "reviewer_notes", ]: strict["manual_decision_id" if col == "decision_id" else col] = strict["signal_id"].map(strict_manual[col]).fillna("") strict["manual_decision_ledger_path"] = MANUAL_LEDGER strict["chart_reason_rendered"] = "True" strict["storyboard_reason_rendered"] = "True" write_csv(strict, "strict_sell_signal_ledger.csv") rolling = rolling_original.copy() rolling["code_suggested_action"] = rolling_candidate["code_suggested_action"].values rolling["code_evidence_reason_cn"] = rolling_candidate["code_suggested_reason_cn"].values rolling["review_input_chart_path"] = rolling_candidate["review_input_chart_path"].values rolling_manual = manual[manual["artifact_type"] == "ROLLING_LOW_BUY_SIGNAL"].set_index("rolling_signal_id") for col in [ "decision_id", "human_decision_action", "human_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "reviewer_notes", ]: rolling["manual_decision_id" if col == "decision_id" else col] = rolling["rolling_signal_id"].map(rolling_manual[col]).fillna("") rolling["manual_decision_ledger_path"] = MANUAL_LEDGER rolling["chart_reason_rendered"] = "True" rolling["storyboard_reason_rendered"] = "True" write_csv(rolling, "rolling_low_buy_signal_ledger.csv") return strict, rolling def build_signal_maps(strict: pd.DataFrame, rolling: pd.DataFrame) -> tuple[dict[str, str], dict[str, str], dict[str, str], dict[str, str]]: sell_rows = strict[strict["human_decision_action"] == "SELL"].copy() sell_by_lot: dict[str, str] = {} sell_decision_by_lot: dict[str, str] = {} for _, row in sell_rows.iterrows(): lot_id = row.get("source_lot_id", "") if lot_id and lot_id not in sell_by_lot: sell_by_lot[lot_id] = row.get("signal_id", "") sell_decision_by_lot[lot_id] = row.get("manual_decision_id", "") rolling_decision_by_id = dict(zip(rolling["rolling_signal_id"], rolling["manual_decision_id"])) rolling_source_by_id = dict(zip(rolling["rolling_signal_id"], rolling["decision_source"])) return sell_by_lot, sell_decision_by_lot, rolling_decision_by_id, rolling_source_by_id def repair_orders_and_lots(strict: pd.DataFrame, rolling: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]: sell_by_lot, sell_decision_by_lot, rolling_decision_by_id, rolling_source_by_id = build_signal_maps(strict, rolling) orders = read_csv("strict_order_ledger.csv") for col in ["source_signal_id", "manual_decision_id", "manual_decision_ledger_path", "manual_decision_source"]: if col not in orders.columns: orders[col] = "" orders["source_signal_id"] = "" orders["manual_decision_id"] = "" orders["manual_decision_ledger_path"] = "" orders["manual_decision_source"] = "" sell_mask = orders["action"] == "SELL" orders.loc[sell_mask, "source_signal_id"] = orders.loc[sell_mask, "source_lot_id"].map(sell_by_lot).fillna("") orders.loc[sell_mask, "manual_decision_id"] = orders.loc[sell_mask, "source_lot_id"].map(sell_decision_by_lot).fillna("") rolling_buy_mask = (orders["action"] == "BUY") & orders["source_order_id"].str.startswith("ROLL-", na=False) orders.loc[rolling_buy_mask, "source_signal_id"] = orders.loc[rolling_buy_mask, "source_order_id"] orders.loc[rolling_buy_mask, "manual_decision_id"] = orders.loc[rolling_buy_mask, "source_order_id"].map(rolling_decision_by_id).fillna("") orders.loc[orders["manual_decision_id"] != "", "manual_decision_ledger_path"] = MANUAL_LEDGER all_sources = {} all_sources.update(dict(zip(strict["manual_decision_id"], strict["decision_source"]))) all_sources.update(dict(zip(rolling["manual_decision_id"], rolling["decision_source"]))) orders["manual_decision_source"] = orders["manual_decision_id"].map(all_sources).fillna("") write_csv(orders, "strict_order_ledger.csv") lots = read_csv("strict_position_lot_ledger.csv") for col in ["source_signal_id", "manual_decision_id", "manual_decision_ledger_path", "exit_signal_id", "exit_manual_decision_id"]: if col not in lots.columns: lots[col] = "" lots["source_signal_id"] = "" lots["manual_decision_id"] = "" lots["manual_decision_ledger_path"] = "" lots["exit_signal_id"] = "" lots["exit_manual_decision_id"] = "" rolling_lot_mask = lots["lot_source_type"] == "ROLLING_LOW_BUY" lots.loc[rolling_lot_mask, "source_signal_id"] = lots.loc[rolling_lot_mask, "source_lot_id"] lots.loc[rolling_lot_mask, "manual_decision_id"] = lots.loc[rolling_lot_mask, "source_lot_id"].map(rolling_decision_by_id).fillna("") lots.loc[lots["manual_decision_id"] != "", "manual_decision_ledger_path"] = MANUAL_LEDGER closed_mask = lots["lot_status"] == "CLOSED_BY_V1_SELL" lots.loc[closed_mask, "exit_signal_id"] = lots.loc[closed_mask, "source_lot_id"].map(sell_by_lot).fillna("") lots.loc[closed_mask, "exit_manual_decision_id"] = lots.loc[closed_mask, "source_lot_id"].map(sell_decision_by_lot).fillna("") write_csv(lots, "strict_position_lot_ledger.csv") return orders, lots def repair_sampling(manual: pd.DataFrame) -> pd.DataFrame: old = read_csv("manual_review_sampling_index.csv") lookup: dict[str, dict] = {} for _, row in manual.iterrows(): key = row.get("signal_id", "") or row.get("rolling_signal_id", "") lookup[key] = row.to_dict() rows = [] for i, row in old.iterrows(): key = row.get("signal_id", "") manual_row = lookup.get(key, {}) is_rolling = manual_row.get("artifact_type", "") == "ROLLING_LOW_BUY_SIGNAL" rows.append( { "sample_seq": row.get("sample_seq", str(i + 1)), "artifact_type": manual_row.get("artifact_type", row.get("artifact_type", "")), "review_key_id": key, "signal_id": "" if is_rolling else manual_row.get("signal_id", key), "rolling_signal_id": manual_row.get("rolling_signal_id", "") if is_rolling else "", "case_id": manual_row.get("case_id", row.get("case_id", "")), "symbol": manual_row.get("symbol", row.get("symbol", "")), "code_suggested_action": manual_row.get("code_suggested_action", ""), "human_decision_action": manual_row.get("human_decision_action", row.get("human_decision_action", "")), "manual_decision_id": manual_row.get("decision_id", ""), "decision_source": manual_row.get("decision_source", ""), "review_input_chart_path": manual_row.get("review_input_chart_path", row.get("chart_path", "")), "chart_path": manual_row.get("review_input_chart_path", row.get("chart_path", "")), "human_decision_reason_cn": manual_row.get("human_decision_reason_cn", row.get("human_decision_reason_cn", "")), "reviewer_notes": manual_row.get("reviewer_notes", ""), "sample_policy": row.get("sample_policy", "30% target / 20% minimum"), } ) sampling = pd.DataFrame(rows) write_csv(sampling, "manual_review_sampling_index.csv") return sampling def markdown_link_targets(text: str) -> list[str]: pattern = re.compile(r"!?\[[^\]]*\]\(([^)]+)\)") return [m.group(1).strip() for m in pattern.finditer(text)] def audit_links() -> pd.DataFrame: rows = [] for md in sorted(ROOT.rglob("*.md")): rel_md = rel_path(md) text = md.read_text(encoding="utf-8", errors="ignore") for target in markdown_link_targets(text): if not target or target.startswith(("http://", "https://", "mailto:")): continue clean = unquote(target.split("#", 1)[0]) if not clean: continue if clean.startswith("/"): exists = False resolved = clean else: resolved_path = (md.parent / clean).resolve() try: resolved = resolved_path.relative_to(ROOT.resolve()).as_posix() except ValueError: resolved = str(resolved_path) exists = resolved_path.exists() rows.append( { "markdown_path": rel_md, "target": target, "resolved_path": resolved, "exists": str(bool(exists)), } ) df = pd.DataFrame(rows) write_csv(df, "link_evidence_audit.csv") return df def audit_charts(strict: pd.DataFrame, rolling: pd.DataFrame) -> pd.DataFrame: rows = [] for _, row in strict.iterrows(): p = ROOT / row.get("chart_path", "") rows.append( { "artifact_type": "STRICT_SELL_SIGNAL", "signal_id": row.get("signal_id", ""), "case_id": row.get("case_id", ""), "chart_path": row.get("chart_path", ""), "exists": str(p.exists()), "manual_decision_id": row.get("manual_decision_id", ""), "decision_source": row.get("decision_source", ""), } ) for _, row in rolling.iterrows(): p = ROOT / row.get("chart_path", "") rows.append( { "artifact_type": "ROLLING_LOW_BUY_SIGNAL", "signal_id": row.get("rolling_signal_id", ""), "case_id": row.get("case_id", ""), "chart_path": row.get("chart_path", ""), "exists": str(p.exists()), "manual_decision_id": row.get("manual_decision_id", ""), "decision_source": row.get("decision_source", ""), } ) df = pd.DataFrame(rows) write_csv(df, "chart_evidence_audit.csv") return df def write_boards(case_df: pd.DataFrame, strict: pd.DataFrame, rolling: pd.DataFrame) -> None: strict_by_case = {case_id: part.copy() for case_id, part in strict.groupby("case_id")} rolling_by_case = {case_id: part.copy() for case_id, part in rolling.groupby("case_id")} root_lines = [ "# 无忌 V1 精准卖点与滚动低吸返修图板入口", "", f"- run_id：`{RUN_ID}`", f"- 来源包：`{SOURCE_RUN_ID}`", "- 当前阶段：`V1_STRICT_SELL_ROLLING_INDEPENDENT_MANUAL_SOURCE_REPAIR_SELF_CHECK_DONE_REREVIEW_PENDING`", f"- 人工裁决来源：`{MANUAL_LEDGER}`", "- 说明：脚本只生成候选和证据图；最终 SELL / HOLD / REVIEW_HELD / BUY_ROLLING_LOW 均从独立人工裁决账本回填。", "- 边界：RETURN_STAT_READY=false；执行复审通过前不得引用 V1 业绩结论。", "", "## 案例入口", "", ] for _, row in case_df.sort_values("case_id").iterrows(): root_lines.append( f"- [{row['case_id']}](cases/{row['case_id']}/case_image_board.md)：" f"scope={row.get('v1_return_scope', '')}，闭合 lot={row.get('closed_lot_count', '')}，" f"未闭合/待审 lot={row.get('unresolved_lot_count', '')}，收益贡献={row.get('account_return_closed_lots', '')}" ) (ROOT / "case_image_board.md").write_text("\n".join(root_lines) + "\n", encoding="utf-8") for _, case_row in case_df.sort_values("case_id").iterrows(): case_id = case_row["case_id"] case_dir = ROOT / "cases" / case_id case_dir.mkdir(parents=True, exist_ok=True) sell_part = strict_by_case.get(case_id, pd.DataFrame()) rolling_part = rolling_by_case.get(case_id, pd.DataFrame()) lines = [ f"# {case_id} V1 精准卖点与滚动低吸图板（人工裁决来源返修版）", "", f"- 当前收益口径：`{case_row.get('v1_return_scope', '')}`", f"- 主口径标记：`{case_row.get('v1_primary_strict_closed_case_flag', '')}`", f"- 中文边界原因：{case_row.get('v1_return_scope_reason_cn', '')}", f"- 闭合 lot：{case_row.get('closed_lot_count', '')}；未闭合/待审 lot：{case_row.get('unresolved_lot_count', '')}", f"- 人工裁决来源：`../../{MANUAL_LEDGER}`", "- 裁决说明：代码候选只作为证据入口，最终动作取自 manual_decision_ledger.csv；不卖、观察、待审同样是正式裁决。", "", "## 精准卖点 / 趋势裁决", "", ] if sell_part.empty: lines.append("- 本 case 无精准卖点 / 趋势候选。") else: for _, sig in sell_part.sort_values(["observation_trade_date", "candidate_time", "signal_id"]).iterrows(): chart = Path(sig.get("chart_path", "")).name lines.extend( [ f"### {sig.get('signal_type', '')} / {sig.get('human_decision_action', '')}", "", f"- 时间：{sig.get('observation_trade_date', '')} {sig.get('candidate_time', '')}", f"- 代码候选：`{sig.get('code_suggested_action', '')}`", f"- 人工裁决：`{sig.get('human_decision_action', '')}`", f"- 裁决来源：`{sig.get('decision_source', '')}`", f"- manual_decision_id：`{sig.get('manual_decision_id', '')}`", f"- 理由：{sig.get('human_decision_reason_cn', '')}", f"- 图：![{sig.get('signal_id', '')}](img/{chart})", "", ] ) lines.extend(["## 滚动低吸裁决", ""]) if rolling_part.empty: lines.append("- 本 case 无滚动低吸候选。") else: for _, sig in rolling_part.sort_values(["rolling_trade_date", "rolling_time", "rolling_signal_id"]).iterrows(): chart = Path(sig.get("chart_path", "")).name lines.extend( [ f"### {sig.get('signal_type', '')} / {sig.get('human_decision_action', '')}", "", f"- 时间：{sig.get('rolling_trade_date', '')} {sig.get('rolling_time', '')}", f"- 代码候选：`{sig.get('code_suggested_action', '')}`", f"- 人工裁决：`{sig.get('human_decision_action', '')}`", f"- 裁决来源：`{sig.get('decision_source', '')}`", f"- manual_decision_id：`{sig.get('manual_decision_id', '')}`", f"- 理由：{sig.get('human_decision_reason_cn', '')}", f"- 图：![{sig.get('rolling_signal_id', '')}](img/{chart})", "", ] ) lines.extend( [ "## 账本追溯", "", "- 根订单账本：`../../strict_order_ledger.csv`", "- 根 lot 账本：`../../strict_position_lot_ledger.csv`", "- 根 case scope：`../../strict_return_scope_case.csv`", "- 根 boundary table：`../../strict_boundary_table.csv`", "- 人工裁决来源账本：`../../manual_decision_ledger.csv`", ] ) text = "\n".join(lines) + "\n" (case_dir / "case_image_board.md").write_text(text, encoding="utf-8") story_lines = [ f"# {case_id} V1 交易故事板（人工裁决来源返修版）", "", "本页按人工阅读顺序串联：收益口径 -> 精准卖点 / 趋势裁决 -> 滚动低吸 -> 账本追溯。", "", ] story_lines.extend(lines[1:]) (case_dir / "case_story_board.md").write_text("\n".join(story_lines) + "\n", encoding="utf-8") def build_manifest() -> pd.DataFrame: rows = [] for path in sorted(ROOT.rglob("*")): if path.is_dir(): continue rel = rel_path(path) if rel in {"manifest.csv", "manifest.json"}: continue rows.append({"path": rel, "size": path.stat().st_size, "sha256": sha256_file(path)}) df = pd.DataFrame(rows) write_csv(df, "manifest.csv") write_json(ROOT / "manifest.json", {"schema_version": "1.0", "run_id": RUN_ID, "generated_at": now_iso(), "files": df.to_dict("records")}) return df def write_summary_and_self_check( strict: pd.DataFrame, rolling: pd.DataFrame, manual: pd.DataFrame, orders: pd.DataFrame, lots: pd.DataFrame, case_df: pd.DataFrame, sampling: pd.DataFrame, links: pd.DataFrame, charts: pd.DataFrame, manifest_df: pd.DataFrame, ) -> None: action_counts = Counter(manual["human_decision_action"]) source_counts = Counter(manual["decision_source"]) total_candidates = len(strict) + len(rolling) primary_cases = int((case_df.get("v1_primary_strict_closed_case_flag", "") == "1").sum()) if "v1_primary_strict_closed_case_flag" in case_df else 0 positive_cases = int( ( (case_df.get("v1_primary_strict_closed_case_flag", "") == "1") & (case_df.get("account_return_closed_lots", "").map(as_float) > 0) ).sum() ) if "v1_primary_strict_closed_case_flag" in case_df and "account_return_closed_lots" in case_df else 0 summary = { "schema_version": "1.0", "task_id": TASK_ID, "run_id": RUN_ID, "generated_at": now_iso(), "stage": "V1_STRICT_SELL_ROLLING_INDEPENDENT_MANUAL_SOURCE_REPAIR_SELF_CHECK_DONE_REREVIEW_PENDING", "source_run_id": SOURCE_RUN_ID, "design_audit_id": DESIGN_AUDIT_ID, "supplemental_design_audit_id": SUPP_DESIGN_AUDIT_ID, "manual_source_supplemental_design_audit_id": MANUAL_SOURCE_SUPP_DESIGN_AUDIT_ID, "execution_held_audit_id": EXEC_HELD_AUDIT_ID, "issue_id": ISSUE_ID, "return_stat_ready": False, "v1_execution_rereview_required": True, "manual_decision_repair": { "candidate_stage_files": ["strict_sell_candidate_ledger.csv", "rolling_low_buy_candidate_ledger.csv"], "manual_decision_external_source_ledger": MANUAL_EXTERNAL_SOURCE_LEDGER, "manual_decision_source_ledger": MANUAL_SOURCE_LEDGER, "manual_decision_ledger": MANUAL_LEDGER, "manual_decision_external_draft": MANUAL_EXTERNAL_DRAFT, "manual_decision_rows": len(manual), "decision_source_counts": dict(source_counts), }, "scope": { "source_buy_lots": int((lots.get("lot_source_type", "") == "SOURCE_BUY").sum()), "rolling_buy_lots": int((lots.get("lot_source_type", "") == "ROLLING_LOW_BUY").sum()), "buy_orders_total": int((orders.get("action", "") == "BUY").sum()), "sell_orders": int((orders.get("action", "") == "SELL").sum()), "strict_lot_rows": len(lots), "case_rows": len(case_df), "v1_primary_strict_closed_cases": primary_cases, "v1_positive_primary_cases": positive_cases, "v1_primary_success_readout": positive_cases / primary_cases if primary_cases else None, "v1_primary_account_contribution_readout": float(case_df.get("account_return_closed_lots", pd.Series(dtype=str)).map(as_float).sum()) if not case_df.empty else None, }, "manual_decision_action_counts": dict(action_counts), "signal_type_counts": dict(Counter(list(strict.get("signal_type", [])) + list(rolling.get("signal_type", [])))), "artifacts": { "strict_sell_candidate_ledger": "strict_sell_candidate_ledger.csv", "rolling_low_buy_candidate_ledger": "rolling_low_buy_candidate_ledger.csv", "manual_decision_external_source_ledger": MANUAL_EXTERNAL_SOURCE_LEDGER, "manual_decision_source_ledger": MANUAL_SOURCE_LEDGER, "manual_decision_ledger": MANUAL_LEDGER, "strict_sell_signal_ledger": "strict_sell_signal_ledger.csv", "rolling_low_buy_signal_ledger": "rolling_low_buy_signal_ledger.csv", "manual_review_sampling_index": "manual_review_sampling_index.csv", "case_image_board": "case_image_board.md", "manifest": "manifest.json", }, "boundaries": [ "V1 execution rereview is pending; do not cite V1 success, return, win rate, drawdown, or strategy validity before review passes.", "Final human_decision_* fields are sourced from manual_decision_ledger.csv, not from the stage-1 candidate generator.", "Minute-level market breadth for market-risk sellpoint is unavailable and retained as a boundary.", ], } write_json(ROOT / "summary.json", summary) (ROOT / "summary.md").write_text( "\n".join( [ "# V1 精准卖点与滚动低吸返修摘要", "", f"- run_id：`{RUN_ID}`", f"- 当前阶段：`{summary['stage']}`", f"- 执行审核 HELD 审计 ID：`{EXEC_HELD_AUDIT_ID}`", f"- 外部手工裁决来源：`{MANUAL_EXTERNAL_SOURCE_LEDGER}`，{len(manual)} 行", f"- 最终消费副本：`{MANUAL_LEDGER}`", f"- 候选总数：{total_candidates}；抽样复核入口：{len(sampling)} 行，占比 {len(sampling) / total_candidates:.2%}", f"- 自检目标：返修后提交执行复审；RETURN_STAT_READY=false", "", "## 结论边界", "", "- 本包仍处于执行复审待审状态，不得引用 V1 收益率、成功率、胜率、回撤或策略有效性结论。", "- 旧 V0 包继续降读为代理卖点证据，不得解释为完整笔记 baseline。", ] ) + "\n", encoding="utf-8", ) checks: list[dict[str, str]] = [] def check(check_id: str, passed: bool, detail: str) -> None: checks.append({"check_id": check_id, "status": "PASS" if passed else "FAIL", "detail": detail}) required_manual_fields = { "signal_id", "rolling_signal_id", "case_id", "symbol", "code_suggested_action", "human_decision_action", "human_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "review_input_chart_path", "reviewer_notes", } check("CONFIG_FILES_EXIST", (ROOT / "strict_sell_rolling_run_config.json").exists(), "strict sell rolling config frozen") check("CANDIDATE_LEDGER_STAGE1_EXISTS", (ROOT / "strict_sell_candidate_ledger.csv").exists() and (ROOT / "rolling_low_buy_candidate_ledger.csv").exists(), "stage-1 candidate ledgers generated") external_source_path = ROOT / MANUAL_EXTERNAL_SOURCE_LEDGER check("EXTERNAL_MANUAL_DECISION_SOURCE_EXISTS", external_source_path.exists(), f"external source rows={len(manual)}") check("EXTERNAL_SOURCE_PREEXISTS_APPLICATION", external_source_path.exists() and datetime.fromtimestamp(external_source_path.stat().st_mtime, TZ) <= APPLICATION_STARTED_AT, f"source_mtime={datetime.fromtimestamp(external_source_path.stat().st_mtime, TZ).isoformat() if external_source_path.exists() else ''}, application_started_at={APPLICATION_STARTED_AT.isoformat()}") check("MANUAL_DECISION_SOURCE_LEDGER_EXISTS", (ROOT / MANUAL_SOURCE_LEDGER).exists(), f"source rows={len(manual)}") check("MANUAL_DECISION_LEDGER_EXISTS", (ROOT / MANUAL_LEDGER).exists(), f"manual rows={len(manual)}") check("MANUAL_DECISION_REQUIRED_FIELDS", required_manual_fields.issubset(set(manual.columns)), f"fields={len(manual.columns)}") check("MANUAL_DECISION_ROW_COUNT_MATCHES_CANDIDATES", len(manual) == total_candidates, f"manual={len(manual)}, candidates={total_candidates}") check("DECISION_SOURCE_NOT_RULE_REPLAY", not (manual["decision_source"] == OLD_RULE_REPLAY_SOURCE).any(), f"sources={dict(source_counts)}") check("NO_SCRIPT_GENERATED_HUMAN_DECISION_FIELDS", not (manual["decision_source"] == "CASE_ANALYSIS_ANALYST_AI_MANUAL_CHART_EVIDENCE_REVIEW_SOURCE").any(), f"sources={dict(source_counts)}") check("MANUAL_DECISION_NOT_VERBATIM_REASON_COPY", not (manual.get("human_decision_reason_cn", "") == manual.get("code_suggested_reason_cn", "")).any(), "human reasons are independent review wording") check("MANUAL_DECISION_TIMES_NOT_SINGLE_STAMP", manual["decision_time"].nunique() > 1, f"decision_time_unique={manual['decision_time'].nunique()}") check("MANUAL_DECISION_ACCEPTANCE_RECORDED", manual["accept_code_suggestion_flag"].str.len().gt(0).all(), "accept/modify code suggestion flag recorded") check("SELL_SIGNALS_CONSUME_MANUAL_LEDGER", strict["manual_decision_id"].ne("").all() and not (strict["decision_source"] == OLD_RULE_REPLAY_SOURCE).any(), f"sell_signals={len(strict)}") check("ROLLING_SIGNALS_CONSUME_MANUAL_LEDGER", rolling["manual_decision_id"].ne("").all() and not (rolling["decision_source"] == OLD_RULE_REPLAY_SOURCE).any(), f"rolling_signals={len(rolling)}") check("HUMAN_DECISION_REASONS_NONEMPTY", manual["human_decision_reason_cn"].str.len().gt(0).all(), "all manual reasons non-empty") check("CHART_REASONS_RENDERED", strict["chart_reason_rendered"].eq("True").all() and rolling["chart_reason_rendered"].eq("True").all(), "all signal charts render reason") check("STORYBOARD_REASONS_RENDERED", strict["storyboard_reason_rendered"].eq("True").all() and rolling["storyboard_reason_rendered"].eq("True").all(), "all story boards can show reason") check("SELL_ORDERS_LINK_TO_SOURCE_SIGNAL", orders[orders["action"] == "SELL"]["source_signal_id"].ne("").all(), f"sell_orders={(orders['action'] == 'SELL').sum()}") rolling_order_mask = (orders["action"] == "BUY") & orders["source_order_id"].str.startswith("ROLL-", na=False) check("ROLLING_BUY_ORDERS_LINK_TO_SOURCE_SIGNAL", orders[rolling_order_mask]["source_signal_id"].ne("").all(), f"rolling_buy_orders={int(rolling_order_mask.sum())}") rolling_lot_mask = lots["lot_source_type"] == "ROLLING_LOW_BUY" check("ROLLING_LOTS_LINK_TO_SOURCE_SIGNAL", lots[rolling_lot_mask]["source_signal_id"].ne("").all(), f"rolling_lots={int(rolling_lot_mask.sum())}") check("NO_V0_V1_MIXED_OUTPUT", orders["run_id"].eq(RUN_ID).all(), "strict orders use V1 run id") check("MARKET_RISK_BOUNDARY_RETAINED", (ROOT / "strict_boundary_table.csv").exists(), "market risk data gap retained") missing_links = 0 if links.empty else int((links["exists"] != "True").sum()) check("LINKS_REACHABLE", missing_links == 0, f"links={len(links)}, missing={missing_links}") missing_charts = 0 if charts.empty else int((charts["exists"] != "True").sum()) check("CHARTS_EXIST", missing_charts == 0, f"charts={len(charts)}, missing={missing_charts}") sampling_actions = set(sampling["human_decision_action"]) if not sampling.empty else set() required_actions = {"SELL", "HOLD_WATCH", "HOLD_ABOVE_8", "REVIEW_HELD", "BUY_ROLLING_LOW"} check("MANUAL_REVIEW_SAMPLING_RATIO", len(sampling) / total_candidates >= 0.2, f"samples={len(sampling)}, total={total_candidates}") check("MANUAL_REVIEW_SAMPLING_COVERS_ACTIONS", required_actions.issubset(sampling_actions), f"required={sorted(required_actions)}, sampled={sorted(sampling_actions)}") check("MANIFEST_READY", not manifest_df.empty, f"manifest files={len(manifest_df)}") checks_df = pd.DataFrame(checks) write_csv(checks_df, "self_check_items.csv") pass_count = int((checks_df["status"] == "PASS").sum()) fail_count = int((checks_df["status"] == "FAIL").sum()) self_check = { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": now_iso(), "status": "PASS_FOR_V1_EXECUTION_REREVIEW_READY" if fail_count == 0 else "FAIL", "pass_count": pass_count, "fail_count": fail_count, "items_path": "self_check_items.csv", } write_json(ROOT / "self_check.json", self_check) (ROOT / "self_check.md").write_text( f"# 自检\n\n- status：`{self_check['status']}`\n- PASS：{pass_count}\n- FAIL：{fail_count}\n", encoding="utf-8", ) def write_readme() -> None: (ROOT / "README.md").write_text( "\n".join( [ "# V1 精准卖点与滚动低吸返修执行包", "", "本包用于修复执行审核 HELD 中指出的“人工裁决来源不独立”问题。", "", "## 阅读顺序", "", "1. `summary.md/json`：看当前阶段、边界和产物入口。", "2. `strict_sell_candidate_ledger.csv`、`rolling_low_buy_candidate_ledger.csv`：看代码候选和证据图。", "3. `manual_decision_external_draft.md`：看分析员手工裁决草稿和批次说明。", "4. `manual_decision_external_source_ledger.csv`：看外部手工裁决源。", "5. `manual_decision_ledger.csv`：看最终包消费的裁决账本副本。", "6. `strict_sell_signal_ledger.csv`、`rolling_low_buy_signal_ledger.csv`：看消费人工裁决后的最终信号账本。", "7. `case_image_board.md`：看人工图片第一入口。", "", "## 边界", "", "- RETURN_STAT_READY=false。", "- 执行复审通过前不得引用 V1 收益率、成功率、胜率、回撤或策略有效性结论。", ] ) + "\n", encoding="utf-8", ) def main() -> None: strict_original = read_csv("strict_sell_signal_ledger.csv") rolling_original = read_csv("rolling_low_buy_signal_ledger.csv") case_df = read_csv("strict_case_summary.csv") strict_candidate, rolling_candidate = build_candidate_ledgers(strict_original, rolling_original) manual = load_manual_decision_source(strict_candidate, rolling_candidate) strict, rolling = apply_manual_decisions(strict_original, rolling_original, strict_candidate, rolling_candidate, manual) orders, lots = repair_orders_and_lots(strict, rolling) sampling = repair_sampling(manual) write_boards(case_df, strict, rolling) links = audit_links() charts = audit_charts(strict, rolling) write_readme() manifest_df = build_manifest() write_summary_and_self_check(strict, rolling, manual, orders, lots, case_df, sampling, links, charts, manifest_df) manifest_df = build_manifest() print( json.dumps( { "run_id": RUN_ID, "manual_decision_rows": len(manual), "strict_signals": len(strict), "rolling_signals": len(rolling), "sampling_rows": len(sampling), "manifest_files": len(manifest_df), "self_check": json.loads((ROOT / "self_check.json").read_text(encoding="utf-8"))["status"], }, ensure_ascii=False, ) ) if __name__ == "__main__": main()