from __future__ import annotations import hashlib import json import os import re from datetime import datetime, timezone, timedelta from pathlib import Path import pandas as pd import pymysql from PIL import Image, ImageDraw, ImageFont RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001" ROOT = Path(__file__).resolve().parents[1] LOCAL_DB_INDEX = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md") TZ = timezone(timedelta(hours=8)) 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=240, ) def font(size: int): for p in [ Path("C:/Windows/Fonts/msyh.ttc"), Path("C:/Windows/Fonts/simhei.ttf"), Path("C:/Windows/Fonts/simsun.ttc"), ]: if p.exists(): return ImageFont.truetype(str(p), size) return ImageFont.load_default() FONT_TITLE = font(24) FONT_MID = font(16) FONT_SMALL = font(13) def normalize_time(value) -> str: text = str(value) if "days" in text: text = text.split()[-1] if "." in text: text = text.split(".")[0] parts = text.split(":") if len(parts) >= 3: return f"{int(parts[0]):02d}:{int(parts[1]):02d}:{int(float(parts[2])):02d}" return text 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, 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 y_price(value: float, low: float, high: float, top: int, bottom: int) -> int: if high <= low: return (top + bottom) // 2 return bottom - int((value - low) / (high - low) * (bottom - top)) def draw_intraday_review_chart(cand: pd.Series, minute: pd.DataFrame, ma5: float | None, out_path: Path) -> None: w, h = 1500, 860 img = Image.new("RGB", (w, h), "#fbfbf7") d = ImageDraw.Draw(img) d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1") title = f"严格版买点人工复核图：{cand.symbol} {cand.entry_trade_date.strftime('%Y-%m-%d')}" d.text((30, 22), title, fill="#111827", font=FONT_TITLE) d.text((30, 56), f"case: {cand.case_id} candidate: {cand.candidate_id}", fill="#334155", font=FONT_SMALL) plot_left, plot_top, plot_right, plot_bottom = 80, 105, 1080, 575 vol_top, vol_bottom = 625, 785 note_left, note_top = 1110, 110 d.rectangle([plot_left, plot_top, plot_right, plot_bottom], outline="#94a3b8") d.rectangle([plot_left, vol_top, plot_right, vol_bottom], outline="#94a3b8") m = minute.sort_values("trade_time").reset_index(drop=True) open_ref = float(m.open_price.iloc[0]) refs = [open_ref, open_ref * 1.03, open_ref * 1.05, open_ref * 1.08] if ma5 and ma5 > 0: refs.append(float(ma5)) price_low = min(float(m.low_price.min()), min(refs)) * 0.998 price_high = max(float(m.high_price.max()), max(refs)) * 1.002 max_vol = max(float(m.volume.max()), 1.0) n = len(m) gap = (plot_right - plot_left) / max(n - 1, 1) pts = [] for i, row in m.iterrows(): x = int(plot_left + gap * i) y = y_price(float(row.close_price), price_low, price_high, plot_top, plot_bottom) pts.append((x, y)) vh = int(float(row.volume) / max_vol * (vol_bottom - vol_top)) color = "#dc2626" if float(row.close_price) >= float(row.open_price) else "#16a34a" d.line([x, vol_bottom, x, vol_bottom - vh], fill=color, width=2) if i % max(1, n // 7) == 0: d.text((x - 20, vol_bottom + 8), str(row.trade_time)[:5], fill="#64748b", font=FONT_SMALL) if len(pts) > 1: d.line(pts, fill="#2563eb", width=2) for label, value, color in [ ("开盘价", open_ref, "#0f172a"), ("+3%", open_ref * 1.03, "#f59e0b"), ("+5%", open_ref * 1.05, "#dc2626"), ("+8%", open_ref * 1.08, "#7c3aed"), ]: yy = y_price(value, price_low, price_high, plot_top, plot_bottom) d.line([plot_left, yy, plot_right, yy], fill=color, width=2) d.text((plot_right + 8, yy - 8), f"{label} {value:.2f}", fill=color, font=FONT_SMALL) if ma5 and ma5 > 0: yy = y_price(float(ma5), price_low, price_high, plot_top, plot_bottom) d.line([plot_left, yy, plot_right, yy], fill="#059669", width=2) d.text((plot_right + 8, yy - 8), f"日MA5 {float(ma5):.2f}", fill="#059669", font=FONT_SMALL) d.rounded_rectangle([note_left, note_top, 1465, 785], radius=8, outline="#334155", fill="#ffffff") notes = [ "人工裁决待填", "本图只打包证据，不自动买入。", "需人工/AI人工判断：", "1. 底部承接是否强", "2. 买点是否成立", "3. 是否追高或假承接", "", "冻结硬筛证据：", f"倍量：{float(cand.volume_ratio):.2f}", f"回调：{float(cand.pullback_from_latest_limitup_close_pct):.2f}%", f"长上影：{float(cand.upper_shadow_pct):.2f}%", f"闸门：{cand.market_gate_status}", "", "最终动作不得由脚本代填。", ] yy = note_top + 16 for i, line in enumerate(notes): d.text((note_left + 16, yy), line, fill="#111827" if i == 0 else "#334155", font=FONT_TITLE if i == 0 else FONT_SMALL) yy += 32 if i == 0 else (24 if line else 12) d.text((30, 820), "decision_input_view：用于外部人工/AI人工裁决，不能反推成脚本自动 BUY。", fill="#334155", font=FONT_MID) out_path.parent.mkdir(parents=True, exist_ok=True) img.save(out_path) def load_minute_and_ma5(eligible: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]: if eligible.empty: return pd.DataFrame(), pd.DataFrame() with get_conn() as conn: minute_parts = [] for trade_date, group in eligible.groupby(eligible.entry_trade_date.dt.strftime("%Y-%m-%d")): symbols = sorted(group.symbol.unique().tolist()) ph = ",".join(["%s"] * len(symbols)) minute_parts.append( pd.read_sql( f""" SELECT trade_date, trade_time, symbol, open_price, high_price, low_price, close_price, volume FROM a_share_minute_price WHERE trade_date=%s AND symbol IN ({ph}) ORDER BY symbol, trade_date, trade_time """, conn, params=[trade_date, *symbols], ) ) minute = pd.concat(minute_parts, ignore_index=True) if minute_parts else pd.DataFrame() symbols = sorted(eligible.symbol.unique().tolist()) ph = ",".join(["%s"] * len(symbols)) min_date = (eligible.signal_trade_date.min() - pd.Timedelta(days=20)).strftime("%Y-%m-%d") max_date = eligible.signal_trade_date.max().strftime("%Y-%m-%d") daily = pd.read_sql( f""" SELECT trade_date, symbol, close_price FROM a_share_daily_price WHERE symbol IN ({ph}) AND trade_date BETWEEN %s AND %s ORDER BY symbol, trade_date """, conn, params=[*symbols, min_date, max_date], ) if not minute.empty: minute["trade_date"] = pd.to_datetime(minute["trade_date"]) minute["trade_time"] = minute["trade_time"].map(normalize_time) for col in ["open_price", "high_price", "low_price", "close_price", "volume"]: minute[col] = pd.to_numeric(minute[col], errors="coerce") if not daily.empty: daily["trade_date"] = pd.to_datetime(daily["trade_date"]) daily["close_price"] = pd.to_numeric(daily["close_price"], errors="coerce") daily["ma5"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(5, min_periods=5).mean()) return minute, daily 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": path.relative_to(ROOT).as_posix(), "size": path.stat().st_size, "sha256": sha256_file(path)}) return pd.DataFrame(rows) def main() -> None: selected = pd.read_csv(ROOT / "strict_note_selected_candidate_ledger.csv", encoding="utf-8-sig") selected["entry_trade_date"] = pd.to_datetime(selected["entry_trade_date"]) selected["signal_trade_date"] = pd.to_datetime(selected["signal_trade_date"]) selected["eligible_for_buy_replay_flag"] = selected["market_gate_open_flag"].astype(str).str.lower().isin(["true", "1"]) selected["buy_replay_boundary_status"] = selected["eligible_for_buy_replay_flag"].map( {True: "ELIGIBLE_FOR_BUY_POINT_MANUAL_REVIEW", False: "NO_TRADE_MARKET_GATE_CLOSED_BOUNDARY"} ) eligible = selected[selected["eligible_for_buy_replay_flag"]].copy() boundary = selected[~selected["eligible_for_buy_replay_flag"]].copy() minute, daily = load_minute_and_ma5(eligible) chart_rows = [] missing_rows = [] template_rows = [] data_gap_candidate_ids = set() for _, cand in eligible.iterrows(): m = minute[(minute["symbol"] == cand.symbol) & (minute["trade_date"] == cand.entry_trade_date)].copy() ma_part = daily[(daily["symbol"] == cand.symbol) & (daily["trade_date"] <= cand.signal_trade_date)].tail(1) ma5 = None if ma_part.empty or pd.isna(ma_part.iloc[0].ma5) else float(ma_part.iloc[0].ma5) chart_path = Path("charts") / "buy_point_review" / cand.case_id / f"buy_point_review_{cand.candidate_id}.png" if m.empty: data_gap_candidate_ids.add(cand.candidate_id) missing_rows.append( { "case_id": cand.case_id, "candidate_id": cand.candidate_id, "symbol": cand.symbol, "entry_trade_date": cand.entry_trade_date.strftime("%Y-%m-%d"), "missing_reason": "ENTRY_DAY_MINUTE_DATA_MISSING", } ) chart_rel = "" chart_sha = "" else: draw_intraday_review_chart(cand, m, ma5, ROOT / chart_path) chart_rel = chart_path.as_posix() chart_sha = sha256_file(ROOT / chart_path) chart_rows.append( { "case_id": cand.case_id, "candidate_id": cand.candidate_id, "symbol": cand.symbol, "entry_trade_date": cand.entry_trade_date.strftime("%Y-%m-%d"), "chart_role": "strict_note_buy_point_manual_review_input", "path": chart_rel, "sha256": chart_sha, "status": "PASS", } ) if chart_rel: template_rows.append( { "external_decision_id": f"EXT-BUY-{cand.candidate_id}", "case_id": cand.case_id, "candidate_id": cand.candidate_id, "symbol": cand.symbol, "signal_trade_date": cand.signal_trade_date.strftime("%Y-%m-%d"), "entry_trade_date": cand.entry_trade_date.strftime("%Y-%m-%d"), "code_suggested_action": "BUY_POINT_REVIEW_REQUIRED", "human_decision_action": "", "human_decision_reason_cn": "", "decision_operator": "", "decision_time": "", "decision_source": "", "review_input_chart_path": chart_rel, "review_input_chart_sha256": chart_sha, "accept_code_suggestion_flag": "", "reviewer_notes": "", } ) selected["buy_point_replay_status"] = selected.apply( lambda r: "NO_TRADE_MARKET_GATE_CLOSED_BOUNDARY" if not bool(r["eligible_for_buy_replay_flag"]) else ( "BUY_POINT_MINUTE_DATA_GAP_HELD" if r["candidate_id"] in data_gap_candidate_ids else "BUY_POINT_MANUAL_REVIEW_READY" ), axis=1, ) ready = selected[selected["buy_point_replay_status"].eq("BUY_POINT_MANUAL_REVIEW_READY")].copy() data_gap = selected[selected["buy_point_replay_status"].eq("BUY_POINT_MINUTE_DATA_GAP_HELD")].copy() write_csv(selected, "strict_note_buy_replay_scope.csv") write_csv(ready, "strict_note_buy_point_review_candidate_ledger.csv") write_csv(boundary, "strict_note_market_gate_boundary_ledger.csv") write_csv(data_gap, "strict_note_buy_point_data_gap_boundary_ledger.csv") write_csv(pd.DataFrame(template_rows), "manual_buy_decision_external_template.csv") write_csv(pd.DataFrame(chart_rows), "chart_evidence_audit.csv") write_csv(pd.DataFrame(missing_rows, columns=["case_id", "candidate_id", "symbol", "entry_trade_date", "missing_reason"]), "missing_chart_inputs.csv") generated_at = now_iso() items = [ ("SCOPE_TOTAL_MATCHES_SELECTED", len(selected) == 3420, f"scope={len(selected)}"), ("READY_PLUS_BOUNDARIES_MATCH_SCOPE", len(ready) + len(data_gap) + len(boundary) == len(selected), f"ready={len(ready)}, data_gap={len(data_gap)}, market_closed={len(boundary)}"), ("MARKET_CLOSED_NOT_ELIGIBLE", boundary["eligible_for_buy_replay_flag"].eq(False).all(), f"boundary={len(boundary)}"), ("MANUAL_TEMPLATE_NO_FINAL_ACTION_PREFILL", pd.DataFrame(template_rows)["human_decision_action"].eq("").all(), "human_decision_action blank"), ("MANUAL_TEMPLATE_NO_REASON_PREFILL", pd.DataFrame(template_rows)["human_decision_reason_cn"].eq("").all(), "human_decision_reason_cn blank"), ("CHARTS_FOR_READY_CANDIDATES", len(chart_rows) == len(ready), f"charts={len(chart_rows)}, ready={len(ready)}"), ("MISSING_CHART_INPUTS_RECORDED_AS_HELD", len(missing_rows) == len(data_gap), f"missing={len(missing_rows)}, data_gap={len(data_gap)}"), ] self_items = pd.DataFrame([{"item": k, "status": "PASS" if ok else "FAIL", "detail": detail} for k, ok, detail in items]) write_csv(self_items, "buy_review_self_check_items.csv") status = "PASS_FOR_BUY_POINT_MANUAL_REVIEW_PREP_READY" if self_items["status"].eq("PASS").all() else "FAIL" write_json({"run_id": RUN_ID, "generated_at": generated_at, "stage": status, "pass_count": int(self_items["status"].eq("PASS").sum()), "fail_count": int(self_items["status"].eq("FAIL").sum())}, "buy_review_self_check.json") summary = { "run_id": RUN_ID, "generated_at": generated_at, "stage": status, "counts": { "selected_candidates": int(len(selected)), "eligible_market_gate_open_before_data_gap": int(len(eligible)), "buy_point_manual_review_ready": int(len(ready)), "buy_point_minute_data_gap_held": int(len(data_gap)), "market_gate_closed_boundary": int(len(boundary)), "buy_point_review_charts": int(len(chart_rows)), "missing_chart_inputs": int(len(missing_rows)), }, "boundary": [ "No final BUY action is generated by this package.", "manual_buy_decision_external_template.csv is intentionally blank for final human decision fields.", "Market-gate-closed candidates are retained as boundary evidence only.", ], } write_json(summary, "buy_review_summary.json") (ROOT / "buy_review_summary.md").write_text( "# Strict Note Buy Point Manual Review Prep\n\n" f"- generated_at: {generated_at}\n" f"- selected candidates: {len(selected)}\n" f"- market gate open before minute-data check: {len(eligible)}\n" f"- buy point manual review ready: {len(ready)}\n" f"- buy point minute data gap held: {len(data_gap)}\n" f"- market gate closed boundary: {len(boundary)}\n" f"- charts: {len(chart_rows)}\n" f"- missing chart inputs: {len(missing_rows)}\n\n" "Boundary: this package prepares external manual/AI-manual buy point review inputs only. It does not generate BUY orders or returns.\n", encoding="utf-8", ) 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()