from __future__ import annotations import hashlib import json import sys from datetime import datetime, timedelta, timezone from pathlib import Path import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent)) import record_p2_resolution as record # noqa: E402 RUN_ID = "RUN-ANA-WUJI-V1-BUY-POINT-SECOND-REVIEW-20260615-001" TZ = timezone(timedelta(hours=8)) ROOT = Path(__file__).resolve().parents[1] PACKET_ROOT = ROOT / "p2_step_review_packets" HARD_PULLBACK_PCTPT = 5.0 STRONG_HIGH_PCT = 5.0 WEAK_CLOSE_PCT = 2.0 WEAK_TAIL_MAX_PCT = 2.0 def now_iso() -> str: return datetime.now(TZ).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_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8-sig") def num(value: object) -> float | None: try: value = float(value) except Exception: return None if pd.isna(value): return None return value def fmt(value: object, digits: int = 2) -> str: value = num(value) if value is None: return "" return f"{value:.{digits}f}" def classify(row: pd.Series) -> tuple[str, str, str]: max_ret = num(row.get("max_ret_pct")) close_ret = num(row.get("close_ret_pct")) tail_max = num(row.get("tail_max_ret_pct")) if max_ret is None or close_ret is None: return "DATA_INSUFFICIENT", "P2_PULLBACK_DATA_INSUFFICIENT", "缺少盘中高点或收盘涨幅，继续待审。" pullback = max_ret - close_ret tail_pullback = max_ret - tail_max if tail_max is not None else None hard_pullback = pullback >= HARD_PULLBACK_PCTPT weak_tail_fade = ( max_ret >= STRONG_HIGH_PCT and close_ret <= WEAK_CLOSE_PCT and tail_max is not None and tail_max <= WEAK_TAIL_MAX_PCT ) if hard_pullback or weak_tail_fade: reason = ( f"按 P2 批量口径：盘中最高 {fmt(max_ret)}%，收盘 {fmt(close_ret)}%，" f"高点回落 {fmt(pullback)} 个百分点" ) if tail_pullback is not None: reason += f"，尾段最高较高点回落 {fmt(tail_pullback)} 个百分点" reason += "，属于冲高后大幅回落，标记 REVIEW_HELD。" return "REVIEW_HELD", "P2_BIG_INTRADAY_PULLBACK", reason reason = ( f"按 P2 批量口径：盘中最高 {fmt(max_ret)}%，收盘 {fmt(close_ret)}%，" f"高点回落 {fmt(pullback)} 个百分点" ) if tail_max is not None: reason += f"，14:40 后最高 {fmt(tail_max)}%" reason += "，未触发大幅回落阈值，标记 BUY。" return "BUY", "P2_NO_BIG_INTRADAY_PULLBACK", reason def build_decisions() -> pd.DataFrame: index = pd.read_csv(PACKET_ROOT / "p2_step_review_index.csv", dtype=str, encoding="utf-8-sig").fillna("") recheck = pd.read_csv(ROOT / "buy_point_second_review_recheck_list.csv", dtype=str, encoding="utf-8-sig").fillna("") recheck = recheck[recheck["human_recheck_priority"].astype(str).str.startswith("P2_", na=False)].copy() metric_cols = [ "candidate_id", "max_ret_pct", "max_ret_time", "close_ret_pct", "tail_min_ret_pct", "tail_max_ret_pct", "tail_max_ret_time", "above_open_ratio", "human_decision_action", "human_decision_reason_cn", "second_review_issue_code", "second_review_reason_cn", ] merged = index.merge(recheck[metric_cols], on="candidate_id", how="left", suffixes=("", "_detail")) rows = [] for _, row in merged.iterrows(): final_action, rule_code, reason = classify(row) max_ret = num(row.get("max_ret_pct")) close_ret = num(row.get("close_ret_pct")) tail_max = num(row.get("tail_max_ret_pct")) rows.append( { "packet_order": row["order"], "case_id": row["case_id"], "candidate_id": row["candidate_id"], "symbol": row["symbol"], "entry_trade_date": row["entry_trade_date"], "original_action": row["original_action"], "final_action": final_action, "rule_code": rule_code, "rule_reason_cn": reason, "max_ret_pct": row.get("max_ret_pct", ""), "max_ret_time": row.get("max_ret_time", ""), "close_ret_pct": row.get("close_ret_pct", ""), "tail_max_ret_pct": row.get("tail_max_ret_pct", ""), "tail_max_ret_time": row.get("tail_max_ret_time", ""), "tail_min_ret_pct": row.get("tail_min_ret_pct", ""), "pullback_from_day_high_pctpt": "" if max_ret is None or close_ret is None else max_ret - close_ret, "tail_pullback_from_day_high_pctpt": "" if max_ret is None or tail_max is None else max_ret - tail_max, "above_open_ratio": row.get("above_open_ratio", ""), "packet_path": row["packet_path"], } ) decisions = pd.DataFrame(rows) decisions["packet_order_num"] = pd.to_numeric(decisions["packet_order"], errors="coerce") decisions = decisions.sort_values("packet_order_num").drop(columns=["packet_order_num"]) return decisions def apply_decisions(decisions: pd.DataFrame) -> None: index = pd.read_csv(PACKET_ROOT / "p2_step_review_index.csv", dtype=str, encoding="utf-8-sig").fillna("") by_order = {int(row["order"]): row for _, row in index.iterrows()} for _, decision in decisions.iterrows(): order = int(decision["packet_order"]) row = by_order[order] final_action = decision["final_action"] reason = decision["rule_reason_cn"] record.update_packet(row, final_action, reason) record.update_ledger(row, final_action, reason) record.update_rollup(row, final_action, reason) rollup = pd.read_csv(record.ROLLUP_PATH, dtype=str, encoding="utf-8-sig").fillna("") record.update_p2_summary() record.update_resolution_summary(rollup) record.update_repair_files(rollup) def write_summary(decisions: pd.DataFrame) -> None: counts = decisions["final_action"].value_counts().sort_index() rule_counts = decisions["rule_code"].value_counts().sort_index() lines = [ "# P2 冲高回落批量裁决说明", "", f"- updated_at: {now_iso()}", "- 口径：冲高大幅回落标记 REVIEW_HELD；未触发大幅回落阈值标记 BUY。", f"- 大幅回落阈值 1：盘中最高涨幅 - 收盘涨幅 >= {HARD_PULLBACK_PCTPT:.1f} 个百分点。", f"- 大幅回落阈值 2：盘中最高涨幅 >= {STRONG_HIGH_PCT:.1f}% 且收盘 <= {WEAK_CLOSE_PCT:.1f}% 且 14:40 后最高 <= {WEAK_TAIL_MAX_PCT:.1f}%。", "", "## 最终动作分布", "", "| final_action | count |", "|---|---:|", ] for action, count in counts.items(): lines.append(f"| `{action}` | {int(count)} |") lines += ["", "## 规则命中分布", "", "| rule_code | count |", "|---|---:|"] for code, count in rule_counts.items(): lines.append(f"| `{code}` | {int(count)} |") lines += [ "", "## 明细", "", "| # | case | symbol | date | final | max% | close% | pullback | reason |", "|---:|---|---|---|---|---:|---:|---:|---|", ] for _, row in decisions.iterrows(): lines.append( f"| {row['packet_order']} | {row['case_id']} | {row['symbol']} | {row['entry_trade_date']} | " f"{row['final_action']} | {fmt(row['max_ret_pct'])} | {fmt(row['close_ret_pct'])} | " f"{fmt(row['pullback_from_day_high_pctpt'])} | {row['rule_code']} |" ) write_text(PACKET_ROOT / "p2_pullback_rule_summary.md", "\n".join(lines) + "\n") def update_manifest() -> None: rows = [] for path in sorted(ROOT.rglob("*")): if path.is_file() and "__pycache__" not in path.parts: rows.append({"path": path.relative_to(ROOT).as_posix(), "size": path.stat().st_size, "sha256": sha256_file(path)}) write_csv(pd.DataFrame(rows), ROOT / "manifest.csv") (ROOT / "manifest.json").write_text( json.dumps({"run_id": RUN_ID, "updated_at": now_iso(), "file_count": len(rows), "files": rows}, ensure_ascii=False, indent=2), encoding="utf-8", ) def main() -> None: decisions = build_decisions() write_csv(decisions, PACKET_ROOT / "p2_pullback_rule_decision.csv") write_summary(decisions) apply_decisions(decisions) update_manifest() if __name__ == "__main__": main()