from __future__ import annotations import hashlib import json import os import re from pathlib import Path import pandas as pd import pymysql from PIL import Image, ImageDraw, ImageFont RUN_ID = "RUN-ANA-WUJI-BASELINE-PILOT-20260607-001" ROOT = Path(__file__).resolve().parents[1] LOCAL_DB_INDEX = Path( r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md" ) 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=120, ) 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 font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont: 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(26) FONT_MID = font(17) 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 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_chart(df: pd.DataFrame, cand: dict, window_name: str, out_path: Path, ma20_ref: float | None) -> 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"买点复核1分钟K图：{cand['symbol']} {cand['entry_trade_date']} {window_name}" d.text((32, 24), title, fill="#111827", font=FONT_TITLE) d.text((32, 58), "当前为人工复核材料，不自动判定 BUY；买点必须由后续人工/AI 看图确认。", fill="#334155", font=FONT_MID) plot_left, plot_top, plot_right, plot_bottom = 80, 105, 1060, 575 vol_top, vol_bottom = 625, 780 note_left, note_top = 1090, 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") if df.empty: d.text((plot_left + 160, plot_top + 180), "该窗口无分钟线数据", fill="#b91c1c", font=FONT_TITLE) price_low, price_high = 0.0, 1.0 else: price_low = float(df["low_price"].min()) * 0.998 price_high = float(df["high_price"].max()) * 1.002 refs = [float(df["open_price"].iloc[0])] if ma20_ref: refs.append(ma20_ref) price_low = min(price_low, min(refs) * 0.998) price_high = max(price_high, max(refs) * 1.002) max_vol = max(float(df["volume"].max()), 1.0) n = len(df) gap = (plot_right - plot_left) / max(n, 1) body_w = max(3, int(gap * 0.55)) for i, row in df.reset_index(drop=True).iterrows(): cx = int(plot_left + gap * i + gap / 2) op, hi, lo, cl = [float(row[c]) for c in ["open_price", "high_price", "low_price", "close_price"]] color = "#dc2626" if cl >= op else "#16a34a" d.line([cx, y_price(lo, price_low, price_high, plot_top, plot_bottom), cx, y_price(hi, price_low, price_high, plot_top, plot_bottom)], fill=color, width=2) y1, y2 = y_price(op, price_low, price_high, plot_top, plot_bottom), y_price(cl, price_low, price_high, plot_top, plot_bottom) d.rectangle([cx - body_w // 2, min(y1, y2), cx + body_w // 2, max(y1, y2)], fill=color, outline=color) vh = int(float(row["volume"]) / max_vol * (vol_bottom - vol_top)) d.rectangle([cx - body_w // 2, vol_bottom - vh, cx + body_w // 2, vol_bottom], fill=color, outline=color) if i % max(1, n // 6) == 0: d.text((cx - 24, vol_bottom + 8), str(row["trade_time"])[:5], fill="#64748b", font=FONT_SMALL) open_ref = float(df["open_price"].iloc[0]) y_open = y_price(open_ref, price_low, price_high, plot_top, plot_bottom) d.line([plot_left, y_open, plot_right, y_open], fill="#0f172a", width=2) d.text((plot_right + 8, y_open - 8), f"开盘价 {open_ref:.2f}", fill="#0f172a", font=FONT_SMALL) if ma20_ref: y_ma20 = y_price(ma20_ref, price_low, price_high, plot_top, plot_bottom) d.line([plot_left, y_ma20, plot_right, y_ma20], fill="#f59e0b", width=2) d.text((plot_right + 8, y_ma20 - 8), f"日MA20 {ma20_ref:.2f}", fill="#b45309", font=FONT_SMALL) for i in range(5): p = price_low + (price_high - price_low) * i / 4 y = y_price(p, price_low, price_high, plot_top, plot_bottom) d.line([plot_left, y, plot_right, y], fill="#e2e8f0") d.text((18, y - 8), f"{p:.2f}", fill="#64748b", font=FONT_SMALL) d.rounded_rectangle([note_left, note_top, 1460, 780], radius=8, outline="#334155", fill="#ffffff") d.text((note_left + 18, note_top + 18), "买点复核提示", fill="#111827", font=FONT_TITLE) notes = [ f"候选排名：{cand['candidate_rank']}", f"候选状态：{cand['candidate_status']}", f"窗口：{window_name}", "允许买点：10:40前或14:40后", "看图重点：", "1. 回踩20日均线附近", "2. 回踩开盘价附近", "3. 缩量有支撑", "4. 放量回踩均线禁买", "5. 急拉放量不追", "", "当前动作：ENTRY_REVIEW_PENDING", "不写 BUY，不计收益。", ] yy = note_top + 58 for line in notes: d.text((note_left + 18, yy), line, fill="#334155", font=FONT_SMALL) yy += 24 if line else 12 d.text((32, 820), "无忌 baseline：买点由1分钟K图和人工确认裁决；本图只准备证据，不产生交易结论。", fill="#334155", font=FONT_MID) img.save(out_path) def main() -> None: selected = pd.read_csv(ROOT / "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"]) open_selected = selected[selected["market_gate_status"] == "MKT_GATE_OPEN_PREV_DAY_UP_3000"].copy() symbols = sorted(open_selected["symbol"].unique().tolist()) dates = sorted(open_selected["entry_trade_date"].dt.strftime("%Y-%m-%d").unique().tolist()) min_signal = (open_selected["signal_trade_date"].min() - pd.Timedelta(days=80)).strftime("%Y-%m-%d") max_signal = open_selected["signal_trade_date"].max().strftime("%Y-%m-%d") minute = pd.DataFrame() daily = pd.DataFrame() if symbols and dates: sym_ph = ",".join(["%s"] * len(symbols)) date_ph = ",".join(["%s"] * len(dates)) with get_conn() as conn: minute = pd.read_sql( f""" SELECT trade_date, trade_time, symbol, open_price, high_price, low_price, close_price, volume, amount FROM a_share_minute_price WHERE symbol IN ({sym_ph}) AND trade_date IN ({date_ph}) ORDER BY symbol, trade_date, trade_time """, conn, params=[*symbols, *dates], ) daily = pd.read_sql( f""" SELECT trade_date, symbol, close_price FROM a_share_daily_price WHERE symbol IN ({sym_ph}) AND trade_date BETWEEN %s AND %s ORDER BY symbol, trade_date """, conn, params=[*symbols, min_signal, max_signal], ) 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", "amount"]: 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["ma20"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(20, min_periods=1).mean()) manifest_rows = [] decision_rows = [] morning_count = 0 late_count = 0 data_gap_count = 0 root_manifest_path = ROOT / "image_manifest.csv" if root_manifest_path.exists(): root_manifest = pd.read_csv(root_manifest_path, encoding="utf-8-sig") else: root_manifest = pd.DataFrame() for _, cand in selected.iterrows(): case_id = cand["case_id"] case_dir = ROOT / "cases" / case_id img_dir = case_dir / "img" img_dir.mkdir(parents=True, exist_ok=True) entry_str = cand["entry_trade_date"].strftime("%Y-%m-%d") signal_str = cand["signal_trade_date"].strftime("%Y-%m-%d") if cand["market_gate_status"] != "MKT_GATE_OPEN_PREV_DAY_UP_3000": decision_rows.append( { "case_id": case_id, "candidate_id": cand["candidate_id"], "symbol": cand["symbol"], "entry_trade_date": entry_str, "decision_stage": "ENTRY_GATE", "action_status": "NO_TRADE_MARKET_GATE_CLOSED", "review_required": False, "evidence_image_path": "", "decision_reason_cn": "前一交易日全A上涨家数未达到3000，按baseline不开新仓。", "lookahead_violation_flag": False, } ) continue ma20_rows = daily[(daily["symbol"] == cand["symbol"]) & (daily["trade_date"] <= cand["signal_trade_date"])].tail(1) ma20_ref = float(ma20_rows["ma20"].iloc[0]) if not ma20_rows.empty else None m = minute[(minute["symbol"] == cand["symbol"]) & (minute["trade_date"] == cand["entry_trade_date"])].copy() windows = [ ("早盘窗口", "morning", "09:30:00", "10:40:00"), ("尾盘窗口", "late", "14:40:00", "15:00:00"), ] candidate_paths = [] missing_windows = [] for label, key, start, end in windows: part = m[(m["trade_time"] >= start) & (m["trade_time"] <= end)].copy() file_name = f"03_entry_1m_{key}_review_{cand['symbol'].replace('.', '_')}_{entry_str.replace('-', '')}.png" out_path = img_dir / file_name cand_dict = cand.to_dict() cand_dict["entry_trade_date"] = entry_str cand_dict["signal_trade_date"] = signal_str draw_chart(part, cand_dict, label, out_path, ma20_ref) if key == "morning": morning_count += 1 else: late_count += 1 if part.empty: data_gap_count += 1 missing_windows.append(label) rel = out_path.relative_to(ROOT).as_posix() candidate_paths.append(rel) manifest_rows.append( { "case_id": case_id, "symbol": cand["symbol"], "trade_date": entry_str, "event_id": f"{cand['candidate_id']}_{key}", "chart_role": f"entry_1m_{key}_review_view", "decision_time": f"{entry_str} {end}", "path": rel, "sha256": sha256_file(out_path), "status": "PASS" if not part.empty else "DATA_GAP_HELD", "note": "1分钟买点复核图；不自动判定BUY。", } ) decision_rows.append( { "case_id": case_id, "candidate_id": cand["candidate_id"], "symbol": cand["symbol"], "entry_trade_date": entry_str, "decision_stage": "ENTRY_REVIEW", "action_status": "ENTRY_REVIEW_PENDING" if not missing_windows else "ENTRY_REVIEW_DATA_GAP", "review_required": True, "evidence_image_path": ";".join(candidate_paths), "decision_reason_cn": ( "已生成1分钟买点复核图，需人工/AI按缩量支撑、回踩均线/开盘价、急拉放量禁追等规则裁决。" if not missing_windows else "部分买点窗口缺少分钟线：" + "、".join(missing_windows) ), "lookahead_violation_flag": False, } ) new_manifest = pd.DataFrame(manifest_rows) if not root_manifest.empty: combined_manifest = pd.concat([root_manifest, new_manifest], ignore_index=True) combined_manifest = combined_manifest.drop_duplicates(subset=["case_id", "symbol", "event_id", "chart_role"], keep="last") else: combined_manifest = new_manifest combined_manifest.to_csv(ROOT / "image_manifest.csv", index=False, encoding="utf-8-sig") decision_log = pd.DataFrame(decision_rows) decision_log.to_csv(ROOT / "decision_log.csv", index=False, encoding="utf-8-sig") # Append entry section to case boards. for case_id, group in new_manifest.groupby("case_id"): case_dir = ROOT / "cases" / case_id board_path = case_dir / "case_image_board.md" existing = board_path.read_text(encoding="utf-8") if board_path.exists() else f"# {case_id} 图片审核板\n" lines = [existing.rstrip(), "", "## 3. 买点 1分钟K 复核图", ""] for _, row in group.iterrows(): rel = Path(row["path"]).relative_to(f"cases/{case_id}").as_posix() lines.extend( [ f"### {row['symbol']} / {row['chart_role']}", "", f"![{row['symbol']}]({rel})", "", f"- 状态：`{row['status']}`", "- 本图只用于买点复核，不自动写 BUY。", "", ] ) board_path.write_text("\n".join(lines) + "\n", encoding="utf-8") case_manifest = combined_manifest[combined_manifest["case_id"] == case_id] case_manifest.to_csv(case_dir / "image_manifest.csv", index=False, encoding="utf-8-sig") summary = { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": "2026-06-08T01:05:00+08:00", "stage": "ENTRY_1M_REVIEW_IMAGES_READY", "decision_rows": int(len(decision_log)), "entry_review_images": int(len(new_manifest)), "morning_images": morning_count, "late_images": late_count, "data_gap_windows": data_gap_count, "no_trade_market_gate_closed_rows": int((decision_log["action_status"] == "NO_TRADE_MARKET_GATE_CLOSED").sum()), "entry_review_pending_rows": int((decision_log["action_status"] == "ENTRY_REVIEW_PENDING").sum()), "entry_review_data_gap_rows": int((decision_log["action_status"] == "ENTRY_REVIEW_DATA_GAP").sum()), "artifacts": { "decision_log.csv": { "size": (ROOT / "decision_log.csv").stat().st_size, "sha256": sha256_file(ROOT / "decision_log.csv"), }, "image_manifest.csv": { "size": (ROOT / "image_manifest.csv").stat().st_size, "sha256": sha256_file(ROOT / "image_manifest.csv"), }, }, "next_step": "Perform manual/AI entry review from 1-minute K charts. Do not create order_ledger until BUY/NO_BUY decisions are confirmed.", } (ROOT / "entry_review_generation_summary.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) (ROOT / "entry_review_generation_summary.md").write_text( "\n".join( [ "# entry_review_generation_summary", "", f"run_id：`{RUN_ID}`", "阶段：`ENTRY_1M_REVIEW_IMAGES_READY`", "", f"- decision rows：{summary['decision_rows']}", f"- 1分钟买点复核图：{summary['entry_review_images']}", f"- 数据缺口窗口：{summary['data_gap_windows']}", f"- 闸门关闭不交易行：{summary['no_trade_market_gate_closed_rows']}", f"- 待人工/AI买点复核行：{summary['entry_review_pending_rows']}", "", "当前不产生订单、不产生收益结论。", "", ] ), encoding="utf-8", ) if __name__ == "__main__": main()