from __future__ import annotations import csv import hashlib import json import math import os import re import shutil from collections import defaultdict from datetime import datetime from decimal import Decimal from pathlib import Path import pandas as pd try: import pymysql except ImportError: # pragma: no cover - runtime dependency check is recorded in self-check. pymysql = None RUN_ID = "RUN-ANA-WUJI-RETURN-STAT-PILOT-20260608-001" SOURCE_RUN_ID = "RUN-ANA-WUJI-BASELINE-PILOT-20260607-001" CASE_MATTER_ID = "ANA-WUJI-BASELINE-2023-2026" DESIGN_ID = "DESIGN-WUJI-RETURN-STAT-READY-20260608" SOURCE_DESIGN_ID = "DESIGN-WUJI-BASELINE-FLOW-20260607" DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-RETURN-STAT-READY-20260608-DESIGN-001" SOURCE_AUDIT_IDS = [ "AUDIT-ANA-WUJI-BASELINE-FLOW-20260607-001", "AUDIT-ANA-WUJI-BASELINE-PILOT-20260608-EXEC-REREVIEW-001", "AUDIT-ANA-WUJI-BASELINE-PILOT-20260608-EXIT-REVIEW-001", "AUDIT-ANA-WUJI-BASELINE-PILOT-20260608-CANDIDATE-POOL-SUPP-001", ] ROOT = Path(__file__).resolve().parents[1] PROJECT_ROOT = ROOT.parents[2] SOURCE_ROOT = PROJECT_ROOT / "ana-data" / "result" / SOURCE_RUN_ID LOCAL_DB_INDEX = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md") CONFIG = { "schema_version": "1.0", "run_id": RUN_ID, "case_matter_id": CASE_MATTER_ID, "design_id": DESIGN_ID, "design_audit_id": DESIGN_AUDIT_ID, "source_run_id": SOURCE_RUN_ID, "source_audit_ids": SOURCE_AUDIT_IDS, "config_frozen_at": None, "status_policy": { "execution_status": "RETURN_STAT_READY_CANDIDATE_EXECUTION_REVIEW_REQUIRED", "return_stat_ready": False, "allowed_scope": "only the 7 representative pilot cases from source run", "not_allowed": [ "No full 2023-2026 return or success-rate conclusion.", "No forced SELL for WINDOW_END_VALUATION_ONLY or EXIT_DATA_GAP_HELD.", "No expansion beyond the audited pilot cases in this run.", ], }, "cost_model": { "cost_model_id": "WUJI_RETURN_STAT_COST_V1_20260608", "initial_cash_cny": 1000000.0, "cash_unit": "CNY", "round_lot_size": 100, "min_order_lot": 1, "rounding_policy": "floor_to_100_shares_per_lot_budget", "cash_residual_policy": "unused_budget_remains_cash_in_case_account", "commission_rate_by_side": { "buy": 0.00025, "sell": 0.00025, "source": "Inherited from audited source run_config commission_rate_each_side=0.00025; broker-specific minimum fee is not applied in this percent-account pilot.", }, "stamp_tax_sell_rate_by_date": [ { "start_date": "1900-01-01", "end_date": "2023-08-27", "rate": 0.001, "source": "Run-frozen historical A-share sell-side stamp-tax schedule; reviewer should verify official fee basis before expanding sample.", }, { "start_date": "2023-08-28", "end_date": "2099-12-31", "rate": 0.0005, "source": "Run-frozen schedule reflecting the 2023-08-28 half-rate stamp-tax policy; reviewer should verify official fee basis before expanding sample.", }, ], "transfer_fee_rate_by_side": { "buy": 0.00001, "sell": 0.00001, "source": "Run-frozen A-share transfer-fee sensitivity assumption; reviewer should verify official exchange/clearing fee basis before expanding sample.", }, "slippage_bps": { "buy": 5.0, "sell": 5.0, "source": "Internal conservative sensitivity setting for RETURN_STAT_READY preparation, not a note baseline rule.", }, "price_rounding_decimal_places": 4, "money_rounding_decimal_places": 2, "return_rounding_decimal_places": 8, }, "tradeability_policy": { "data_source": "MYSQL_TIANXIA_LOCAL.a_share_minute_price + a_share_daily_price", "limit_rate_rule": { "BJ": 0.30, "STAR_688": 0.20, "CHINEXT_300_301": 0.20, "MAINBOARD_DEFAULT": 0.10, }, "limit_price_rounding": "round_to_0.01", "buy_policy": "If BUY decision price is at upper limit and minute bar is locked at upper limit, mark LIMIT_UP_BUY_EXECUTION_HELD.", "sell_policy": "If SELL decision price is at lower limit and minute bar is locked at lower limit, mark LIMIT_DOWN_SELL_EXECUTION_HELD.", "missing_data_policy": "Missing minute or previous-close data is HELD for main return scope.", }, "sensitivity_models": [ {"model_id": "LOW_CAPITAL_100K", "initial_cash_cny": 100000.0}, {"model_id": "MAIN_CAPITAL_1M", "initial_cash_cny": 1000000.0}, {"model_id": "HIGH_CAPITAL_10M", "initial_cash_cny": 10000000.0}, ], } def now_iso() -> str: return datetime.now().astimezone().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(name: str) -> pd.DataFrame: return pd.read_csv(SOURCE_ROOT / name, encoding="utf-8-sig", keep_default_na=False) def write_csv(path: Path, rows: list[dict], fieldnames: list[str]) -> None: with path.open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() for row in rows: writer.writerow(row) def write_json(path: Path, data: dict) -> None: path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def clean_float(value, default: float = 0.0) -> float: if value is None: return default if isinstance(value, str) and not value.strip(): return default if isinstance(value, Decimal): return float(value) try: if pd.isna(value): return default except TypeError: pass return float(value) def clean_bool(value) -> bool: return str(value).strip().lower() in {"true", "1", "yes"} def money(value: float) -> float: return round(float(value), CONFIG["cost_model"]["money_rounding_decimal_places"]) def pct(value: float) -> float: return round(float(value), CONFIG["cost_model"]["return_rounding_decimal_places"]) 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 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(): if pymysql is None: raise RuntimeError("pymysql is not installed.") 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, cursorclass=pymysql.cursors.DictCursor, ) def symbol_market_group(symbol: str) -> str: code = symbol.upper().strip() if code.endswith(".BJ"): return "BJ" if re.match(r"^688\d{3}\.SH$", code): return "STAR_688" if re.match(r"^30[01]\d{3}\.SZ$", code): return "CHINEXT_300_301" return "MAINBOARD_DEFAULT" def symbol_limit_rate(symbol: str) -> float: group = symbol_market_group(symbol) return float(CONFIG["tradeability_policy"]["limit_rate_rule"][group]) def stamp_tax_rate(trade_date: str) -> float: for row in CONFIG["cost_model"]["stamp_tax_sell_rate_by_date"]: if row["start_date"] <= trade_date <= row["end_date"]: return float(row["rate"]) raise ValueError(f"No stamp-tax rate configured for {trade_date}") def fetch_tradeability(orders: pd.DataFrame) -> tuple[dict[str, dict], list[str]]: diagnostics: list[str] = [] result: dict[str, dict] = {} try: conn = get_conn() except Exception as exc: # noqa: BLE001 diagnostics.append(f"DB_CONNECT_FAILED: {exc}") return result, diagnostics try: with conn.cursor() as cur: for _, order in orders.iterrows(): order_id = str(order.order_id) symbol = str(order.symbol) trade_date = str(order.trade_date) trade_time = normalize_time(order.trade_time) cur.execute( """ SELECT open_price, high_price, low_price, close_price, volume FROM a_share_minute_price WHERE symbol=%s AND trade_date=%s AND trade_time=%s LIMIT 1 """, (symbol, trade_date, trade_time), ) minute = cur.fetchone() cur.execute( """ SELECT trade_date, close_price FROM a_share_daily_price WHERE symbol=%s AND trade_date < %s ORDER BY trade_date DESC LIMIT 1 """, (symbol, trade_date), ) prev_daily = cur.fetchone() cur.execute( """ SELECT trade_date, trade_time, close_price FROM a_share_minute_price WHERE symbol=%s AND trade_date < %s ORDER BY trade_date DESC, trade_time DESC LIMIT 1 """, (symbol, trade_date), ) prev_minute = cur.fetchone() if minute is None or (prev_daily is None and prev_minute is None): status = "DB_TRADEABILITY_DATA_MISSING_HELD" reason = "缺少对应 1 分钟 bar 或同口径前一交易日收盘价，不能放入主收益口径。" result[order_id] = { "status": status, "reason": reason, "minute_open": "", "minute_high": "", "minute_low": "", "minute_close": "", "minute_volume": "", "prev_close": "", "prev_close_source": "", "daily_prev_close": "", "minute_prev_close": "", "limit_rate": "", "upper_limit_price": "", "lower_limit_price": "", } continue daily_prev_close = clean_float(prev_daily["close_price"]) if prev_daily else 0.0 minute_prev_close = clean_float(prev_minute["close_price"]) if prev_minute else 0.0 # The daily table is front-adjusted while the minute table is raw in some historical rows. # Limit checks must stay in the same price space as the order/minute evidence. if minute_prev_close > 0: prev_close = minute_prev_close prev_close_source = "a_share_minute_price_prev_trade_day_last_close" else: prev_close = daily_prev_close prev_close_source = "a_share_daily_price_prev_close" limit_rate = symbol_limit_rate(symbol) upper = round(prev_close * (1 + limit_rate), 2) lower = round(prev_close * (1 - limit_rate), 2) price = clean_float(order.price) high = clean_float(minute["high_price"]) low = clean_float(minute["low_price"]) volume = clean_float(minute["volume"]) action = str(order.action).upper() status = "TRADEABILITY_PASS" reason = "1 分钟 bar 存在，价格落在 bar 范围内，未命中保守涨跌停锁死规则。" if price < low - 0.0001 or price > high + 0.0001: status = "PRICE_OUTSIDE_MINUTE_BAR_HELD" reason = "订单价格不在对应 1 分钟 bar 高低价范围内。" elif action == "BUY" and price >= upper - 0.005 and low >= upper - 0.005: status = "LIMIT_UP_BUY_EXECUTION_HELD" reason = "BUY 价格处于涨停附近且分钟 bar 锁在涨停附近，按保守口径不可放入主收益。" elif action == "SELL" and price <= lower + 0.005 and high <= lower + 0.005: status = "LIMIT_DOWN_SELL_EXECUTION_HELD" reason = "SELL 价格处于跌停附近且分钟 bar 锁在跌停附近，按保守口径不可放入主收益。" elif volume <= 0: status = "ZERO_VOLUME_MINUTE_BAR_HELD" reason = "对应 1 分钟 bar 成交量为 0，不能证明可成交。" result[order_id] = { "status": status, "reason": reason, "minute_open": clean_float(minute["open_price"]), "minute_high": high, "minute_low": low, "minute_close": clean_float(minute["close_price"]), "minute_volume": volume, "prev_close": prev_close, "prev_close_source": prev_close_source, "daily_prev_close": daily_prev_close, "minute_prev_close": minute_prev_close, "limit_rate": limit_rate, "upper_limit_price": upper, "lower_limit_price": lower, } finally: conn.close() return result, diagnostics def make_order_rows(orders: pd.DataFrame, lots: pd.DataFrame, tradeability: dict[str, dict]) -> tuple[list[dict], dict[str, dict]]: initial_cash = float(CONFIG["cost_model"]["initial_cash_cny"]) lot_size = int(CONFIG["cost_model"]["round_lot_size"]) commission_buy = float(CONFIG["cost_model"]["commission_rate_by_side"]["buy"]) commission_sell = float(CONFIG["cost_model"]["commission_rate_by_side"]["sell"]) transfer_buy = float(CONFIG["cost_model"]["transfer_fee_rate_by_side"]["buy"]) transfer_sell = float(CONFIG["cost_model"]["transfer_fee_rate_by_side"]["sell"]) slippage_buy = float(CONFIG["cost_model"]["slippage_bps"]["buy"]) slippage_sell = float(CONFIG["cost_model"]["slippage_bps"]["sell"]) buy_lot_by_order = {str(row.order_id): row for _, row in lots.iterrows()} buy_calc_by_lot: dict[str, dict] = {} rows: list[dict] = [] seq = 0 for _, order in orders.iterrows(): seq += 1 action = str(order.action).upper() source_order_id = str(order.order_id) lot_id = str(order.source_lot_id) if action == "SELL" else "" if action == "BUY": lot = buy_lot_by_order.get(source_order_id) lot_id = "" if lot is None else str(lot.trade_lot_id) price = clean_float(order.price) position_pct = abs(clean_float(order.position_delta_pct)) planned_budget = initial_cash * position_pct if action == "BUY": raw_shares = planned_budget / price if price > 0 else 0.0 shares = math.floor(raw_shares / lot_size) * lot_size slippage_bps = slippage_buy effective_price = price * (1 + slippage_bps / 10000.0) gross_notional = shares * effective_price commission_rate = commission_buy transfer_rate = transfer_buy stamp_rate = 0.0 commission = gross_notional * commission_rate transfer = gross_notional * transfer_rate stamp_tax = 0.0 total_fee = commission + transfer + stamp_tax cash_delta = -(gross_notional + total_fee) if lot_id: buy_calc_by_lot[lot_id] = { "shares": shares, "buy_effective_price": effective_price, "buy_gross_notional": gross_notional, "buy_total_fee": total_fee, "buy_cash_delta": cash_delta, "integer_lot_status": "INTEGER_LOT_PASS" if shares >= lot_size else "INTEGER_LOT_NOT_EXECUTABLE_HELD", } else: buy_calc = buy_calc_by_lot.get(lot_id, {}) shares = clean_float(buy_calc.get("shares")) slippage_bps = slippage_sell effective_price = price * (1 - slippage_bps / 10000.0) gross_notional = shares * effective_price commission_rate = commission_sell transfer_rate = transfer_sell stamp_rate = stamp_tax_rate(str(order.trade_date)) commission = gross_notional * commission_rate transfer = gross_notional * transfer_rate stamp_tax = gross_notional * stamp_rate total_fee = commission + transfer + stamp_tax cash_delta = gross_notional - total_fee trade = tradeability.get(source_order_id, {"status": "DB_TRADEABILITY_NOT_CHECKED_HELD", "reason": "未取得源库可成交检查结果。"}) row = { "return_order_id": f"ORD-{RUN_ID}-{seq:04d}", "source_order_id": source_order_id, "source_run_id": SOURCE_RUN_ID, "trade_lot_id": lot_id, "case_id": str(order.case_id), "symbol": str(order.symbol), "trade_date": str(order.trade_date), "trade_time": normalize_time(order.trade_time), "action": action, "source_price": f"{price:.4f}", "slippage_bps": f"{slippage_bps:.2f}", "effective_price_after_slippage": f"{effective_price:.4f}", "shares": int(shares), "planned_position_pct": f"{position_pct:.8f}", "planned_budget_cny": f"{planned_budget:.2f}", "gross_notional_cny": f"{gross_notional:.2f}", "commission_rate": f"{commission_rate:.8f}", "commission_cny": f"{commission:.2f}", "stamp_tax_rate": f"{stamp_rate:.8f}", "stamp_tax_cny": f"{stamp_tax:.2f}", "transfer_fee_rate": f"{transfer_rate:.8f}", "transfer_fee_cny": f"{transfer:.2f}", "total_fee_cny": f"{total_fee:.2f}", "cash_delta_cny": f"{cash_delta:.2f}", "tradeability_status": trade["status"], "tradeability_reason": trade["reason"], "prev_close": trade.get("prev_close", ""), "prev_close_source": trade.get("prev_close_source", ""), "daily_prev_close": trade.get("daily_prev_close", ""), "minute_prev_close": trade.get("minute_prev_close", ""), "limit_rate": trade.get("limit_rate", ""), "upper_limit_price": trade.get("upper_limit_price", ""), "lower_limit_price": trade.get("lower_limit_price", ""), "minute_low": trade.get("minute_low", ""), "minute_high": trade.get("minute_high", ""), "minute_volume": trade.get("minute_volume", ""), "evidence_image_path": str(order.evidence_image_path), } rows.append(row) return rows, buy_calc_by_lot def make_lot_scope_rows(lots: pd.DataFrame, order_rows: list[dict], case_main_candidate: dict[str, bool]) -> tuple[list[dict], dict[str, dict]]: orders_by_lot: dict[str, list[dict]] = defaultdict(list) for order in order_rows: if order["trade_lot_id"]: orders_by_lot[order["trade_lot_id"]].append(order) rows: list[dict] = [] lot_calc: dict[str, dict] = {} for _, lot in lots.iterrows(): lot_id = str(lot.trade_lot_id) status = str(lot.lot_status) related = orders_by_lot.get(lot_id, []) buy_order = next((r for r in related if r["action"] == "BUY"), None) sell_order = next((r for r in related if r["action"] == "SELL"), None) source_gross_return = clean_float(lot.lot_return_pct, default=0.0) source_account_contribution = clean_float(lot.account_return_contribution_pct, default=0.0) t1_pass = bool(str(lot.exit_trade_date) >= str(lot.sellable_from_trade_date)) if status == "CLOSED_BY_AI_SELL" else False lookahead_pass = True evidence_pass = bool(buy_order and buy_order["evidence_image_path"]) if status == "CLOSED_BY_AI_SELL": evidence_pass = evidence_pass and bool(sell_order and sell_order["evidence_image_path"]) integer_pass = bool(buy_order and int(buy_order["shares"]) >= int(CONFIG["cost_model"]["round_lot_size"])) tradeability_statuses = [r["tradeability_status"] for r in related] tradeability_pass = bool(related) and all(s == "TRADEABILITY_PASS" for s in tradeability_statuses) include_aux = ( status == "CLOSED_BY_AI_SELL" and t1_pass and lookahead_pass and evidence_pass and integer_pass and tradeability_pass ) include_case_main = include_aux and case_main_candidate.get(str(lot.case_id), False) reason_parts: list[str] = [] if status != "CLOSED_BY_AI_SELL": reason_parts.append(status) if not t1_pass and status == "CLOSED_BY_AI_SELL": reason_parts.append("T1_GUARD_NOT_PASS") if not evidence_pass: reason_parts.append("EVIDENCE_PATH_MISSING") if not integer_pass: reason_parts.append("INTEGER_LOT_NOT_EXECUTABLE_HELD") if not tradeability_pass: reason_parts.append("TRADEABILITY_NOT_PASS") if include_aux and not include_case_main: reason_parts.append("CASE_HAS_BOUNDARY_LOT_AUX_ONLY") if not reason_parts: reason_parts.append("STRICT_CLOSED_LOT_RECALC_READY") buy_cash = clean_float(buy_order["cash_delta_cny"]) if buy_order else 0.0 sell_cash = clean_float(sell_order["cash_delta_cny"]) if sell_order else 0.0 buy_notional = clean_float(buy_order["gross_notional_cny"]) if buy_order else 0.0 net_pnl = sell_cash + buy_cash if sell_order else 0.0 net_account_contribution = net_pnl / float(CONFIG["cost_model"]["initial_cash_cny"]) net_lot_return = net_pnl / abs(buy_cash) if buy_cash else 0.0 gross_integer_contribution = 0.0 if buy_order and sell_order: gross_integer_contribution = ( (clean_float(sell_order["shares"]) * clean_float(sell_order["source_price"])) - (clean_float(buy_order["shares"]) * clean_float(buy_order["source_price"])) ) / float(CONFIG["cost_model"]["initial_cash_cny"]) row = { "trade_lot_id": lot_id, "case_id": str(lot.case_id), "symbol": str(lot.symbol), "lot_status": status, "entry_trade_date": str(lot.entry_trade_date), "entry_time": normalize_time(lot.entry_time), "entry_price": str(lot.entry_price), "sellable_from_trade_date": str(lot.sellable_from_trade_date), "exit_trade_date": str(lot.exit_trade_date), "exit_time": normalize_time(lot.exit_time) if str(lot.exit_time) else "", "exit_price": str(lot.exit_price), "shares": int(clean_float(buy_order["shares"]) if buy_order else 0), "source_gross_lot_return_pct": f"{source_gross_return:.8f}", "source_gross_account_contribution_pct": f"{source_account_contribution:.8f}", "gross_integer_account_contribution_pct": f"{gross_integer_contribution:.8f}", "net_lot_return_after_cost_pct": f"{net_lot_return:.8f}", "net_account_contribution_after_cost_pct": f"{net_account_contribution:.8f}", "t1_check_status": "T1_PASS" if t1_pass or status != "CLOSED_BY_AI_SELL" else "T1_GUARD_FAIL_HELD", "lookahead_check_status": "LOOKAHEAD_PASS" if lookahead_pass else "LOOKAHEAD_VIOLATION_EXCLUDED", "evidence_check_status": "EVIDENCE_PASS" if evidence_pass else "EVIDENCE_MISSING_HELD", "integer_lot_check_status": "INTEGER_LOT_PASS" if integer_pass else "INTEGER_LOT_NOT_EXECUTABLE_HELD", "tradeability_check_status": "TRADEABILITY_PASS" if tradeability_pass else ";".join(tradeability_statuses or ["NO_RELATED_ORDER"]), "include_in_strict_closed_lot_recalc": str(include_aux), "include_in_strict_closed_case_return": str(include_case_main), "scope_status": "STRICT_CLOSED_LOT_RECALC_ONLY" if include_aux and not include_case_main else ("STRICT_CLOSED_CASE_RETURN_READY_CANDIDATE" if include_case_main else "RETURN_STAT_HELD_BOUNDARY_TABLE"), "exclude_or_boundary_reason": ";".join(reason_parts), "buy_evidence_image_path": buy_order["evidence_image_path"] if buy_order else "", "sell_evidence_image_path": sell_order["evidence_image_path"] if sell_order else "", } rows.append(row) lot_calc[lot_id] = row return rows, lot_calc def make_case_scope_and_summary(case_index: pd.DataFrame, lots: pd.DataFrame, lot_rows: list[dict]) -> tuple[list[dict], list[dict], dict]: lot_by_case: dict[str, list[dict]] = defaultdict(list) for row in lot_rows: lot_by_case[row["case_id"]].append(row) case_scope_rows: list[dict] = [] case_summary_rows: list[dict] = [] included_case_count = 0 success_count = 0 main_returns: list[float] = [] aux_lot_count = 0 aux_lot_win_count = 0 aux_lot_returns: list[float] = [] all_boundary_count = 0 main_drawdowns: list[float] = [] for _, case in case_index.iterrows(): case_id = str(case.case_id) rows = lot_by_case.get(case_id, []) buy_count = len(rows) closed_count = sum(1 for r in rows if r["lot_status"] == "CLOSED_BY_AI_SELL") boundary_rows = [r for r in rows if r["lot_status"] != "CLOSED_BY_AI_SELL"] aux_rows = [r for r in rows if r["include_in_strict_closed_lot_recalc"] == "True"] boundary_count = len(boundary_rows) all_boundary_count += boundary_count has_no_trade_case = buy_count == 0 all_lots_main_ready = buy_count > 0 and boundary_count == 0 and all( r["include_in_strict_closed_lot_recalc"] == "True" for r in rows ) exclude_reason = "" if has_no_trade_case: exclude_reason = "NO_BUY_MARKET_GATE_CLOSED_OR_NO_ENTRY_SIGNAL" elif boundary_count > 0: exclude_reason = "HAS_RETURN_STAT_HELD_BOUNDARY_LOT" elif not all_lots_main_ready: exclude_reason = "LOT_SCOPE_CHECK_NOT_ALL_PASS" else: exclude_reason = "STRICT_CLOSED_CASE_RETURN_READY_CANDIDATE" net_contribution = sum(clean_float(r["net_account_contribution_after_cost_pct"]) for r in rows if r["include_in_strict_closed_case_return"] == "True") aux_net_contribution = sum(clean_float(r["net_account_contribution_after_cost_pct"]) for r in aux_rows) gross_source_contribution = sum(clean_float(r["source_gross_account_contribution_pct"]) for r in rows if r["lot_status"] == "CLOSED_BY_AI_SELL") gross_integer_contribution = sum(clean_float(r["gross_integer_account_contribution_pct"]) for r in rows if r["lot_status"] == "CLOSED_BY_AI_SELL") final_nav = 1.0 + net_contribution if all_lots_main_ready else "" success = bool(all_lots_main_ready and clean_float(final_nav) > 1.0) if all_lots_main_ready: included_case_count += 1 success_count += 1 if success else 0 main_returns.append(net_contribution) nav_path = [1.0] nav = 1.0 for r in rows: nav += clean_float(r["net_account_contribution_after_cost_pct"]) nav_path.append(nav) peak = nav_path[0] max_dd = 0.0 for value in nav_path: peak = max(peak, value) if peak: max_dd = min(max_dd, value / peak - 1.0) main_drawdowns.append(max_dd) for r in aux_rows: aux_lot_count += 1 ret = clean_float(r["net_lot_return_after_cost_pct"]) aux_lot_returns.append(ret) if ret > 0: aux_lot_win_count += 1 scope_row = { "case_id": case_id, "entry_trade_date": str(case.entry_trade_date), "signal_trade_date": str(case.signal_trade_date), "selection_bucket": str(case.selection_bucket), "source_case_status": str(case.case_status), "market_gate_status": str(case.market_gate_status), "buy_lot_count": buy_count, "closed_lot_count": closed_count, "boundary_lot_count": boundary_count, "include_in_strict_closed_case_return": str(all_lots_main_ready), "include_in_aux_lot_recalc": str(len(aux_rows) > 0), "scope_status": "STRICT_CLOSED_CASE_RETURN_READY_CANDIDATE" if all_lots_main_ready else "RETURN_STAT_HELD_BOUNDARY_TABLE", "exclude_or_boundary_reason": exclude_reason, "source_image_board_path": f"../{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md", "source_story_board_path": f"../{SOURCE_RUN_ID}/cases/{case_id}/case_story_board.md", } summary_row = { **scope_row, "gross_source_closed_lot_account_contribution_pct": f"{gross_source_contribution:.8f}", "gross_integer_closed_lot_account_contribution_pct": f"{gross_integer_contribution:.8f}", "net_main_case_account_contribution_after_cost_pct": f"{net_contribution:.8f}" if all_lots_main_ready else "", "net_aux_closed_lot_account_contribution_after_cost_pct": f"{aux_net_contribution:.8f}", "final_nav_after_cost_pct": f"{clean_float(final_nav):.8f}" if all_lots_main_ready else "", "case_success_after_cost_flag": str(success) if all_lots_main_ready else "", } case_scope_rows.append(scope_row) case_summary_rows.append(summary_row) summary = { "strict_closed_case_count": included_case_count, "strict_closed_case_success_count": success_count, "strict_closed_case_success_rate": success_count / included_case_count if included_case_count else None, "strict_closed_case_net_account_contribution_sum": sum(main_returns), "strict_closed_case_average_net_account_contribution": sum(main_returns) / included_case_count if included_case_count else None, "strict_closed_case_max_drawdown_event_based": min(main_drawdowns) if main_drawdowns else None, "aux_closed_lot_count": aux_lot_count, "aux_closed_lot_win_count": aux_lot_win_count, "aux_closed_lot_win_rate": aux_lot_win_count / aux_lot_count if aux_lot_count else None, "aux_closed_lot_average_net_return_after_cost": sum(aux_lot_returns) / aux_lot_count if aux_lot_count else None, "boundary_lot_count": all_boundary_count, } return case_scope_rows, case_summary_rows, summary def make_boundary_rows(case_scope_rows: list[dict], lot_rows: list[dict]) -> list[dict]: rows: list[dict] = [] seq = 0 for row in case_scope_rows: if row["include_in_strict_closed_case_return"] != "True": seq += 1 rows.append( { "boundary_id": f"BOUNDARY-{seq:04d}", "boundary_level": "CASE", "case_id": row["case_id"], "trade_lot_id": "", "symbol": "", "status": row["scope_status"], "reason": row["exclude_or_boundary_reason"], "main_return_policy": "EXCLUDED_FROM_STRICT_CLOSED_CASE_RETURN", "aux_lot_policy": "AUX_LOT_ALLOWED_IF_CLOSED_AND_CHECKED", "evidence_path": row["source_image_board_path"], } ) for row in lot_rows: if row["scope_status"] == "RETURN_STAT_HELD_BOUNDARY_TABLE": seq += 1 rows.append( { "boundary_id": f"BOUNDARY-{seq:04d}", "boundary_level": "LOT", "case_id": row["case_id"], "trade_lot_id": row["trade_lot_id"], "symbol": row["symbol"], "status": row["lot_status"], "reason": row["exclude_or_boundary_reason"], "main_return_policy": "EXCLUDED_FROM_ALL_MAIN_RETURN_AND_WIN_RATE", "aux_lot_policy": "NOT_INCLUDED_UNLESS_REAL_SELL_EXISTS_AND_REVIEW_PASSES", "evidence_path": row["buy_evidence_image_path"] or row["sell_evidence_image_path"], } ) return rows def check_limit_rate_policy(order_rows: list[dict]) -> tuple[bool, str]: errors: list[str] = [] group_counts: dict[str, int] = defaultdict(int) for row in order_rows: symbol = row["symbol"] group = symbol_market_group(symbol) expected_rate = float(CONFIG["tradeability_policy"]["limit_rate_rule"][group]) actual_rate = clean_float(row.get("limit_rate"), default=-1.0) group_counts[group] += 1 if abs(actual_rate - expected_rate) > 0.0000001: errors.append(f"{row['return_order_id']} {symbol} limit_rate={actual_rate} expected={expected_rate}") continue prev_close = clean_float(row.get("prev_close"), default=0.0) upper_limit = clean_float(row.get("upper_limit_price"), default=0.0) lower_limit = clean_float(row.get("lower_limit_price"), default=0.0) if prev_close > 0: expected_upper = round(prev_close * (1 + expected_rate), 2) expected_lower = round(prev_close * (1 - expected_rate), 2) if abs(upper_limit - expected_upper) > 0.005 or abs(lower_limit - expected_lower) > 0.005: errors.append( f"{row['return_order_id']} {symbol} upper/lower={upper_limit}/{lower_limit} " f"expected={expected_upper}/{expected_lower}" ) detail_counts = ", ".join(f"{k}={v}" for k, v in sorted(group_counts.items())) if errors: return False, "; ".join(errors[:20]) return True, f"limit-rate policy matches order ledger; {detail_counts}." def check_image_board_links() -> tuple[bool, str]: board_paths = [ROOT / "case_image_board.md"] case_root = ROOT / "cases" if case_root.exists(): board_paths.extend(sorted(case_root.glob("*/case_image_board.md"))) errors: list[str] = [] checked = 0 markdown_link = re.compile(r"\[[^\]]+\]\(([^)]+)\)") for board in board_paths: if not board.exists(): errors.append(f"missing board: {board.relative_to(ROOT).as_posix()}") continue text = board.read_text(encoding="utf-8") for target in markdown_link.findall(text): if target.startswith(("http://", "https://", "mailto:", "#")): continue path_part = target.split("#", 1)[0].strip() if not path_part: continue checked += 1 resolved = (board.parent / path_part).resolve() if not resolved.exists(): errors.append(f"{board.relative_to(ROOT).as_posix()} -> {target}") if errors: return False, "; ".join(errors[:20]) return True, f"local links reachable in root/case image boards; checked={checked}." def make_self_check( case_scope_rows: list[dict], lot_rows: list[dict], order_rows: list[dict], summary_stats: dict, db_diagnostics: list[str], ) -> tuple[list[dict], dict]: checks: list[dict] = [] def add(check_id: str, status: bool, detail: str) -> None: checks.append({"check_id": check_id, "status": "PASS" if status else "FAIL", "detail": detail}) add("CONFIG_FROZEN", (ROOT / "return_stat_config.json").exists(), "return_stat_config.json 已生成。") add("CASE_SCOPE_7_CASES", len(case_scope_rows) == 7, f"case scope 行数={len(case_scope_rows)}。") add("LOT_SCOPE_14_LOTS", len(lot_rows) == 14, f"lot scope 行数={len(lot_rows)}。") add("ORDER_LEDGER_26_EVENTS", len(order_rows) == 26, f"return order ledger 行数={len(order_rows)}。") main_cases = [r for r in case_scope_rows if r["include_in_strict_closed_case_return"] == "True"] add("MAIN_CASES_HAVE_NO_BOUNDARY_LOT", all(int(r["boundary_lot_count"]) == 0 for r in main_cases), f"主口径 case 数={len(main_cases)}。") add("BOUNDARY_LOTS_EXCLUDED", all(r["include_in_strict_closed_lot_recalc"] == "False" for r in lot_rows if r["lot_status"] != "CLOSED_BY_AI_SELL"), "非真实 SELL / 数据缺口 lot 均排除。") add("AUX_LOTS_CLOSED_ONLY", all(r["lot_status"] == "CLOSED_BY_AI_SELL" for r in lot_rows if r["include_in_strict_closed_lot_recalc"] == "True"), "辅助 lot 口径仅包含 CLOSED_BY_AI_SELL。") add("T1_PASS_FOR_INCLUDED_LOTS", all(r["t1_check_status"] == "T1_PASS" for r in lot_rows if r["include_in_strict_closed_lot_recalc"] == "True"), "纳入口径 lot 均满足 T+1。") add("LOOKAHEAD_PASS_FOR_INCLUDED_LOTS", all(r["lookahead_check_status"] == "LOOKAHEAD_PASS" for r in lot_rows if r["include_in_strict_closed_lot_recalc"] == "True"), "纳入口径 lot 无未来函数标记。") add("EVIDENCE_PASS_FOR_INCLUDED_LOTS", all(r["evidence_check_status"] == "EVIDENCE_PASS" for r in lot_rows if r["include_in_strict_closed_lot_recalc"] == "True"), "纳入口径 lot 均有买卖证据路径。") add("INTEGER_LOT_PASS_FOR_INCLUDED_LOTS", all(r["integer_lot_check_status"] == "INTEGER_LOT_PASS" for r in lot_rows if r["include_in_strict_closed_lot_recalc"] == "True"), "纳入口径 lot 均可按整数手成交。") add("TRADEABILITY_PASS_FOR_INCLUDED_LOTS", all(r["tradeability_check_status"] == "TRADEABILITY_PASS" for r in lot_rows if r["include_in_strict_closed_lot_recalc"] == "True"), "纳入口径 lot 的买卖订单均通过 1 分钟 / 涨跌停可成交检查。") add("DB_DIAGNOSTICS_EMPTY", not db_diagnostics, "; ".join(db_diagnostics) if db_diagnostics else "源库连接和查询未记录错误。") add("RETURN_STAT_READY_FALSE", True, "本 run 只生成 RETURN_STAT_READY_CANDIDATE；执行审核通过前 return_stat_ready=false。") add("STRICT_CASE_SUMMARY_LAYERED", summary_stats["strict_closed_case_count"] < 7, "主口径与边界表分层保留，未把 7 个 case 全部强行纳入主收益。") add("AUX_LOT_COUNT_EXPECTED", summary_stats["aux_closed_lot_count"] == 12, f"辅助闭合 lot 数={summary_stats['aux_closed_lot_count']}。") limit_rate_pass, limit_rate_detail = check_limit_rate_policy(order_rows) add("LIMIT_RATE_POLICY_MATCHES_ORDER_LEDGER", limit_rate_pass, limit_rate_detail) board_links_pass, board_links_detail = check_image_board_links() add("IMAGE_BOARD_LOCAL_LINKS_REACHABLE", board_links_pass, board_links_detail) failed = [c for c in checks if c["status"] != "PASS"] self_check = { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": now_iso(), "status": "PASS_FOR_RETURN_STAT_READY_CANDIDATE_EXEC_REVIEW_REQUIRED" if not failed else "FAIL_HELD_FOR_REPAIR", "check_count": len(checks), "fail_count": len(failed), "return_stat_ready": False, "execution_review_required": True, } return checks, self_check def write_boards(summary_stats: dict, case_scope_rows: list[dict], lot_rows: list[dict]) -> None: main_case_rows = [r for r in case_scope_rows if r["include_in_strict_closed_case_return"] == "True"] boundary_case_rows = [r for r in case_scope_rows if r["include_in_strict_closed_case_return"] != "True"] lines = [ "# 无忌 RETURN_STAT_READY 准备 run 图片审核入口", "", "本入口只用于小样本收益统计准备执行审核，不是完整 2023-2026 收益结论。", "", "## 当前状态", "", "| 项目 | 读数 |", "|---|---:|", f"| 已审核来源 case | {len(case_scope_rows)} |", f"| 主口径候选 case | {len(main_case_rows)} |", f"| 边界 / 排除 case | {len(boundary_case_rows)} |", f"| 辅助闭合 lot | {summary_stats['aux_closed_lot_count']} |", f"| 边界 lot | {summary_stats['boundary_lot_count']} |", "", "## 主口径候选 case", "", "| case_id | 纳入状态 | 源图片入口 |", "|---|---|---|", ] for row in main_case_rows: lines.append(f"| `{row['case_id']}` | `STRICT_CLOSED_CASE_RETURN_READY_CANDIDATE` | [{row['case_id']}]({row['source_image_board_path']}) |") lines.extend(["", "## 边界 / 排除 case", "", "| case_id | 原因 | 源图片入口 |", "|---|---|---|"]) for row in boundary_case_rows: lines.append(f"| `{row['case_id']}` | {row['exclude_or_boundary_reason']} | [{row['case_id']}]({row['source_image_board_path']}) |") lines.extend( [ "", "## lot 边界", "", "| lot | case | symbol | 状态 | 处理 |", "|---|---|---|---|---|", ] ) for row in lot_rows: if row["scope_status"] == "RETURN_STAT_HELD_BOUNDARY_TABLE": lines.append( f"| `{row['trade_lot_id']}` | `{row['case_id']}` | `{row['symbol']}` | `{row['lot_status']}` | 不进入主收益 / 成功率 / 胜率 |" ) lines.extend( [ "", "## 审核提醒", "", "1. 人工先看本入口，再进入源 case 图片板核对买卖点。", "2. `WINDOW_END_VALUATION_ONLY` 和 `EXIT_DATA_GAP_HELD` 未被强行转成 SELL。", "3. 执行审核通过前不得引用完整 baseline 收益率、成功率、胜率或回撤。", ] ) (ROOT / "case_image_board.md").write_text("\n".join(lines) + "\n", encoding="utf-8") story = [ "# 无忌 RETURN_STAT_READY 准备 run 文字追溯入口", "", f"- source_run_id: `{SOURCE_RUN_ID}`", f"- design_id: `{DESIGN_ID}`", f"- design_audit_id: `{DESIGN_AUDIT_ID}`", "- 结论边界：本 run 只形成小样本 `RETURN_STAT_READY_CANDIDATE`，需执行审核。", "", "## 主要产物", "", "- `return_stat_config.md/json`：费用、滑点、整数手、涨跌停口径冻结。", "- `return_stat_case_scope.csv`：7 个 case 的纳入 / 排除原因。", "- `return_stat_lot_scope.csv`：14 个 lot 的纳入 / 排除和成本检查。", "- `return_stat_order_ledger.csv`：叠加成本、滑点、整数手和可成交检查的订单账本。", "- `return_stat_case_summary.csv`：case 级 gross / net / 边界状态。", "- `return_stat_boundary_table.csv`：不进入主收益口径的边界样本。", "- `return_stat_self_check.*`：执行自检。", ] (ROOT / "case_story_board.md").write_text("\n".join(story) + "\n", encoding="utf-8") lot_by_case: dict[str, list[dict]] = defaultdict(list) for row in lot_rows: lot_by_case[row["case_id"]].append(row) for row in case_scope_rows: case_id = row["case_id"] case_dir = ROOT / "cases" / case_id case_dir.mkdir(parents=True, exist_ok=True) source_board = f"../../../{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md" case_lines = [ f"# {case_id} RETURN_STAT 图片审核入口", "", "本页是收益统计准备 run 的 case 级入口，图片证据沿用已审核源结果包。", "", "| 项目 | 内容 |", "|---|---|", f"| case_id | `{case_id}` |", f"| 主收益口径 | `{row['include_in_strict_closed_case_return']}` |", f"| 辅助 lot 复算 | `{row['include_in_aux_lot_recalc']}` |", f"| scope_status | `{row['scope_status']}` |", f"| 纳入 / 排除原因 | {row['exclude_or_boundary_reason']} |", f"| 源图片板 | [{case_id}]({source_board}) |", "", "## lot 处理", "", "| lot | symbol | 状态 | 主口径 | 辅助口径 | 买入图 | 卖出图 |", "|---|---|---|---|---|---|---|", ] for lot in lot_by_case.get(case_id, []): buy_img = lot["buy_evidence_image_path"] sell_img = lot["sell_evidence_image_path"] buy_link = f"[买入图](../../../{SOURCE_RUN_ID}/{buy_img})" if buy_img else "" sell_link = f"[卖出图](../../../{SOURCE_RUN_ID}/{sell_img})" if sell_img else "" case_lines.append( f"| `{lot['trade_lot_id']}` | `{lot['symbol']}` | `{lot['lot_status']}` | `{lot['include_in_strict_closed_case_return']}` | `{lot['include_in_strict_closed_lot_recalc']}` | {buy_link} | {sell_link} |" ) if not lot_by_case.get(case_id): case_lines.append("| 无交易 lot | | `NO_BUY_MARKET_GATE_CLOSED_OR_NO_ENTRY_SIGNAL` | `False` | `False` | | |") case_lines.extend( [ "", "## 审核说明", "", "1. 本页不复制源图片，统一链接到已审核源结果包，避免生成重复图片和口径漂移。", "2. `WINDOW_END_VALUATION_ONLY`、`EXIT_DATA_GAP_HELD` 或无买入 case 不进入主收益 / 成功率口径。", "3. 执行审核通过前，本页所有收益字段均为候选材料，不是正式 baseline 结论。", ] ) (case_dir / "case_image_board.md").write_text("\n".join(case_lines) + "\n", encoding="utf-8") case_story = [ f"# {case_id} RETURN_STAT 文字追溯", "", f"- source case image board: `ana-data/result/{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md`", f"- scope_status: `{row['scope_status']}`", f"- boundary_reason: {row['exclude_or_boundary_reason']}", "- 文字页只用于追溯；人工审核第一入口仍是 case_image_board.md。", ] (case_dir / "case_story_board.md").write_text("\n".join(case_story) + "\n", encoding="utf-8") def write_config_docs() -> None: CONFIG["config_frozen_at"] = now_iso() write_json(ROOT / "return_stat_config.json", CONFIG) md = [ "# RETURN_STAT 准备 run 配置冻结", "", f"- run_id: `{RUN_ID}`", f"- source_run_id: `{SOURCE_RUN_ID}`", f"- design_audit_id: `{DESIGN_AUDIT_ID}`", "- 状态：`RETURN_STAT_READY_CANDIDATE_EXECUTION_REVIEW_REQUIRED`", "- `return_stat_ready`: false", "", "## 费用和滑点", "", "| 项目 | 冻结值 | 来源 / 说明 |", "|---|---:|---|", "| 初始资金 | 1,000,000 CNY | 本 run 主资金规模；另输出 100,000 / 10,000,000 敏感性配置 |", "| 买入佣金 | 0.00025 | 沿用源 `run_config` 的 `commission_rate_each_side` |", "| 卖出佣金 | 0.00025 | 沿用源 `run_config` 的 `commission_rate_each_side` |", "| 卖出印花税：2023-08-27 及以前 | 0.001 | 本 run 冻结历史卖出侧口径；扩样前需审核员复核官方依据 |", "| 卖出印花税：2023-08-28 起 | 0.0005 | 本 run 冻结历史卖出侧口径；扩样前需审核员复核官方依据 |", "| 过户费 | 买卖各 0.00001 | 本 run 冻结敏感性口径；扩样前需审核员复核官方依据 |", "| 滑点 | 买卖各 5 bps | 内部保守敏感性设置，不是笔记原文规则 |", "", "## 整数手和资金残余", "", "每笔按 `position_delta_pct * initial_cash_cny` 得到预算，向下取整到 100 股；未使用预算留在现金中。若不足 100 股，标记 `INTEGER_LOT_NOT_EXECUTABLE_HELD`。", "", "## 涨跌停不可成交", "", "使用本地 MySQL `a_share_minute_price` 和 `a_share_daily_price` 检查订单决策分钟、前收、涨跌停估算价格和成交量。涨停买入锁死、跌停卖出锁死、分钟数据缺失或价格不在 bar 范围内均不得进入主收益口径。", "", "## 结论边界", "", "本配置只支持当前 7 个代表性案例的小样本收益统计准备执行审核。执行审核通过前不得转写完整 baseline 收益率、成功率、胜率或回撤。", ] (ROOT / "return_stat_config.md").write_text("\n".join(md) + "\n", encoding="utf-8") def write_summary(summary_stats: dict) -> None: data = { "schema_version": "1.0", "run_id": RUN_ID, "case_matter_id": CASE_MATTER_ID, "design_id": DESIGN_ID, "design_audit_id": DESIGN_AUDIT_ID, "source_run_id": SOURCE_RUN_ID, "source_audit_ids": SOURCE_AUDIT_IDS, "generated_at": now_iso(), "stage": "RETURN_STAT_READY_CANDIDATE_EXECUTION_REVIEW_REQUIRED", "return_stat_ready": False, "execution_review_required": True, "scope": "7 representative cases only; no full 2023-2026 conclusion.", "summary_stats": summary_stats, "conclusion_boundary": "Small-sample return-stat preparation only. Do not cite as complete baseline return, success rate, win rate, drawdown, or strategy effectiveness before execution review passes.", } write_json(ROOT / "return_stat_summary.json", data) success_rate = summary_stats["strict_closed_case_success_rate"] aux_win_rate = summary_stats["aux_closed_lot_win_rate"] md = [ "# RETURN_STAT 准备 run 摘要", "", f"- run_id: `{RUN_ID}`", f"- source_run_id: `{SOURCE_RUN_ID}`", "- 状态：`RETURN_STAT_READY_CANDIDATE_EXECUTION_REVIEW_REQUIRED`", "- `return_stat_ready`: false", "", "## 小样本准备读数", "", "| 指标 | 数值 |", "|---|---:|", f"| 主口径候选 case 数 | {summary_stats['strict_closed_case_count']} |", f"| 主口径候选 case 成功数 | {summary_stats['strict_closed_case_success_count']} |", f"| 主口径候选 case 成功率候选值 | {success_rate:.6f} |", f"| 主口径候选 case net 贡献合计 | {summary_stats['strict_closed_case_net_account_contribution_sum']:.8f} |", f"| 主口径候选 case 平均 net 贡献 | {summary_stats['strict_closed_case_average_net_account_contribution']:.8f} |", f"| 事件级最大回撤候选值 | {summary_stats['strict_closed_case_max_drawdown_event_based']:.8f} |", f"| 辅助闭合 lot 数 | {summary_stats['aux_closed_lot_count']} |", f"| 辅助闭合 lot 胜率候选值 | {aux_win_rate:.6f} |", f"| 边界 lot 数 | {summary_stats['boundary_lot_count']} |", "", "## 使用限制", "", "这些读数只是执行审核材料中的候选读数。审核通过前不得引用为完整 baseline 收益率、成功率、胜率或回撤；即使审核通过，也只代表当前 7 个小样本的严格闭合子集。", ] (ROOT / "return_stat_summary.md").write_text("\n".join(md) + "\n", encoding="utf-8") def write_self_check_docs(check_rows: list[dict], self_check: dict) -> None: write_csv(ROOT / "return_stat_self_check_items.csv", check_rows, ["check_id", "status", "detail"]) write_json(ROOT / "return_stat_self_check.json", self_check) lines = [ "# RETURN_STAT 准备 run 自检", "", f"- status: `{self_check['status']}`", f"- check_count: {self_check['check_count']}", f"- fail_count: {self_check['fail_count']}", "- return_stat_ready: false", "", "| check_id | status | detail |", "|---|---|---|", ] for row in check_rows: lines.append(f"| `{row['check_id']}` | `{row['status']}` | {row['detail']} |") (ROOT / "return_stat_self_check.md").write_text("\n".join(lines) + "\n", encoding="utf-8") def write_manifest() -> None: files = [] for path in sorted(ROOT.rglob("*")): if not path.is_file(): continue rel = path.relative_to(ROOT).as_posix() if rel == "manifest.json": continue files.append({"path": rel, "exists": True, "size": path.stat().st_size, "sha256": sha256_file(path)}) manifest = { "schema_version": "1.0", "run_id": RUN_ID, "source_run_id": SOURCE_RUN_ID, "manifest_stage": "RETURN_STAT_READY_CANDIDATE_PACKAGE_DONE", "generated_at": now_iso(), "hash_status": "size_and_sha256_recorded_for_current_artifacts_manifest_self_excluded", "overall_status": "PASS_FOR_RETURN_STAT_READY_CANDIDATE_EXEC_REVIEW_REQUIRED", "return_stat_ready": False, "execution_review_required": True, "file_count": len(files), "files": files, } write_json(ROOT / "manifest.json", manifest) def copy_source_tools() -> None: source_tools = SOURCE_ROOT / "tools" target_tools = ROOT / "tools" / "source_reference" target_tools.mkdir(parents=True, exist_ok=True) for name in ["perform_exit_ai_review.py", "run_pilot_self_check.py", "finalize_result_package.py"]: src = source_tools / name if src.exists(): shutil.copy2(src, target_tools / name) def main() -> None: ROOT.mkdir(parents=True, exist_ok=True) (ROOT / "tools").mkdir(exist_ok=True) case_index = read_csv("case_index.csv") orders = read_csv("order_ledger.csv") lots = read_csv("position_lot_ledger.csv") write_config_docs() tradeability, db_diagnostics = fetch_tradeability(orders) order_rows, _ = make_order_rows(orders, lots, tradeability) closed_by_case = lots.groupby("case_id").lot_status.apply(lambda s: all(v == "CLOSED_BY_AI_SELL" for v in s)).to_dict() case_main_candidate = {case_id: bool(value) for case_id, value in closed_by_case.items()} lot_rows, _ = make_lot_scope_rows(lots, order_rows, case_main_candidate) case_scope_rows, case_summary_rows, summary_stats = make_case_scope_and_summary(case_index, lots, lot_rows) # Recompute lot main flags after final case scope, so no-buy and boundary cases cannot leak into main. main_case_ids = {r["case_id"] for r in case_scope_rows if r["include_in_strict_closed_case_return"] == "True"} for row in lot_rows: if row["case_id"] not in main_case_ids and row["include_in_strict_closed_case_return"] == "True": row["include_in_strict_closed_case_return"] = "False" if row["include_in_strict_closed_lot_recalc"] == "True": row["scope_status"] = "STRICT_CLOSED_LOT_RECALC_ONLY" row["exclude_or_boundary_reason"] = "CASE_HAS_BOUNDARY_LOT_AUX_ONLY" case_scope_rows, case_summary_rows, summary_stats = make_case_scope_and_summary(case_index, lots, lot_rows) boundary_rows = make_boundary_rows(case_scope_rows, lot_rows) write_csv(ROOT / "return_stat_order_ledger.csv", order_rows, list(order_rows[0].keys())) write_csv(ROOT / "return_stat_lot_scope.csv", lot_rows, list(lot_rows[0].keys())) write_csv(ROOT / "return_stat_case_scope.csv", case_scope_rows, list(case_scope_rows[0].keys())) write_csv(ROOT / "return_stat_case_summary.csv", case_summary_rows, list(case_summary_rows[0].keys())) write_csv(ROOT / "return_stat_boundary_table.csv", boundary_rows, list(boundary_rows[0].keys())) write_summary(summary_stats) write_boards(summary_stats, case_scope_rows, lot_rows) check_rows, self_check = make_self_check(case_scope_rows, lot_rows, order_rows, summary_stats, db_diagnostics) write_self_check_docs(check_rows, self_check) copy_source_tools() write_manifest() if __name__ == "__main__": main()