from __future__ import annotations import hashlib import json import shutil from datetime import datetime from pathlib import Path from typing import Iterable import pandas as pd RUN_ID = "RUN-ANA-WUJI-EXPAND-30-20260608-001" SOURCE_RUN_ID = "RUN-ANA-WUJI-BASELINE-PILOT-20260607-001" TASK_ID = "ANA-WUJI-BASELINE-2023-2026" DESIGN_ID = "DESIGN-WUJI-EXPAND-30-20260608" DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-EXPAND-30-20260608-DESIGN-001" SOURCE_AUDIT_IDS = [ "AUDIT-ANA-WUJI-BASELINE-FLOW-20260607-001", "AUDIT-ANA-WUJI-BASELINE-PILOT-20260608-EXEC-REREVIEW-001", "AUDIT-ANA-WUJI-BASELINE-PILOT-20260608-EXIT-REVIEW-001", "AUDIT-ANA-WUJI-RETURN-STAT-PILOT-20260608-EXEC-REREVIEW-001", "AUDIT-ANA-WUJI-BASELINE-PILOT-20260608-CANDIDATE-POOL-EVIDENCE-REREVIEW-001", ] ROOT = Path(__file__).resolve().parents[1] PROJECT_ROOT = ROOT.parents[2] SOURCE_ROOT = PROJECT_ROOT / "ana-data" / "result" / SOURCE_RUN_ID NO_FUTURE_SELECTION_STATEMENT = ( "Selection uses candidate_ledger signal/entry fields only. It did not read " "buy/sell decision logs, future prices after the entry date, returns, lot " "outcomes, drawdown, or return statistics." ) def now_iso() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") 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 write_csv(df: pd.DataFrame, path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) df.to_csv(path, index=False, encoding="utf-8-sig") def write_json(data: dict, path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def bool_series(series: pd.Series) -> pd.Series: return series.astype(str).str.lower().isin(["true", "1", "yes"]) def copy_source_inputs() -> None: ROOT.mkdir(parents=True, exist_ok=True) for name in [ "candidate_ledger.csv", "candidate_generation_summary.json", "candidate_generation_summary.md", "baseline_rule_mapping.csv", "baseline_excluded_rule_table.csv", "candidate_pool_evidence_repair_summary.json", "candidate_pool_evidence_repair_summary.md", "candidate_pool_evidence_source_sample_50.csv", ]: src = SOURCE_ROOT / name if src.exists(): shutil.copy2(src, ROOT / name) target_tools = ROOT / "tools" target_tools.mkdir(parents=True, exist_ok=True) for src in sorted((SOURCE_ROOT / "tools").glob("*.py")): text = src.read_text(encoding="utf-8") text = text.replace(SOURCE_RUN_ID, RUN_ID) (target_tools / src.name).write_text(text, encoding="utf-8") def load_inputs() -> tuple[pd.DataFrame, pd.DataFrame, dict]: candidate_ledger = pd.read_csv(SOURCE_ROOT / "candidate_ledger.csv", encoding="utf-8-sig") source_case_index = pd.read_csv(SOURCE_ROOT / "case_index.csv", encoding="utf-8-sig") summary = json.loads((SOURCE_ROOT / "candidate_generation_summary.json").read_text(encoding="utf-8")) return candidate_ledger, source_case_index, summary def build_date_summary(candidate_ledger: pd.DataFrame) -> pd.DataFrame: df = candidate_ledger.copy() df["entry_trade_date"] = df["entry_trade_date"].astype(str) df["signal_trade_date"] = df["signal_trade_date"].astype(str) df["candidate_rank_num"] = pd.to_numeric(df["candidate_rank"], errors="coerce") df["strict_bool"] = bool_series(df["strict_candidate_flag"]) df["market_gate_open_bool"] = bool_series(df["market_gate_open_flag"]) for col in [ "up_count", "down_count", "stock_count", "volume_ratio", "upper_shadow_pct", "amount", ]: df[col] = pd.to_numeric(df[col], errors="coerce") grouped = ( df.groupby("entry_trade_date", as_index=False) .agg( signal_trade_date=("signal_trade_date", "first"), market_gate_status=("market_gate_status", "first"), market_gate_open_flag=("market_gate_open_bool", "max"), stock_count=("stock_count", "first"), up_count=("up_count", "first"), down_count=("down_count", "first"), candidate_count=("candidate_id", "count"), strict_candidate_count=("strict_bool", "sum"), pass_count=("candidate_status", lambda s: int((s == "PASS").sum())), risk_count=("candidate_status", lambda s: int((s == "FAKE_BREAKOUT_RISK_REVIEW").sum())), review_candidate_count=("candidate_status", lambda s: int((s != "PASS").sum())), max_volume_ratio=("volume_ratio", "max"), max_upper_shadow_pct=("upper_shadow_pct", "max"), max_amount=("amount", "max"), min_candidate_rank=("candidate_rank_num", "min"), ) .sort_values("entry_trade_date") .reset_index(drop=True) ) grouped["year"] = grouped["entry_trade_date"].str.slice(0, 4) grouped["market_gate_open_flag"] = grouped["market_gate_open_flag"].astype(bool) return grouped def sort_open_pass(pool: pd.DataFrame) -> pd.DataFrame: return pool.sort_values( ["pass_count", "max_volume_ratio", "max_upper_shadow_pct", "entry_trade_date"], ascending=[False, False, False, True], ) def sort_closed(pool: pd.DataFrame) -> pd.DataFrame: return pool.sort_values( ["candidate_count", "strict_candidate_count", "entry_trade_date"], ascending=[False, False, True], ) def sort_risk(pool: pd.DataFrame) -> pd.DataFrame: p = pool.copy() p["gate_score"] = p["market_gate_open_flag"].astype(int) return p.sort_values( ["gate_score", "risk_count", "max_upper_shadow_pct", "max_volume_ratio", "entry_trade_date"], ascending=[False, False, False, False, True], ) def make_case_row( row: pd.Series, case_id: str, role: str, bucket: str, source_case_id: str, reason: str, substitution_flag: bool = False, substitution_reason: str = "", ) -> dict: gate_open = bool(row["market_gate_open_flag"]) case_status = "SELECTED_FOR_EXPAND_REPLAY" if gate_open else "SELECTED_NO_TRADE_MARKET_GATE_CLOSED" return { "case_id": case_id, "source_case_id": source_case_id, "source_run_id": SOURCE_RUN_ID if source_case_id else "", "expand_case_role": role, "entry_trade_date": str(row["entry_trade_date"]), "signal_trade_date": str(row["signal_trade_date"]), "selection_bucket": bucket, "case_status": case_status, "market_gate_status": str(row["market_gate_status"]), "candidate_count": int(row["candidate_count"]), "strict_candidate_count": int(row["strict_candidate_count"]), "review_candidate_count": int(row["review_candidate_count"]), "pass_count": int(row["pass_count"]), "risk_count": int(row["risk_count"]), "up_count": int(row["up_count"]), "down_count": int(row["down_count"]), "stock_count": int(row["stock_count"]), "max_volume_ratio": float(row["max_volume_ratio"]), "max_upper_shadow_pct": float(row["max_upper_shadow_pct"]), "max_amount": float(row["max_amount"]), "selection_reason": reason, "selection_basis": ( "Deterministic controlled expansion sampling from candidate_ledger.csv, " "using only signal-day and entry-day candidate fields." ), "substitution_flag": bool(substitution_flag), "substitution_reason": substitution_reason, "no_future_selection_statement": NO_FUTURE_SELECTION_STATEMENT, } def add_anchor_rows( date_summary: pd.DataFrame, source_case_index: pd.DataFrame, case_rows: list[dict], selection_log: list[dict], selected_dates: set[str], ) -> None: date_lookup = date_summary.set_index("entry_trade_date") for _, src in source_case_index.sort_values("case_id").iterrows(): entry_date = str(src["entry_trade_date"]) row = date_lookup.loc[entry_date].copy() row["entry_trade_date"] = entry_date case_row = make_case_row( row=row, case_id=str(src["case_id"]), role="ANCHOR_REUSED_FROM_AUDITED_7_CASE_PILOT", bucket=str(src["selection_bucket"]), source_case_id=str(src["case_id"]), reason=( "Reuse audited 7-case structural pilot anchor exactly; this anchor was " "already checked in the prior structure pilot and return-stat prep flow." ), ) case_row["case_status"] = str(src["case_status"]) case_rows.append(case_row) selected_dates.add(entry_date) selection_log.append({**case_row, "selection_action": "ANCHOR_REUSED", "selection_order": len(case_rows)}) def choose_rows( pool: pd.DataFrame, count: int, selected_dates: set[str], sort_fn, per_year: int | None = None, years: Iterable[str] = ("2023", "2024", "2025", "2026"), ) -> list[pd.Series]: chosen: list[pd.Series] = [] available = pool[~pool["entry_trade_date"].isin(selected_dates)].copy() if per_year is not None: for year in years: yp = sort_fn(available[available["year"] == year]) for _, row in yp.head(per_year).iterrows(): if len(chosen) >= count: break date = str(row["entry_trade_date"]) if date not in selected_dates: chosen.append(row) selected_dates.add(date) if len(chosen) >= count: break if len(chosen) < count: remaining = pool[~pool["entry_trade_date"].isin(selected_dates)].copy() for _, row in sort_fn(remaining).iterrows(): if len(chosen) >= count: break date = str(row["entry_trade_date"]) chosen.append(row) selected_dates.add(date) return chosen def choose_boundary_rows(date_summary: pd.DataFrame, selected_dates: set[str]) -> list[pd.Series]: available = date_summary[~date_summary["entry_trade_date"].isin(selected_dates)].copy() ordered = available.sort_values("entry_trade_date") start_pool = ordered.head(17) end_pool = ordered.tail(12).sort_values("entry_trade_date", ascending=False) chosen: list[pd.Series] = [] for pool in [start_pool, end_pool]: for _, row in pool.iterrows(): date = str(row["entry_trade_date"]) if date not in selected_dates: chosen.append(row) selected_dates.add(date) break if len(chosen) < 3: min_date = pd.to_datetime(ordered["entry_trade_date"].min()) max_date = pd.to_datetime(ordered["entry_trade_date"].max()) remaining = date_summary[~date_summary["entry_trade_date"].isin(selected_dates)].copy() remaining["entry_dt"] = pd.to_datetime(remaining["entry_trade_date"]) remaining["boundary_distance"] = remaining["entry_dt"].map( lambda d: min(abs((d - min_date).days), abs((max_date - d).days)) ) remaining = remaining.sort_values( ["boundary_distance", "market_gate_open_flag", "candidate_count", "entry_trade_date"], ascending=[True, False, False, True], ) for _, row in remaining.iterrows(): if len(chosen) >= 3: break date = str(row["entry_trade_date"]) chosen.append(row) selected_dates.add(date) return chosen def build_case_selection(candidate_ledger: pd.DataFrame, source_case_index: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: date_summary = build_date_summary(candidate_ledger) case_rows: list[dict] = [] selection_log: list[dict] = [] selected_dates: set[str] = set() add_anchor_rows(date_summary, source_case_index, case_rows, selection_log, selected_dates) new_specs = [ ( "OPEN_STRONG_CANDIDATE", choose_rows( date_summary[(date_summary["market_gate_open_flag"]) & (date_summary["pass_count"] > 0)], 12, selected_dates, sort_open_pass, per_year=3, ), "Market gate open and PASS candidates are abundant; chosen by pass_count, volume_ratio, upper_shadow and date.", ), ( "MARKET_GATE_CLOSED_WITH_CANDIDATES", choose_rows( date_summary[~date_summary["market_gate_open_flag"]], 3, selected_dates, sort_closed, per_year=1, years=("2023", "2024", "2025"), ), "Market gate closed boundary; retained to prove no-trade status is explicit, not silently dropped.", ), ( "PREV_HIGH_REVIEW_RISK", choose_rows( date_summary[date_summary["risk_count"] > 0], 3, selected_dates, sort_risk, ), "FAKE_BREAKOUT_RISK_REVIEW / previous-high risk coverage; chosen by gate availability and risk evidence.", ), ( "MINUTE_COVERAGE_BOUNDARY", choose_boundary_rows(date_summary, selected_dates), "Near minute replay coverage boundary; chosen by entry date proximity to replay start/end.", ), ( "LOW_STRICT_OR_NO_TRADE_BOUNDARY", choose_rows( date_summary, 2, selected_dates, lambda p: p.sort_values( ["market_gate_open_flag", "strict_candidate_count", "pass_count", "candidate_count", "entry_trade_date"], ascending=[False, True, True, False, True], ), ), "Low strict-candidate / no-trade boundary coverage; chosen by low strict/pass count without outcome data.", ), ] seq = len(case_rows) + 1 for bucket, rows, reason in new_specs: for row in rows: entry_date = str(row["entry_trade_date"]) case_id = f"WUJI-EXPAND-CASE-{seq:02d}-{entry_date.replace('-', '')}" case_row = make_case_row( row=row, case_id=case_id, role="NEW_EXPAND_30_CASE", bucket=bucket, source_case_id="", reason=reason, ) case_rows.append(case_row) selection_log.append({**case_row, "selection_action": "NEW_SELECTED", "selection_order": len(case_rows)}) seq += 1 if len(case_rows) != 30: raise RuntimeError(f"Expected 30 case days, got {len(case_rows)}.") if len({r["entry_trade_date"] for r in case_rows}) != 30: raise RuntimeError("Duplicate entry_trade_date detected in selected expansion cases.") return pd.DataFrame(case_rows), pd.DataFrame(selection_log), date_summary def selected_candidates_for_cases(case_index: pd.DataFrame, candidate_ledger: pd.DataFrame) -> pd.DataFrame: rows: list[pd.DataFrame] = [] for _, case in case_index.iterrows(): day = candidate_ledger[candidate_ledger["entry_trade_date"].astype(str) == str(case["entry_trade_date"])].copy() day["candidate_rank"] = pd.to_numeric(day["candidate_rank"], errors="coerce") if str(case["selection_bucket"]) == "PREV_HIGH_REVIEW_RISK": review = day[day["candidate_status"] != "PASS"].sort_values("candidate_rank").head(2) strict = day[day["candidate_status"] == "PASS"].sort_values("candidate_rank").head(3) chosen = pd.concat([strict, review], ignore_index=True).sort_values("candidate_rank").head(5) else: strict = day[day["candidate_status"] == "PASS"].sort_values("candidate_rank").head(5) chosen = strict if len(strict) >= 5 else day.sort_values("candidate_rank").head(5) chosen = chosen.copy() chosen["case_id"] = case["case_id"] chosen["source_case_id"] = case.get("source_case_id", "") chosen["expand_case_role"] = case["expand_case_role"] chosen["case_status"] = case["case_status"] chosen["selection_bucket"] = case["selection_bucket"] chosen["case_candidate_slot"] = range(1, len(chosen) + 1) chosen["selection_reason"] = case["selection_reason"] chosen["no_future_selection_statement"] = NO_FUTURE_SELECTION_STATEMENT rows.append(chosen) return pd.concat(rows, ignore_index=True) if rows else pd.DataFrame() def build_manifest() -> dict: files = [] for path in sorted(ROOT.rglob("*")): if not path.is_file(): continue if "__pycache__" in path.parts: continue rel = path.relative_to(ROOT).as_posix() if rel == "manifest.json": continue files.append({"path": rel, "size": path.stat().st_size, "sha256": sha256_file(path)}) return { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": now_iso(), "stage": "EXPAND_RUN_CONFIG_FROZEN", "overall_status": "READY_FOR_EXPAND_EXECUTION", "files": files, } def markdown_table(df: pd.DataFrame, cols: list[str]) -> str: head = "| " + " | ".join(cols) + " |" sep = "| " + " | ".join(["---"] * len(cols)) + " |" lines = [head, sep] for _, row in df[cols].iterrows(): values = [str(row[c]).replace("\n", " ") for c in cols] lines.append("| " + " | ".join(values) + " |") return "\n".join(lines) def write_config_docs(case_index: pd.DataFrame, selection_log: pd.DataFrame, date_summary: pd.DataFrame, source_summary: dict) -> None: bucket_counts = case_index["selection_bucket"].value_counts().sort_index().to_dict() role_counts = case_index["expand_case_role"].value_counts().sort_index().to_dict() config = { "schema_version": "1.0", "task_id": TASK_ID, "run_id": RUN_ID, "source_run_id": SOURCE_RUN_ID, "design_id": DESIGN_ID, "design_audit_id": DESIGN_AUDIT_ID, "source_audit_ids": SOURCE_AUDIT_IDS, "generated_at": now_iso(), "stage": "EXPAND_RUN_CONFIG_FROZEN", "execution_scope": { "case_days_total": int(len(case_index)), "anchor_case_days": int((case_index["expand_case_role"] == "ANCHOR_REUSED_FROM_AUDITED_7_CASE_PILOT").sum()), "new_case_days": int((case_index["expand_case_role"] == "NEW_EXPAND_30_CASE").sum()), "allowed_scope": "30 controlled case days only", "not_allowed": [ "2023-2026 full-sample execution", "complete baseline return/success/win/drawdown/effectiveness conclusion", "silent sample substitution", ], }, "selection_policy": { "no_future_selection_statement": NO_FUTURE_SELECTION_STATEMENT, "source_fields": [ "candidate_ledger.csv", "candidate_generation_summary.json", "case_index.csv from audited 7-case pilot for anchors", ], "sort_policy": source_summary.get("candidate_sort_policy", {}), "bucket_counts": bucket_counts, "role_counts": role_counts, }, "source_candidate_pool_snapshot": { "candidate_rows": int(len(pd.read_csv(ROOT / "candidate_ledger.csv", encoding="utf-8-sig"))), "candidate_entry_dates": int(date_summary["entry_trade_date"].nunique()), "non_anchor_entry_dates": int(date_summary["entry_trade_date"].nunique() - 7), "market_gate_open_entry_dates": int(date_summary["market_gate_open_flag"].sum()), "market_gate_closed_entry_dates": int((~date_summary["market_gate_open_flag"]).sum()), "source_date_constraints": source_summary.get("date_constraints", {}), }, "artifacts": { "expand_run_config.md": "Human-readable frozen expansion config.", "expand_run_config.json": "Machine-readable frozen expansion config.", "case_index.csv": "Tool-compatible 30-case index.", "expand_case_index.csv": "Expanded case index with audit fields.", "expand_sample_selection_log.csv": "One row per selected case day with role, bucket, reason, and substitution status.", "expand_candidate_selection_ledger.csv": "Frozen selected candidate evidence, independent of image script rewrites.", "candidate_date_summary.csv": "Entry-date summary used for deterministic selection.", }, } write_json(config, ROOT / "expand_run_config.json") write_json(config, ROOT / "run_config.json") md_lines = [ "# 无忌交易系统 30 案例日扩样执行配置冻结", "", f"- task_id：`{TASK_ID}`", f"- run_id：`{RUN_ID}`", f"- source_run_id：`{SOURCE_RUN_ID}`", f"- design_id：`{DESIGN_ID}`", f"- design_audit_id：`{DESIGN_AUDIT_ID}`", f"- generated_at：`{config['generated_at']}`", "- stage：`EXPAND_RUN_CONFIG_FROZEN`", "", "## 选样边界", "", "- 本轮只允许执行 30 个受控案例日：7 个已审核锚点 + 23 个新增分层案例日。", "- 不直接进入 2023-2026 全量执行。", "- 不得把本轮读数写成完整 baseline 成功率、收益率、胜率、回撤或策略有效性结论。", f"- 后验排除声明：{NO_FUTURE_SELECTION_STATEMENT}", "", "## 分层计数", "", markdown_table( pd.DataFrame( [ {"item": "anchor_case_days", "count": config["execution_scope"]["anchor_case_days"]}, {"item": "new_case_days", "count": config["execution_scope"]["new_case_days"]}, *[{"item": k, "count": v} for k, v in bucket_counts.items()], ] ), ["item", "count"], ), "", "## 30 个案例日冻结表", "", markdown_table( case_index[ [ "case_id", "expand_case_role", "entry_trade_date", "signal_trade_date", "selection_bucket", "market_gate_status", "candidate_count", "pass_count", "risk_count", "substitution_flag", ] ], [ "case_id", "expand_case_role", "entry_trade_date", "signal_trade_date", "selection_bucket", "market_gate_status", "candidate_count", "pass_count", "risk_count", "substitution_flag", ], ), "", "## 审核关注点", "", "- 执行审核应检查本配置、候选账本、选样日志、选股图片、买卖裁决图片、订单账本、lot 账本、账户账本、自检和 manifest 是否互相追溯。", "- 替代样本不得静默发生；如执行阶段出现数据缺口，必须写入 `expand_sample_selection_log.csv` 或问题记录并送审。", "- 所有图片仍按无忌专项流程要求作为人工审核第一入口。", "", ] text = "\n".join(md_lines) (ROOT / "expand_run_config.md").write_text(text, encoding="utf-8") (ROOT / "run_config.md").write_text(text, encoding="utf-8") def main() -> None: copy_source_inputs() candidate_ledger, source_case_index, source_summary = load_inputs() case_index, selection_log, date_summary = build_case_selection(candidate_ledger, source_case_index) selected = selected_candidates_for_cases(case_index, candidate_ledger) write_csv(case_index, ROOT / "case_index.csv") write_csv(case_index, ROOT / "expand_case_index.csv") write_csv(selection_log, ROOT / "expand_sample_selection_log.csv") write_csv(date_summary, ROOT / "candidate_date_summary.csv") write_csv(selected, ROOT / "expand_candidate_selection_ledger.csv") write_csv(selected, ROOT / "selected_candidate_ledger.csv") snapshot = { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": now_iso(), "source_run_id": SOURCE_RUN_ID, "source_candidate_rows": int(len(candidate_ledger)), "source_entry_dates": int(date_summary["entry_trade_date"].nunique()), "selected_case_days": int(case_index["entry_trade_date"].nunique()), "selected_candidate_rows": int(len(selected)), "no_future_selection_statement": NO_FUTURE_SELECTION_STATEMENT, } write_json(snapshot, ROOT / "candidate_source_audit_snapshot.json") write_config_docs(case_index, selection_log, date_summary, source_summary) write_json(build_manifest(), ROOT / "manifest.json") print(json.dumps({ "run_id": RUN_ID, "stage": "EXPAND_RUN_CONFIG_FROZEN", "case_days": int(len(case_index)), "selected_candidate_rows": int(len(selected)), "manifest_files": len(build_manifest()["files"]), }, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()