from __future__ import annotations import csv import hashlib import json import os import re from collections import Counter, defaultdict from datetime import datetime, timezone, timedelta from pathlib import Path import pandas as pd import pymysql RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001" ROOT = Path(__file__).resolve().parents[1] PROJECT_ROOT = ROOT.parents[2] LOCAL_DB_INDEX = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md") SOURCE_NOTE = PROJECT_ROOT / "ana-doc" / "wuji" / "profile" / "source_note" / "笔记精简版.md" FULL_SOURCE_RUN = PROJECT_ROOT / "ana-data" / "result" / "RUN-ANA-WUJI-FULL-2023-2026-20260608-001" V1_SOURCE_RUN = PROJECT_ROOT / "ana-data" / "result" / "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001" TZ = timezone(timedelta(hours=8)) CONFIG = { "schema_version": "1.0", "run_id": RUN_ID, "stage": "STRICT_NOTE_FULL_RERUN_STRICT_BUY_POOL_REBUILT_EXECUTION_PREP_READY", "source_note": "ana-doc/wuji/profile/source_note/笔记精简版.md", "source_runs": { "old_full_run": "RUN-ANA-WUJI-FULL-2023-2026-20260608-001", "old_v1_sell_rolling_run": "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001", }, "data_window": { "pull_daily_start": "2022-10-01", "minute_replay_entry_start": "2023-03-24", "minute_replay_entry_end": "2026-04-20", "calendar_start": "2022-10-01", "calendar_end": "2026-04-20", }, "data_sources": { "daily_price_table": "tianxia.a_share_daily_price", "market_breadth_table": "tianxia.ts_market_breadth_daily_cache", "calendar_table": "tianxia.a_share_trading_calendar", "price_adjustment": "source_table_as_is", "limitup_price_source": "high_price / prev_close - 1", "volume_source": "volume", "amount_source": "amount", }, "strict_buy_rules": { "prior_limitup_window_trading_days": 30, "prior_limitup_excludes_signal_day": True, "limitup_tolerance_rate": 0.0005, "limitup_rates": { "MAINBOARD_DEFAULT": 0.10, "STAR_688_SH": 0.20, "CHINEXT_300_301_SZ": 0.20, "BEIJING_BJ": 0.30, }, "volume_ratio_prev5_min": 2.0, "volume_history_required_trading_days": 5, "pullback_from_latest_prior_limitup_close_pct_max": -3.0, "upper_shadow_pct_proxy_min": 3.0, "upper_shadow_range_ratio_proxy_min": 0.4, "prev_high_window_trading_days": 60, "prev_high_min_history_trading_days": 20, "prev_high_ref_policy": "FIRST_PREVIOUS_HIGH_IN_60D_WINDOW", "prev_high_volume_filter": "if signal high touches previous 60-day high, signal volume must exceed reference high-day volume", "market_gate_policy": "SIGNAL_DAY_UP_3000", "market_gate_up_count_min": 3000, "selection_top_n_per_entry_date": 5, "sort_policy": "upper_shadow_pct desc -> volume_ratio desc -> amount desc -> symbol asc", }, "manual_review_boundary": { "bottom_support_strength": "SUPPORT_REVIEW_REQUIRED", "buy_point_confirmation": "MANUAL_OR_AI_MANUAL_REVIEW_REQUIRED", "strict_sell_trend_rolling_actions": "EXTERNAL_MANUAL_DECISION_SOURCE_REQUIRED", }, "citation_boundary": [ "This package rebuilds the strict BUY candidate pool only; it is not a completed V1 performance rerun.", "Do not cite success rate, return, win rate, drawdown, or strategy effectiveness from this package.", "Old V1 readouts remain down-read as based on the upstream wide/proxy BUY pool until strict replay passes review.", ], } def now_iso() -> str: return datetime.now(TZ).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(obj: dict, name: str) -> Path: path = ROOT / name path.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8") return path def board_group(symbol: str) -> str: if symbol.endswith(".BJ"): return "BEIJING_BJ" if symbol.startswith("688") and symbol.endswith(".SH"): return "STAR_688_SH" if (symbol.startswith("300") or symbol.startswith("301")) and symbol.endswith(".SZ"): return "CHINEXT_300_301_SZ" return "MAINBOARD_DEFAULT" def limit_rate(symbol: str) -> float: return CONFIG["strict_buy_rules"]["limitup_rates"][board_group(symbol)] def load_source_data() -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: w = CONFIG["data_window"] with get_conn() as conn: daily = pd.read_sql( """ SELECT trade_date, symbol, open_price, high_price, low_price, close_price, volume, amount FROM a_share_daily_price WHERE trade_date BETWEEN %(start)s AND %(end)s ORDER BY symbol, trade_date """, conn, params={"start": w["pull_daily_start"], "end": w["minute_replay_entry_end"]}, ) breadth = pd.read_sql( """ SELECT trade_date, stock_count, up_count, flat_count, down_count, run_id, price_source_table FROM ts_market_breadth_daily_cache WHERE trade_date BETWEEN %(start)s AND %(end)s AND scope_type='ALL_A_SHARE' AND scope_value='ALL' ORDER BY trade_date """, conn, params={"start": w["calendar_start"], "end": w["minute_replay_entry_end"]}, ) calendar = pd.read_sql( """ SELECT calendar_date, is_trading_day, trade_date_rank FROM a_share_trading_calendar WHERE calendar_date BETWEEN %(start)s AND %(end)s ORDER BY calendar_date """, conn, params={"start": w["calendar_start"], "end": w["calendar_end"]}, ) return daily, breadth, calendar def enrich_daily(daily: pd.DataFrame) -> pd.DataFrame: daily = daily.copy() 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) grouped = daily.groupby("symbol", group_keys=False) daily["prev_close"] = grouped["close_price"].shift(1) daily["prev5_avg_volume"] = grouped["volume"].transform( lambda s: s.shift(1).rolling(5, min_periods=5).mean() ) daily["volume_ratio"] = daily["volume"] / daily["prev5_avg_volume"] daily["upper_shadow_abs"] = daily["high_price"] - daily[["open_price", "close_price"]].max(axis=1) daily["day_range_abs"] = daily["high_price"] - daily["low_price"] daily["upper_shadow_pct"] = daily["upper_shadow_abs"] / daily["prev_close"] * 100.0 daily["upper_shadow_range_ratio"] = daily["upper_shadow_abs"] / daily["day_range_abs"] daily["daily_return_pct"] = (daily["close_price"] / daily["prev_close"] - 1.0) * 100.0 daily["market_group"] = daily["symbol"].map(board_group) daily["limit_rate"] = daily["symbol"].map(limit_rate) tol = CONFIG["strict_buy_rules"]["limitup_tolerance_rate"] daily["strict_limitup_return_rate"] = daily["high_price"] / daily["prev_close"] - 1.0 daily["strict_limitup_event_flag"] = ( daily["prev_close"].gt(0) & daily["strict_limitup_return_rate"].ge(daily["limit_rate"] - tol) ) prev60_high = [] prev60_high_volume = [] prev60_high_ref_date = [] prior_limitup_flag = [] latest_prior_limitup_date = [] latest_prior_limitup_close = [] pullback_low = [] pullback_low_date = [] prior30_count = [] prev_high_policy = [] reason_counter = Counter() for _symbol, g in daily.groupby("symbol", sort=False): highs = g["high_price"].to_numpy() lows = g["low_price"].to_numpy() closes = g["close_price"].to_numpy() vols = g["volume"].to_numpy() dates = pd.to_datetime(g["trade_date"]).tolist() limit_flags = g["strict_limitup_event_flag"].to_numpy() for i in range(len(g)): p60_start = max(0, i - CONFIG["strict_buy_rules"]["prev_high_window_trading_days"]) if i - p60_start >= CONFIG["strict_buy_rules"]["prev_high_min_history_trading_days"]: wh = highs[p60_start:i] max_pos = int(wh.argmax()) ref_idx = p60_start + max_pos prev60_high.append(highs[ref_idx]) prev60_high_volume.append(vols[ref_idx]) prev60_high_ref_date.append(dates[ref_idx]) prev_high_policy.append(CONFIG["strict_buy_rules"]["prev_high_ref_policy"]) else: prev60_high.append(float("nan")) prev60_high_volume.append(float("nan")) prev60_high_ref_date.append(pd.NaT) prev_high_policy.append("") p30_start = max(0, i - CONFIG["strict_buy_rules"]["prior_limitup_window_trading_days"]) prior30_count.append(i - p30_start) limit_indices = [idx for idx in range(p30_start, i) if bool(limit_flags[idx])] if not limit_indices: prior_limitup_flag.append(False) latest_prior_limitup_date.append(pd.NaT) latest_prior_limitup_close.append(float("nan")) pullback_low.append(float("nan")) pullback_low_date.append(pd.NaT) if i - p30_start < CONFIG["strict_buy_rules"]["prior_limitup_window_trading_days"]: reason_counter["PRIOR_30_TRADING_DAY_WINDOW_INCOMPLETE_HELD"] += 1 else: reason_counter["STRICT_PRIOR_LIMITUP_FAIL"] += 1 continue latest_idx = limit_indices[-1] lo_slice = lows[latest_idx + 1 : i + 1] if len(lo_slice) == 0: lo_slice = lows[latest_idx : i + 1] lo_offset = 0 else: lo_offset = latest_idx + 1 min_pos = int(lo_slice.argmin()) min_idx = lo_offset + min_pos prior_limitup_flag.append(True) latest_prior_limitup_date.append(dates[latest_idx]) latest_prior_limitup_close.append(closes[latest_idx]) pullback_low.append(lows[min_idx]) pullback_low_date.append(dates[min_idx]) daily["prev60_high"] = prev60_high daily["prev60_high_volume"] = prev60_high_volume daily["prev60_high_ref_date"] = prev60_high_ref_date daily["prev60_high_ref_policy"] = prev_high_policy daily["prior30_trading_day_count"] = prior30_count daily["prior_strict_limitup_30_flag"] = prior_limitup_flag daily["latest_prior_strict_limitup_date"] = latest_prior_limitup_date daily["latest_prior_strict_limitup_close"] = latest_prior_limitup_close daily["pullback_low_since_latest_limitup"] = pullback_low daily["pullback_low_since_latest_limitup_date"] = pullback_low_date daily["pullback_from_latest_limitup_close_pct"] = ( daily["pullback_low_since_latest_limitup"] / daily["latest_prior_strict_limitup_close"] - 1.0 ) * 100.0 daily["touch_prev_high_flag"] = daily["prev60_high"].gt(0) & daily["high_price"].ge(daily["prev60_high"] * 0.995) daily["prev_high_volume_pass_flag"] = (~daily["touch_prev_high_flag"]) | daily["volume"].gt(daily["prev60_high_volume"]) return daily def build_entry_links(daily: pd.DataFrame, breadth: pd.DataFrame) -> pd.DataFrame: breadth = breadth.copy() breadth["signal_trade_date"] = pd.to_datetime(breadth["trade_date"]) breadth = breadth.drop(columns=["trade_date"]) trade_dates = sorted(pd.to_datetime(daily["trade_date"].drop_duplicates()).tolist()) links = pd.DataFrame({"signal_trade_date": trade_dates[:-1], "entry_trade_date": trade_dates[1:]}) w = CONFIG["data_window"] links = links[ (links["entry_trade_date"] >= pd.Timestamp(w["minute_replay_entry_start"])) & (links["entry_trade_date"] <= pd.Timestamp(w["minute_replay_entry_end"])) ].copy() links = links.merge(breadth, on="signal_trade_date", how="left") links["market_gate_open_flag"] = links["up_count"].fillna(-1).ge(CONFIG["strict_buy_rules"]["market_gate_up_count_min"]) links["market_gate_status"] = links["market_gate_open_flag"].map( { True: "MKT_GATE_OPEN_SIGNAL_DAY_UP_3000", False: "NO_TRADE_MARKET_GATE_CLOSED_SIGNAL_DAY", } ) return links def build_candidates(daily: pd.DataFrame, links: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: signals = daily.merge( links, left_on="trade_date", right_on="signal_trade_date", how="inner", suffixes=("", "_breadth"), ) r = CONFIG["strict_buy_rules"] checks = { "prev_close_ok": signals["prev_close"].gt(0), "prev5_volume_ok": signals["prev5_avg_volume"].gt(0), "prior_limitup_ok": signals["prior_strict_limitup_30_flag"], "volume_ratio_ok": signals["volume_ratio"].ge(r["volume_ratio_prev5_min"]), "pullback_ok": signals["pullback_from_latest_limitup_close_pct"].le( r["pullback_from_latest_prior_limitup_close_pct_max"] ), "upper_shadow_pct_ok": signals["upper_shadow_pct"].ge(r["upper_shadow_pct_proxy_min"]), "upper_shadow_range_ok": signals["upper_shadow_range_ratio"].ge(r["upper_shadow_range_ratio_proxy_min"]), "prev_high_volume_ok": signals["prev_high_volume_pass_flag"], } for col, value in checks.items(): signals[col] = value.fillna(False) failure_reasons = [] for row in signals.itertuples(index=False): reasons = [] for name in checks: if not bool(getattr(row, name)): reasons.append(name.replace("_ok", "").upper() + "_FAIL") if row.prior30_trading_day_count < r["prior_limitup_window_trading_days"]: reasons.append("PRIOR_30_TRADING_DAY_WINDOW_INCOMPLETE_HELD") if not reasons: reasons.append("STRICT_CODE_PASS_SUPPORT_REVIEW_REQUIRED") failure_reasons.append(";".join(dict.fromkeys(reasons))) signals["strict_code_status"] = [ "STRICT_CODE_PASS_SUPPORT_REVIEW_REQUIRED" if x == "STRICT_CODE_PASS_SUPPORT_REVIEW_REQUIRED" else "STRICT_CODE_FAIL_OR_HELD" for x in failure_reasons ] signals["strict_code_reason"] = failure_reasons pass_mask = signals["strict_code_status"].eq("STRICT_CODE_PASS_SUPPORT_REVIEW_REQUIRED") candidates = signals[pass_mask].copy() candidates = candidates.sort_values( ["entry_trade_date", "upper_shadow_pct", "volume_ratio", "amount", "symbol"], ascending=[True, False, False, False, True], ) candidates["candidate_rank"] = candidates.groupby("entry_trade_date").cumcount() + 1 candidates["candidate_id"] = [ f"STRICT-NOTE-{d:%Y%m%d}-{rank:02d}-{sym.replace('.', '_')}" for d, rank, sym in zip(candidates["entry_trade_date"], candidates["candidate_rank"], candidates["symbol"]) ] candidates["support_manual_decision"] = "SUPPORT_REVIEW_REQUIRED" candidates["buy_point_manual_decision"] = "BUY_POINT_REVIEW_REQUIRED" candidates["source_note_semantics"] = "prior_limitup_30_ex_signal;volume_ratio_ge_2;pullback_ge_3pct;long_upper_shadow_proxy;bottom_support_manual" reason_counter = Counter() for reason in signals["strict_code_reason"].astype(str): for item in reason.split(";"): reason_counter[item] += 1 reason_summary = pd.DataFrame( [{"reason": k, "rows": v} for k, v in sorted(reason_counter.items())] ) failed = signals[~pass_mask].copy() review_sample = ( failed.sort_values(["strict_code_reason", "entry_trade_date", "symbol"]) .groupby("strict_code_reason", dropna=False) .head(50) .reset_index(drop=True) ) review_sample["review_row_id"] = [f"STRICT-NOTE-REVIEW-SAMPLE-{i+1:06d}" for i in range(len(review_sample))] return candidates.reset_index(drop=True), review_sample, reason_summary def build_case_index(candidates: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]: top_n = CONFIG["strict_buy_rules"]["selection_top_n_per_entry_date"] selected = candidates[candidates["candidate_rank"].le(top_n)].copy() selected["case_id"] = [f"WUJI-STRICT-{d:%Y%m%d}" for d in selected["entry_trade_date"]] case_index = ( selected.groupby(["case_id", "entry_trade_date", "signal_trade_date", "market_gate_status", "market_gate_open_flag"], dropna=False) .agg( selected_candidate_count=("candidate_id", "count"), symbols=("symbol", lambda s: ";".join(s)), candidate_ids=("candidate_id", lambda s: ";".join(s)), ) .reset_index() .sort_values("entry_trade_date") ) return selected.reset_index(drop=True), case_index.reset_index(drop=True) def format_dates(df: pd.DataFrame) -> pd.DataFrame: out = df.copy() for col in out.columns: if pd.api.types.is_datetime64_any_dtype(out[col]): out[col] = out[col].dt.strftime("%Y-%m-%d") return out def write_readme(summary: dict) -> Path: path = ROOT / "README.md" text = f"""# {RUN_ID} ## Purpose This package is the execution-prep strict BUY candidate rebuild for the Wuji V1 strict-note full rerun. It rebuilds the BUY candidate pool from source daily data using `笔记精简版.md` strict limit-up and volume conditions. ## What This Package Is - strict BUY candidate evidence package - frozen run configuration - selected top-5-per-entry-date candidate ledger for the next replay stage - self-check, source artifact manifest, and package manifest ## What This Package Is Not - not a completed V1 strict performance rerun - not a buy recommendation - not a success-rate, return-rate, win-rate, drawdown, or strategy-effectiveness conclusion ## Key Counts - strict code-pass candidates: {summary['counts']['strict_code_pass_candidates']} - selected candidates for replay: {summary['counts']['selected_candidates']} - case dates prepared: {summary['counts']['case_dates_prepared']} - market gate open case dates: {summary['counts']['market_gate_open_case_dates']} - market gate closed case dates: {summary['counts']['market_gate_closed_case_dates']} ## Next Required Step Submit this package to `case_analysis.reviewer` for execution-prep review. If it passes, the next stage is minute buy-point review, external manual/AI-manual decision source, V1 sell/trend/rolling replay, lifecycle/readability package generation, and final citation review. """ path.write_text(text, encoding="utf-8") return path def build_source_manifest() -> pd.DataFrame: rows = [] for source_name, path in [ ("source_note", SOURCE_NOTE), ("old_full_summary", FULL_SOURCE_RUN / "summary.json"), ("old_full_candidate_ledger", FULL_SOURCE_RUN / "candidate_ledger.csv"), ("old_v1_summary", V1_SOURCE_RUN / "summary.json"), ]: exists = path.exists() rows.append( { "source_name": source_name, "path": str(path), "exists": exists, "size": path.stat().st_size if exists else "", "sha256": sha256_file(path) if exists and path.is_file() else "", } ) return pd.DataFrame(rows) def build_manifest() -> pd.DataFrame: rows = [] for path in sorted(ROOT.rglob("*")): if path.is_file() and path.name not in {"manifest.csv", "manifest.json"}: 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) daily_raw, breadth, calendar = load_source_data() daily = enrich_daily(daily_raw) links = build_entry_links(daily, breadth) candidates, review_pool, reason_summary = build_candidates(daily, links) selected, case_index = build_case_index(candidates) counts = { "source_daily_rows": int(len(daily_raw)), "source_symbols": int(daily_raw["symbol"].nunique()), "entry_links": int(len(links)), "strict_code_pass_candidates": int(len(candidates)), "selected_candidates": int(len(selected)), "case_dates_prepared": int(case_index["case_id"].nunique()), "market_gate_open_case_dates": int(case_index["market_gate_open_flag"].sum()), "market_gate_closed_case_dates": int((~case_index["market_gate_open_flag"]).sum()), "support_review_required_candidates": int(candidates["support_manual_decision"].eq("SUPPORT_REVIEW_REQUIRED").sum()), } reason_counts = dict(zip(reason_summary["reason"], reason_summary["rows"])) generated_at = now_iso() config = dict(CONFIG) config["generated_at"] = generated_at write_json(config, "strict_note_run_config.json") (ROOT / "strict_note_run_config.md").write_text( "# strict_note_run_config\n\n" f"- run_id: {RUN_ID}\n" f"- generated_at: {generated_at}\n" "- direct_blueprint: ana-doc/wuji/profile/source_note/笔记精简版.md\n" "- strict_limitup: prior 30 trading days, exclude signal day, board-specific 10%/20%/30%, high_price / prev_close - 1, tolerance 0.05 percentage point\n" "- volume_ratio: signal volume / previous 5 trading-day average >= 2.0\n" "- pullback: latest prior strict limit-up close to signal-window low <= -3.0%\n" "- long_upper_shadow: execution proxy upper_shadow_pct >= 3.0 and upper_shadow_range_ratio >= 0.4\n" "- bottom_support: SUPPORT_REVIEW_REQUIRED, not auto-passed by code\n" "- market_gate: signal-day ALL_A_SHARE up_count >= 3000\n" "- selection: top 5 per entry_trade_date after strict code pass; no BUY is generated in this package\n", encoding="utf-8", ) candidate_cols = [ "candidate_id", "symbol", "market_group", "signal_trade_date", "entry_trade_date", "candidate_rank", "market_gate_status", "market_gate_open_flag", "up_count", "open_price", "high_price", "low_price", "close_price", "prev_close", "volume", "amount", "prev5_avg_volume", "volume_ratio", "limit_rate", "strict_limitup_return_rate", "prior_strict_limitup_30_flag", "latest_prior_strict_limitup_date", "latest_prior_strict_limitup_close", "pullback_low_since_latest_limitup", "pullback_low_since_latest_limitup_date", "pullback_from_latest_limitup_close_pct", "upper_shadow_pct", "upper_shadow_range_ratio", "prev60_high", "prev60_high_volume", "prev60_high_ref_date", "prev60_high_ref_policy", "touch_prev_high_flag", "prev_high_volume_pass_flag", "support_manual_decision", "buy_point_manual_decision", "source_note_semantics", ] selected_cols = ["case_id"] + candidate_cols review_cols = [ "review_row_id", "symbol", "market_group", "signal_trade_date", "entry_trade_date", "strict_code_status", "strict_code_reason", "market_gate_status", "up_count", "prev_close", "high_price", "low_price", "close_price", "volume", "prev5_avg_volume", "volume_ratio", "limit_rate", "strict_limitup_return_rate", "prior_strict_limitup_30_flag", "latest_prior_strict_limitup_date", "pullback_from_latest_limitup_close_pct", "upper_shadow_pct", "upper_shadow_range_ratio", "prev_high_volume_pass_flag", ] write_csv(format_dates(candidates[candidate_cols]), "strict_note_candidate_ledger.csv") write_csv(format_dates(selected[selected_cols]), "strict_note_selected_candidate_ledger.csv") write_csv(format_dates(case_index), "strict_note_case_index.csv") write_csv(format_dates(review_pool[review_cols]), "strict_note_review_pool_ledger.csv") write_csv(reason_summary, "strict_note_review_pool_reason_summary.csv") source_manifest = build_source_manifest() write_csv(source_manifest, "source_artifact_manifest.csv") self_items = [ ("SOURCE_NOTE_EXISTS", SOURCE_NOTE.exists(), str(SOURCE_NOTE)), ("STRICT_CONFIG_WRITTEN", (ROOT / "strict_note_run_config.json").exists(), "strict_note_run_config.json"), ("PRIOR_LIMITUP_EXCLUDES_SIGNAL_DAY", True, "computed from rows [i-30:i], signal row excluded"), ("VOLUME_RATIO_MIN_2_FROZEN", CONFIG["strict_buy_rules"]["volume_ratio_prev5_min"] == 2.0, "volume_ratio_prev5_min=2.0"), ("BOTTOM_SUPPORT_NOT_AUTO_PASSED", candidates["support_manual_decision"].eq("SUPPORT_REVIEW_REQUIRED").all(), "all candidates require support review"), ("SELECTED_TOP_N_PER_ENTRY", selected.groupby("entry_trade_date")["candidate_id"].count().le(CONFIG["strict_buy_rules"]["selection_top_n_per_entry_date"]).all(), "top_n<=5"), ("NO_BUY_ORDERS_GENERATED", not (ROOT / "strict_order_ledger.csv").exists(), "candidate package only"), ("SOURCE_ARTIFACTS_EXIST", source_manifest["exists"].all(), "source_artifact_manifest.csv"), ] self_df = pd.DataFrame( [ {"item": item, "status": "PASS" if ok else "FAIL", "detail": detail} for item, ok, detail in self_items ] ) write_csv(self_df, "self_check_items.csv") self_json = { "run_id": RUN_ID, "generated_at": generated_at, "stage": CONFIG["stage"], "overall_status": "PASS_FOR_STRICT_BUY_POOL_EXECUTION_PREP_REVIEW_READY" if self_df["status"].eq("PASS").all() else "FAIL", "pass_count": int(self_df["status"].eq("PASS").sum()), "fail_count": int(self_df["status"].eq("FAIL").sum()), } write_json(self_json, "self_check.json") summary = { "run_id": RUN_ID, "generated_at": generated_at, "stage": CONFIG["stage"], "counts": counts, "review_pool_reason_counts": reason_counts, "citation_boundary": CONFIG["citation_boundary"], "next_step": "submit strict BUY pool execution-prep package to case_analysis.reviewer", } write_json(summary, "summary.json") (ROOT / "summary.md").write_text( "# Strict Note BUY Pool Rebuild Summary\n\n" f"- run_id: {RUN_ID}\n" f"- generated_at: {generated_at}\n" f"- strict code-pass candidates: {counts['strict_code_pass_candidates']}\n" f"- selected candidates: {counts['selected_candidates']}\n" f"- case dates prepared: {counts['case_dates_prepared']}\n" f"- market gate open case dates: {counts['market_gate_open_case_dates']}\n" f"- market gate closed case dates: {counts['market_gate_closed_case_dates']}\n\n" "Boundary: this package rebuilds the strict BUY pool only. It does not contain BUY orders, SELL orders, returns, or strategy conclusions.\n", encoding="utf-8", ) write_readme(summary) manifest = build_manifest() write_csv(manifest, "manifest.csv") write_json( { "run_id": RUN_ID, "generated_at": generated_at, "file_count": int(len(manifest)), "files": manifest.to_dict(orient="records"), }, "manifest.json", ) if __name__ == "__main__": main()