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): 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 evaluate_candidate(cand: pd.Series, minute: pd.DataFrame, daily: pd.DataFrame) -> dict: m = minute[(minute.symbol == cand.symbol) & (minute.trade_date == cand.entry_trade_date)].copy() ma = daily[(daily.symbol == cand.symbol) & (daily.trade_date <= cand.signal_trade_date)].tail(1) ma20 = float(ma.ma20.iloc[0]) if not ma.empty else None if m.empty: return {"action_status": "DATA_GAP_HELD", "reason": "入场日无分钟线，不能判断买点。"} open_ref = float(m.open_price.iloc[0]) opening = m[(m.trade_time >= "09:30:00") & (m.trade_time <= "09:35:00")] had_opening_push = (not opening.empty) and float(opening.high_price.max()) >= open_ref * 1.01 for label, start, end, min_time in [ ("早盘窗口", "09:30:00", "10:40:00", "09:34:00"), ("尾盘窗口", "14:40:00", "15:00:00", "14:40:00"), ]: part = m[(m.trade_time >= start) & (m.trade_time <= end) & (m.trade_time >= min_time)].copy() if part.empty: continue part["vol_prev5"] = part.volume.shift(1).rolling(5, min_periods=3).mean() part["near_open"] = had_opening_push & part.low_price.le(open_ref * 1.010) & part.high_price.ge(open_ref * 0.997) part["near_ma20"] = False if ma20 is None else ( part.low_price.le(ma20 * 1.010) & part.high_price.ge(ma20 * 0.997) ) part["vol_ok"] = part.vol_prev5.notna() & part.volume.le(part.vol_prev5 * 1.10) part["not_chase"] = part.close_price.le(open_ref * 1.055) part["ok"] = (part.near_open | part.near_ma20) & part.vol_ok & part.not_chase ok = part[part.ok] if not ok.empty: row = ok.iloc[0] trigger = "回踩开盘价附近缩量支撑" if bool(row.near_open) else "回踩日MA20附近缩量支撑" return { "action_status": "AI_BUY_CONFIRMED", "decision_time": str(row.trade_time), "price": float(row.close_price), "window": label, "trigger": trigger, "ma20_ref": ma20, "open_ref": open_ref, "reason": f"{label} {row.trade_time} {trigger}，未触发放量回踩禁买或急拉追高禁买。", } return { "action_status": "NO_BUY_AI_REVIEWED", "reason": "允许买入窗口内未看到符合缩量支撑的回踩开盘价或回踩日MA20买点。", "ma20_ref": ma20, "open_ref": open_ref, } def draw_buy_decision(minute: pd.DataFrame, cand: pd.Series, decision: dict, 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"买入决策1分钟K图：{cand.symbol} {cand.entry_trade_date.strftime('%Y-%m-%d')}" d.text((32, 24), title, fill="#111827", font=FONT_TITLE) d.text((32, 58), f"AI裁决：{decision['decision_time']} 买入第一份仓，价格 {decision['price']:.2f}", fill="#7f1d1d", 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") start, end = ("09:30:00", "10:40:00") if decision["window"] == "早盘窗口" else ("14:40:00", "15:00:00") df = minute[ (minute.symbol == cand.symbol) & (minute.trade_date == cand.entry_trade_date) & (minute.trade_time >= start) & (minute.trade_time <= end) & (minute.trade_time <= decision["decision_time"]) ].copy().reset_index(drop=True) refs = [decision["open_ref"]] if decision.get("ma20_ref"): refs.append(decision["ma20_ref"]) price_low = min(float(df.low_price.min()), min(refs)) * 0.998 price_high = max(float(df.high_price.max()), 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)) buy_x = None for i, row in df.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 str(row.trade_time) == decision["decision_time"]: buy_x = cx if i % max(1, n // 6) == 0: d.text((cx - 24, vol_bottom + 8), str(row.trade_time)[:5], fill="#64748b", font=FONT_SMALL) for ref, label, color in [ (decision["open_ref"], "开盘价", "#0f172a"), (decision.get("ma20_ref"), "日MA20", "#f59e0b"), ]: if ref: yy = y_price(float(ref), 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} {float(ref):.2f}", fill=color, font=FONT_SMALL) if buy_x is not None: d.line([buy_x, plot_top, buy_x, vol_bottom], fill="#b91c1c", width=3) d.text((buy_x + 8, plot_top + 8), "买入", fill="#b91c1c", font=FONT_MID) d.rounded_rectangle([note_left, note_top, 1460, 780], radius=8, outline="#334155", fill="#ffffff") notes = [ "买入裁决", f"动作：BUY 第一份仓", f"时间：{decision['decision_time']}", f"价格：{decision['price']:.2f}", "仓位：账户4%", f"触发：{decision['trigger']}", "依据：无忌买点规则", "T+1：当日不可卖", "", "本图为 decision_view，", "只使用买入时间及以前", "的分钟数据裁决。", ] yy = note_top + 18 for i, line in enumerate(notes): d.text((note_left + 18, 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((32, 820), "无忌 baseline：本买入由AI按已冻结规则看图裁决，后续仍需执行审核确认。", 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 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], ) calendar = pd.read_sql( """ SELECT calendar_date FROM a_share_trading_calendar WHERE is_trading_day=1 AND calendar_date BETWEEN '2023-01-01' AND '2026-12-31' ORDER BY calendar_date """, conn, ) 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") 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()) calendar.calendar_date = pd.to_datetime(calendar.calendar_date) trading_dates = list(calendar.calendar_date.dt.strftime("%Y-%m-%d")) def next_trade_date(date_str: str) -> str: idx = trading_dates.index(date_str) return trading_dates[idx + 1] decision_rows = [] order_rows = [] lot_rows = [] manifest_rows = [] order_seq = 0 for _, cand in selected.iterrows(): entry_str = cand.entry_trade_date.strftime("%Y-%m-%d") if cand.market_gate_status != "MKT_GATE_OPEN_PREV_DAY_UP_3000": decision_rows.append( { "case_id": cand.case_id, "candidate_id": cand.candidate_id, "symbol": cand.symbol, "entry_trade_date": entry_str, "decision_stage": "ENTRY_AI_REVIEW", "action_status": "NO_TRADE_MARKET_GATE_CLOSED", "decision_time": "", "price": "", "position_delta_pct": 0, "review_required": False, "evidence_image_path": "", "decision_reason_cn": "前一交易日全A上涨家数未达到3000，按baseline不开新仓。", "lookahead_violation_flag": False, } ) continue decision = evaluate_candidate(cand, minute, daily) evidence = "" if decision["action_status"] == "AI_BUY_CONFIRMED": case_dir = ROOT / "cases" / cand.case_id img_dir = case_dir / "img" img_dir.mkdir(parents=True, exist_ok=True) out_path = img_dir / f"04_entry_1m_decision_{cand.symbol.replace('.', '_')}_{entry_str.replace('-', '')}_{decision['decision_time'].replace(':', '')}.png" draw_buy_decision(minute, cand, decision, out_path) evidence = out_path.relative_to(ROOT).as_posix() manifest_rows.append( { "case_id": cand.case_id, "symbol": cand.symbol, "trade_date": entry_str, "event_id": f"{cand.candidate_id}_buy_decision", "chart_role": "entry_1m_buy_decision_view", "decision_time": f"{entry_str} {decision['decision_time']}", "path": evidence, "sha256": sha256_file(out_path), "status": "PASS", "note": "AI买入裁决图，标出买入点；仍需执行审核确认。", } ) order_seq += 1 order_id = f"ORD-{RUN_ID}-{order_seq:04d}" lot_id = f"LOT-{RUN_ID}-{order_seq:04d}" sellable = next_trade_date(entry_str) order_rows.append( { "order_id": order_id, "case_id": cand.case_id, "candidate_id": cand.candidate_id, "variant_id": "V0A_STRICT_TIME_WINDOW", "symbol": cand.symbol, "trade_date": entry_str, "trade_time": decision["decision_time"], "action": "BUY", "price": f"{decision['price']:.4f}", "position_delta_pct": "0.04", "tranche_index": 1, "planned_tranche_count": 5, "decision_reason_cn": decision["reason"], "evidence_image_path": evidence, "t1_sellable_from_trade_date": sellable, "lookahead_violation_flag": False, } ) lot_rows.append( { "trade_lot_id": lot_id, "order_id": order_id, "case_id": cand.case_id, "variant_id": "V0A_STRICT_TIME_WINDOW", "symbol": cand.symbol, "entry_trade_date": entry_str, "entry_time": decision["decision_time"], "entry_price": f"{decision['price']:.4f}", "position_pct": "0.04", "tranche_index": 1, "sellable_from_trade_date": sellable, "lot_status": "OPEN_PENDING_EXIT_REVIEW", "exit_trade_date": "", "exit_time": "", "exit_price": "", "lot_return_pct": "", "account_return_contribution_pct": "", } ) decision_rows.append( { "case_id": cand.case_id, "candidate_id": cand.candidate_id, "symbol": cand.symbol, "entry_trade_date": entry_str, "decision_stage": "ENTRY_AI_REVIEW", "action_status": decision["action_status"], "decision_time": decision.get("decision_time", ""), "price": "" if "price" not in decision else f"{decision['price']:.4f}", "position_delta_pct": "0.04" if decision["action_status"] == "AI_BUY_CONFIRMED" else 0, "review_required": False, "evidence_image_path": evidence, "decision_reason_cn": decision["reason"], "lookahead_violation_flag": False, } ) decision_log = pd.DataFrame(decision_rows) decision_log.to_csv(ROOT / "decision_log.csv", index=False, encoding="utf-8-sig") pd.DataFrame(order_rows).to_csv(ROOT / "order_ledger.csv", index=False, encoding="utf-8-sig") pd.DataFrame(lot_rows).to_csv(ROOT / "position_lot_ledger.csv", index=False, encoding="utf-8-sig") root_manifest_path = ROOT / "image_manifest.csv" root_manifest = pd.read_csv(root_manifest_path, encoding="utf-8-sig") if root_manifest_path.exists() else pd.DataFrame() if manifest_rows: new_manifest = pd.DataFrame(manifest_rows) combined = pd.concat([root_manifest, new_manifest], ignore_index=True) combined = combined.drop_duplicates(subset=["case_id", "symbol", "event_id", "chart_role"], keep="last") else: combined = root_manifest combined.to_csv(ROOT / "image_manifest.csv", index=False, encoding="utf-8-sig") # Append buy-decision image references to case boards and per-case manifests. if manifest_rows: new_manifest = pd.DataFrame(manifest_rows) 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(), "", "## 4. AI 买入决策图", ""] for _, row in group.iterrows(): rel = Path(row.path).relative_to(f"cases/{case_id}").as_posix() lines.extend( [ f"### {row.symbol} / BUY", "", f"![{row.symbol}]({rel})", "", f"- 决策时间：`{row.decision_time}`", "- 本图标出买入点，后续仍需执行审核确认。", "", ] ) board_path.write_text("\n".join(lines) + "\n", encoding="utf-8") case_manifest = combined[combined.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:25:00+08:00", "stage": "ENTRY_AI_REVIEW_DONE", "decision_counts": decision_log.action_status.value_counts().to_dict(), "order_rows": len(order_rows), "open_lot_rows": len(lot_rows), "buy_decision_image_count": len(manifest_rows), "artifacts": { "decision_log.csv": {"size": (ROOT / "decision_log.csv").stat().st_size, "sha256": sha256_file(ROOT / "decision_log.csv")}, "order_ledger.csv": {"size": (ROOT / "order_ledger.csv").stat().st_size, "sha256": sha256_file(ROOT / "order_ledger.csv")}, "position_lot_ledger.csv": {"size": (ROOT / "position_lot_ledger.csv").stat().st_size, "sha256": sha256_file(ROOT / "position_lot_ledger.csv")}, "image_manifest.csv": {"size": (ROOT / "image_manifest.csv").stat().st_size, "sha256": sha256_file(ROOT / "image_manifest.csv")}, }, "boundary": "Entry AI review only; no sell decisions and no return statistics yet.", "next_step": "Run T+1-safe holding/exit review for open lots.", } (ROOT / "entry_ai_review_summary.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) (ROOT / "entry_ai_review_summary.md").write_text( "\n".join( [ "# entry_ai_review_summary", "", f"run_id：`{RUN_ID}`", "阶段：`ENTRY_AI_REVIEW_DONE`", "", f"- BUY 订单：{len(order_rows)}", f"- open lots：{len(lot_rows)}", f"- 买入决策图：{len(manifest_rows)}", "", "当前只完成买入裁决；卖点、持仓和收益复算尚未完成。", "", ] ), encoding="utf-8", ) if __name__ == "__main__": main()