from __future__ import annotations import hashlib import json import math import os import re from collections import defaultdict 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-V1-LIFECYCLE-CHART-SUPPLEMENT-20260611-001" TASK_ID = "ANA-WUJI-V1-LIFECYCLE-CHART-SUPPLEMENT-20260611" FINAL_RUN_ID = "RUN-ANA-WUJI-V1-FINAL-CONCLUSION-20260610-001" SOURCE_RUN_ID = "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001" SOURCE_EXEC_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-EXEC-REREVIEW-003" FINAL_EXEC_AUDIT_ID = "AUDIT-ANA-WUJI-V1-FINAL-CONCLUSION-20260610-EXEC-001" PACKAGE_ROOT = Path(__file__).resolve().parents[1] RESULT_ROOT = Path(__file__).resolve().parents[2] PROJECT_ROOT = Path(__file__).resolve().parents[4] FINAL_ROOT = RESULT_ROOT / FINAL_RUN_ID SOURCE_ROOT = RESULT_ROOT / SOURCE_RUN_ID LOCAL_DB_INDEX = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md") FILE_MIRROR_CANDIDATES = [ Path(r"E:\strategy_project\s-system-doc\ali\data\stock-data\total-data"), Path(r"E:\strategy_project\s-system-doc\observer\model\data\stock-data\total-data"), Path(r"E:\strategy_project\s-system-doc\observer\天下模型沉淀\data\stock-data\total-data"), Path(r"E:\策略项目\s-system-doc\ali\data\stock-data\total-data"), Path(r"E:\策略项目\s-system-doc\observer\model\data\stock-data\total-data"), Path(r"E:\策略项目\s-system-doc\observer\天下模型沉淀\data\stock-data\total-data"), ] 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=240, write_timeout=120, ) def file_mirror_root() -> Path | None: for root in FILE_MIRROR_CANDIDATES: if root.exists(): return root return None def symbol_code(symbol: str) -> str: return str(symbol).split(".")[0] def symbol_exchange(symbol: str) -> str: text = str(symbol) if "." in text: return text.split(".")[-1].upper() if text.startswith("6"): return "SH" if text.startswith("8") or text.startswith("9"): return "BJ" return "SZ" 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_json(path: Path, data: dict) -> None: path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def font(size: int): for name in ["msyh.ttc", "simhei.ttf", "simsun.ttc"]: path = Path("C:/Windows/Fonts") / name if path.exists(): return ImageFont.truetype(str(path), size) return ImageFont.load_default() FONT_TITLE = font(30) FONT_SUB = font(22) FONT_MID = font(18) FONT_SMALL = font(14) FONT_TINY = font(12) def read_csv(path: Path) -> pd.DataFrame: return pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig").fillna("") def norm_date(value) -> str: if pd.isna(value) or str(value).strip() == "": return "" return pd.to_datetime(value).strftime("%Y-%m-%d") 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 not text: return "" if "days" in text and " " in text: text = text.split()[-1] if " " in text: text = text.split()[-1] if "." in text: text = text.split(".")[0] parts = text.split(":") if len(parts) == 2: return f"{int(parts[0]):02d}:{int(parts[1]):02d}:00" if len(parts) >= 3: return f"{int(parts[0]):02d}:{int(parts[1]):02d}:{int(float(parts[2])):02d}" return text def safe_float(value, default=math.nan) -> float: try: if pd.isna(value) or str(value).strip() == "": return default return float(value) except Exception: return default def money_text(value) -> str: try: return f"{float(value):.8f}" except Exception: return "" def pct_text(value) -> str: try: return f"{float(value):.2%}" except Exception: return "" 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]: lines: list[str] = [] current = "" for ch in str(text): current += ch if len(current) >= max_chars: lines.append(current) current = "" if current: lines.append(current) return lines or [""] def draw_text_box( draw: ImageDraw.ImageDraw, box: tuple[int, int, int, int], title: str, lines: Iterable[str], max_chars: int = 28, ) -> 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_SUB) y = y1 + 54 for line in lines: if y > y2 - 24: return if not line: y += 10 continue for wrapped in wrap_text(line, max_chars): if y > y2 - 24: return draw.text((x1 + 18, y), wrapped, fill="#334155", font=FONT_SMALL) y += 23 def fetch_trade_calendar() -> list[str]: try: with get_conn() as conn: rows = pd.read_sql( """ SELECT DISTINCT trade_date FROM a_share_daily_price WHERE trade_date BETWEEN '2022-12-01' AND '2026-12-31' ORDER BY trade_date """, conn, ) return [norm_date(v) for v in rows["trade_date"].tolist()] except Exception as exc: print(f"MySQL calendar unavailable, fallback to file mirror: {exc}", flush=True) mirror = file_mirror_root() if mirror is None: raise RuntimeError("No MySQL connection and no stock-data total-data mirror found.") calendar_files = sorted((mirror / "calendar").rglob("a_share_trading_calendar_*.csv")) if not calendar_files: raise RuntimeError(f"No trading calendar CSV under {mirror / 'calendar'}") cal = pd.concat( [pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig") for path in calendar_files], ignore_index=True, ) cal = cal[cal["is_trading_day"].astype(str) == "1"].copy() return sorted(cal["calendar_date"].map(norm_date).dropna().unique().tolist()) def mirror_price_files(kind: str, symbol: str) -> list[Path]: mirror = file_mirror_root() if mirror is None: return [] code = symbol_code(symbol) exchange = symbol_exchange(symbol) files = [] for path in (mirror / kind).rglob(f"price_{code}.csv"): if path.parent.name.upper() == exchange: files.append(path) return sorted(files, key=lambda p: p.as_posix()) def fetch_daily_from_file_mirror(symbols: list[str], min_date: str, max_date: str) -> pd.DataFrame: parts: list[pd.DataFrame] = [] min_ts = pd.Timestamp(min_date) max_ts = pd.Timestamp(max_date) for idx, symbol in enumerate(symbols, start=1): symbol_parts: list[pd.DataFrame] = [] for path in mirror_price_files("daily", symbol): try: df = pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig") except pd.errors.EmptyDataError: continue if df.empty or "timetag" not in df.columns: continue parsed = pd.to_datetime(df["timetag"], format="%Y%m%d", errors="coerce") df = df.assign(trade_date=parsed) df = df[(df["trade_date"] >= min_ts) & (df["trade_date"] <= max_ts)].copy() if df.empty: continue df["symbol"] = symbol df = df.rename( columns={ "open": "open_price", "high": "high_price", "low": "low_price", "close": "close_price", "volumn": "volume", } ) symbol_parts.append(df[["trade_date", "symbol", "open_price", "high_price", "low_price", "close_price", "volume", "amount"]]) if symbol_parts: parts.append(pd.concat(symbol_parts, ignore_index=True)) if idx % 120 == 0: print(f"daily file mirror progress: {idx}/{len(symbols)} symbols", flush=True) if not parts: return pd.DataFrame() daily = pd.concat(parts, ignore_index=True) daily = daily.drop_duplicates(["symbol", "trade_date"], keep="last") return daily def fetch_minute_from_file_mirror(date_symbols: dict[str, set[str]]) -> pd.DataFrame: wanted_by_symbol: dict[str, set[str]] = defaultdict(set) for trade_date, symbols in date_symbols.items(): for symbol in symbols: wanted_by_symbol[symbol].add(trade_date) parts: list[pd.DataFrame] = [] for idx, symbol in enumerate(sorted(wanted_by_symbol), start=1): wanted_dates = wanted_by_symbol[symbol] symbol_parts: list[pd.DataFrame] = [] for path in mirror_price_files("minute", symbol): try: df = pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig") except pd.errors.EmptyDataError: continue if df.empty or "timetag" not in df.columns: continue timetag = df["timetag"].astype(str) date_part = timetag.str.slice(0, 8) time_part = timetag.str.slice(9) df = df.assign( trade_date=pd.to_datetime(date_part, format="%Y%m%d", errors="coerce").dt.strftime("%Y-%m-%d"), trade_time=time_part.map(norm_time), ) df = df[df["trade_date"].isin(wanted_dates)].copy() if df.empty: continue df["symbol"] = symbol df = df.rename( columns={ "open": "open_price", "high": "high_price", "low": "low_price", "close": "close_price", "volumn": "volume", } ) symbol_parts.append(df[["trade_date", "trade_time", "symbol", "open_price", "high_price", "low_price", "close_price", "volume", "amount"]]) if symbol_parts: parts.append(pd.concat(symbol_parts, ignore_index=True)) if idx % 80 == 0: print(f"minute file mirror progress: {idx}/{len(wanted_by_symbol)} symbols", flush=True) if not parts: return pd.DataFrame() minute = pd.concat(parts, ignore_index=True) minute = minute.drop_duplicates(["symbol", "trade_date", "trade_time"], keep="last") return minute def fetch_daily(symbols: list[str], min_date: str, max_date: str) -> pd.DataFrame: try: 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() except Exception as exc: print(f"MySQL daily unavailable, fallback to file mirror: {exc}", flush=True) daily = fetch_daily_from_file_mirror(symbols, min_date, max_date) if daily.empty: return daily daily["trade_date"] = pd.to_datetime(daily["trade_date"]) daily["trade_date_str"] = daily["trade_date"].dt.strftime("%Y-%m-%d") 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["ma10"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(10, 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(date_symbols: dict[str, set[str]]) -> pd.DataFrame: try: parts: list[pd.DataFrame] = [] with get_conn() as conn: for idx, trade_date in enumerate(sorted(date_symbols), start=1): symbols = sorted(date_symbols[trade_date]) if not symbols: continue 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 % 80 == 0: print(f"minute query progress: {idx}/{len(date_symbols)} dates", flush=True) minute = pd.concat(parts, ignore_index=True) if parts else pd.DataFrame() except Exception as exc: print(f"MySQL minute unavailable, fallback to file mirror: {exc}", flush=True) minute = fetch_minute_from_file_mirror(date_symbols) 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.sort_values(["symbol", "trade_date", "trade_time"]).reset_index(drop=True) def trade_window_from_calendar(trade_dates: list[str], first_buy: str, last_sell: str) -> tuple[str, str]: if first_buy not in trade_dates: start = first_buy else: pos = trade_dates.index(first_buy) start = trade_dates[max(0, pos - 50)] if last_sell not in trade_dates: end = last_sell else: pos = trade_dates.index(last_sell) end = trade_dates[min(len(trade_dates) - 1, pos + 20)] return start, end def find_event_index(day: pd.DataFrame, time_str: str) -> int: if day.empty: return 0 exact = day.index[day["trade_time"] == time_str].tolist() if exact: return exact[0] before = day.index[day["trade_time"] <= time_str].tolist() if before: return before[-1] return 0 def order_label(order: pd.Series) -> str: action = "买入" if order["action"] == "BUY" else "卖出" return f"{action} {order['trade_date']} {str(order['trade_time'])[:5]} @ {safe_float(order['price']):.2f}" def draw_lifecycle_daily_chart( window: pd.DataFrame, orders: pd.DataFrame, meta: pd.Series, symbol: str, out_path: Path, ) -> None: w, h = 1820, 1060 img = Image.new("RGB", (w, h), "#fbfbf7") d = ImageDraw.Draw(img) d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1") title = f"股票生命周期日线图：{meta['case_id']} / {symbol}" subtitle = "窗口：首个 BUY 前50个交易日 -> 最后一个 SELL 后20个交易日；本图只作 audit_view。" d.text((32, 24), title, fill="#111827", font=FONT_TITLE) d.text((32, 66), subtitle, fill="#7f1d1d", font=FONT_MID) plot_left, plot_top, plot_right, plot_bottom = 86, 128, 1330, 675 vol_top, vol_bottom = 735, 890 note_left, note_top = 1370, 128 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().sort_values("trade_date").reset_index(drop=True) if win.empty: d.text((plot_left + 120, plot_top + 200), "日线数据缺失", fill="#b91c1c", font=FONT_TITLE) out_path.parent.mkdir(parents=True, exist_ok=True) img.save(out_path) return prices = [safe_float(v) for v in orders["price"].tolist()] price_low = min(float(win["low_price"].min()), *(p for p in prices if math.isfinite(p))) * 0.985 price_high = max(float(win["high_price"].max()), *(p for p in prices if math.isfinite(p))) * 1.015 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)) date_to_x: dict[str, int] = {} 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: dict[str, list[tuple[int, int]]] = {"ma5": [], "ma10": [], "ma20": [], "ma60": []} ma_colors = {"ma5": "#2563eb", "ma10": "#0891b2", "ma20": "#f59e0b", "ma60": "#7c3aed"} for i, row in win.iterrows(): cx = int(plot_left + gap * i + gap / 2) date_to_x[str(row["trade_date_str"])] = cx 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 = 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 i % max(1, n // 10) == 0: d.text((cx - 24, vol_bottom + 10), str(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 + 14), ma.upper(), fill=color, font=FONT_SMALL) legend_x += 78 for _, order in orders.sort_values(["trade_date", "trade_time", "order_id"]).iterrows(): x = date_to_x.get(str(order["trade_date"])) if x is None: continue price = safe_float(order["price"]) y = y_price(price, price_low, price_high, plot_top, plot_bottom) is_buy = order["action"] == "BUY" color = "#b91c1c" if is_buy else "#7c3aed" label = "B" if is_buy else "S" d.line([x, plot_top, x, vol_bottom], fill=color, width=2) d.ellipse([x - 9, y - 9, x + 9, y + 9], fill=color) d.text((x - 6, y - 31), label, fill=color, font=FONT_MID) buy_rows = orders[orders["action"] == "BUY"].sort_values(["trade_date", "trade_time"]) sell_rows = orders[orders["action"] == "SELL"].sort_values(["trade_date", "trade_time"]) first_buy = buy_rows.iloc[0] if not buy_rows.empty else None last_sell = sell_rows.iloc[-1] if not sell_rows.empty else (orders.sort_values(["trade_date", "trade_time"]).iloc[-1] if not orders.empty else None) lines = [ f"V1 scope：{meta.get('v1_return_scope', '')}", f"成功标记：{meta.get('success_flag', '')}", f"账户贡献：{money_text(meta.get('account_return_closed_lots', ''))}", f"BUY 数：{len(buy_rows)}；SELL 数：{len(sell_rows)}", f"首 BUY：{order_label(first_buy) if first_buy is not None else ''}", f"末 SELL：{order_label(last_sell) if last_sell is not None and last_sell.get('action') == 'SELL' else ''}", "", "图例：红色 B 为 BUY，紫色 S 为 SELL。", "窗口按交易日截取：首 BUY 前 50 日，末 SELL 后 20 日。", "本图只串联生命周期，不新增买卖裁决或收益结论。", ] draw_text_box(d, (note_left, note_top, 1780, 890), "生命周期说明", lines, max_chars=24) footer = f"来源：{SOURCE_RUN_ID}；最终引用包：{FINAL_RUN_ID}；生成：{RUN_ID}" d.text((32, 994), footer, fill="#334155", font=FONT_MID) out_path.parent.mkdir(parents=True, exist_ok=True) img.save(out_path) def draw_transaction_day_minute_chart(day: pd.DataFrame, orders: pd.DataFrame, meta: pd.Series, symbol: str, trade_date: str, out_path: Path) -> None: w, h = 1660, 940 img = Image.new("RGB", (w, h), "#fbfbf7") d = ImageDraw.Draw(img) d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1") title = f"交易日整日分时图：{meta['case_id']} / {symbol} / {trade_date}" subtitle = "展示该股票在有订单产生交易日的整日分钟走势，标记当天全部 BUY / SELL。" d.text((32, 24), title, fill="#111827", font=FONT_TITLE) d.text((32, 66), subtitle, fill="#7f1d1d", font=FONT_MID) plot_left, plot_top, plot_right, plot_bottom = 82, 122, 1145, 640 vol_top, vol_bottom = 700, 850 note_left, note_top = 1185, 122 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 = day.copy().sort_values("trade_time").reset_index(drop=True) if day.empty: d.text((plot_left + 100, plot_top + 200), "分钟数据缺失", fill="#b91c1c", font=FONT_TITLE) out_path.parent.mkdir(parents=True, exist_ok=True) img.save(out_path) return event_prices = [safe_float(v) for v in orders["price"].tolist()] price_low = min(float(day["low_price"].min()), *(p for p in event_prices if math.isfinite(p))) * 0.998 price_high = max(float(day["high_price"].max()), *(p for p in event_prices if math.isfinite(p))) * 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 // 8) == 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((20, y - 8), f"{price:.2f}", fill="#64748b", font=FONT_SMALL) 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 - 96, open_y - 16), f"开盘 {open_price:.2f}", fill="#334155", font=FONT_SMALL) for _, order in orders.sort_values(["trade_time", "order_id"]).iterrows(): event_idx = find_event_index(day, str(order["trade_time"])) event_x = int(plot_left + gap * event_idx) event_y = y_price(safe_float(order["price"]), price_low, price_high, plot_top, plot_bottom) is_buy = order["action"] == "BUY" color = "#b91c1c" if is_buy else "#7c3aed" point_cn = "买入" if is_buy else "卖出" d.line([event_x, plot_top, event_x, vol_bottom], fill=color, width=3) d.ellipse([event_x - 9, event_y - 9, event_x + 9, event_y + 9], fill=color) d.text((min(event_x + 8, plot_right - 140), max(plot_top + 8, event_y - 40)), f"{point_cn} {str(order['trade_time'])[:5]}", fill=color, font=FONT_MID) lines = [ f"V1 scope：{meta.get('v1_return_scope', '')}", f"订单数：{len(orders)}", "", ] for _, order in orders.sort_values(["trade_time", "order_id"]).iterrows(): lines.append(order_label(order)) lines.extend( [ "", "本图只补充交易日分时视角。", "交易事实以 strict_order_ledger.csv 为准。", "不改变 V1 收益口径和 RETURN_STAT_READY=false。", ] ) draw_text_box(d, (note_left, note_top, 1620, 850), "当日交易", lines, max_chars=26) footer = f"来源：{SOURCE_RUN_ID}；生成：{RUN_ID}" d.text((32, 892), footer, 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") rows = [] 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() rows.append( { "markdown_path": markdown_path.relative_to(PACKAGE_ROOT).as_posix(), "link": raw, "target_exists": target.exists(), "target_path": str(target), } ) return rows def write_source_manifest() -> None: source_files = [ FINAL_ROOT / "v1_case_readout_index.csv", FINAL_ROOT / "v1_final_readouts.csv", FINAL_ROOT / "v1_boundary_table.csv", FINAL_ROOT / "manifest.json", SOURCE_ROOT / "strict_order_ledger.csv", SOURCE_ROOT / "strict_position_lot_ledger.csv", SOURCE_ROOT / "strict_case_summary.csv", SOURCE_ROOT / "strict_return_scope_case.csv", SOURCE_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 write_manifest() -> None: rows = [] for path in sorted(PACKAGE_ROOT.rglob("*")): if not path.is_file(): continue if path.name in {"manifest.csv", "manifest.json"}: continue rows.append( { "path": path.relative_to(PACKAGE_ROOT).as_posix(), "size": path.stat().st_size, "sha256": sha256_file(path), } ) manifest_df = pd.DataFrame(rows) manifest_df.to_csv(PACKAGE_ROOT / "manifest.csv", index=False, encoding="utf-8-sig") write_json( PACKAGE_ROOT / "manifest.json", { "schema_version": "1.0", "run_id": RUN_ID, "task_id": TASK_ID, "generated_at": GENERATED_AT, "source_run_id": SOURCE_RUN_ID, "final_run_id": FINAL_RUN_ID, "manifest_self_hash_excluded": True, "file_count": len(rows), "files": rows, }, ) def build_case_board(case_id: str, meta: pd.Series, symbol_rows: list[dict], tx_rows: list[dict]) -> None: case_dir = PACKAGE_ROOT / "cases" / case_id lines = [ f"# {case_id} V1 生命周期补充图", "", f"- 补充包：`{RUN_ID}`", f"- V1 最终引用包：`{FINAL_RUN_ID}`", f"- V1 来源包：`{SOURCE_RUN_ID}`", f"- V1 scope：`{meta.get('v1_return_scope', '')}`", f"- V1 账户贡献：`{money_text(meta.get('account_return_closed_lots', ''))}`", f"- 来源图片板：[打开](../../../{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md)", f"- 来源故事板：[打开](../../../{SOURCE_RUN_ID}/cases/{case_id}/case_story_board.md)", "", "本板只补充图片阅读视角，不重跑候选池、买卖裁决、订单、lot 或账户账本。", "", ] for symbol_row in symbol_rows: symbol = symbol_row["symbol"] lines.extend( [ f"## {symbol}", "", f"- 首 BUY：`{symbol_row['first_buy_datetime']}`", f"- 末 SELL：`{symbol_row['last_sell_datetime']}`", f"- BUY / SELL：`{symbol_row['buy_count']}` / `{symbol_row['sell_count']}`", f"- 生命周期日线图：", "", ] ) if symbol_row.get("lifecycle_chart_path"): rel = Path(symbol_row["lifecycle_chart_path"]).relative_to(f"cases/{case_id}").as_posix() lines.extend([f"![{symbol} 生命周期]({rel})", ""]) else: lines.extend(["- 生命周期日线图缺失：本机行情文件镜像没有覆盖该窗口。", ""]) lines.extend(["### 交易日分时图", ""]) symbol_tx = [r for r in tx_rows if r["symbol"] == symbol] for tx in sorted(symbol_tx, key=lambda r: r["trade_date"]): rel = Path(tx["minute_chart_path"]).relative_to(f"cases/{case_id}").as_posix() lines.extend( [ f"#### {symbol} {tx['trade_date']}，订单数 {tx['order_count']}", "", f"![{symbol} {tx['trade_date']} 分时]({rel})", "", ] ) lines.extend( [ "## 边界", "", "- 生命周期日线窗口是首个 BUY 前 50 个交易日到最后一个 SELL 后 20 个交易日。", "- 交易日分时图只展示有 BUY / SELL 订单产生的日期。", "- 所有买卖点来自 `strict_order_ledger.csv`，本包不新增交易结论。", ] ) (case_dir / "case_lifecycle_board.md").write_text("\n".join(lines) + "\n", encoding="utf-8") def build_root_docs(case_rows: list[dict], symbol_rows: list[dict], tx_rows: list[dict], self_summary: dict | None = None) -> None: readme = [ "# 无忌 V1 生命周期补充图片包", "", f"- 生成时间：`{GENERATED_AT}`", f"- 补充包：`{RUN_ID}`", f"- V1 最终引用包：`{FINAL_RUN_ID}`", f"- V1 来源包：`{SOURCE_RUN_ID}`", f"- 来源执行复审 ID：`{SOURCE_EXEC_AUDIT_ID}`", f"- 最终引用包执行审核 ID：`{FINAL_EXEC_AUDIT_ID}`", "", "## 补充了什么", "", "1. 按 `case_id + symbol` 串联股票生命周期：首个 BUY 前 50 个交易日到最后一个 SELL 后 20 个交易日的日线图。", "2. 按 `case_id + symbol + trade_date` 生成有交易发生日期的整日分时图，标记当天全部 BUY / SELL。", "", "## 怎么看", "", "1. 先打开 `case_lifecycle_board.md`，按 case 进入单个生命周期板。", "2. 单个 case 里先看生命周期日线，再看交易日分时。", "3. 需要复核数字时回到 V1 来源包的 `strict_order_ledger.csv` 和 `strict_position_lot_ledger.csv`。", "", ] if self_summary: readme.extend( [ "## 自检状态", "", f"- 当前状态：`{self_summary['status']}`", f"- V1 有 BUY 的 case：`{self_summary['v1_buy_case_count']}`", f"- 生命周期日线图：`{self_summary['case_symbol_lifecycle_count']}` / `{self_summary.get('case_symbol_lifecycle_expected', self_summary['case_symbol_lifecycle_count'])}`", f"- 交易日分时图：`{self_summary['transaction_day_chart_count']}` / `{self_summary.get('transaction_day_chart_expected', self_summary['transaction_day_chart_count'])}`", f"- 缺失行情输入：`{self_summary.get('missing_chart_input_count', 0)}`，集中在本地行情镜像未覆盖的 2026-04-03 及 2026-04-08 之后交易日；详见 `missing_chart_inputs.csv`。", "", ] ) readme.extend( [ "## 重要边界", "", "- 本包只补图片阅读入口，不重跑候选、裁决、订单、lot 或账户流水。", "- 本包不改变 V1 主口径、边界表和 `RETURN_STAT_READY=false`。", "- 日线后 20 个交易日属于事后 audit_view，只用于人工复盘，不得反推当时决策。", ] ) (PACKAGE_ROOT / "README.md").write_text("\n".join(readme) + "\n", encoding="utf-8") lines = [ "# 无忌 V1 生命周期补充图总入口", "", f"- 补充包：`{RUN_ID}`", f"- 股票生命周期图：`{len(symbol_rows)}`", f"- 交易日分时图：`{len(tx_rows)}`", "", "| case_id | V1 scope | 股票数 | BUY | SELL | 账户贡献 | 生命周期板 |", "|---|---|---:|---:|---:|---:|---|", ] for row in sorted(case_rows, key=lambda r: r["case_id"]): lines.append( f"| `{row['case_id']}` | `{row['v1_return_scope']}` | {row['symbol_count']} | {row['buy_order_count']} | {row['sell_order_count']} | {money_text(row['account_return_closed_lots'])} | [打开](cases/{row['case_id']}/case_lifecycle_board.md) |" ) (PACKAGE_ROOT / "case_lifecycle_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() v1_index = read_csv(FINAL_ROOT / "v1_case_readout_index.csv") order = read_csv(SOURCE_ROOT / "strict_order_ledger.csv") for col in ["trade_date", "t1_sellable_from_trade_date"]: if col in order.columns: order[col] = order[col].map(norm_date) order["trade_time"] = order["trade_time"].map(norm_time) for col in ["price", "position_delta_pct"]: order[col] = pd.to_numeric(order[col], errors="coerce") order = order[order["action"].isin(["BUY", "SELL"])].copy() buy_case_ids = set(order[order["action"] == "BUY"]["case_id"].tolist()) v1_index = v1_index[v1_index["case_id"].isin(buy_case_ids)].copy() v1_by_case = v1_index.set_index("case_id", drop=False).to_dict("index") order = order[order["case_id"].isin(v1_by_case)].copy() group_rows = [] date_symbols: dict[str, set[str]] = defaultdict(set) for (case_id, symbol), group in order.groupby(["case_id", "symbol"]): buy_rows = group[group["action"] == "BUY"].sort_values(["trade_date", "trade_time", "order_id"]) if buy_rows.empty: continue sell_rows = group[group["action"] == "SELL"].sort_values(["trade_date", "trade_time", "order_id"]) last_rows = sell_rows if not sell_rows.empty else group.sort_values(["trade_date", "trade_time", "order_id"]) first_buy = buy_rows.iloc[0] last_sell = last_rows.iloc[-1] group_rows.append( { "case_id": case_id, "symbol": symbol, "first_buy_date": first_buy["trade_date"], "first_buy_time": first_buy["trade_time"], "first_buy_datetime": f"{first_buy['trade_date']} {first_buy['trade_time']}", "last_sell_date": last_sell["trade_date"] if last_sell["action"] == "SELL" else "", "last_sell_time": last_sell["trade_time"] if last_sell["action"] == "SELL" else "", "last_sell_datetime": f"{last_sell['trade_date']} {last_sell['trade_time']}" if last_sell["action"] == "SELL" else "", "buy_count": int(len(buy_rows)), "sell_count": int(len(sell_rows)), } ) for _, row in group.iterrows(): date_symbols[row["trade_date"]].add(symbol) trade_dates = fetch_trade_calendar() min_fetch_dates: list[str] = [] max_fetch_dates: list[str] = [] for row in group_rows: last_date = row["last_sell_date"] or row["first_buy_date"] start, end = trade_window_from_calendar(trade_dates, row["first_buy_date"], last_date) row["daily_window_start"] = start row["daily_window_end"] = end min_fetch_dates.append(start) max_fetch_dates.append(end) symbols = sorted({r["symbol"] for r in group_rows}) min_date = min(min_fetch_dates) max_date = max(max_fetch_dates) print(f"V1 buy cases={len(v1_index)}, case-symbol groups={len(group_rows)}, symbols={len(symbols)}", flush=True) daily = fetch_daily(symbols, min_date, max_date) minute = fetch_minute(date_symbols) print(f"market rows: daily={len(daily)}, minute={len(minute)}", flush=True) chart_rows: list[dict] = [] missing_rows: list[dict] = [] symbol_index_rows: list[dict] = [] tx_index_rows: list[dict] = [] case_rows: list[dict] = [] case_symbol_rows: dict[str, list[dict]] = defaultdict(list) case_tx_rows: dict[str, list[dict]] = defaultdict(list) for idx, row in enumerate(group_rows, start=1): case_id = row["case_id"] symbol = row["symbol"] meta = pd.Series(v1_by_case[case_id]) case_dir = PACKAGE_ROOT / "cases" / case_id img_dir = case_dir / "img" img_dir.mkdir(parents=True, exist_ok=True) group = order[(order["case_id"] == case_id) & (order["symbol"] == symbol)].copy() daily_window = daily[ (daily["symbol"] == symbol) & (daily["trade_date_str"] >= row["daily_window_start"]) & (daily["trade_date_str"] <= row["daily_window_end"]) ].copy() safe_symbol = symbol.replace(".", "_") life_out = img_dir / f"09_lifecycle_daily_50pre_20post_{safe_symbol}_{row['first_buy_date'].replace('-', '')}_{(row['last_sell_date'] or row['first_buy_date']).replace('-', '')}.png" if daily_window.empty: missing_rows.append({"case_id": case_id, "symbol": symbol, "kind": "daily_lifecycle", "date": row["first_buy_date"]}) lifecycle_path = "" else: draw_lifecycle_daily_chart(daily_window, group, meta, symbol, life_out) lifecycle_path = life_out.relative_to(PACKAGE_ROOT).as_posix() chart_rows.append( { "case_id": case_id, "symbol": symbol, "chart_role": "symbol_lifecycle_daily_50pre_20post_audit_view", "action": "LIFECYCLE", "trade_date": row["first_buy_date"], "trade_time": row["first_buy_time"], "path": lifecycle_path, "source_order_id": "", "status": "PASS", "note": "首个BUY前50个交易日至最后SELL后20个交易日；仅作audit_view。", "size": life_out.stat().st_size, "sha256": sha256_file(life_out), } ) tx_count = 0 for trade_date_key, day_orders in group.groupby(["trade_date"]): trade_date = trade_date_key[0] if isinstance(trade_date_key, tuple) else trade_date_key minute_day = minute[ (minute["symbol"] == symbol) & (minute["trade_date"] == trade_date) ].copy() tx_out = img_dir / f"10_transaction_day_minute_line_{safe_symbol}_{str(trade_date).replace('-', '')}.png" if minute_day.empty: missing_rows.append({"case_id": case_id, "symbol": symbol, "kind": "transaction_minute", "date": trade_date}) continue draw_transaction_day_minute_chart(minute_day, day_orders, meta, symbol, trade_date, tx_out) tx_path = tx_out.relative_to(PACKAGE_ROOT).as_posix() tx_count += 1 tx_row = { "case_id": case_id, "symbol": symbol, "trade_date": trade_date, "order_count": int(len(day_orders)), "buy_order_count": int((day_orders["action"] == "BUY").sum()), "sell_order_count": int((day_orders["action"] == "SELL").sum()), "minute_chart_path": tx_path, } tx_index_rows.append(tx_row) case_tx_rows[case_id].append(tx_row) chart_rows.append( { "case_id": case_id, "symbol": symbol, "chart_role": "transaction_day_minute_line_audit_view", "action": "BUY_SELL_DAY", "trade_date": trade_date, "trade_time": "", "path": tx_path, "source_order_id": ";".join(day_orders["order_id"].tolist()), "status": "PASS", "note": "有交易产生的交易日整日分时；标记当天全部BUY/SELL订单。", "size": tx_out.stat().st_size, "sha256": sha256_file(tx_out), } ) symbol_row = { **row, "v1_return_scope": meta.get("v1_return_scope", ""), "success_flag": meta.get("success_flag", ""), "account_return_closed_lots": meta.get("account_return_closed_lots", ""), "lifecycle_chart_path": lifecycle_path, "transaction_day_chart_count": tx_count, "case_lifecycle_board": f"cases/{case_id}/case_lifecycle_board.md", "source_case_image_board": f"{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md", } symbol_index_rows.append(symbol_row) case_symbol_rows[case_id].append(symbol_row) if idx % 80 == 0: print(f"chart progress: {idx}/{len(group_rows)} case-symbol groups", flush=True) for case_id, symbols_for_case in case_symbol_rows.items(): meta = pd.Series(v1_by_case[case_id]) build_case_board(case_id, meta, symbols_for_case, case_tx_rows[case_id]) for case_id, meta_dict in v1_by_case.items(): case_orders = order[order["case_id"] == case_id] if case_orders[case_orders["action"] == "BUY"].empty: continue case_rows.append( { "case_id": case_id, "v1_return_scope": meta_dict.get("v1_return_scope", ""), "success_flag": meta_dict.get("success_flag", ""), "account_return_closed_lots": meta_dict.get("account_return_closed_lots", ""), "symbol_count": len(case_symbol_rows.get(case_id, [])), "buy_order_count": int((case_orders["action"] == "BUY").sum()), "sell_order_count": int((case_orders["action"] == "SELL").sum()), "case_lifecycle_board": f"cases/{case_id}/case_lifecycle_board.md", } ) pd.DataFrame(symbol_index_rows).to_csv(PACKAGE_ROOT / "lifecycle_symbol_index.csv", index=False, encoding="utf-8-sig") pd.DataFrame(tx_index_rows).to_csv(PACKAGE_ROOT / "transaction_day_chart_index.csv", index=False, encoding="utf-8-sig") pd.DataFrame(chart_rows).to_csv(PACKAGE_ROOT / "chart_evidence_audit.csv", index=False, encoding="utf-8-sig") pd.DataFrame(missing_rows).to_csv(PACKAGE_ROOT / "missing_chart_inputs.csv", index=False, encoding="utf-8-sig") pd.DataFrame(case_rows).to_csv(PACKAGE_ROOT / "case_lifecycle_index.csv", index=False, encoding="utf-8-sig") build_root_docs(case_rows, symbol_index_rows, tx_index_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_lifecycle = len(group_rows) expected_tx = len( order[order["case_id"].isin(v1_by_case)] .groupby(["case_id", "symbol", "trade_date"]) .size() ) lifecycle_actual = int((pd.DataFrame(chart_rows)["chart_role"] == "symbol_lifecycle_daily_50pre_20post_audit_view").sum()) if chart_rows else 0 tx_actual = int((pd.DataFrame(chart_rows)["chart_role"] == "transaction_day_minute_line_audit_view").sum()) if chart_rows else 0 link_missing = 0 if link_df.empty else int((~link_df["target_exists"]).sum()) self_checks = [ ("FINAL_V1_PACKAGE_EXISTS", FINAL_ROOT.exists(), str(FINAL_ROOT)), ("SOURCE_V1_PACKAGE_EXISTS", SOURCE_ROOT.exists(), str(SOURCE_ROOT)), ("V1_BUY_CASES_251", len(v1_index) == 251, f"buy_cases={len(v1_index)}"), ("LIFECYCLE_CHARTS_COMPLETE", lifecycle_actual == expected_lifecycle, f"actual={lifecycle_actual}, expected={expected_lifecycle}"), ("TRANSACTION_DAY_CHARTS_COMPLETE", tx_actual == expected_tx, f"actual={tx_actual}, expected={expected_tx}"), ("MISSING_CHART_INPUTS_ZERO", len(missing_rows) == 0, f"missing={len(missing_rows)}"), ("MARKDOWN_LOCAL_LINKS_REACHABLE", link_missing == 0, f"links={len(link_df)}, missing={link_missing}"), ("RETURN_STAT_READY_FALSE_PRESERVED", True, "supplement package does not change V1 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, "final_run_id": FINAL_RUN_ID, "status": "PASS_FOR_V1_LIFECYCLE_CHART_SUPPLEMENT_READY" if fail_count == 0 else "PARTIAL_WITH_MISSING_MARKET_INPUTS", "pass_count": int((self_df["status"] == "PASS").sum()), "fail_count": fail_count, "v1_buy_case_count": len(v1_index), "case_symbol_lifecycle_count": lifecycle_actual, "case_symbol_lifecycle_expected": expected_lifecycle, "transaction_day_chart_count": tx_actual, "transaction_day_chart_expected": expected_tx, "missing_chart_input_count": len(missing_rows), "return_stat_ready": False, } write_json(PACKAGE_ROOT / "self_check.json", self_json) build_root_docs(case_rows, symbol_index_rows, tx_index_rows, self_json) (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"- V1 有 BUY case：`{self_json['v1_buy_case_count']}`", f"- 生命周期日线图：`{self_json['case_symbol_lifecycle_count']}` / `{self_json['case_symbol_lifecycle_expected']}`", f"- 交易日分时图：`{self_json['transaction_day_chart_count']}` / `{self_json['transaction_day_chart_expected']}`", f"- 缺失行情输入：`{self_json['missing_chart_input_count']}`", "", "本包只补充人工审阅图片，不改变 V1 账本、收益口径和 `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()