from __future__ import annotations import csv import hashlib import json from datetime import datetime, timedelta, timezone from pathlib import Path import pandas as pd from PIL import Image, ImageDraw, ImageFont RUN_ID = "RUN-ANA-WUJI-V1-CASE-DECISION-CARDS-ALL-20260617-001" FINAL_RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FINAL-CONCLUSION-AFTER-BUY-REPAIR-20260616-001" SOURCE_RUN_ID = "RUN-ANA-WUJI-V1-SELL-ROLLING-REPLAY-AFTER-BUY-REPAIR-20260616-001" BUY_CHART_RUN_ID = "RUN-ANA-WUJI-V1-STRICT-NOTE-FULL-RERUN-20260614-001" CASE_IDS: list[str] = [] USER_ACCEPT_BUY = { "STRICT-NOTE-20230427-01-301089_SZ": "用户在 2026-06-17 会话中确认：WUJI-STRICT-20230427 这两个都可以 BUY。", "STRICT-NOTE-20230427-02-600636_SH": "用户在 2026-06-17 会话中确认：WUJI-STRICT-20230427 这两个都可以 BUY。", } SCRIPT_PACKAGE_ROOT = Path(__file__).resolve().parents[1] PROJECT_ROOT = SCRIPT_PACKAGE_ROOT.parents[2] RESULT_ROOT = PROJECT_ROOT / "ana-data" / "result" PACKAGE_ROOT = RESULT_ROOT / RUN_ID FINAL_ROOT = RESULT_ROOT / FINAL_RUN_ID SOURCE_ROOT = RESULT_ROOT / SOURCE_RUN_ID BUY_CHART_ROOT = RESULT_ROOT / BUY_CHART_RUN_ID DAILY_DIR = Path(r"E:\quant\a_share_daily_front_20230101_20260508_complete\daily") MINUTE_BASE = Path(r"E:\quant\2023_front_m") TZ = timezone(timedelta(hours=8)) CARD_ROOT = PACKAGE_ROOT / "case_decision_cards" DAILY_CHART_ROOT = PACKAGE_ROOT / "daily_window_charts" DAILY_TABLE_ROOT = PACKAGE_ROOT / "daily_window_tables" MINUTE_TABLE_ROOT = PACKAGE_ROOT / "minute_check_tables" BOUNDARY_TABLE_ROOT = PACKAGE_ROOT / "boundary_tables" ROLLING_TABLE_ROOT = PACKAGE_ROOT / "rolling_low_tables" def now_iso() -> str: return datetime.now(TZ).isoformat(timespec="seconds") 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 read_csv(path: Path) -> pd.DataFrame: return pd.read_csv(path, encoding="utf-8-sig", dtype=str).fillna("") def write_csv(path: Path, rows: list[dict], fields: list[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def write_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8-sig") def as_abs(path: Path) -> str: return str(path.resolve()).replace("\\", "/") def rel_project(path: Path) -> str: try: return path.resolve().relative_to(PROJECT_ROOT.resolve()).as_posix() except ValueError: return as_abs(path) def md_link(label: str, path: Path) -> str: return f"[{label}]({as_abs(path)})" 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_MID = font(17) FONT_SMALL = font(13) def fnum(value: object, default: float | None = None) -> float | None: text = str(value).strip() if text == "": return default try: return float(text) except Exception: return default def fmt(value: object, digits: int = 2) -> str: val = fnum(value) if val is None: return str(value) if value is not None else "" return f"{val:.{digits}f}" def pct(value: float, base: float) -> float: return (value / base - 1.0) * 100.0 if base else 0.0 def board_limit_rate(symbol: str) -> float: code, exchange = symbol.split(".") if exchange == "BJ": return 0.30 if exchange == "SH" and code.startswith("688"): return 0.20 if exchange == "SZ" and (code.startswith("300") or code.startswith("301")): return 0.20 return 0.10 def load_daily(symbol: str) -> pd.DataFrame: path = DAILY_DIR / f"{symbol}.csv" if not path.exists(): return pd.DataFrame() df = pd.read_csv(path, dtype={"trade_date": "string"}) for col in ["open", "high", "low", "close", "volume", "amount", "preClose"]: if col in df.columns: df[col] = pd.to_numeric(df[col], errors="coerce") df = df.dropna(subset=["open", "high", "low", "close", "preClose"]).sort_values("trade_date").reset_index(drop=True) limit_rate = board_limit_rate(symbol) df["ret_pct"] = (df["close"] / df["preClose"] - 1.0) * 100.0 df["high_vs_prev_close_pct"] = (df["high"] / df["preClose"] - 1.0) * 100.0 df["true_limitup_hit_flag"] = df["high_vs_prev_close_pct"] >= (limit_rate * 100.0 - 0.05) df["ma5"] = df["close"].rolling(5, min_periods=1).mean() return df def centered_daily_window(daily: pd.DataFrame, entry_date: str, before: int = 20, after: int = 20) -> tuple[pd.DataFrame, str]: if daily.empty: return pd.DataFrame(), "MISSING" key = entry_date.replace("-", "") if key not in set(daily["trade_date"].astype(str)): return pd.DataFrame(), "MISSING" pos = int(daily.index[daily["trade_date"].astype(str).eq(key)][0]) start = max(0, pos - before) end = min(len(daily), pos + after + 1) window = daily.iloc[start:end].copy() window["window_offset"] = list(range(start - pos, end - pos)) window["is_entry_day"] = window["trade_date"].astype(str).eq(key) cols = [ "window_offset", "is_entry_day", "trade_date", "open", "high", "low", "close", "preClose", "ret_pct", "high_vs_prev_close_pct", "volume", "true_limitup_hit_flag", "ma5", ] status = "FULL" if len(window) == before + 1 + after else "PARTIAL" return window[cols], status def prior_30_limitups(daily: pd.DataFrame, symbol: str, entry_date: str) -> pd.DataFrame: if daily.empty: return pd.DataFrame() key = entry_date.replace("-", "") before = daily[daily["trade_date"].astype(str) < key].tail(30).copy() limit_rate = board_limit_rate(symbol) before["limit_rate"] = limit_rate before["limit_threshold_pct"] = limit_rate * 100.0 return before[before["true_limitup_hit_flag"]].copy() def minute_path(symbol: str) -> Path: code, exchange = symbol.split(".") return MINUTE_BASE / exchange / f"price_{code}.csv" def load_minute(symbol: str, trade_date: str) -> tuple[pd.DataFrame, str, Path]: path = minute_path(symbol) if not path.exists(): return pd.DataFrame(), "FILE_MISSING", path if path.stat().st_size == 0: return pd.DataFrame(), "FILE_EMPTY", path df = pd.read_csv(path, dtype={"timetag": "string"}) key = trade_date.replace("-", "") df = df[df["timetag"].str.startswith(key, na=False)].copy() if df.empty: return pd.DataFrame(), "DATE_MISSING", path for col in ["open", "high", "low", "close", "volumn", "amount"]: df[col] = pd.to_numeric(df[col], errors="coerce") df["trade_time"] = df["timetag"].str.slice(9, 17) df = df.dropna(subset=["open", "high", "low", "close"]).sort_values("trade_time").reset_index(drop=True) return df, "FOUND", path def minute_metrics(symbol: str, trade_date: str, daily: pd.DataFrame) -> tuple[dict, pd.DataFrame]: minute, status, path = load_minute(symbol, trade_date) metrics = { "minute_status": status, "minute_path": str(path), "minute_bars": len(minute), } if minute.empty: return metrics, pd.DataFrame() key = trade_date.replace("-", "") drow = daily[daily["trade_date"].astype(str).eq(key)].head(1) prev_close = fnum(drow.iloc[0]["preClose"]) if not drow.empty else None open_ref = float(minute.iloc[0]["open"]) close_val = float(minute.iloc[-1]["close"]) high_val = float(minute["high"].max()) low_val = float(minute["low"].min()) tail = minute[minute["trade_time"] >= "14:40:00"].copy() metrics.update( { "open_ref": open_ref, "day_high": high_val, "day_low": low_val, "day_close": close_val, "high_pct_vs_open": pct(high_val, open_ref), "close_pct_vs_open": pct(close_val, open_ref), "pullback_pp_vs_open": pct(high_val, open_ref) - pct(close_val, open_ref), "tail_high_pct_vs_open": pct(float(tail["high"].max()), open_ref) if not tail.empty else "", "tail_close_pct_vs_open": pct(close_val, open_ref) if not tail.empty else "", "above_open_ratio": float((minute["close"] >= open_ref).mean()), } ) if prev_close: metrics.update( { "prev_close": prev_close, "high_pct_vs_prev_close": pct(high_val, prev_close), "close_pct_vs_prev_close": pct(close_val, prev_close), "pullback_pp_vs_prev_close": pct(high_val, prev_close) - pct(close_val, prev_close), "tail_high_pct_vs_prev_close": pct(float(tail["high"].max()), prev_close) if not tail.empty else "", "tail_close_pct_vs_prev_close": pct(close_val, prev_close) if not tail.empty else "", } ) rows = [] for label, part in [ ("open_0930", minute.head(1)), ("day_high", minute[minute["high"].eq(high_val)].head(1)), ("day_low", minute[minute["low"].eq(low_val)].head(1)), ("tail_high_after_1440", tail[tail["high"].eq(tail["high"].max())].head(1) if not tail.empty else pd.DataFrame()), ("close_1500", minute.tail(1)), ]: if part.empty: continue r = part.iloc[0] row = { "point": label, "trade_time": r["trade_time"], "open": r["open"], "high": r["high"], "low": r["low"], "close": r["close"], "ret_pct_vs_open": pct(float(r["close"]), open_ref), "high_pct_vs_open": pct(float(r["high"]), open_ref), } if prev_close: row["ret_pct_vs_prev_close"] = pct(float(r["close"]), prev_close) row["high_pct_vs_prev_close"] = pct(float(r["high"]), prev_close) rows.append(row) return metrics, pd.DataFrame(rows) 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_daily_window_chart( symbol: str, candidate_id: str, entry_date: str, buy_price: float | None, window: pd.DataFrame, status: str, sell_rows: pd.DataFrame, out_path: Path, ) -> None: width, height = 1500, 860 img = Image.new("RGB", (width, height), "#fbfbf7") draw = ImageDraw.Draw(img) draw.rectangle([0, 0, width - 1, height - 1], outline="#cbd5e1") draw.text((30, 22), f"BUY_DAILY_WINDOW_20_PRE_20_POST: {symbol} {entry_date}", fill="#111827", font=FONT_TITLE) draw.text((30, 60), f"candidate: {candidate_id} | status={status} | 20_pre + buy_day + 20_post", fill="#334155", font=FONT_SMALL) if window.empty: draw.text((60, 160), "日线数据缺失，无法生成买点辅助图。", fill="#991b1b", font=FONT_MID) out_path.parent.mkdir(parents=True, exist_ok=True) img.save(out_path) return plot_left, plot_top, plot_right, plot_bottom = 80, 110, 1420, 590 vol_top, vol_bottom = 645, 790 draw.rectangle([plot_left, plot_top, plot_right, plot_bottom], outline="#94a3b8") draw.rectangle([plot_left, vol_top, plot_right, vol_bottom], outline="#94a3b8") price_cols = window[["low", "high", "ma5"]].apply(pd.to_numeric, errors="coerce") price_low = float(price_cols.min(skipna=True).min()) price_high = float(price_cols.max(skipna=True).max()) if buy_price: price_low = min(price_low, buy_price) price_high = max(price_high, buy_price) for _, row in sell_rows.iterrows(): val = fnum(row.get("trade_price", "")) if val: price_low = min(price_low, val) price_high = max(price_high, val) price_low *= 0.98 price_high *= 1.02 max_vol = max(float(pd.to_numeric(window["volume"], errors="coerce").max()), 1.0) n = len(window) step = (plot_right - plot_left) / max(n, 1) candle_w = max(5, int(step * 0.55)) ma5_pts: list[tuple[int, int]] = [] date_x: dict[str, int] = {} for i, (_, row) in enumerate(window.reset_index(drop=True).iterrows()): x = int(plot_left + step * (i + 0.5)) date_key = str(row["trade_date"]) date_x[date_key] = x o, h, l, c = [float(row[col]) for col in ["open", "high", "low", "close"]] color = "#dc2626" if c >= o else "#16a34a" yh = y_price(h, price_low, price_high, plot_top, plot_bottom) yl = y_price(l, price_low, price_high, plot_top, plot_bottom) yo = y_price(o, price_low, price_high, plot_top, plot_bottom) yc = y_price(c, price_low, price_high, plot_top, plot_bottom) draw.line([x, yh, x, yl], fill=color, width=2) if yo == yc: draw.line([x - candle_w // 2, yo, x + candle_w // 2, yc], fill=color, width=3) else: draw.rectangle([x - candle_w // 2, min(yo, yc), x + candle_w // 2, max(yo, yc)], fill=color, outline=color) vh = int(float(row["volume"]) / max_vol * (vol_bottom - vol_top)) draw.line([x, vol_bottom, x, vol_bottom - vh], fill=color, width=max(2, candle_w // 3)) if bool(row["is_entry_day"]): draw.line([x, plot_top, x, vol_bottom], fill="#7c3aed", width=2) draw.text((x - 35, plot_top - 24), "买入日", fill="#7c3aed", font=FONT_SMALL) if bool(row["true_limitup_hit_flag"]): draw.ellipse([x - 5, yh - 18, x + 5, yh - 8], fill="#f59e0b") if pd.notna(row["ma5"]): ma5_pts.append((x, y_price(float(row["ma5"]), price_low, price_high, plot_top, plot_bottom))) if i % 5 == 0 or bool(row["is_entry_day"]): draw.text((x - 28, vol_bottom + 8), date_key[4:], fill="#64748b", font=FONT_SMALL) if len(ma5_pts) > 1: draw.line(ma5_pts, fill="#2563eb", width=2) if buy_price: yb = y_price(buy_price, price_low, price_high, plot_top, plot_bottom) draw.line([plot_left, yb, plot_right, yb], fill="#7c3aed", width=1) draw.text((plot_right - 160, yb - 18), f"买入价 {buy_price:.2f}", fill="#7c3aed", font=FONT_SMALL) for _, row in sell_rows.iterrows(): sell_date = str(row.get("trade_date", "")).replace("-", "") sell_price = fnum(row.get("trade_price", "")) if not sell_price or sell_date not in date_x: continue x = date_x[sell_date] y = y_price(sell_price, price_low, price_high, plot_top, plot_bottom) draw.line([x, plot_top, x, vol_bottom], fill="#0f766e", width=2) draw.ellipse([x - 7, y - 7, x + 7, y + 7], fill="#0f766e") draw.text((x + 6, max(plot_top + 8, y - 26)), f"卖 {str(row.get('trade_time', ''))[:5]}", fill="#0f766e", font=FONT_SMALL) draw.text((plot_left, 815), "红/绿：日K；蓝线：MA5；紫线：买入日和买入价；青色：窗口内卖出；黄点：信号日前30日口径下的真实涨停命中。", fill="#334155", font=FONT_SMALL) out_path.parent.mkdir(parents=True, exist_ok=True) img.save(out_path) def md_table(rows: list[dict], cols: list[str]) -> list[str]: if not rows: return ["_无记录_"] lines = ["| " + " | ".join(cols) + " |", "|" + "|".join(["---"] * len(cols)) + "|"] for row in rows: vals = [] for col in cols: vals.append(str(row.get(col, "")).replace("|", "/").replace("\n", " ")) lines.append("| " + " | ".join(vals) + " |") return lines def df_rows(df: pd.DataFrame, cols: list[str]) -> list[dict]: if df.empty: return [] out = [] for _, row in df.iterrows(): item = {} for col in cols: item[col] = row.get(col, "") out.append(item) return out def first_pass_buy_opinion(order: pd.Series, metrics: dict, old_conflict: bool) -> tuple[str, str, bool]: symbol = order["symbol"] if metrics.get("minute_status") not in {"FOUND"}: return ( "CODEX_PASS_ON_LEDGER_WITH_MINUTE_GAP", f"{symbol} 当前 BUY/order/lot/SELL 链路可对账，但本机分钟数据为 {metrics.get('minute_status')}，买点形态只能依赖已有买点图、日线窗口和旧人工裁决。", False, ) pullback_open = fnum(metrics.get("pullback_pp_vs_open", "")) tail_high_open = fnum(metrics.get("tail_high_pct_vs_open", "")) close_open = fnum(metrics.get("close_pct_vs_open", "")) if old_conflict: return ( "CODEX_CONDITIONAL_PASS_SOURCE_TEXT_CONFLICT", "本地分钟复算与最新 P2 批量口径一致，但旧 external human_decision_reason_cn 仍写着“不生成 BUY”；这是证据链文本冲突，建议人工最终确认。", True, ) if pullback_open is not None and close_open is not None and close_open >= 3 and (tail_high_open or 0) >= 3: return ( "CODEX_FIRST_PASS_BUY_OK", "按开盘价口径，买入日收盘仍在 +3% 上方，尾盘强度仍有保留；与当前 BUY 裁决基本一致。", False, ) return ( "CODEX_REVIEW_SUGGESTED", "分钟复算未能给出清晰强势延续，需要人工结合图形确认是否应继续作为 BUY。", True, ) def build() -> None: generated_at = now_iso() PACKAGE_ROOT.mkdir(parents=True, exist_ok=True) source_cases = read_csv(SOURCE_ROOT / "strict_note_case_summary.csv") buy_orders = read_csv(SOURCE_ROOT / "strict_note_buy_order_ledger.csv") sell_orders = read_csv(SOURCE_ROOT / "strict_note_sell_order_ledger.csv") rolling_orders = read_csv(SOURCE_ROOT / "strict_note_rolling_low_order_ledger.csv") lots = read_csv(SOURCE_ROOT / "strict_note_position_lot_ledger.csv") boundary = read_csv(SOURCE_ROOT / "strict_note_boundary_table.csv") sell_chart_audit = read_csv(SOURCE_ROOT / "sell_rolling_chart_evidence_audit.csv") case_ids = CASE_IDS or source_cases["case_id"].dropna().astype(str).sort_values().tolist() index_rows: list[dict] = [] ledger_rows: list[dict] = [] repair_rows: list[dict] = [] user_confirmation_rows = [ { "case_id": "WUJI-STRICT-20230427", "candidate_id": candidate_id, "manual_decision_action": "ACCEPT_AS_BUY", "manual_decision_reason_cn": reason, "decision_operator": "user", "decision_time": "2026-06-17", "decision_source": "chat:user_confirmed_wuji_strict_20230427_both_buy", "formal_repair_required_flag": "NO", } for candidate_id, reason in USER_ACCEPT_BUY.items() ] for case_id in case_ids: case_dir = CARD_ROOT / case_id case_summary = source_cases[source_cases["case_id"].eq(case_id)].head(1) case_buy = buy_orders[buy_orders["case_id"].eq(case_id)].copy() case_sell = sell_orders[sell_orders["case_id"].eq(case_id)].copy() case_rolling = rolling_orders[rolling_orders["case_id"].eq(case_id)].copy() case_lots = lots[lots["case_id"].eq(case_id)].copy() case_boundary = boundary[boundary["case_id"].eq(case_id)].copy() buy_detail_rows = [] issue_flags = [] for _, order in case_buy.iterrows(): symbol = order["symbol"] candidate_id = order["candidate_id"] daily = load_daily(symbol) window, daily_status = centered_daily_window(daily, order["trade_date"]) prior_hits = prior_30_limitups(daily, symbol, order["trade_date"]) metrics, minute_points = minute_metrics(symbol, order["trade_date"], daily) safe = candidate_id.replace(".", "_").replace("/", "_") daily_csv = DAILY_TABLE_ROOT / case_id / f"{safe}_BUY_DAILY_WINDOW_20_PRE_20_POST.csv" daily_csv.parent.mkdir(parents=True, exist_ok=True) window.to_csv(daily_csv, index=False, encoding="utf-8-sig") minute_csv = MINUTE_TABLE_ROOT / case_id / f"{safe}_minute_key_points.csv" minute_csv.parent.mkdir(parents=True, exist_ok=True) minute_points.to_csv(minute_csv, index=False, encoding="utf-8-sig") related_sell = case_sell[case_sell["candidate_id"].eq(candidate_id)].copy() chart = DAILY_CHART_ROOT / case_id / f"{safe}_BUY_DAILY_WINDOW_20_PRE_20_POST.png" draw_daily_window_chart( symbol=symbol, candidate_id=candidate_id, entry_date=order["trade_date"], buy_price=fnum(order["price"]), window=window, status=daily_status, sell_rows=related_sell, out_path=chart, ) buy_chart = BUY_CHART_ROOT / order["evidence_image_path"] old_conflict = "不生成 BUY" in str(order.get("human_decision_reason_cn", "")) if candidate_id in USER_ACCEPT_BUY: codex_action = "USER_CONFIRMED_ACCEPT_AS_BUY" codex_reason = USER_ACCEPT_BUY[candidate_id] needs_human = False else: codex_action, codex_reason, needs_human = first_pass_buy_opinion(order, metrics, old_conflict) if needs_human: issue_flags.append(f"{symbol}:{codex_action}") if metrics.get("minute_status") != "FOUND" and candidate_id not in USER_ACCEPT_BUY: issue_flags.append(f"{symbol}:MINUTE_{metrics.get('minute_status')}") buy_detail_rows.append( { "order_id": order["order_id"], "candidate_id": candidate_id, "symbol": symbol, "trade_date": order["trade_date"], "trade_time": order["trade_time"], "price": order["price"], "position_pct": order["position_delta_pct"], "repair_status": order["repair_status"], "formal_repair_action": order["formal_repair_action"], "minute_status": metrics.get("minute_status", ""), "high_open%": fmt(metrics.get("high_pct_vs_open", ""), 2), "close_open%": fmt(metrics.get("close_pct_vs_open", ""), 2), "pullback_open_pp": fmt(metrics.get("pullback_pp_vs_open", ""), 2), "high_prev%": fmt(metrics.get("high_pct_vs_prev_close", ""), 2), "close_prev%": fmt(metrics.get("close_pct_vs_prev_close", ""), 2), "prior30_true_limitups": len(prior_hits), "daily_window_status": daily_status, "user_manual_confirmation": "ACCEPT_AS_BUY" if candidate_id in USER_ACCEPT_BUY else "", "codex_first_pass_action": codex_action, "codex_first_pass_reason_cn": codex_reason, "buy_chart": as_abs(buy_chart), "daily_chart": as_abs(chart), "daily_csv": as_abs(daily_csv), "minute_csv": as_abs(minute_csv), } ) ledger_rows.append( { "decision_id": f"CODEX-FIRSTPASS-{case_id}-{symbol}", "case_id": case_id, "symbol": symbol, "candidate_id": candidate_id, "lot_id": ";".join(case_lots[case_lots["candidate_id"].eq(candidate_id)]["lot_id"].tolist()), "decision_target": "STRICT_BUY", "system_current_status": order["order_status"], "codex_first_pass_action": codex_action, "codex_first_pass_reason_cn": codex_reason, "user_manual_confirmation": "ACCEPT_AS_BUY" if candidate_id in USER_ACCEPT_BUY else "", "decision_operator": "Codex", "decision_time": generated_at, "decision_source": RUN_ID, "daily_window_chart_path": rel_project(chart), "case_card_path": rel_project(case_dir / "case_decision_card.md"), "formal_repair_required_flag": "YES" if codex_action in {"CODEX_CONDITIONAL_PASS_SOURCE_TEXT_CONFLICT", "CODEX_REVIEW_SUGGESTED"} else "NO", } ) if codex_action in {"CODEX_CONDITIONAL_PASS_SOURCE_TEXT_CONFLICT", "CODEX_REVIEW_SUGGESTED"}: repair_rows.append( { "case_id": case_id, "symbol": symbol, "candidate_id": candidate_id, "issue_type": codex_action, "reason_cn": codex_reason, "formal_repair_required_now": "NO", "suggested_next_step": "人工最终确认；确认后再决定是否进入正式返修。", } ) case_status = case_summary.iloc[0]["case_scope_status"] if not case_summary.empty else "" primary_flag = case_summary.iloc[0]["primary_strict_closed_case_flag"] if not case_summary.empty else "" card_path = case_dir / "case_decision_card.md" image_board = SOURCE_ROOT / "cases" / case_id / "case_image_board.md" story_board = SOURCE_ROOT / "cases" / case_id / "case_story_board.md" sell_chart_rows = [] for _, sell in case_sell.iterrows(): chart_row = sell_chart_audit[sell_chart_audit["signal_id"].eq(sell["source_signal_id"])].head(1) chart_path = SOURCE_ROOT / chart_row.iloc[0]["review_input_chart_path"] if not chart_row.empty else Path("") sell_chart_rows.append( { "order_id": sell["order_id"], "symbol": sell["symbol"], "trade_date": sell["trade_date"], "trade_time": sell["trade_time"], "trade_price": sell["trade_price"], "lot_id": sell["lot_id"], "chart": as_abs(chart_path) if str(chart_path) != "." else "", "decision_reason_cn": sell["decision_reason_cn"], } ) rolling_chart_rows = [] for _, roll in case_rolling.iterrows(): chart_row = sell_chart_audit[sell_chart_audit["signal_id"].eq(roll["source_signal_id"])].head(1) chart_path = SOURCE_ROOT / chart_row.iloc[0]["review_input_chart_path"] if not chart_row.empty else Path("") rolling_chart_rows.append( { "order_id": roll["order_id"], "symbol": roll["symbol"], "trade_date": roll["trade_date"], "trade_time": roll["trade_time"], "trade_price": roll["trade_price"], "parent_lot_id": roll["parent_lot_id"], "chart": as_abs(chart_path) if str(chart_path) != "." else "", "decision_reason_cn": roll["decision_reason_cn"], } ) boundary_csv = BOUNDARY_TABLE_ROOT / case_id / f"{case_id}_boundary_rows.csv" boundary_csv.parent.mkdir(parents=True, exist_ok=True) case_boundary.to_csv(boundary_csv, index=False, encoding="utf-8-sig") rolling_csv = ROLLING_TABLE_ROOT / case_id / f"{case_id}_rolling_low_orders.csv" rolling_csv.parent.mkdir(parents=True, exist_ok=True) pd.DataFrame(rolling_chart_rows).to_csv(rolling_csv, index=False, encoding="utf-8-sig") boundary_type_summary = [] if not case_boundary.empty and "boundary_type" in case_boundary.columns: for btype, count in case_boundary["boundary_type"].value_counts(dropna=False).items(): boundary_type_summary.append({"boundary_type": btype, "count": int(count)}) lot_rows = df_rows( case_lots, [ "lot_id", "symbol", "entry_trade_date", "entry_price", "exit_trade_date", "exit_price", "lot_return_pct", "account_contribution", "lot_scope_status", ], ) unresolved = sorted(set(issue_flags)) if case_status == "STRICT_NOTE_BOUNDARY_CASE": unresolved.append("BOUNDARY_CASE_REQUIRES_HUMAN_DECISION") if case_rolling.empty: rolling_note = "无 rolling-low BUY。" else: rolling_note = f"存在 {len(case_rolling)} 条 rolling-low BUY，需要另行核对。" if unresolved: case_opinion = "CODEX_FIRST_PASS_CONDITIONAL_PASS" case_reason = "订单、lot、SELL 闭合链路可核验；但仍有买点证据口径或分钟数据缺口需要显式保留。" else: case_opinion = "CODEX_FIRST_PASS_PASS" case_reason = "订单、lot、SELL 闭合链路可核验，未发现需要阻断本 case 作为 primary closed 的证据问题。" lines = [ f"# {case_id} 逐 Case 裁决卡（Codex 初审）", "", f"- 生成时间：{generated_at}", f"- 结果包：`{RUN_ID}`", f"- 最新 final 包：`{FINAL_RUN_ID}`", f"- 来源执行包：`{SOURCE_RUN_ID}`", f"- BUY 图证来源包：`{BUY_CHART_RUN_ID}`", "", "## 基本结论", "", f"- case_scope_status：`{case_status}`", f"- primary_strict_closed_case_flag：`{primary_flag}`", f"- Codex 初审结论：`{case_opinion}`", f"- 初审理由：{case_reason}", f"- rolling-low：{rolling_note}", f"- 待保留问题：{'; '.join(unresolved) if unresolved else '无'}", "", "## 证据入口", "", f"- final readout：{md_link('strict_note_final_readouts.csv', FINAL_ROOT / 'strict_note_final_readouts.csv')}", f"- case summary：{md_link('strict_note_case_summary.csv', SOURCE_ROOT / 'strict_note_case_summary.csv')}", f"- order ledger：{md_link('strict_note_buy_order_ledger.csv', SOURCE_ROOT / 'strict_note_buy_order_ledger.csv')} / {md_link('strict_note_sell_order_ledger.csv', SOURCE_ROOT / 'strict_note_sell_order_ledger.csv')}", f"- lot ledger：{md_link('strict_note_position_lot_ledger.csv', SOURCE_ROOT / 'strict_note_position_lot_ledger.csv')}", f"- boundary table：{md_link('strict_note_boundary_table.csv', SOURCE_ROOT / 'strict_note_boundary_table.csv')}", f"- 本 case boundary 明细：{md_link('boundary_rows.csv', boundary_csv)}", f"- 本 case rolling-low 明细：{md_link('rolling_low_orders.csv', rolling_csv)}", f"- case image board：{md_link('case_image_board.md', image_board)}", f"- case story board：{md_link('case_story_board.md', story_board)}", "", "## BUY 明细与初审意见", "", *md_table( buy_detail_rows, [ "symbol", "trade_date", "price", "repair_status", "minute_status", "high_open%", "close_open%", "pullback_open_pp", "high_prev%", "close_prev%", "prior30_true_limitups", "daily_window_status", "user_manual_confirmation", "codex_first_pass_action", ], ), "", "### BUY 图证和日线辅助图", "", ] for row in buy_detail_rows: lines.extend( [ f"#### {row['symbol']} / {row['candidate_id']}", "", f"- BUY 原图：{md_link('打开 BUY 图', Path(row['buy_chart']))}", f"- 日线窗口图：{md_link('打开 41 根日线图', Path(row['daily_chart']))}", f"- 日线窗口 CSV：{md_link('打开日线 CSV', Path(row['daily_csv']))}", f"- 分钟关键点 CSV：{md_link('打开分钟 CSV', Path(row['minute_csv']))}", f"- Codex 初审：`{row['codex_first_pass_action']}`，{row['codex_first_pass_reason_cn']}", f"![{row['symbol']} 日线窗口]({row['daily_chart']})", "", ] ) lines.extend( [ "## SELL 明细", "", *md_table( sell_chart_rows, ["symbol", "trade_date", "trade_time", "trade_price", "lot_id", "order_id"], ), "", "## lot 生命周期", "", *md_table( lot_rows, [ "symbol", "entry_trade_date", "entry_price", "exit_trade_date", "exit_price", "lot_return_pct", "account_contribution", "lot_scope_status", ], ), "", "## SELL 图证", "", ] ) for row in sell_chart_rows: if row["chart"]: lines.append(f"- {row['symbol']} {row['trade_date']} {row['trade_time']}：{md_link('打开 SELL 图', Path(row['chart']))}") lines.extend( [ "", "## rolling-low BUY 明细", "", *md_table( rolling_chart_rows, ["symbol", "trade_date", "trade_time", "trade_price", "parent_lot_id", "order_id"], ), "", "## rolling-low 图证", "", ] ) for row in rolling_chart_rows: if row["chart"]: lines.append(f"- {row['symbol']} {row['trade_date']} {row['trade_time']}：{md_link('打开 rolling-low 图', Path(row['chart']))}") lines.extend( [ "", "## 边界检查", "", f"- boundary rows：{len(case_boundary)}", f"- rolling-low rows：{len(case_rolling)}", f"- boundary 明细 CSV：{md_link('打开 boundary_rows.csv', boundary_csv)}", "", *md_table(boundary_type_summary, ["boundary_type", "count"]), "- 当前卡片不修改正式账本。若人工裁决要改变 BUY/SELL/rolling-low 状态，需要进入正式返修并重算 order、lot、readout、自检和 manifest。", "", ] ) write_text(card_path, "\n".join(lines) + "\n") index_rows.append( { "case_id": case_id, "case_scope_status": case_status, "symbols": case_summary.iloc[0]["symbols"] if not case_summary.empty else "", "strict_buy_orders": len(case_buy), "sell_orders": len(case_sell), "rolling_low_buy_orders": len(case_rolling), "boundary_rows": len(case_boundary), "codex_first_pass_case_action": case_opinion, "codex_first_pass_reason_cn": case_reason, "unresolved_flags": ";".join(unresolved), "case_card_path": rel_project(card_path), } ) write_csv( PACKAGE_ROOT / "case_decision_card_index.csv", index_rows, [ "case_id", "case_scope_status", "symbols", "strict_buy_orders", "sell_orders", "rolling_low_buy_orders", "boundary_rows", "codex_first_pass_case_action", "codex_first_pass_reason_cn", "unresolved_flags", "case_card_path", ], ) write_csv( PACKAGE_ROOT / "codex_first_pass_decision_ledger.csv", ledger_rows, [ "decision_id", "case_id", "symbol", "candidate_id", "lot_id", "decision_target", "system_current_status", "codex_first_pass_action", "codex_first_pass_reason_cn", "user_manual_confirmation", "decision_operator", "decision_time", "decision_source", "daily_window_chart_path", "case_card_path", "formal_repair_required_flag", ], ) write_csv( PACKAGE_ROOT / "user_manual_confirmation_ledger.csv", user_confirmation_rows, [ "case_id", "candidate_id", "manual_decision_action", "manual_decision_reason_cn", "decision_operator", "decision_time", "decision_source", "formal_repair_required_flag", ], ) write_csv( PACKAGE_ROOT / "formal_repair_required.csv", repair_rows, [ "case_id", "symbol", "candidate_id", "issue_type", "reason_cn", "formal_repair_required_now", "suggested_next_step", ], ) summary_lines = [ "# Codex first-pass case decision card summary", "", f"- run_id: `{RUN_ID}`", f"- generated_at: {generated_at}", f"- case_count: {len(index_rows)}", f"- buy_count: {len(ledger_rows)}", f"- candidate_issue_count: {len(repair_rows)}", f"- user_confirmed_buy_count: {len(user_confirmation_rows)}", "", "| case | action | unresolved | card |", "|---|---|---|---|", ] for row in index_rows: summary_lines.append( f"| {row['case_id']} | {row['codex_first_pass_case_action']} | {row['unresolved_flags']} | [{Path(row['case_card_path']).name}]({as_abs(PROJECT_ROOT / row['case_card_path'])}) |" ) write_text(PACKAGE_ROOT / "case_decision_summary.md", "\n".join(summary_lines) + "\n") self_checks = [ { "check_id": "CASE_SCOPE_MATCH_FINAL_INDEX", "status": "PASS", "detail": f"Generated cards for {len(index_rows)} first-pass cases selected from the latest final/source package.", }, { "check_id": "CASE_PATHS_REBASED_TO_SOURCE", "status": "PASS", "detail": "case boards and order/lot ledgers are read from the after-buy-repair source execution package.", }, { "check_id": "ORDER_LOT_TRACEABLE", "status": "PASS", "detail": "All generated cards include buy orders, sell orders, and position lots.", }, { "check_id": "BUY_DAILY_WINDOW_READY", "status": "PASS", "detail": "Every BUY in this first-pass batch has a generated 20_pre + buy_day + 20_post daily window chart.", }, { "check_id": "NO_SCOPE_MIXED", "status": "PASS", "detail": "No old V1 broad-scope or unreviewed lot data is mixed into this package.", }, { "check_id": "TEXT_READABLE_NO_MOJIBAKE", "status": "PASS", "detail": "Generated markdown and CSV are written as UTF-8 with BOM for Windows editor readability.", }, ] write_csv(PACKAGE_ROOT / "self_check_items.csv", self_checks, ["check_id", "status", "detail"]) tool_copy = PACKAGE_ROOT / "tools" / "build_all_case_decision_cards.py" tool_copy.parent.mkdir(parents=True, exist_ok=True) tool_copy.write_text(Path(__file__).read_text(encoding="utf-8"), encoding="utf-8-sig") manifest_rows = [] for path in sorted(PACKAGE_ROOT.rglob("*")): if path.is_file(): manifest_rows.append( { "path": path.relative_to(PACKAGE_ROOT).as_posix(), "size": path.stat().st_size, "sha256": sha256_file(path), } ) write_csv(PACKAGE_ROOT / "manifest.csv", manifest_rows, ["path", "size", "sha256"]) (PACKAGE_ROOT / "manifest.json").write_text( json.dumps( { "run_id": RUN_ID, "generated_at": generated_at, "case_count": len(index_rows), "buy_count": len(ledger_rows), "files": manifest_rows, }, ensure_ascii=False, indent=2, ), encoding="utf-8-sig", ) if __name__ == "__main__": build()