from __future__ import annotations import hashlib import json import os import re from datetime import datetime from pathlib import Path import pandas as pd import pymysql from PIL import Image, ImageDraw, ImageFont RUN_ID = "RUN-ANA-WUJI-BASELINE-PILOT-20260607-001" ROOT = Path(__file__).resolve().parents[1] LOCAL_DB_INDEX = Path( r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md" ) def read_password() -> str: env = os.environ.get("TIANXIA_MYSQL_PASSWORD") or os.environ.get("MYSQL_PWD") if env: return env text = LOCAL_DB_INDEX.read_text(encoding="utf-8") match = re.search(r"^\s*-\s*密码：`([^`]+)`", text, re.MULTILINE) if not match: raise RuntimeError("Unable to read local MySQL credential from approved local index.") return match.group(1) def get_conn(): return pymysql.connect( host="127.0.0.1", port=3306, user="root", password=read_password(), database="tianxia", charset="utf8mb4", connect_timeout=5, read_timeout=120, ) def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def font(size: int): for p in [ Path("C:/Windows/Fonts/msyh.ttc"), Path("C:/Windows/Fonts/simhei.ttf"), Path("C:/Windows/Fonts/simsun.ttc"), ]: if p.exists(): return ImageFont.truetype(str(p), size) return ImageFont.load_default() FONT_TITLE = font(26) FONT_MID = font(17) FONT_SMALL = font(13) def normalize_time(value) -> str: text = str(value) if "days" in text: text = text.split()[-1] if "." in text: text = text.split(".")[0] parts = text.split(":") if len(parts) >= 3: return f"{int(parts[0]):02d}:{int(parts[1]):02d}:{int(float(parts[2])):02d}" return text def y_price(value: float, low: float, high: float, top: int, bottom: int) -> int: if high <= low: return (top + bottom) // 2 return bottom - int((value - low) / (high - low) * (bottom - top)) def short_id(value: object) -> str: return str(value).replace(f"LOT-{RUN_ID}-", "LOT-").replace(f"ORD-{RUN_ID}-", "ORD-") def signal_label(signal_type: str) -> str: return { "STOP5_SELL_SIGNAL_CANDIDATE": "-5%止损卖出", "TREND_TAKE_PROFIT_SIGNAL_CANDIDATE": "趋势止盈卖出", "THREE_HIGH_NOT_RISING_SIGNAL_CANDIDATE": "三高不升卖出", "HOLD_REVIEW_CANDIDATE": "继续持仓", }.get(signal_type, signal_type) def wrap_text(text: object, max_chars: int = 22) -> list[str]: lines: list[str] = [] for raw in str(text).splitlines(): item = raw.strip() if not item: lines.append("") continue while len(item) > max_chars: lines.append(item[:max_chars]) item = item[max_chars:] lines.append(item) return lines def clean_zero(value: float) -> float: return 0.0 if abs(value) < 5e-10 else value def evaluate_sell(signal: pd.Series, minute: pd.DataFrame) -> dict: day = minute[ (minute.symbol == signal.symbol) & (minute.trade_date == signal.signal_trade_date) ].copy().reset_index(drop=True) entry = float(signal.entry_price) stop_price = entry * 0.95 target_price = entry * 1.05 if day.empty: return { "action_status": "EXIT_DATA_GAP_HELD", "reason": "信号日无1分钟数据，不能伪造真实SELL；保留为数据缺口待审。", "lookahead_violation_flag": False, } if str(signal.signal_trade_date) < str(signal.sellable_from_trade_date): return { "action_status": "T1_GUARD_FAIL_HELD", "reason": "信号日在T+1可卖日之前，不能真实卖出。", "lookahead_violation_flag": False, } signal_type = str(signal.signal_type) if signal_type == "STOP5_SELL_SIGNAL_CANDIDATE": hit = day[day.low_price.le(stop_price)] if hit.empty: return { "action_status": "SELL_REVIEW_DATA_MISMATCH_HELD", "reason": "日K触发-5%止损候选，但1分钟线未复现跌破价位，保留复核。", "lookahead_violation_flag": False, } row = hit.iloc[0] return { "action_status": "AI_SELL_CONFIRMED", "decision_time": str(row.trade_time), "price": float(row.close_price), "sell_reason_type": "SELL_RISK_STOP5", "trigger_ref_price": stop_price, "reason": f"{row.trade_time} 分钟K最低价触及买入价-5%硬止损线，按止损铁律卖出；成交价按该分钟收盘价记录。", "lookahead_violation_flag": False, } if signal_type == "TREND_TAKE_PROFIT_SIGNAL_CANDIDATE": open_ref = float(day.open_price.iloc[0]) candidates = day[ (day.trade_time >= "09:35:00") & day.close_price.ge(target_price) & day.close_price.ge(open_ref * 1.01) ].copy() if candidates.empty: candidates = day[ day.close_price.ge(target_price) ].copy() if candidates.empty: return { "action_status": "SELL_REVIEW_DATA_MISMATCH_HELD", "reason": "日K趋势止盈候选成立，但1分钟线未找到达到买入价+5%的裁决点，保留复核。", "lookahead_violation_flag": False, } row = candidates.iloc[0] return { "action_status": "AI_SELL_CONFIRMED", "decision_time": str(row.trade_time), "price": float(row.close_price), "sell_reason_type": "SELL_TREND_TAKE_PROFIT", "trigger_ref_price": target_price, "reason": f"{row.trade_time} 价格达到买入价+5%并保持强于开盘方向，按“第二天走出趋势性上涨则卖出”裁决止盈。", "lookahead_violation_flag": False, } return { "action_status": "EXIT_REVIEW_HELD", "reason": "当前信号需要更强人工语义确认，本轮不写真实SELL。", "lookahead_violation_flag": False, } def observation_dates(trade_dates: list[str], entry_trade_date: str, sellable_from_trade_date: str) -> list[str]: idx = trade_dates.index(str(entry_trade_date)) obs = trade_dates[idx : idx + 11] return [d for d in obs if d >= str(sellable_from_trade_date)] def evaluate_sell_window(signal: pd.Series, minute: pd.DataFrame, trade_dates: list[str]) -> dict: """Resolve SELL by scanning the full approved observation window with 1m evidence.""" entry = float(signal.entry_price) stop_price = entry * 0.95 target_price = entry * 1.05 dates = observation_dates(trade_dates, str(signal.entry_trade_date), str(signal.sellable_from_trade_date)) gap_dates: list[str] = [] checked_dates: list[str] = [] first_mismatch_date = "" for trade_date in dates: day = minute[ (minute.symbol == signal.symbol) & (minute.trade_date == trade_date) ].copy().reset_index(drop=True) if day.empty: gap_dates.append(trade_date) continue checked_dates.append(trade_date) open_ref = float(day.open_price.iloc[0]) day_stop_possible = False fallback_trend_row = None for _, row in day.iterrows(): trade_time = str(row.trade_time) low = float(row.low_price) close = float(row.close_price) if low <= stop_price: return { "action_status": "AI_SELL_CONFIRMED", "signal_type": "STOP5_SELL_SIGNAL_CANDIDATE", "signal_trade_date": trade_date, "decision_time": trade_time, "price": close, "sell_reason_type": "SELL_RISK_STOP5", "trigger_ref_price": stop_price, "reason": f"{trade_time} 分钟K最低价触及买入价-5%硬止损线，按止损铁律卖出；成交价按该分钟收盘价记录。", "lookahead_violation_flag": False, "resolution_note_cn": "分钟线确认观察窗口内最早止损卖点。", "checked_dates": ",".join(checked_dates), "gap_dates": ",".join(gap_dates), } if close <= stop_price: day_stop_possible = True if trade_time >= "09:35:00" and close >= target_price and close >= open_ref * 1.01: return { "action_status": "AI_SELL_CONFIRMED", "signal_type": "TREND_TAKE_PROFIT_SIGNAL_CANDIDATE", "signal_trade_date": trade_date, "decision_time": trade_time, "price": close, "sell_reason_type": "SELL_TREND_TAKE_PROFIT", "trigger_ref_price": target_price, "reason": f"{trade_time} 价格达到买入价+5%并保持强于开盘方向，按“第二天走出趋势性上涨则卖出”裁决止盈。", "lookahead_violation_flag": False, "resolution_note_cn": "分钟线确认观察窗口内最早趋势止盈卖点。", "checked_dates": ",".join(checked_dates), "gap_dates": ",".join(gap_dates), } if close >= target_price and fallback_trend_row is None: fallback_trend_row = row if day_stop_possible and not first_mismatch_date: first_mismatch_date = trade_date if fallback_trend_row is not None: trade_time = str(fallback_trend_row.trade_time) close = float(fallback_trend_row.close_price) return { "action_status": "AI_SELL_CONFIRMED", "signal_type": "TREND_TAKE_PROFIT_SIGNAL_CANDIDATE", "signal_trade_date": trade_date, "decision_time": trade_time, "price": close, "sell_reason_type": "SELL_TREND_TAKE_PROFIT", "trigger_ref_price": target_price, "reason": f"{trade_time} 价格达到买入价+5%，按“第二天走出趋势性上涨则卖出”裁决止盈；本交易日未找到强于开盘1%的更强趋势点，作为趋势止盈复核点记录。", "lookahead_violation_flag": False, "resolution_note_cn": "分钟线确认观察窗口内最早趋势止盈卖点。", "checked_dates": ",".join(checked_dates), "gap_dates": ",".join(gap_dates), } if gap_dates: return { "action_status": "EXIT_DATA_GAP_HELD", "signal_type": signal.signal_type, "signal_trade_date": str(signal.signal_trade_date), "reason": "观察窗口内存在分钟数据缺口，不能伪造真实SELL；保留为数据缺口待审。", "lookahead_violation_flag": False, "resolution_note_cn": "分钟线覆盖不足，不能闭合 lot。", "checked_dates": ",".join(checked_dates), "gap_dates": ",".join(gap_dates), } return { "action_status": "WINDOW_END_VALUATION_ONLY", "signal_type": "HOLD_REVIEW_CANDIDATE", "signal_trade_date": dates[-1] if dates else "", "reason": "观察窗口内分钟线未确认原文卖点；按流程只能记录窗口末估值/继续持仓，不写真实SELL。", "lookahead_violation_flag": False, "resolution_note_cn": "已完成观察窗口分钟线复核，无原文卖点。", "checked_dates": ",".join(checked_dates), "gap_dates": "", "first_mismatch_date": first_mismatch_date, } def draw_sell_decision(minute: pd.DataFrame, signal: pd.Series, decision: dict, out_path: Path) -> None: w, h = 1500, 860 img = Image.new("RGB", (w, h), "#fbfbf7") d = ImageDraw.Draw(img) d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1") title = f"卖出决策1分钟K图：{signal.symbol} {signal.signal_trade_date}" d.text((32, 24), title, fill="#111827", font=FONT_TITLE) d.text( (32, 58), f"AI裁决：{decision['decision_time']} 卖出第一份仓，价格 {decision['price']:.2f}", fill="#7f1d1d", font=FONT_MID, ) plot_left, plot_top, plot_right, plot_bottom = 80, 105, 1060, 575 vol_top, vol_bottom = 625, 780 note_left, note_top = 1090, 110 d.rectangle([plot_left, plot_top, plot_right, plot_bottom], outline="#94a3b8") d.rectangle([plot_left, vol_top, plot_right, vol_bottom], outline="#94a3b8") df = minute[ (minute.symbol == signal.symbol) & (minute.trade_date == signal.signal_trade_date) & (minute.trade_time <= decision["decision_time"]) ].copy().reset_index(drop=True) entry = float(signal.entry_price) stop_price = entry * 0.95 target_price = entry * 1.05 refs = [entry, stop_price, target_price, float(decision["price"])] price_low = min(float(df.low_price.min()), min(refs)) * 0.998 price_high = max(float(df.high_price.max()), max(refs)) * 1.002 max_vol = max(float(df.volume.max()), 1.0) n = len(df) gap = (plot_right - plot_left) / max(n, 1) body_w = max(3, int(gap * 0.55)) sell_x = None for i, row in df.iterrows(): cx = int(plot_left + gap * i + gap / 2) op, hi, lo, cl = [float(row[c]) for c in ["open_price", "high_price", "low_price", "close_price"]] color = "#dc2626" if cl >= op else "#16a34a" d.line( [cx, y_price(lo, price_low, price_high, plot_top, plot_bottom), cx, y_price(hi, price_low, price_high, plot_top, plot_bottom)], fill=color, width=2, ) y1, y2 = y_price(op, price_low, price_high, plot_top, plot_bottom), y_price(cl, price_low, price_high, plot_top, plot_bottom) d.rectangle([cx - body_w // 2, min(y1, y2), cx + body_w // 2, max(y1, y2)], fill=color, outline=color) vh = int(float(row.volume) / max_vol * (vol_bottom - vol_top)) d.rectangle([cx - body_w // 2, vol_bottom - vh, cx + body_w // 2, vol_bottom], fill=color, outline=color) if str(row.trade_time) == decision["decision_time"]: sell_x = cx if i % max(1, n // 6) == 0: d.text((cx - 24, vol_bottom + 8), str(row.trade_time)[:5], fill="#64748b", font=FONT_SMALL) for ref, label, color in [ (entry, "买入价", "#2563eb"), (stop_price, "-5%止损", "#b91c1c"), (target_price, "+5%趋势", "#7c3aed"), ]: yy = y_price(ref, price_low, price_high, plot_top, plot_bottom) d.line([plot_left, yy, plot_right, yy], fill=color, width=2) d.text((plot_right - 110, yy - 8), f"{label} {ref:.2f}", fill=color, font=FONT_SMALL) if sell_x is not None: d.line([sell_x, plot_top, sell_x, vol_bottom], fill="#b91c1c", width=3) d.text((sell_x + 8, plot_top + 8), "卖出", fill="#b91c1c", font=FONT_MID) lot_ret = float(decision["price"]) / entry - 1 d.rounded_rectangle([note_left, note_top, 1460, 780], radius=8, outline="#334155", fill="#ffffff") notes = [ "卖出裁决", f"动作：SELL 第一份仓", f"lot：{short_id(signal.trade_lot_id)}", f"时间：{decision['decision_time']}", f"价格：{decision['price']:.2f}", f"收益：{lot_ret:.2%}", f"原因：{signal_label(signal.signal_type)}", "说明：", *wrap_text(decision["reason"], 18), "", "T+1：已过可卖日", "本图为 decision_view，", "只使用卖出时间及以前数据。", ] yy = note_top + 18 for i, line in enumerate(notes): d.text( (note_left + 18, yy), line, fill="#111827" if i == 0 else "#334155", font=FONT_TITLE if i == 0 else FONT_SMALL, ) yy += 32 if i == 0 else (24 if line else 12) d.text((32, 820), "无忌 baseline：卖出由AI按已冻结规则看图裁决，执行审核通过前不得引用收益结论。", fill="#334155", font=FONT_MID) img.save(out_path) def next_order_id(seq: int) -> str: return f"ORD-{RUN_ID}-SELL-{seq:04d}" def append_case_board(case_id: str, rows: list[dict]) -> None: case_dir = ROOT / "cases" / case_id board_path = case_dir / "case_image_board.md" existing = board_path.read_text(encoding="utf-8") if board_path.exists() else f"# {case_id} 图片审核板\n" existing = re.sub(r"\n## 6\. 卖出1分钟裁决图\n[\s\S]*$", "", existing.rstrip()) lines = [existing.rstrip(), "", "## 6. 卖出1分钟裁决图", ""] if not rows: lines.append("- 本案例无真实 SELL 裁决图。") for row in rows: rel = Path(row["path"]).relative_to(f"cases/{case_id}").as_posix() lines.extend( [ f"### {row['symbol']} {row['decision_time']}", "", f"![{row['symbol']}]({rel})", "", f"- 卖出原因：{row['note']}", "", ] ) board_path.write_text("\n".join(lines) + "\n", encoding="utf-8") def main() -> None: signals = pd.read_csv(ROOT / "sell_signal_candidates.csv", encoding="utf-8-sig") lots = pd.read_csv(ROOT / "position_lot_ledger.csv", encoding="utf-8-sig") orders = pd.read_csv(ROOT / "order_ledger.csv", encoding="utf-8-sig") for col in ["exit_trade_date", "exit_time", "exit_price", "lot_return_pct", "account_return_contribution_pct", "lot_status"]: if col in lots.columns: lots[col] = lots[col].fillna("").astype("object") signals["signal_trade_date"] = signals["signal_trade_date"].astype(str) signals["sellable_from_trade_date"] = signals["sellable_from_trade_date"].astype(str) for col in ["entry_price"]: signals[col] = pd.to_numeric(signals[col], errors="coerce") symbols = sorted(signals.symbol.dropna().unique().tolist()) with get_conn() as conn: trade_dates_df = pd.read_sql( """ SELECT DISTINCT trade_date FROM a_share_daily_price WHERE trade_date BETWEEN '2023-01-01' AND '2026-12-31' ORDER BY trade_date """, conn, ) trade_dates = pd.to_datetime(trade_dates_df.trade_date).dt.strftime("%Y-%m-%d").tolist() dates: list[str] = sorted( { d for _, signal in signals.iterrows() for d in observation_dates(trade_dates, str(signal.entry_trade_date), str(signal.sellable_from_trade_date)) } ) minute = pd.DataFrame() if symbols and dates: sym_ph = ",".join(["%s"] * len(symbols)) date_ph = ",".join(["%s"] * len(dates)) with get_conn() as conn: minute = pd.read_sql( f""" SELECT trade_date, trade_time, symbol, open_price, high_price, low_price, close_price, volume FROM a_share_minute_price WHERE symbol IN ({sym_ph}) AND trade_date IN ({date_ph}) ORDER BY symbol, trade_date, trade_time """, conn, params=[*symbols, *dates], ) if not minute.empty: minute.trade_date = pd.to_datetime(minute.trade_date).dt.strftime("%Y-%m-%d") minute.trade_time = minute.trade_time.map(normalize_time) for col in ["open_price", "high_price", "low_price", "close_price", "volume"]: minute[col] = pd.to_numeric(minute[col], errors="coerce") sell_decisions = [] resolution_rows = [] sell_orders = [] manifest_rows = [] sold_chart_rows_by_case: dict[str, list[dict]] = {} seq = 1 for _, signal in signals.iterrows(): decision = evaluate_sell_window(signal, minute, trade_dates) decision_signal_type = decision.get("signal_type", signal.signal_type) decision_signal_date = decision.get("signal_trade_date", signal.signal_trade_date) evidence_path = "" price = "" position_delta = "0" action_status = decision["action_status"] if action_status == "AI_SELL_CONFIRMED": case_dir = ROOT / "cases" / signal.case_id img_dir = case_dir / "img" img_dir.mkdir(parents=True, exist_ok=True) safe_time = str(decision["decision_time"]).replace(":", "") out_path = img_dir / f"06_exit_1m_decision_{signal.symbol.replace('.', '_')}_{decision_signal_date}_{safe_time}.png" chart_signal = signal.copy() chart_signal["signal_type"] = decision_signal_type chart_signal["signal_trade_date"] = decision_signal_date draw_sell_decision(minute, chart_signal, decision, out_path) evidence_path = out_path.relative_to(ROOT).as_posix() price = f"{float(decision['price']):.4f}" position_delta = f"{-float(lots[lots.trade_lot_id == signal.trade_lot_id].iloc[0].position_pct):.4f}" order_id = next_order_id(seq) seq += 1 sell_orders.append( { "order_id": order_id, "case_id": signal.case_id, "candidate_id": signal.order_id, "variant_id": lots[lots.trade_lot_id == signal.trade_lot_id].iloc[0].variant_id, "symbol": signal.symbol, "trade_date": decision_signal_date, "trade_time": decision["decision_time"], "action": "SELL", "price": price, "position_delta_pct": position_delta, "tranche_index": "1", "planned_tranche_count": "5", "decision_reason_cn": decision["reason"], "evidence_image_path": evidence_path, "t1_sellable_from_trade_date": signal.sellable_from_trade_date, "lookahead_violation_flag": str(decision["lookahead_violation_flag"]), "source_lot_id": signal.trade_lot_id, "source_order_id": signal.order_id, "exit_signal_type": decision_signal_type, } ) manifest_row = { "case_id": signal.case_id, "symbol": signal.symbol, "trade_date": decision_signal_date, "event_id": f"{signal.trade_lot_id}_exit_sell_decision", "chart_role": "exit_1m_sell_decision_view", "decision_time": f"{decision_signal_date} {decision['decision_time']}", "path": evidence_path, "sha256": sha256_file(out_path), "status": "PASS", "note": signal_label(decision_signal_type), } manifest_rows.append(manifest_row) sold_chart_rows_by_case.setdefault(signal.case_id, []).append(manifest_row) sell_decisions.append( { "trade_lot_id": signal.trade_lot_id, "source_order_id": signal.order_id, "case_id": signal.case_id, "symbol": signal.symbol, "entry_trade_date": signal.entry_trade_date, "entry_time": signal.entry_time, "entry_price": f"{float(signal.entry_price):.4f}", "sellable_from_trade_date": signal.sellable_from_trade_date, "original_signal_type": signal.signal_type, "original_signal_trade_date": signal.signal_trade_date, "signal_type": decision_signal_type, "signal_trade_date": decision_signal_date, "action_status": action_status, "decision_time": decision.get("decision_time", ""), "price": price, "position_delta_pct": position_delta, "evidence_image_path": evidence_path, "decision_reason_cn": decision["reason"], "lookahead_violation_flag": str(decision["lookahead_violation_flag"]), } ) resolution_rows.append( { "trade_lot_id": signal.trade_lot_id, "case_id": signal.case_id, "symbol": signal.symbol, "entry_trade_date": signal.entry_trade_date, "entry_price": f"{float(signal.entry_price):.4f}", "sellable_from_trade_date": signal.sellable_from_trade_date, "original_signal_type": signal.signal_type, "original_signal_trade_date": signal.signal_trade_date, "resolved_signal_type": decision_signal_type, "resolved_signal_trade_date": decision_signal_date, "final_action_status": action_status, "decision_time": decision.get("decision_time", ""), "price": price, "checked_dates": decision.get("checked_dates", ""), "gap_dates": decision.get("gap_dates", ""), "resolution_note_cn": decision.get("resolution_note_cn", decision["reason"]), "decision_reason_cn": decision["reason"], } ) sell_decisions_df = pd.DataFrame(sell_decisions) sell_decisions_df.to_csv(ROOT / "sell_decision_log.csv", index=False, encoding="utf-8-sig") pd.DataFrame(resolution_rows).to_csv(ROOT / "exit_resolution_log.csv", index=False, encoding="utf-8-sig") # Keep BUY orders and replace this stage's SELL orders idempotently. if "action" in orders.columns: orders = orders[orders.action != "SELL"].copy() orders = pd.concat([orders, pd.DataFrame(sell_orders)], ignore_index=True, sort=False) orders.to_csv(ROOT / "order_ledger.csv", index=False, encoding="utf-8-sig") lots = lots.copy() for _, decision in sell_decisions_df.iterrows(): idx = lots.trade_lot_id == decision.trade_lot_id if decision.action_status == "AI_SELL_CONFIRMED": entry_price = float(lots.loc[idx, "entry_price"].iloc[0]) pos = float(lots.loc[idx, "position_pct"].iloc[0]) exit_price = float(decision.price) lot_return = exit_price / entry_price - 1 lots.loc[idx, "lot_status"] = "CLOSED_BY_AI_SELL" lots.loc[idx, "exit_trade_date"] = decision.signal_trade_date lots.loc[idx, "exit_time"] = decision.decision_time lots.loc[idx, "exit_price"] = f"{exit_price:.4f}" lots.loc[idx, "lot_return_pct"] = f"{lot_return:.8f}" lots.loc[idx, "account_return_contribution_pct"] = f"{lot_return * pos:.8f}" else: lots.loc[idx, "lot_status"] = decision.action_status lots.to_csv(ROOT / "position_lot_ledger.csv", index=False, encoding="utf-8-sig") decision_log = pd.read_csv(ROOT / "decision_log.csv", encoding="utf-8-sig") if "decision_stage" in decision_log.columns: decision_log = decision_log[decision_log.decision_stage != "EXIT_AI_REVIEW"].copy() exit_decision_for_main = pd.DataFrame( [ { "case_id": r["case_id"], "candidate_id": r["source_order_id"], "symbol": r["symbol"], "entry_trade_date": r["entry_trade_date"], "decision_stage": "EXIT_AI_REVIEW", "action_status": r["action_status"], "decision_time": f"{r['signal_trade_date']} {r['decision_time']}" if r["decision_time"] else "", "price": r["price"], "position_delta_pct": r["position_delta_pct"], "review_required": "False" if r["action_status"] == "AI_SELL_CONFIRMED" else "True", "evidence_image_path": r["evidence_image_path"], "decision_reason_cn": r["decision_reason_cn"], "lookahead_violation_flag": r["lookahead_violation_flag"], } for r in sell_decisions ] ) decision_log = pd.concat([decision_log, exit_decision_for_main], ignore_index=True, sort=False) decision_log.to_csv(ROOT / "decision_log.csv", index=False, encoding="utf-8-sig") if manifest_rows: image_manifest = pd.read_csv(ROOT / "image_manifest.csv", encoding="utf-8-sig") image_manifest = pd.concat([image_manifest, pd.DataFrame(manifest_rows)], ignore_index=True) image_manifest = image_manifest.drop_duplicates(subset=["case_id", "symbol", "event_id", "chart_role"], keep="last") image_manifest.to_csv(ROOT / "image_manifest.csv", index=False, encoding="utf-8-sig") for case_id in sorted(lots.case_id.unique()): rows = sold_chart_rows_by_case.get(case_id, []) append_case_board(case_id, rows) case_dir = ROOT / "cases" / case_id image_manifest[image_manifest.case_id == case_id].to_csv(case_dir / "image_manifest.csv", index=False, encoding="utf-8-sig") account_rows = [] for case_id, group in lots.groupby("case_id"): events = [] buys = orders[(orders.case_id == case_id) & (orders.action == "BUY")].copy() sells = orders[(orders.case_id == case_id) & (orders.action == "SELL")].copy() for _, row in buys.iterrows(): events.append((row.trade_date, row.trade_time, "BUY", float(row.position_delta_pct), 0.0, row.symbol)) for _, row in sells.iterrows(): lot = group[group.trade_lot_id == row.source_lot_id].iloc[0] events.append((row.trade_date, row.trade_time, "SELL", float(lot.position_pct), float(lot.account_return_contribution_pct), row.symbol)) cash = 1.0 open_pos = 0.0 for date, time, action, pos_delta, realized, symbol in sorted(events): if action == "BUY": cash -= pos_delta open_pos += pos_delta realized_delta = 0.0 else: cash += pos_delta + realized open_pos -= pos_delta realized_delta = realized cash = clean_zero(cash) open_pos = clean_zero(open_pos) realized_delta = clean_zero(realized_delta) nav = cash + open_pos nav = clean_zero(nav) account_rows.append( { "case_id": case_id, "event_date": date, "event_time": time, "symbol": symbol, "action": action, "cash_pct_after_event": f"{cash:.8f}", "open_position_pct_after_event": f"{open_pos:.8f}", "realized_return_delta": f"{realized_delta:.8f}", "account_nav_after_event": f"{nav:.8f}", } ) account_df = pd.DataFrame(account_rows) account_df.to_csv(ROOT / "daily_account_ledger.csv", index=False, encoding="utf-8-sig") case_rows = [] for case_id, group in lots.groupby("case_id"): closed = group[group.lot_status == "CLOSED_BY_AI_SELL"].copy() unresolved = group[group.lot_status != "CLOSED_BY_AI_SELL"].copy() account_return = pd.to_numeric(closed.account_return_contribution_pct, errors="coerce").fillna(0).sum() case_rows.append( { "case_id": case_id, "buy_lot_count": len(group), "closed_lot_count": len(closed), "unresolved_lot_count": len(unresolved), "account_return_closed_lots": f"{account_return:.8f}", "strict_baseline_return_ready_flag": "0", "return_boundary": "STRUCTURE_PILOT_AI_SELL_REVIEWED__EXEC_AUDIT_PENDING__NOT_RETURN_STAT_READY", } ) case_summary = pd.DataFrame(case_rows) case_summary.to_csv(ROOT / "case_summary.csv", index=False, encoding="utf-8-sig") summary = { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": datetime.now().astimezone().isoformat(timespec="seconds"), "stage": "EXIT_AI_REVIEW_DONE", "sell_decision_counts": sell_decisions_df.action_status.value_counts().to_dict(), "sell_order_count": len(sell_orders), "closed_lot_count": int((lots.lot_status == "CLOSED_BY_AI_SELL").sum()), "unresolved_lot_count": int((lots.lot_status != "CLOSED_BY_AI_SELL").sum()), "closed_lot_account_return_sum": float(case_summary.account_return_closed_lots.astype(float).sum()), "strict_baseline_return_ready_flag": False, "boundary": "AI sell decisions are recorded for structure pilot. Execution audit is still required; return statistics are not ready.", "artifacts": { "sell_decision_log.csv": { "size": (ROOT / "sell_decision_log.csv").stat().st_size, "sha256": sha256_file(ROOT / "sell_decision_log.csv"), }, "exit_resolution_log.csv": { "size": (ROOT / "exit_resolution_log.csv").stat().st_size, "sha256": sha256_file(ROOT / "exit_resolution_log.csv"), }, "order_ledger.csv": { "size": (ROOT / "order_ledger.csv").stat().st_size, "sha256": sha256_file(ROOT / "order_ledger.csv"), }, "position_lot_ledger.csv": { "size": (ROOT / "position_lot_ledger.csv").stat().st_size, "sha256": sha256_file(ROOT / "position_lot_ledger.csv"), }, "daily_account_ledger.csv": { "size": (ROOT / "daily_account_ledger.csv").stat().st_size, "sha256": sha256_file(ROOT / "daily_account_ledger.csv"), }, "case_summary.csv": { "size": (ROOT / "case_summary.csv").stat().st_size, "sha256": sha256_file(ROOT / "case_summary.csv"), }, }, } (ROOT / "exit_ai_review_summary.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) (ROOT / "exit_ai_review_summary.md").write_text( "\n".join( [ "# exit_ai_review_summary", "", f"run_id：`{RUN_ID}`", "阶段：`EXIT_AI_REVIEW_DONE`", "", f"- SELL 订单数：{len(sell_orders)}", f"- 已闭合 lot：{summary['closed_lot_count']}", f"- 未闭合 / 待审 lot：{summary['unresolved_lot_count']}", f"- 闭合 lot 账户贡献合计：{summary['closed_lot_account_return_sum']:.4%}", "", "边界：本轮为结构试点 AI 卖出裁决，执行审核未通过前不得引用收益统计。", "", ] ), encoding="utf-8", ) if __name__ == "__main__": main()