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()
|
