from __future__ import annotations import hashlib import json import math import os import re from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Iterable import pandas as pd import pymysql from PIL import Image, ImageDraw, ImageFont RUN_ID = "RUN-ANA-WUJI-STRICT-CLOSED-234-CHART-ENHANCE-20260609-001" SOURCE_RUN_ID = "RUN-ANA-WUJI-FULL-2023-2026-20260608-001" GUIDE_RUN_ID = "RUN-ANA-WUJI-STRICT-CLOSED-234-REVIEW-GUIDE-20260608-001" TASK_ID = "ANA-WUJI-STRICT-CLOSED-234-CHART-ENHANCE-20260609" PACKAGE_ROOT = Path(__file__).resolve().parents[1] RESULT_ROOT = Path(__file__).resolve().parents[2] PROJECT_ROOT = Path(__file__).resolve().parents[4] SOURCE_ROOT = RESULT_ROOT / SOURCE_RUN_ID GUIDE_ROOT = RESULT_ROOT / GUIDE_RUN_ID LOCAL_DB_INDEX = Path( r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md" ) def now_iso() -> str: return datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds") GENERATED_AT = now_iso() def read_password() -> str: env = os.environ.get("TIANXIA_MYSQL_PASSWORD") or os.environ.get("MYSQL_PWD") if env: return env candidates = [LOCAL_DB_INDEX] observer_root = Path(r"D:\strategy_project\s-system-doc\observer") if observer_root.exists(): candidates.extend(observer_root.glob("*/数据库索引数据.md")) for path in candidates: if not path.exists(): continue text = path.read_text(encoding="utf-8") match = re.search(r"^\s*-\s*密码：`([^`]+)`", text, re.MULTILINE) if match: return match.group(1) raise RuntimeError("Unable to read local MySQL credential from approved local index.") 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 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(28) FONT_SUB = font(20) FONT_MID = font(17) FONT_SMALL = font(14) FONT_TINY = font(12) def read_csv(name: str) -> pd.DataFrame: return pd.read_csv(SOURCE_ROOT / name, encoding="utf-8-sig") def norm_time(value) -> str: if pd.isna(value): return "" if hasattr(value, "total_seconds"): seconds = int(value.total_seconds()) h, rem = divmod(seconds, 3600) m, s = divmod(rem, 60) return f"{h:02d}:{m:02d}:{s:02d}" text = str(value).strip() if "days" in text and " " in text: text = text.split()[-1] if re.fullmatch(r"\d{2}:\d{2}$", text): return f"{text}:00" if re.fullmatch(r"\d{2}:\d{2}:\d{2}(\.\d+)?", text): return text[:8] return text def to_bool_int(series: pd.Series) -> pd.Series: return pd.to_numeric(series, errors="coerce").fillna(0).astype(int) def y_price(value: float, low: float, high: float, top: int, bottom: int) -> int: if not math.isfinite(value) or high <= low: return (top + bottom) // 2 return bottom - int((value - low) / (high - low) * (bottom - top)) def wrap_text(text: str, max_chars: int) -> list[str]: text = "" if pd.isna(text) else str(text) lines: list[str] = [] current = "" for ch in text: current += ch if len(current) >= max_chars: lines.append(current) current = "" if current: lines.append(current) return lines or [""] def pct_text(value) -> str: try: return f"{float(value):.2%}" except Exception: return "" def money_text(value) -> str: try: return f"{float(value):.6f}" except Exception: return "" def gate_cn(value: str) -> str: return "市场闸门打开" if value == "MKT_GATE_OPEN_PREV_DAY_UP_3000" else "市场闸门关闭" def status_cn(value: str) -> str: if value == "PASS": return "候选通过" if value == "FAKE_BREAKOUT_RISK_REVIEW": return "前高假突破风险复核" return str(value) def draw_text_box( draw: ImageDraw.ImageDraw, box: tuple[int, int, int, int], title: str, lines: Iterable[str], ) -> None: x1, y1, x2, y2 = box draw.rounded_rectangle([x1, y1, x2, y2], radius=8, outline="#334155", fill="#ffffff") draw.text((x1 + 18, y1 + 16), title, fill="#111827", font=FONT_TITLE) y = y1 + 58 for line in lines: if not line: y += 10 continue for wrapped in wrap_text(line, 25): if y > y2 - 22: return draw.text((x1 + 18, y), wrapped, fill="#334155", font=FONT_SMALL) y += 23 def fetch_daily(selected: pd.DataFrame) -> pd.DataFrame: symbols = sorted(selected["symbol"].dropna().unique().tolist()) min_date = (pd.to_datetime(selected["signal_trade_date"]).min() - pd.Timedelta(days=230)).strftime("%Y-%m-%d") max_date = (pd.to_datetime(selected["signal_trade_date"]).max() + pd.Timedelta(days=45)).strftime("%Y-%m-%d") parts: list[pd.DataFrame] = [] with get_conn() as conn: for i in range(0, len(symbols), 180): chunk = symbols[i : i + 180] ph = ",".join(["%s"] * len(chunk)) parts.append( pd.read_sql( f""" SELECT trade_date, symbol, open_price, high_price, low_price, close_price, volume, amount FROM a_share_daily_price WHERE trade_date BETWEEN %s AND %s AND symbol IN ({ph}) ORDER BY symbol, trade_date """, conn, params=[min_date, max_date, *chunk], ) ) daily = pd.concat(parts, ignore_index=True) if parts else pd.DataFrame() if daily.empty: return daily 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) daily["ma5"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(5, min_periods=1).mean()) daily["ma20"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(20, min_periods=1).mean()) daily["ma60"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(60, min_periods=1).mean()) return daily def fetch_minute(orders: pd.DataFrame) -> pd.DataFrame: date_symbol_map: dict[str, set[str]] = {} for _, row in orders.iterrows(): date_symbol_map.setdefault(str(row["trade_date"]), set()).add(str(row["symbol"])) parts: list[pd.DataFrame] = [] with get_conn() as conn: for idx, trade_date in enumerate(sorted(date_symbol_map), start=1): symbols = sorted(date_symbol_map[trade_date]) ph = ",".join(["%s"] * len(symbols)) parts.append( 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 trade_date = %s AND symbol IN ({ph}) ORDER BY symbol, trade_date, trade_time """, conn, params=[trade_date, *symbols], ) ) if idx % 100 == 0: print(f"minute query progress: {idx}/{len(date_symbol_map)} dates", flush=True) minute = pd.concat(parts, ignore_index=True) if parts else pd.DataFrame() if minute.empty: return minute minute["trade_date"] = pd.to_datetime(minute["trade_date"]).dt.strftime("%Y-%m-%d") minute["trade_time"] = minute["trade_time"].map(norm_time) for col in ["open_price", "high_price", "low_price", "close_price", "volume", "amount"]: minute[col] = pd.to_numeric(minute[col], errors="coerce") return minute def daily_window(daily: pd.DataFrame, symbol: str, signal_date: str) -> pd.DataFrame: sd = pd.Timestamp(signal_date) rows = daily[daily["symbol"] == symbol].sort_values("trade_date").reset_index(drop=True) hits = rows.index[rows["trade_date"] == sd].tolist() if not hits: return pd.DataFrame() pos = hits[0] start = max(0, pos - 90) end = min(len(rows), pos + 11) return rows.iloc[start:end].copy().reset_index(drop=True) def draw_daily_enhanced_chart(window: pd.DataFrame, cand: pd.Series, out_path: Path) -> None: w, h = 1680, 980 img = Image.new("RGB", (w, h), "#fbfbf7") d = ImageDraw.Draw(img) d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1") signal_date = str(cand["signal_trade_date"]) entry_date = str(cand["entry_trade_date"]) title = f"选股日K图增强审阅：{cand['symbol']} 信号日 {signal_date}" subtitle = "窗口：信号日前90个交易日 + 信号日后10个交易日；信号后部分只作 audit_view，不参与当时选股。" d.text((32, 22), title, fill="#111827", font=FONT_TITLE) d.text((32, 62), subtitle, fill="#7f1d1d", font=FONT_MID) plot_left, plot_top, plot_right, plot_bottom = 80, 118, 1190, 640 vol_top, vol_bottom = 700, 850 note_left, note_top = 1220, 118 d.rectangle([plot_left, plot_top, plot_right, plot_bottom], outline="#94a3b8") d.rectangle([plot_left, vol_top, plot_right, vol_bottom], outline="#94a3b8") win = window.copy().reset_index(drop=True) win["trade_date_str"] = win["trade_date"].dt.strftime("%Y-%m-%d") price_low = float(win["low_price"].min()) * 0.98 price_high = float(win["high_price"].max()) * 1.02 max_vol = max(float(win["volume"].max()), 1.0) n = len(win) gap = (plot_right - plot_left) / max(n, 1) body_w = max(3, int(gap * 0.58)) signal_positions = win.index[win["trade_date_str"] == signal_date].tolist() entry_positions = win.index[win["trade_date_str"] == entry_date].tolist() signal_pos = signal_positions[0] if signal_positions else None if signal_pos is not None and signal_pos + 1 < n: x1 = int(plot_left + gap * (signal_pos + 1)) d.rectangle([x1, plot_top, plot_right, vol_bottom], fill="#fff7ed") d.text((x1 + 8, plot_top + 8), "信号后10日审阅区", fill="#9a3412", font=FONT_SMALL) for i in range(5): price = price_low + (price_high - price_low) * i / 4 y = y_price(price, price_low, price_high, plot_top, plot_bottom) d.line([plot_left, y, plot_right, y], fill="#e2e8f0") d.text((18, y - 8), f"{price:.2f}", fill="#64748b", font=FONT_SMALL) ma_points = {"ma5": [], "ma20": [], "ma60": []} ma_colors = {"ma5": "#2563eb", "ma20": "#f59e0b", "ma60": "#7c3aed"} for i, row in win.iterrows(): cx = int(plot_left + gap * i + gap / 2) op = float(row["open_price"]) hi = float(row["high_price"]) lo = float(row["low_price"]) cl = float(row["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 = y_price(op, price_low, price_high, plot_top, plot_bottom) y2 = 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) for ma in ma_points: if pd.notna(row[ma]): ma_points[ma].append((cx, y_price(float(row[ma]), price_low, price_high, plot_top, plot_bottom))) if row["trade_date_str"] == signal_date: d.line([cx, plot_top, cx, vol_bottom], fill="#0f172a", width=3) d.text((cx + 5, plot_top + 30), "信号日", fill="#0f172a", font=FONT_SMALL) if row["trade_date_str"] == entry_date: d.line([cx, plot_top, cx, vol_bottom], fill="#b91c1c", width=2) d.text((cx + 5, plot_top + 55), "买入日", fill="#b91c1c", font=FONT_SMALL) if i % max(1, n // 9) == 0: d.text((cx - 24, vol_bottom + 8), row["trade_date_str"][5:], fill="#64748b", font=FONT_SMALL) for ma, pts in ma_points.items(): if len(pts) > 1: d.line(pts, fill=ma_colors[ma], width=2) legend_x = plot_left + 8 for ma, color in ma_colors.items(): d.text((legend_x, plot_bottom + 12), ma.upper(), fill=color, font=FONT_SMALL) legend_x += 75 lines = [ f"候选排名：{cand['candidate_rank']}", f"候选状态：{status_cn(cand['candidate_status'])}", f"市场闸门：{gate_cn(cand['market_gate_status'])}", f"上涨/下跌家数：{cand.get('up_count', '')}/{cand.get('down_count', '')}", f"量比：{float(cand['volume_ratio']):.2f}，长上影：{float(cand['upper_shadow_pct']):.2f}%", f"近30日涨停：{cand.get('last_limitup_date', '')}", f"前高引用：{cand.get('prev60_high_ref_date', '')}", f"前高量能通过：{cand.get('prev_high_volume_pass_flag', '')}", "", "重要边界：信号日后10天仅帮助同事看后续走势，不能反推当时是否入池、买入或卖出。", "原决策仍以已审核 decision_view、订单账本和 scope 表为准。", ] draw_text_box(d, (note_left, note_top, 1640, 850), "图上说明", lines) footer = f"来源：{SOURCE_RUN_ID}；本图为增强 audit_view，不新增交易结论。" d.text((32, 916), footer, fill="#334155", font=FONT_MID) out_path.parent.mkdir(parents=True, exist_ok=True) img.save(out_path) def find_event_index(df: pd.DataFrame, time_str: str) -> int: if df.empty: return 0 hits = df.index[df["trade_time"] == time_str].tolist() if hits: return hits[0] before = df.index[df["trade_time"] <= time_str].tolist() if before: return before[-1] return 0 def draw_minute_line_chart(df: pd.DataFrame, order: pd.Series, action: str, out_path: Path) -> None: w, h = 1540, 860 img = Image.new("RGB", (w, h), "#fbfbf7") d = ImageDraw.Draw(img) d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1") action_cn = "买入" if action == "BUY" else "卖出" point_cn = "买点" if action == "BUY" else "卖点" color = "#b91c1c" if action == "BUY" else "#7c3aed" title = f"{action_cn}日整日分钟行情折线图：{order['symbol']} {order['trade_date']}" d.text((32, 24), title, fill="#111827", font=FONT_TITLE) d.text((32, 60), f"标记：{point_cn} {order['trade_time']} @ {float(order['price']):.2f}；整日走势只作人工审阅，不改写原裁决。", fill=color, font=FONT_MID) plot_left, plot_top, plot_right, plot_bottom = 80, 110, 1080, 590 vol_top, vol_bottom = 645, 790 note_left, note_top = 1115, 112 d.rectangle([plot_left, plot_top, plot_right, plot_bottom], outline="#94a3b8") d.rectangle([plot_left, vol_top, plot_right, vol_bottom], outline="#94a3b8") day = df.copy().sort_values("trade_time").reset_index(drop=True) event_price = float(order["price"]) if day.empty: d.text((plot_left + 80, plot_top + 180), "分钟线数据缺失", fill="#b91c1c", font=FONT_TITLE) img.save(out_path) return price_low = min(float(day["low_price"].min()), event_price) * 0.998 price_high = max(float(day["high_price"].max()), event_price) * 1.002 max_vol = max(float(day["volume"].max()), 1.0) n = len(day) gap = (plot_right - plot_left) / max(n - 1, 1) points: list[tuple[int, int]] = [] for i, row in day.iterrows(): cx = int(plot_left + gap * i) cy = y_price(float(row["close_price"]), price_low, price_high, plot_top, plot_bottom) points.append((cx, cy)) vh = int(float(row["volume"]) / max_vol * (vol_bottom - vol_top)) d.line([cx, vol_bottom, cx, vol_bottom - vh], fill="#cbd5e1", width=1) if i % max(1, n // 7) == 0: d.text((cx - 20, vol_bottom + 8), str(row["trade_time"])[:5], fill="#64748b", font=FONT_SMALL) if len(points) > 1: d.line(points, fill="#0f766e", width=3) for i in range(5): price = price_low + (price_high - price_low) * i / 4 y = y_price(price, price_low, price_high, plot_top, plot_bottom) d.line([plot_left, y, plot_right, y], fill="#e2e8f0") d.text((18, y - 8), f"{price:.2f}", fill="#64748b", font=FONT_SMALL) event_idx = find_event_index(day, str(order["trade_time"])) event_x = int(plot_left + gap * event_idx) event_y = y_price(event_price, price_low, price_high, plot_top, plot_bottom) d.line([event_x, plot_top, event_x, vol_bottom], fill=color, width=3) d.ellipse([event_x - 8, event_y - 8, event_x + 8, event_y + 8], fill=color) d.text((min(event_x + 8, plot_right - 120), max(plot_top + 8, event_y - 42)), f"{point_cn} {order['trade_time'][:5]}", fill=color, font=FONT_SUB) open_price = float(day.iloc[0]["open_price"]) open_y = y_price(open_price, price_low, price_high, plot_top, plot_bottom) d.line([plot_left, open_y, plot_right, open_y], fill="#334155", width=1) d.text((plot_right - 95, open_y - 16), f"开盘 {open_price:.2f}", fill="#334155", font=FONT_SMALL) lines = [ f"动作：{action_cn}", f"订单：{order['order_id']}", f"时间：{order['trade_time']}", f"价格：{float(order['price']):.2f}", f"仓位变化：{float(order['position_delta_pct']):.2%}", f"证据来源：{order.get('evidence_image_path', '')}", "", "本图新增整日折线视角，目的是让同事看清买卖点处于全天什么位置。", "原始买卖裁决仍以来源包的1分钟K图、订单账本和已审核scope为准。", "本图不新增收益率或策略有效性结论。", ] draw_text_box(d, (note_left, note_top, 1500, 790), f"{action_cn}说明", lines) d.text((32, 820), f"来源：{SOURCE_RUN_ID}；增强包：{RUN_ID}", fill="#334155", font=FONT_MID) out_path.parent.mkdir(parents=True, exist_ok=True) img.save(out_path) def local_link_targets(markdown_path: Path) -> list[dict]: text = markdown_path.read_text(encoding="utf-8") targets = [] for match in re.finditer(r"!?\[[^\]]*\]\(([^)]+)\)", text): raw = match.group(1).strip() if not raw or raw.startswith("#") or "://" in raw: continue no_anchor = raw.split("#", 1)[0] if not no_anchor: continue target = (markdown_path.parent / no_anchor).resolve() targets.append( { "markdown_path": markdown_path.relative_to(PACKAGE_ROOT).as_posix(), "link": raw, "target_exists": target.exists(), "target_path": str(target), } ) return targets def write_manifest() -> None: files = [] for path in sorted(PACKAGE_ROOT.rglob("*")): if not path.is_file(): continue if path.name == "manifest.json": continue rel = path.relative_to(PACKAGE_ROOT).as_posix() files.append( { "path": rel, "size": path.stat().st_size, "sha256": sha256_file(path), } ) manifest = { "run_id": RUN_ID, "task_id": TASK_ID, "generated_at": GENERATED_AT, "source_run_id": SOURCE_RUN_ID, "source_guide_run_id": GUIDE_RUN_ID, "manifest_self_hash_excluded": True, "file_count": len(files), "files": files, } (PACKAGE_ROOT / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def write_source_manifest() -> None: source_files = [ SOURCE_ROOT / "full_return_stat_case_scope.csv", SOURCE_ROOT / "full_selected_candidate_ledger.csv", SOURCE_ROOT / "order_ledger.csv", SOURCE_ROOT / "position_lot_ledger.csv", SOURCE_ROOT / "case_summary.csv", SOURCE_ROOT / "full_return_stat_summary.json", SOURCE_ROOT / "manifest.json", GUIDE_ROOT / "strict_closed_234_case_index.csv", GUIDE_ROOT / "strict_closed_234_file_map.csv", GUIDE_ROOT / "manifest.json", ] rows = [] for path in source_files: rows.append( { "source_path": path.relative_to(PROJECT_ROOT).as_posix(), "exists": path.exists(), "size": path.stat().st_size if path.exists() else "", "sha256": sha256_file(path) if path.exists() else "", } ) pd.DataFrame(rows).to_csv(PACKAGE_ROOT / "source_artifact_manifest.csv", index=False, encoding="utf-8-sig") def build_case_board( case_id: str, case_scope: pd.Series, case_summary: pd.Series | None, candidate_rows: pd.DataFrame, buy_rows: pd.DataFrame, sell_rows: pd.DataFrame, chart_rows: list[dict], ) -> None: case_dir = PACKAGE_ROOT / "cases" / case_id source_case_dir = SOURCE_ROOT / "cases" / case_id lines = [ f"# {case_id} 增强图片审核板", "", f"- 增强包：`{RUN_ID}`", f"- 来源全量包：`{SOURCE_RUN_ID}`", f"- 来源原图片板：[打开原 case_image_board.md](../../../{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md)", f"- 当前收益口径：`PRIMARY_STRICT_CLOSED_CASE`", f"- RETURN_STAT_READY：`false`", "", "本板是给同事看的图片增强入口，不重跑候选池、买卖裁决或账本，不新增收益结论。", "选股日K图包含信号日后10个交易日的事后审阅区；该区域只能帮助复盘，不能反推当时选股或买卖决策。", "", "## 1. 选股日K图：信号日前90天 + 后10天", "", ] for _, cand in candidate_rows.sort_values("candidate_rank").iterrows(): rows = [r for r in chart_rows if r["case_id"] == case_id and r["chart_role"] == "candidate_daily_90pre_10post_audit_view" and r["symbol"] == cand["symbol"]] if rows: rel = Path(rows[0]["path"]).relative_to(f"cases/{case_id}").as_posix() lines.extend( [ f"### 候选排名 {int(cand['candidate_rank'])}：{cand['symbol']}", "", f"![{cand['symbol']}]({rel})", "", f"- 信号日：`{cand['signal_trade_date']}`；买入日：`{cand['entry_trade_date']}`", f"- 候选状态：`{cand['candidate_status']}`；量比 `{float(cand['volume_ratio']):.2f}`；长上影 `{float(cand['upper_shadow_pct']):.2f}%`", "", ] ) lines.extend(["## 2. 买入日整日分钟行情折线图", ""]) if buy_rows.empty: lines.append("- 本案例无 BUY 订单。") for _, order in buy_rows.sort_values(["trade_date", "trade_time", "symbol"]).iterrows(): rows = [r for r in chart_rows if r["event_id"] == order["order_id"] and r["chart_role"] == "entry_full_day_minute_line_audit_view"] if rows: rel = Path(rows[0]["path"]).relative_to(f"cases/{case_id}").as_posix() lines.extend( [ f"### 买点：{order['symbol']} {order['trade_date']} {order['trade_time']}", "", f"![买点 {order['symbol']}]({rel})", "", f"- 买入价格：`{float(order['price']):.4f}`；仓位变化：`{float(order['position_delta_pct']):.2%}`", f"- 来源订单：`{order['order_id']}`", "", ] ) lines.extend(["## 3. 卖出日整日分钟行情折线图", ""]) if sell_rows.empty: lines.append("- 本案例无 SELL 订单。") for _, order in sell_rows.sort_values(["trade_date", "trade_time", "symbol"]).iterrows(): rows = [r for r in chart_rows if r["event_id"] == order["order_id"] and r["chart_role"] == "exit_full_day_minute_line_audit_view"] if rows: rel = Path(rows[0]["path"]).relative_to(f"cases/{case_id}").as_posix() lines.extend( [ f"### 卖点：{order['symbol']} {order['trade_date']} {order['trade_time']}", "", f"![卖点 {order['symbol']}]({rel})", "", f"- 卖出价格：`{float(order['price']):.4f}`；仓位变化：`{float(order['position_delta_pct']):.2%}`", f"- 来源订单：`{order['order_id']}`；来源 lot：`{order.get('source_lot_id', '')}`", "", ] ) ret = "" if case_summary is None else money_text(case_summary.get("account_return_closed_lots", "")) success = "" if case_scope is None else case_scope.get("primary_case_success_flag", "") lines.extend( [ "## 4. 账本和收益口径追溯", "", f"- 主口径成功标记：`{success}`", f"- 闭合 lot 账户贡献合计：`{ret}`", f"- 来源 case scope：`{SOURCE_RUN_ID}/full_return_stat_case_scope.csv`", f"- 来源订单账本：`{SOURCE_RUN_ID}/order_ledger.csv`", f"- 来源 lot 账本：`{SOURCE_RUN_ID}/position_lot_ledger.csv`", f"- 来源账户流水：`{SOURCE_RUN_ID}/daily_account_ledger.csv`", "", "禁止过读：本 case 属于 234 个严格闭合主口径子集；不得把该子集直接说成完整 743 个 entry date 的无边界表现。", ] ) (case_dir / "case_image_board.md").write_text("\n".join(lines) + "\n", encoding="utf-8") def build_root_docs(case_index: pd.DataFrame, chart_rows: list[dict]) -> None: readme = [ "# 无忌交易系统 234 个严格闭合案例增强图片包", "", f"- 生成时间：`{GENERATED_AT}`", f"- 增强包：`{RUN_ID}`", f"- 来源全量包：`{SOURCE_RUN_ID}`", f"- 来源说明包：`{GUIDE_RUN_ID}`", "", "## 这个包解决什么问题", "", "同事反馈原 `case_image_board.md` 还需要三类更方便的图：", "", "1. 选股日K图按信号日前90个交易日 + 信号日后10个交易日展示。", "2. 增加买入日整日分钟行情折线图，并清楚标记买点。", "3. 增加卖出日整日分钟行情折线图，并清楚标记卖点。", "", "本包只增强图片阅读入口，不重跑候选池、买卖裁决或账本，不新增收益结论。", "信号日后的10个交易日和整日分钟折线图均属于 `audit_view`，用于人工复核，不得反推当时决策。", "", "## 怎么看", "", "1. 先打开本目录 `case_image_board.md`，从234个 case 列表进入单个 case。", "2. 单个 case 先看增强选股日K图，再看买入日整日分钟折线图，最后看卖出日整日分钟折线图。", "3. 需要核对数字时回到来源全量包的 `order_ledger.csv`、`position_lot_ledger.csv`、`daily_account_ledger.csv` 和 `full_return_stat_case_scope.csv`。", "", "## 关键边界", "", "- 当前范围仍是 `PRIMARY_STRICT_CLOSED_CASE` 的 234 个严格闭合 case。", "- `RETURN_STAT_READY=false` 保留。", "- 不得把 234 个 case 写成完整 743 个 entry date 的无边界收益或成功率。", "- 不得声称无忌 baseline 策略有效性已经被证明。", ] (PACKAGE_ROOT / "README.md").write_text("\n".join(readme) + "\n", encoding="utf-8") lines = [ "# 234 个严格闭合案例增强图片总入口", "", f"- 增强包：`{RUN_ID}`", f"- 来源全量包：`{SOURCE_RUN_ID}`", f"- 增强图数量：`{len(chart_rows)}`", "", "## 入口列表", "", "| case_id | 入场日 | 主口径收益贡献 | 买入图 | 卖出图 | 图片板 |", "|---|---:|---:|---:|---:|---|", ] for _, row in case_index.sort_values("entry_trade_date").iterrows(): lines.append( f"| `{row['case_id']}` | {row['entry_trade_date']} | {money_text(row['account_return_closed_lots'])} | {int(row['buy_chart_count'])} | {int(row['sell_chart_count'])} | [打开](cases/{row['case_id']}/case_image_board.md) |" ) (PACKAGE_ROOT / "case_image_board.md").write_text("\n".join(lines) + "\n", encoding="utf-8") def main() -> None: PACKAGE_ROOT.mkdir(parents=True, exist_ok=True) (PACKAGE_ROOT / "cases").mkdir(parents=True, exist_ok=True) write_source_manifest() scope = read_csv("full_return_stat_case_scope.csv") strict = scope[ (scope["case_scope_status"] == "PRIMARY_STRICT_CLOSED_CASE") & (to_bool_int(scope["primary_strict_closed_case_flag"]) == 1) ].copy() strict_ids = set(strict["case_id"]) selected = read_csv("full_selected_candidate_ledger.csv") selected = selected[selected["case_id"].isin(strict_ids)].copy() selected["signal_trade_date"] = pd.to_datetime(selected["signal_trade_date"]).dt.strftime("%Y-%m-%d") selected["entry_trade_date"] = pd.to_datetime(selected["entry_trade_date"]).dt.strftime("%Y-%m-%d") selected["candidate_rank"] = pd.to_numeric(selected["candidate_rank"], errors="coerce") orders = read_csv("order_ledger.csv") orders = orders[orders["case_id"].isin(strict_ids) & orders["action"].isin(["BUY", "SELL"])].copy() orders["trade_date"] = pd.to_datetime(orders["trade_date"]).dt.strftime("%Y-%m-%d") orders["trade_time"] = orders["trade_time"].map(norm_time) orders["price"] = pd.to_numeric(orders["price"], errors="coerce") orders["position_delta_pct"] = pd.to_numeric(orders["position_delta_pct"], errors="coerce") case_summary = read_csv("case_summary.csv") case_summary = case_summary[case_summary["case_id"].isin(strict_ids)].copy() print(f"strict cases={len(strict)}, selected={len(selected)}, orders={len(orders)}", flush=True) daily = fetch_daily(selected) minute = fetch_minute(orders) print(f"daily rows={len(daily)}, minute rows={len(minute)}", flush=True) chart_rows: list[dict] = [] missing_rows: list[dict] = [] case_index_rows: list[dict] = [] for seq, (_, case_scope) in enumerate(strict.sort_values("entry_trade_date").iterrows(), start=1): case_id = case_scope["case_id"] case_dir = PACKAGE_ROOT / "cases" / case_id img_dir = case_dir / "img" img_dir.mkdir(parents=True, exist_ok=True) candidate_rows = selected[selected["case_id"] == case_id].copy() buy_rows = orders[(orders["case_id"] == case_id) & (orders["action"] == "BUY")].copy() sell_rows = orders[(orders["case_id"] == case_id) & (orders["action"] == "SELL")].copy() summary_rows = case_summary[case_summary["case_id"] == case_id] summary_row = summary_rows.iloc[0] if not summary_rows.empty else None for _, cand in candidate_rows.sort_values("candidate_rank").iterrows(): window = daily_window(daily, str(cand["symbol"]), str(cand["signal_trade_date"])) out = img_dir / f"01_candidate_daily_90pre_10post_rank{int(cand['candidate_rank']):02d}_{str(cand['symbol']).replace('.', '_')}_{str(cand['signal_trade_date']).replace('-', '')}.png" if window.empty: missing_rows.append({"case_id": case_id, "symbol": cand["symbol"], "kind": "daily", "date": cand["signal_trade_date"]}) continue draw_daily_enhanced_chart(window, cand, out) chart_rows.append( { "case_id": case_id, "symbol": cand["symbol"], "event_id": cand["candidate_id"], "chart_role": "candidate_daily_90pre_10post_audit_view", "action": "CANDIDATE", "trade_date": cand["signal_trade_date"], "trade_time": "close", "path": out.relative_to(PACKAGE_ROOT).as_posix(), "source_order_id": "", "source_candidate_id": cand["candidate_id"], "status": "PASS", "note": "信号日前90个交易日 + 信号日后10个交易日；信号后区间仅作audit_view。", "size": out.stat().st_size, "sha256": sha256_file(out), } ) for _, order in pd.concat([buy_rows, sell_rows], ignore_index=True).sort_values(["trade_date", "trade_time", "symbol", "order_id"]).iterrows(): action = str(order["action"]) role = "entry_full_day_minute_line_audit_view" if action == "BUY" else "exit_full_day_minute_line_audit_view" prefix = "07_entry_full_day_minute_line" if action == "BUY" else "08_exit_full_day_minute_line" day = minute[ (minute["symbol"] == order["symbol"]) & (minute["trade_date"] == order["trade_date"]) ].copy() safe_order = re.sub(r"[^A-Za-z0-9]+", "_", str(order["order_id"]))[-36:] out = img_dir / f"{prefix}_{str(order['symbol']).replace('.', '_')}_{str(order['trade_date']).replace('-', '')}_{str(order['trade_time']).replace(':', '')}_{safe_order}.png" if day.empty: missing_rows.append({"case_id": case_id, "symbol": order["symbol"], "kind": action.lower(), "date": order["trade_date"]}) continue draw_minute_line_chart(day, order, action, out) chart_rows.append( { "case_id": case_id, "symbol": order["symbol"], "event_id": order["order_id"], "chart_role": role, "action": action, "trade_date": order["trade_date"], "trade_time": order["trade_time"], "path": out.relative_to(PACKAGE_ROOT).as_posix(), "source_order_id": order["order_id"], "source_candidate_id": order.get("candidate_id", ""), "status": "PASS", "note": "整日分钟行情折线图；仅作audit_view，标记来源订单买卖点。", "size": out.stat().st_size, "sha256": sha256_file(out), } ) build_case_board(case_id, case_scope, summary_row, candidate_rows, buy_rows, sell_rows, chart_rows) case_chart_rows = [r for r in chart_rows if r["case_id"] == case_id] (case_dir / "image_manifest.csv").write_text( pd.DataFrame(case_chart_rows).to_csv(index=False), encoding="utf-8-sig", ) case_index_rows.append( { "case_id": case_id, "entry_trade_date": case_scope["entry_trade_date"], "signal_trade_date": case_scope["signal_trade_date"], "account_return_closed_lots": case_scope["account_return_closed_lots"], "primary_case_success_flag": case_scope["primary_case_success_flag"], "candidate_chart_count": len([r for r in case_chart_rows if r["chart_role"] == "candidate_daily_90pre_10post_audit_view"]), "buy_chart_count": len([r for r in case_chart_rows if r["chart_role"] == "entry_full_day_minute_line_audit_view"]), "sell_chart_count": len([r for r in case_chart_rows if r["chart_role"] == "exit_full_day_minute_line_audit_view"]), "enhanced_case_image_board": f"cases/{case_id}/case_image_board.md", "source_case_image_board": f"{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md", } ) if seq % 25 == 0: print(f"case progress: {seq}/{len(strict)}", flush=True) chart_df = pd.DataFrame(chart_rows) chart_df.to_csv(PACKAGE_ROOT / "chart_evidence_audit.csv", index=False, encoding="utf-8-sig") missing_df = pd.DataFrame(missing_rows) missing_df.to_csv(PACKAGE_ROOT / "missing_chart_inputs.csv", index=False, encoding="utf-8-sig") case_index = pd.DataFrame(case_index_rows).sort_values("entry_trade_date") case_index.to_csv(PACKAGE_ROOT / "enhanced_case_index.csv", index=False, encoding="utf-8-sig") build_root_docs(case_index, chart_rows) link_rows = [] for md in sorted(PACKAGE_ROOT.rglob("*.md")): link_rows.extend(local_link_targets(md)) link_df = pd.DataFrame(link_rows) link_df.to_csv(PACKAGE_ROOT / "link_evidence_audit.csv", index=False, encoding="utf-8-sig") expected_daily = len(strict) * 5 expected_buy = len(orders[orders["action"] == "BUY"]) expected_sell = len(orders[orders["action"] == "SELL"]) self_checks = [ ("SOURCE_FULL_RUN_EXISTS", SOURCE_ROOT.exists(), str(SOURCE_ROOT)), ("SOURCE_REVIEW_GUIDE_EXISTS", GUIDE_ROOT.exists(), str(GUIDE_ROOT)), ("STRICT_CASE_COUNT_234", len(strict) == 234, f"strict_cases={len(strict)}"), ("SELECTED_CANDIDATE_ROWS_1170", len(selected) == expected_daily, f"selected={len(selected)}, expected={expected_daily}"), ("DAILY_AUDIT_CHARTS_COMPLETE", len(chart_df[chart_df["chart_role"] == "candidate_daily_90pre_10post_audit_view"]) == expected_daily, f"actual={len(chart_df[chart_df['chart_role'] == 'candidate_daily_90pre_10post_audit_view'])}, expected={expected_daily}"), ("BUY_FULL_DAY_LINE_CHARTS_COMPLETE", len(chart_df[chart_df["chart_role"] == "entry_full_day_minute_line_audit_view"]) == expected_buy, f"actual={len(chart_df[chart_df['chart_role'] == 'entry_full_day_minute_line_audit_view'])}, expected={expected_buy}"), ("SELL_FULL_DAY_LINE_CHARTS_COMPLETE", len(chart_df[chart_df["chart_role"] == "exit_full_day_minute_line_audit_view"]) == expected_sell, f"actual={len(chart_df[chart_df['chart_role'] == 'exit_full_day_minute_line_audit_view'])}, expected={expected_sell}"), ("CASE_IMAGE_BOARDS_234", len(list((PACKAGE_ROOT / "cases").glob("*/case_image_board.md"))) == 234, f"boards={len(list((PACKAGE_ROOT / 'cases').glob('*/case_image_board.md')))}"), ("MISSING_CHART_INPUTS_ZERO", missing_df.empty, f"missing={len(missing_df)}"), ("MARKDOWN_LOCAL_LINKS_REACHABLE", bool(link_df.empty or link_df["target_exists"].all()), f"links={len(link_df)}, missing={0 if link_df.empty else int((~link_df['target_exists']).sum())}"), ("RETURN_STAT_READY_FALSE_PRESERVED", True, "enhancement package does not change source return_stat_ready=false"), ] self_df = pd.DataFrame( [ { "check_id": check_id, "status": "PASS" if passed else "FAIL", "detail": detail, } for check_id, passed, detail in self_checks ] ) self_df.to_csv(PACKAGE_ROOT / "self_check_items.csv", index=False, encoding="utf-8-sig") fail_count = int((self_df["status"] != "PASS").sum()) self_json = { "run_id": RUN_ID, "task_id": TASK_ID, "generated_at": GENERATED_AT, "source_run_id": SOURCE_RUN_ID, "status": "PASS_FOR_STRICT_CLOSED_234_CHART_ENHANCEMENT_REVIEW_READY" if fail_count == 0 else "FAIL", "pass_count": int((self_df["status"] == "PASS").sum()), "fail_count": fail_count, "strict_case_count": len(strict), "daily_audit_chart_count": int((chart_df["chart_role"] == "candidate_daily_90pre_10post_audit_view").sum()), "buy_full_day_line_chart_count": int((chart_df["chart_role"] == "entry_full_day_minute_line_audit_view").sum()), "sell_full_day_line_chart_count": int((chart_df["chart_role"] == "exit_full_day_minute_line_audit_view").sum()), "return_stat_ready": False, "full_baseline_conclusion_allowed": False, } (PACKAGE_ROOT / "self_check.json").write_text(json.dumps(self_json, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") (PACKAGE_ROOT / "self_check.md").write_text( "\n".join( [ "# 自检结果", "", f"- 状态：`{self_json['status']}`", f"- PASS：`{self_json['pass_count']}`", f"- FAIL：`{self_json['fail_count']}`", f"- 选股增强日K图：`{self_json['daily_audit_chart_count']}`", f"- 买入日整日分钟折线图：`{self_json['buy_full_day_line_chart_count']}`", f"- 卖出日整日分钟折线图：`{self_json['sell_full_day_line_chart_count']}`", "", "本包只增强人工图片入口，不改变来源全量包的收益口径和 `RETURN_STAT_READY=false`。", ] ) + "\n", encoding="utf-8", ) write_manifest() print(json.dumps(self_json, ensure_ascii=False, indent=2), flush=True) if __name__ == "__main__": main()