from __future__ import annotations import csv import hashlib import json import os import re from datetime import datetime from pathlib import Path import pandas as pd import pymysql RUN_ID = "RUN-ANA-WUJI-V1-ORIGINAL-BUY-LIMITUP-SCOPE-CHECK-20260614-001" TASK_ID = "ANA-WUJI-V1-ORIGINAL-BUY-LIMITUP-SCOPE-20260614" DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-V1-ORIGINAL-BUY-LIMITUP-SCOPE-20260614-DESIGN-001" ISSUE_ID = "ANA-ISSUE-WUJI-V1-ORIGINAL-BUY-LIMITUP-SCOPE-20260614-001" ROOT = Path(__file__).resolve().parents[1] PROJECT_ROOT = ROOT.parents[2] SOURCE_V1 = PROJECT_ROOT / "ana-data/result/RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001" SOURCE_FULL = PROJECT_ROOT / "ana-data/result/RUN-ANA-WUJI-FULL-2023-2026-20260608-001" LOCAL_DB_INDEX = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md") STRICT_POLICY = { "window_trading_days": 30, "exclude_signal_day": True, "price_source_table": "a_share_daily_price", "limit_hit_field": "high_price", "previous_close_field": "prev_close", "limit_rate_policy": { "BJ": 0.30, "STAR_688": 0.20, "CHINEXT_300_301": 0.20, "MAINBOARD_DEFAULT": 0.10, }, "tolerance_pct_points": 0.05, "limit_hit_formula": "((high_price / prev_close) - 1) * 100 >= limit_rate_pct - tolerance_pct_points", "source_buy_rule": "strict_position_lot_ledger.lot_source_type == SOURCE_BUY and tranche_index == 1", "rolling_buy_rule": "strict_position_lot_ledger.lot_source_type == ROLLING_LOW_BUY and tranche_index > 1", } def now_iso() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") def read_password() -> str: env = os.environ.get("TIANXIA_MYSQL_PASSWORD") or os.environ.get("MYSQL_PWD") if env: return env text = LOCAL_DB_INDEX.read_text(encoding="utf-8") match = re.search(r"^\s*-\s*密码：`([^`]+)`", text, re.MULTILINE) if not match: raise RuntimeError("Unable to read local MySQL credential from approved local index.") return match.group(1) def get_conn(): return pymysql.connect( host="127.0.0.1", port=3306, user="root", password=read_password(), database="tianxia", charset="utf8mb4", connect_timeout=5, read_timeout=180, ) 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.relative_to(ROOT).as_posix() def write_csv(df: pd.DataFrame, name: str) -> Path: path = ROOT / name path.parent.mkdir(parents=True, exist_ok=True) df.to_csv(path, index=False, encoding="utf-8-sig") return path def write_json(data: dict, name: str) -> Path: path = ROOT / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return path def read_csv(path: Path) -> pd.DataFrame: return pd.read_csv(path, encoding="utf-8-sig") def board_policy(symbol: str) -> tuple[str, float]: s = str(symbol).upper() code = s.split(".")[0] suffix = s.split(".")[-1] if "." in s else "" if suffix == "BJ" or code.startswith(("8", "4", "9")): return "BJ", 0.30 if suffix == "SH" and code.startswith("688"): return "STAR_688", 0.20 if suffix == "SZ" and code.startswith(("300", "301")): return "CHINEXT_300_301", 0.20 return "MAINBOARD_DEFAULT", 0.10 def clean_str(value: object) -> str: text = "" if pd.isna(value) else str(value).strip() return "" if text.lower() in {"nan", "nat", "none"} else text def load_scope() -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: lots = read_csv(SOURCE_V1 / "strict_position_lot_ledger.csv") orders = read_csv(SOURCE_V1 / "strict_order_ledger.csv") selected = read_csv(SOURCE_FULL / "full_selected_candidate_ledger.csv") lots["tranche_index"] = pd.to_numeric(lots["tranche_index"], errors="coerce") source_lots = lots[(lots["lot_source_type"] == "SOURCE_BUY") & (lots["tranche_index"] == 1)].copy() rolling_lots = lots[(lots["lot_source_type"] == "ROLLING_LOW_BUY") & (lots["tranche_index"] > 1)].copy() buy_orders = orders[orders["action"] == "BUY"].copy() source_buy_orders = buy_orders[pd.to_numeric(buy_orders["tranche_index"], errors="coerce") == 1].copy() rolling_buy_orders = buy_orders[pd.to_numeric(buy_orders["tranche_index"], errors="coerce") > 1].copy() source = source_lots.merge( source_buy_orders[ [ "order_id", "source_order_id", "case_id", "candidate_id", "symbol", "trade_date", "trade_time", "price", "position_delta_pct", ] ], left_on=["source_order_id", "case_id", "symbol"], right_on=["source_order_id", "case_id", "symbol"], how="left", suffixes=("_lot", "_order"), ) source = source.merge( selected[ [ "candidate_id", "case_id", "symbol", "signal_trade_date", "entry_trade_date", "candidate_rank", "candidate_status", "strict_candidate_flag", "recent_limitup_30_flag", "last_limitup_date", "days_since_last_limitup", "run_id", "price_source_table", ] ], on=["candidate_id", "case_id", "symbol"], how="left", suffixes=("", "_candidate"), ) return source, rolling_lots, rolling_buy_orders def query_daily(symbols: list[str], min_date: str, max_date: str) -> pd.DataFrame: placeholders = ",".join(["%s"] * len(symbols)) sql = f""" SELECT trade_date, symbol, open_price, high_price, low_price, close_price, volume, amount FROM a_share_daily_price WHERE trade_date BETWEEN %s AND %s AND symbol IN ({placeholders}) ORDER BY symbol, trade_date """ params = [min_date, max_date] + symbols with get_conn() as conn: daily = pd.read_sql(sql, conn, params=params) daily["trade_date"] = pd.to_datetime(daily["trade_date"]) for col in ["open_price", "high_price", "low_price", "close_price", "volume", "amount"]: daily[col] = pd.to_numeric(daily[col], errors="coerce") daily = daily.sort_values(["symbol", "trade_date"]).reset_index(drop=True) daily["prev_close"] = daily.groupby("symbol")["close_price"].shift(1) return daily def evaluate_source_buys(source: pd.DataFrame, daily: pd.DataFrame) -> pd.DataFrame: daily_by_symbol = {symbol: g.reset_index(drop=True) for symbol, g in daily.groupby("symbol")} rows = [] tol = STRICT_POLICY["tolerance_pct_points"] for row in source.itertuples(index=False): case_id = clean_str(getattr(row, "case_id", "")) symbol = clean_str(getattr(row, "symbol", "")) candidate_id = clean_str(getattr(row, "candidate_id", "")) signal_date_raw = clean_str(getattr(row, "signal_trade_date", "")) entry_date_raw = clean_str(getattr(row, "entry_trade_date_candidate", "")) or clean_str(getattr(row, "entry_trade_date", "")) board, limit_rate = board_policy(symbol) status = "STRICT_LIMITUP_PASS" reason = "" evidence_date = "" evidence_return_pct = "" evidence_high = "" evidence_prev_close = "" window_start = "" window_end = "" window_rows = 0 max_return_pct = "" data_gap_reason = "" if not signal_date_raw: status = "SIGNAL_DATE_MISSING_HELD" reason = "候选账本缺 signal_trade_date，无法复核信号日前 30 个交易日涨停。" data_gap_reason = "signal_trade_date_missing" window = pd.DataFrame() elif symbol not in daily_by_symbol: status = "DAILY_DATA_GAP_HELD" reason = "日线源缺该 symbol，无法复核严格涨停记忆。" data_gap_reason = "symbol_daily_missing" window = pd.DataFrame() else: signal_ts = pd.Timestamp(signal_date_raw) g = daily_by_symbol[symbol] prior = g[g["trade_date"] < signal_ts].tail(30).copy() window = prior window_rows = len(prior) if not prior.empty: window_start = prior["trade_date"].iloc[0].date().isoformat() window_end = prior["trade_date"].iloc[-1].date().isoformat() prior["limit_return_pct"] = (prior["high_price"] / prior["prev_close"] - 1.0) * 100.0 prior["limit_rate_pct"] = limit_rate * 100.0 prior["strict_hit"] = prior["prev_close"].gt(0) & prior["limit_return_pct"].ge( prior["limit_rate_pct"] - tol ) max_return_pct = "" if prior["limit_return_pct"].dropna().empty else round(float(prior["limit_return_pct"].max()), 6) hits = prior[prior["strict_hit"]].copy() else: hits = pd.DataFrame() if window_rows < 30: status = "PRIOR_30_TRADING_DAY_WINDOW_INCOMPLETE_HELD" reason = f"信号日前可用日线不足 30 个交易日，实际 {window_rows}。" data_gap_reason = "prior_window_incomplete" elif hits.empty: status = "STRICT_LIMITUP_FAIL" reason = "信号日前 30 个交易日内未发现按板块阈值触及涨停。" else: hit = hits.iloc[-1] evidence_date = hit["trade_date"].date().isoformat() evidence_return_pct = round(float(hit["limit_return_pct"]), 6) evidence_high = round(float(hit["high_price"]), 6) evidence_prev_close = round(float(hit["prev_close"]), 6) reason = f"信号日前 30 个交易日内于 {evidence_date} 触及板块涨停阈值。" rows.append( { "run_id": RUN_ID, "case_id": case_id, "symbol": symbol, "strict_lot_id": clean_str(getattr(row, "strict_lot_id", "")), "source_lot_id": clean_str(getattr(row, "source_lot_id", "")), "source_order_id": clean_str(getattr(row, "source_order_id", "")), "v1_order_id": clean_str(getattr(row, "order_id", "")), "candidate_id": candidate_id, "candidate_rank": clean_str(getattr(row, "candidate_rank", "")), "signal_trade_date": signal_date_raw, "entry_trade_date": entry_date_raw, "board_policy": board, "limit_rate": limit_rate, "limit_rate_pct": limit_rate * 100.0, "tolerance_pct_points": tol, "limit_hit_field": STRICT_POLICY["limit_hit_field"], "previous_close_field": STRICT_POLICY["previous_close_field"], "window_start_trade_date": window_start, "window_end_trade_date": window_end, "window_trading_days": window_rows, "window_excludes_signal_day_flag": True, "strict_limitup_status": status, "strict_limitup_pass_flag": status == "STRICT_LIMITUP_PASS", "strict_limitup_evidence_date": evidence_date, "strict_limitup_evidence_return_pct": evidence_return_pct, "strict_limitup_evidence_high_price": evidence_high, "strict_limitup_evidence_prev_close": evidence_prev_close, "max_prior_30_high_return_pct": max_return_pct, "data_gap_reason": data_gap_reason, "upstream_recent_limitup_30_flag": clean_str(getattr(row, "recent_limitup_30_flag", "")), "upstream_last_limitup_date": clean_str(getattr(row, "last_limitup_date", "")), "upstream_days_since_last_limitup": clean_str(getattr(row, "days_since_last_limitup", "")), "evaluation_reason_cn": reason, } ) return pd.DataFrame(rows) def make_case_summary(detail: pd.DataFrame) -> pd.DataFrame: grouped = detail.groupby("case_id", dropna=False) summary = grouped.agg( source_buy_lots=("strict_lot_id", "count"), strict_pass_lots=("strict_limitup_pass_flag", "sum"), strict_fail_lots=("strict_limitup_status", lambda s: int((s == "STRICT_LIMITUP_FAIL").sum())), held_lots=("strict_limitup_status", lambda s: int(s.astype(str).str.endswith("_HELD").sum())), ).reset_index() summary["case_has_any_strict_fail_or_held"] = (summary["strict_fail_lots"] + summary["held_lots"]) > 0 summary["case_strict_all_source_buys_pass_flag"] = summary["strict_pass_lots"].eq(summary["source_buy_lots"]) summary["case_scope_status"] = summary["case_strict_all_source_buys_pass_flag"].map( {True: "ALL_ORIGINAL_BUY_STRICT_LIMITUP_PASS", False: "HAS_ORIGINAL_BUY_STRICT_LIMITUP_FAIL_OR_HELD"} ) return summary def build_manifest(files: list[Path]) -> pd.DataFrame: rows = [] for path in sorted(files, key=lambda p: rel(p)): rows.append( { "path": rel(path), "size": path.stat().st_size, "sha256": sha256_file(path), } ) return pd.DataFrame(rows) def main() -> None: ROOT.mkdir(parents=True, exist_ok=True) (ROOT / "tools").mkdir(parents=True, exist_ok=True) source, rolling_lots, rolling_orders = load_scope() symbols = sorted(source["symbol"].dropna().astype(str).unique()) min_signal = pd.to_datetime(source["signal_trade_date"]).min() max_signal = pd.to_datetime(source["signal_trade_date"]).max() pull_start = (min_signal - pd.Timedelta(days=80)).date().isoformat() pull_end = max_signal.date().isoformat() daily = query_daily(symbols, pull_start, pull_end) detail = evaluate_source_buys(source, daily) case_summary = make_case_summary(detail) board_rows = [] for symbol in sorted(detail["symbol"].unique()): board, rate = board_policy(symbol) board_rows.append({"symbol": symbol, "board_policy": board, "limit_rate": rate}) board_df = pd.DataFrame(board_rows) colleague_expected = { "source_buy_lots": 710, "rolling_low_buy_lots": 25, "strict_pass_lots": 235, "strict_fail_or_nonpass_lots": 475, "cases_with_any_nonpass": 222, } pass_count = int(detail["strict_limitup_pass_flag"].sum()) fail_or_held = int(len(detail) - pass_count) cases_nonpass = int((~case_summary["case_strict_all_source_buys_pass_flag"]).sum()) mismatch = pd.DataFrame( [ { "metric": "source_buy_lots", "colleague_readout": colleague_expected["source_buy_lots"], "this_run_readout": int(len(detail)), "match_flag": int(len(detail)) == colleague_expected["source_buy_lots"], }, { "metric": "rolling_low_buy_lots", "colleague_readout": colleague_expected["rolling_low_buy_lots"], "this_run_readout": int(len(rolling_lots)), "match_flag": int(len(rolling_lots)) == colleague_expected["rolling_low_buy_lots"], }, { "metric": "strict_pass_lots", "colleague_readout": colleague_expected["strict_pass_lots"], "this_run_readout": pass_count, "match_flag": pass_count == colleague_expected["strict_pass_lots"], }, { "metric": "strict_fail_or_nonpass_lots", "colleague_readout": colleague_expected["strict_fail_or_nonpass_lots"], "this_run_readout": fail_or_held, "match_flag": fail_or_held == colleague_expected["strict_fail_or_nonpass_lots"], }, { "metric": "cases_with_any_nonpass", "colleague_readout": colleague_expected["cases_with_any_nonpass"], "this_run_readout": cases_nonpass, "match_flag": cases_nonpass == colleague_expected["cases_with_any_nonpass"], }, ] ) source_manifest = [] for path in [ SOURCE_V1 / "strict_position_lot_ledger.csv", SOURCE_V1 / "strict_order_ledger.csv", SOURCE_FULL / "full_selected_candidate_ledger.csv", SOURCE_FULL / "candidate_generation_summary.json", ]: source_manifest.append( { "source_run_id": path.parent.name, "path": path.relative_to(PROJECT_ROOT).as_posix(), "size": path.stat().st_size, "sha256": sha256_file(path), } ) source_manifest_df = pd.DataFrame(source_manifest) files: list[Path] = [] run_config = { "schema_version": "1.0", "run_id": RUN_ID, "task_id": TASK_ID, "generated_at": now_iso(), "design_audit_id": DESIGN_AUDIT_ID, "issue_id": ISSUE_ID, "source_runs": { "v1_run": "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001", "full_run": "RUN-ANA-WUJI-FULL-2023-2026-20260608-001", }, "source_buy_identification": { "source_buy_rule": STRICT_POLICY["source_buy_rule"], "rolling_buy_exclusion_rule": STRICT_POLICY["rolling_buy_rule"], "expected_source_buy_lots": 710, "expected_rolling_low_buy_lots": 25, }, "signal_date_policy": { "primary_source": "full_selected_candidate_ledger.signal_trade_date joined by candidate_id/case_id/symbol", "entry_date_source": "full_selected_candidate_ledger.entry_trade_date", "missing_or_conflict_status": "SIGNAL_DATE_MISSING_HELD", }, "strict_limitup_policy": STRICT_POLICY, "data_sources": { "daily_price_table": "tianxia.a_share_daily_price", "daily_price_fields": ["trade_date", "symbol", "open_price", "high_price", "low_price", "close_price", "volume", "amount"], "source_v1_lot_ledger": (SOURCE_V1 / "strict_position_lot_ledger.csv").relative_to(PROJECT_ROOT).as_posix(), "source_v1_order_ledger": (SOURCE_V1 / "strict_order_ledger.csv").relative_to(PROJECT_ROOT).as_posix(), "source_full_selected_candidate_ledger": (SOURCE_FULL / "full_selected_candidate_ledger.csv").relative_to(PROJECT_ROOT).as_posix(), }, "rounding_and_tolerance": { "reason": "A-share price-limit hits are rounded to price ticks; strict threshold uses a 0.05 percentage-point tolerance and exposes exact high/prev_close return.", "tolerance_pct_points": STRICT_POLICY["tolerance_pct_points"], }, "boundary_statuses": [ "STRICT_LIMITUP_PASS", "STRICT_LIMITUP_FAIL", "SIGNAL_DATE_MISSING_HELD", "DAILY_DATA_GAP_HELD", "PRIOR_30_TRADING_DAY_WINDOW_INCOMPLETE_HELD", ], "citation_boundary": "Execution review is required before this run's readouts can be cited as formal conclusions.", } files.append(write_json(run_config, "run_config.json")) run_config_md = ROOT / "run_config.md" run_config_md.write_text( "\n".join( [ f"# {RUN_ID} run_config", "", f"- 设计审计 ID：`{DESIGN_AUDIT_ID}`", f"- 关联问题：`{ISSUE_ID}`", "- 原始新开仓 BUY：`lot_source_type == SOURCE_BUY` 且 `tranche_index == 1`。", "- 滚动低吸 BUY：`lot_source_type == ROLLING_LOW_BUY` 且 `tranche_index > 1`，本轮从新开仓严格涨停复核排除。", "- 信号日来源：`full_selected_candidate_ledger.signal_trade_date`，按 `candidate_id / case_id / symbol` 追溯。", "- 严格涨停窗口：信号日前 30 个交易日，不含信号日。", "- 涨停判定：`a_share_daily_price.high_price / prev_close - 1` 达到板块阈值，容差 0.05 个百分点。", "- 板块阈值：`.BJ` / 8、4、9 开头为 30%；`688*.SH` 为 20%；`300*.SZ` / `301*.SZ` 为 20%；其他默认 10%。", "- 边界：执行审核通过前，不引用本包读数为正式审计结论。", ] ), encoding="utf-8", ) files.append(run_config_md) files.append(write_csv(detail, "original_buy_limitup_scope_check.csv")) files.append(write_csv(case_summary, "case_limitup_scope_summary.csv")) files.append(write_csv(board_df, "symbol_board_policy_check.csv")) files.append(write_csv(mismatch, "mismatch_with_colleague_readout.csv")) files.append(write_csv(source_manifest_df, "source_artifact_manifest.csv")) self_checks = [ { "check_id": "SOURCE_BUY_LOT_COUNT_IS_710", "status": "PASS" if len(detail) == 710 else "FAIL", "detail": f"source_buy_lots={len(detail)}", }, { "check_id": "ROLLING_LOW_BUY_EXCLUDED_COUNT_IS_25", "status": "PASS" if len(rolling_lots) == 25 else "FAIL", "detail": f"rolling_low_buy_lots={len(rolling_lots)}; rolling_buy_orders={len(rolling_orders)}", }, { "check_id": "STATUS_TOTAL_EQUALS_710", "status": "PASS" if int(detail["strict_limitup_status"].count()) == 710 else "FAIL", "detail": detail["strict_limitup_status"].value_counts().to_dict(), }, { "check_id": "SOURCE_BUY_TRACE_FIELDS_COMPLETE", "status": "PASS" if not detail[["case_id", "symbol", "candidate_id", "signal_trade_date", "entry_trade_date", "strict_lot_id", "v1_order_id"]] .replace("", pd.NA) .isna() .any() .any() else "FAIL", "detail": "case_id/symbol/candidate_id/signal_trade_date/entry_trade_date/strict_lot_id/v1_order_id present for all source BUY rows.", }, { "check_id": "WINDOW_EXCLUDES_SIGNAL_DAY", "status": "PASS" if bool(detail["window_excludes_signal_day_flag"].all()) else "FAIL", "detail": "All rows use daily.trade_date < signal_trade_date.", }, { "check_id": "STRICT_WINDOW_DAY_COUNT_OR_HELD", "status": "PASS" if bool((detail["window_trading_days"].ge(30) | detail["strict_limitup_status"].eq("PRIOR_30_TRADING_DAY_WINDOW_INCOMPLETE_HELD")).all()) else "FAIL", "detail": "Rows with fewer than 30 prior trading days are marked HELD.", }, { "check_id": "BOARD_POLICY_ASSIGNED", "status": "PASS" if not detail["board_policy"].isna().any() else "FAIL", "detail": board_df["board_policy"].value_counts().to_dict(), }, { "check_id": "MATCHES_COLLEAGUE_READOUT_TABLE_CREATED", "status": "PASS" if not mismatch.empty else "FAIL", "detail": mismatch[["metric", "match_flag"]].to_dict("records"), }, { "check_id": "SOURCE_ARTIFACT_MANIFEST_CREATED", "status": "PASS" if len(source_manifest_df) == 4 else "FAIL", "detail": f"source_artifacts={len(source_manifest_df)}", }, ] self_check_df = pd.DataFrame(self_checks) files.append(write_csv(self_check_df, "self_check_items.csv")) summary = { "schema_version": "1.0", "run_id": RUN_ID, "task_id": TASK_ID, "generated_at": now_iso(), "stage": "ORIGINAL_BUY_LIMITUP_SCOPE_CHECK_DONE_READY_FOR_EXEC_REVIEW", "design_audit_id": DESIGN_AUDIT_ID, "issue_id": ISSUE_ID, "source_runs": [ "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001", "RUN-ANA-WUJI-FULL-2023-2026-20260608-001", ], "strict_policy": STRICT_POLICY, "readouts": { "source_buy_lots": int(len(detail)), "rolling_low_buy_lots_excluded": int(len(rolling_lots)), "strict_limitup_pass_lots": pass_count, "strict_limitup_fail_or_held_lots": fail_or_held, "strict_limitup_fail_lots": int((detail["strict_limitup_status"] == "STRICT_LIMITUP_FAIL").sum()), "held_lots": int(detail["strict_limitup_status"].astype(str).str.endswith("_HELD").sum()), "cases_total": int(case_summary["case_id"].nunique()), "cases_all_source_buys_pass": int(case_summary["case_strict_all_source_buys_pass_flag"].sum()), "cases_with_any_fail_or_held": cases_nonpass, }, "colleague_readout_comparison": mismatch.to_dict("records"), "conclusion_boundary": [ "This package only checks original SOURCE_BUY strict prior limit-up memory.", "ROLLING_LOW_BUY lots are excluded from original new-position buy screening.", "This package does not change V1 sell/rolling-ledger correctness or audited V1 return arithmetic.", "Execution review is required before these readouts can be cited as formal audit conclusions.", ], } files.append(write_json(summary, "summary.json")) summary_md = ROOT / "summary.md" summary_md.write_text( "\n".join( [ f"# {RUN_ID} 摘要", "", f"- 生成时间：{summary['generated_at']}", f"- 设计审计 ID：`{DESIGN_AUDIT_ID}`", f"- 关联问题：`{ISSUE_ID}`", f"- 原始新开仓 BUY lot：{len(detail)}", f"- 已排除滚动低吸 BUY lot：{len(rolling_lots)}", f"- 严格涨停通过 lot：{pass_count}", f"- 严格涨停未通过或待审 lot：{fail_or_held}", f"- 至少一只原始 BUY 未通过或待审的 case：{cases_nonpass}", "", "## 口径", "", "严格涨停窗口为信号日前 30 个交易日，不含信号日；价源使用 `a_share_daily_price.high_price / prev_close` 是否达到板块阈值，容差 0.05 个百分点。", "", "## 边界", "", "本包只复核 V1 原始新开仓 BUY 的严格近期涨停记忆，不改变 V1 卖点、滚动低吸、人工裁决、订单、lot、账户账本或已审核收益读数。执行审核通过前不得把本包读数作为正式审计结论引用。", ] ), encoding="utf-8", ) files.append(summary_md) self_check = { "run_id": RUN_ID, "generated_at": summary["generated_at"], "overall_status": "PASS_FOR_EXECUTION_REVIEW_READY" if all(item["status"] == "PASS" for item in self_checks) else "FAIL", "pass_count": sum(1 for item in self_checks if item["status"] == "PASS"), "fail_count": sum(1 for item in self_checks if item["status"] != "PASS"), "items": self_checks, } files.append(write_json(self_check, "self_check.json")) # Add this script after generated artifacts exist. files.append(Path(__file__)) manifest_df = build_manifest(files) manifest_csv = write_csv(manifest_df, "manifest.csv") write_json({"run_id": RUN_ID, "generated_at": now_iso(), "manifest_self_included": False, "files": manifest_df.to_dict("records")}, "manifest.json") # Manifest files are intentionally not self-listed; self-hashes are not stable after write. manifest_df.to_csv(manifest_csv, index=False, encoding="utf-8-sig") print(json.dumps(summary["readouts"], ensure_ascii=False, indent=2)) if __name__ == "__main__": main()