from __future__ import annotations

import csv
import hashlib
import json
import math
import os
import re
import shutil
from collections import Counter, defaultdict
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path

import pandas as pd
import pymysql
from PIL import Image, ImageDraw, ImageFont


RUN_ID = "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001"
TASK_ID = "ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609"
DESIGN_ID = "DESIGN-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609"
DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-DESIGN-002"
SUPP_DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-SUPP-DESIGN-003"
SOURCE_RUN_ID = "RUN-ANA-WUJI-FULL-2023-2026-20260608-001"
SOURCE_EXEC_AUDIT_ID = "AUDIT-ANA-WUJI-FULL-2023-2026-20260608-EXEC-REREVIEW-002"
ISSUE_ID = "ANA-ISSUE-WUJI-STRICT-SELL-ROLLING-GAP-20260609-001"

ROOT = Path(__file__).resolve().parents[1]
SOURCE_ROOT = ROOT.parent / SOURCE_RUN_ID
PROJECT_ROOT = ROOT.parents[2]

OBSERVATION_TRADING_DAYS = 10
FAST_BREAKOUT_MAX_MINUTES = 10
GAIN_3 = 0.03
GAIN_5 = 0.05
GAIN_8 = 0.08
SUPPORT_BREAK_TOLERANCE = 0.003
OPEN_DOWN_TOLERANCE = 0.003
BREAKOUT_TOLERANCE = 0.001
ROLLING_NEAR_MA5_PCT = 0.012
ROLLING_VOLUME_MULTIPLE = 1.20
ROLLING_START_TIME = "10:40:00"
ROLLING_END_TIME = "14:40:00"
MAX_TRANCHES_PER_CASE_SYMBOL = 5
POSITION_PCT_PER_TRANCHE = 0.04
AUDIT_SAMPLE_MIN_RATIO = 0.20
AUDIT_SAMPLE_TARGET_RATIO = 0.30


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 write_json(path: Path, data: dict) -> None:
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def normalize_date(value) -> str:
    return pd.to_datetime(value).strftime("%Y-%m-%d")


def normalize_time(value) -> str:
    text = str(value)
    if " " in text:
        text = text.split(" ")[-1]
    if "." in text:
        text = text.split(".")[0]
    parts = text.split(":")
    if len(parts) == 2:
        return f"{int(parts[0]):02d}:{int(parts[1]):02d}:00"
    if len(parts) >= 3:
        return f"{int(parts[0]):02d}:{int(parts[1]):02d}:{int(float(parts[2])):02d}"
    return text


def read_mysql_password() -> str:
    env = os.environ.get("TIANXIA_MYSQL_PASSWORD") or os.environ.get("MYSQL_PWD")
    if env:
        return env
    index = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md")
    text = 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_mysql_password(),
        database="tianxia",
        charset="utf8mb4",
        connect_timeout=5,
        read_timeout=120,
        write_timeout=120,
    )


def font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
    for name in ["msyh.ttc", "simhei.ttf", "simsun.ttc"]:
        path = Path("C:/Windows/Fonts") / name
        if path.exists():
            return ImageFont.truetype(str(path), size)
    return ImageFont.load_default()


FONT_18 = font(18)
FONT_20 = font(20)
FONT_24 = font(24)
FONT_30 = font(30)


@dataclass
class LotInput:
    source_lot_id: str
    source_order_id: str
    case_id: str
    symbol: str
    entry_trade_date: str
    entry_time: str
    entry_price: float
    position_pct: float
    tranche_index: int
    sellable_from_trade_date: str
    candidate_id: str
    variant_id: str
    decision_reason_cn: str
    evidence_image_path: str
    is_rolling: bool = False
    parent_lot_id: str = ""


def safe_float(value, default=math.nan) -> float:
    try:
        if pd.isna(value):
            return default
        return float(value)
    except Exception:
        return default


def pct(value: float) -> str:
    if pd.isna(value):
        return ""
    return f"{value * 100:.2f}%"


def load_inputs() -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, list[LotInput]]:
    order = pd.read_csv(SOURCE_ROOT / "order_ledger.csv", encoding="utf-8-sig")
    lots = pd.read_csv(SOURCE_ROOT / "position_lot_ledger.csv", encoding="utf-8-sig")
    selected = pd.read_csv(SOURCE_ROOT / "selected_candidate_ledger.csv", encoding="utf-8-sig")

    for df, date_cols in [
        (order, ["trade_date", "t1_sellable_from_trade_date"]),
        (lots, ["entry_trade_date", "sellable_from_trade_date", "exit_trade_date"]),
        (selected, ["entry_trade_date", "signal_trade_date"]),
    ]:
        for col in date_cols:
            if col in df.columns:
                df[col] = df[col].map(lambda v: "" if pd.isna(v) or str(v) == "" else normalize_date(v))
        if "trade_time" in df.columns:
            df["trade_time"] = df["trade_time"].map(lambda v: "" if pd.isna(v) else normalize_time(v))
        if "entry_time" in df.columns:
            df["entry_time"] = df["entry_time"].map(lambda v: "" if pd.isna(v) else normalize_time(v))

    buy_order_by_id = order[order["action"] == "BUY"].set_index("order_id").to_dict("index")
    source_lots: list[LotInput] = []
    for _, row in lots.iterrows():
        order_row = buy_order_by_id.get(row["order_id"], {})
        source_lots.append(
            LotInput(
                source_lot_id=row["trade_lot_id"],
                source_order_id=row["order_id"],
                case_id=row["case_id"],
                symbol=row["symbol"],
                entry_trade_date=row["entry_trade_date"],
                entry_time=row["entry_time"],
                entry_price=safe_float(row["entry_price"]),
                position_pct=safe_float(row["position_pct"], POSITION_PCT_PER_TRANCHE),
                tranche_index=int(safe_float(row["tranche_index"], 1)),
                sellable_from_trade_date=row["sellable_from_trade_date"],
                candidate_id=str(order_row.get("candidate_id", "")),
                variant_id=str(order_row.get("variant_id", "")),
                decision_reason_cn=str(order_row.get("decision_reason_cn", "")),
                evidence_image_path=str(order_row.get("evidence_image_path", "")),
            )
        )
    return order, selected, lots, source_lots


def fetch_trade_calendar() -> list[str]:
    with get_conn() as conn:
        dates = 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,
        )
    return [normalize_date(v) for v in dates["trade_date"].tolist()]


def previous_trade_dates(trade_dates: list[str], trade_date: str, count: int) -> list[str]:
    if trade_date not in trade_dates:
        return []
    idx = trade_dates.index(trade_date)
    return trade_dates[max(0, idx - count) : idx]


def window_dates(trade_dates: list[str], entry_date: str, sellable_from: str) -> list[str]:
    if entry_date not in trade_dates or sellable_from not in trade_dates:
        return []
    entry_idx = trade_dates.index(entry_date)
    start_idx = trade_dates.index(sellable_from)
    end_idx = min(len(trade_dates) - 1, entry_idx + OBSERVATION_TRADING_DAYS)
    if start_idx > end_idx:
        return []
    return trade_dates[start_idx : end_idx + 1]


def build_fetch_maps(lots: list[LotInput], trade_dates: list[str]) -> tuple[dict[str, set[str]], set[str], str, str]:
    date_symbols: dict[str, set[str]] = defaultdict(set)
    symbols: set[str] = set()
    all_dates: set[str] = set()
    for lot in lots:
        symbols.add(lot.symbol)
        all_dates.add(lot.entry_trade_date)
        date_symbols[lot.entry_trade_date].add(lot.symbol)
        for d in window_dates(trade_dates, lot.entry_trade_date, lot.sellable_from_trade_date):
            date_symbols[d].add(lot.symbol)
            all_dates.add(d)
            for prev in previous_trade_dates(trade_dates, d, 5):
                date_symbols[prev].add(lot.symbol)
                all_dates.add(prev)
    min_date = min(all_dates)
    max_date = max(all_dates)
    return date_symbols, symbols, min_date, max_date


def fetch_market_data(date_symbols: dict[str, set[str]], symbols: set[str], min_date: str, max_date: str) -> tuple[pd.DataFrame, pd.DataFrame]:
    minute_parts: list[pd.DataFrame] = []
    with get_conn() as conn:
        for trade_date in sorted(date_symbols):
            day_symbols = sorted(date_symbols[trade_date])
            if not day_symbols:
                continue
            ph = ",".join(["%s"] * len(day_symbols))
            minute_parts.append(
                pd.read_sql(
                    f"""
                    SELECT trade_date, trade_time, symbol, open_price, high_price, low_price, close_price, volume, amount
                    FROM a_share_minute_price
                    WHERE trade_date = %s AND symbol IN ({ph})
                    ORDER BY symbol, trade_date, trade_time
                    """,
                    conn,
                    params=[trade_date, *day_symbols],
                )
            )
        sym_list = sorted(symbols)
        ph = ",".join(["%s"] * len(sym_list))
        daily = pd.read_sql(
            f"""
            SELECT trade_date, symbol, open_price, high_price, low_price, close_price, volume, amount
            FROM a_share_daily_price
            WHERE symbol IN ({ph}) AND trade_date BETWEEN %s AND %s
            ORDER BY symbol, trade_date
            """,
            conn,
            params=[*sym_list, min_date, max_date],
        )
    minute = pd.concat(minute_parts, ignore_index=True) if minute_parts else pd.DataFrame()
    if not minute.empty:
        minute["trade_date"] = minute["trade_date"].map(normalize_date)
        minute["trade_time"] = minute["trade_time"].map(normalize_time)
        for col in ["open_price", "high_price", "low_price", "close_price", "volume", "amount"]:
            minute[col] = pd.to_numeric(minute[col], errors="coerce")
    if not daily.empty:
        daily["trade_date"] = daily["trade_date"].map(normalize_date)
        for col in ["open_price", "high_price", "low_price", "close_price", "volume", "amount"]:
            daily[col] = pd.to_numeric(daily[col], errors="coerce")
        daily = daily.sort_values(["symbol", "trade_date"]).reset_index(drop=True)
        daily["ma5_close"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(5, min_periods=3).mean())
    return minute, daily


def make_lookup(df: pd.DataFrame) -> dict[tuple[str, str], pd.DataFrame]:
    lookup = {}
    if df.empty:
        return lookup
    for (symbol, trade_date), g in df.groupby(["symbol", "trade_date"]):
        lookup[(symbol, trade_date)] = g.sort_values("trade_time").reset_index(drop=True)
    return lookup


def daily_lookup_map(daily: pd.DataFrame) -> dict[tuple[str, str], dict]:
    return {(r["symbol"], r["trade_date"]): r.to_dict() for _, r in daily.iterrows()}


def time_minutes(start: str, end: str) -> float:
    base = datetime(2000, 1, 1)
    a = datetime.strptime(start, "%H:%M:%S")
    b = datetime.strptime(end, "%H:%M:%S")
    return ((base.replace(hour=b.hour, minute=b.minute, second=b.second) - base.replace(hour=a.hour, minute=a.minute, second=a.second)).total_seconds() / 60)


def support_price_for_lot(lot: LotInput, minute_lookup: dict[tuple[str, str], pd.DataFrame]) -> tuple[float, str]:
    day = minute_lookup.get((lot.symbol, lot.entry_trade_date), pd.DataFrame())
    if day.empty:
        return lot.entry_price, "ENTRY_PRICE_FALLBACK"
    prior = day[day["trade_time"] <= lot.entry_time]
    if prior.empty:
        prior = day.head(1)
    open_price = safe_float(day.iloc[0]["open_price"], lot.entry_price)
    prior_low = safe_float(prior["low_price"].min(), lot.entry_price)
    support = min(open_price, prior_low, lot.entry_price)
    return support, "ENTRY_DAY_OPEN_OR_PRE_BUY_LOW"


def day_prev_close(symbol: str, trade_date: str, trade_dates: list[str], daily_lookup: dict[tuple[str, str], dict]) -> float:
    prevs = previous_trade_dates(trade_dates, trade_date, 1)
    if not prevs:
        return math.nan
    return safe_float(daily_lookup.get((symbol, prevs[-1]), {}).get("close_price"))


def previous_first_minute_avg(symbol: str, trade_date: str, trade_dates: list[str], minute_lookup: dict[tuple[str, str], pd.DataFrame]) -> float:
    vols = []
    for d in previous_trade_dates(trade_dates, trade_date, 5):
        day = minute_lookup.get((symbol, d), pd.DataFrame())
        if not day.empty:
            vols.append(safe_float(day.iloc[0]["volume"]))
    return float(pd.Series(vols).mean()) if vols else math.nan


def find_row_at_or_before(day: pd.DataFrame, t: str) -> pd.Series | None:
    frame = day[day["trade_time"] <= t]
    if frame.empty:
        return None
    return frame.iloc[-1]


def high_rising_three_day(symbol: str, trade_date: str, day: pd.DataFrame, trade_dates: list[str], daily_lookup: dict[tuple[str, str], dict]) -> tuple[bool, str]:
    prevs = previous_trade_dates(trade_dates, trade_date, 2)
    if len(prevs) < 2:
        return False, "THREE_DAY_SOURCE_GAP"
    high1 = safe_float(daily_lookup.get((symbol, prevs[0]), {}).get("high_price"))
    high2 = safe_float(daily_lookup.get((symbol, prevs[1]), {}).get("high_price"))
    morning = day[day["trade_time"] <= "10:40:00"]
    if morning.empty or math.isnan(high1) or math.isnan(high2):
        return False, "THREE_DAY_SOURCE_GAP"
    high3 = safe_float(morning["high_price"].max())
    ok = (high2 >= high1 * (1 - BREAKOUT_TOLERANCE)) and (high3 >= high2 * (1 - BREAKOUT_TOLERANCE))
    detail = f"前两日高点 {high1:.4f}->{high2:.4f}，当日10:40前高点 {high3:.4f}"
    return ok, detail


def add_signal(signals: list[dict], *, lot: LotInput, trade_date: str, time: str, signal_type: str, code_action: str, human_action: str, reason: str, price: float, gain_pct: float, support_price: float, support_type: str, extra: dict | None = None) -> dict:
    extra = extra or {}
    signal_id = f"SIG-{RUN_ID}-{len(signals)+1:06d}"
    row = {
        "signal_id": signal_id,
        "run_id": RUN_ID,
        "source_run_id": SOURCE_RUN_ID,
        "source_lot_id": lot.source_lot_id,
        "source_order_id": lot.source_order_id,
        "case_id": lot.case_id,
        "candidate_id": lot.candidate_id,
        "symbol": lot.symbol,
        "entry_trade_date": lot.entry_trade_date,
        "entry_time": lot.entry_time,
        "entry_price": f"{lot.entry_price:.4f}",
        "sellable_from_trade_date": lot.sellable_from_trade_date,
        "observation_trade_date": trade_date,
        "candidate_time": time,
        "signal_type": signal_type,
        "code_suggested_action": code_action,
        "human_decision_action": human_action,
        "human_decision_reason_cn": reason,
        "decision_operator": "case_analysis.analyst",
        "decision_time": now_iso(),
        "decision_source": "ANALYST_RULE_REPLAY_WITH_FROZEN_THRESHOLDS",
        "action_price": "" if math.isnan(price) else f"{price:.4f}",
        "gain_pct": "" if math.isnan(gain_pct) else f"{gain_pct:.8f}",
        "entry_support_price": "" if math.isnan(support_price) else f"{support_price:.4f}",
        "entry_support_policy": support_type,
        "chart_path": "",
        "chart_reason_rendered": "False",
        "storyboard_reason_rendered": "False",
        "v1_stats_inclusion_status": "INCLUDED_IF_ORDER_EXECUTED" if human_action in {"SELL", "BUY_ROLLING_LOW"} else "BOUNDARY_OR_HOLD_NOT_A_RETURN_EVENT",
    }
    row.update(extra)
    signals.append(row)
    return row


def evaluate_lot(lot: LotInput, trade_dates: list[str], minute_lookup: dict[tuple[str, str], pd.DataFrame], daily_lookup: dict[tuple[str, str], dict], allow_rolling: bool = True) -> tuple[list[dict], dict | None, list[dict]]:
    signals: list[dict] = []
    rolling_signals: list[dict] = []
    support_price, support_type = support_price_for_lot(lot, minute_lookup)
    dates = window_dates(trade_dates, lot.entry_trade_date, lot.sellable_from_trade_date)
    if not dates:
        add_signal(
            signals,
            lot=lot,
            trade_date=lot.sellable_from_trade_date,
            time="",
            signal_type="SELL_REVIEW_DATA_GAP_HELD",
            code_action="REVIEW_HELD",
            human_action="REVIEW_HELD",
            reason="交易日窗口缺失，无法按 V1 卖点规则裁决，保留待审。",
            price=math.nan,
            gain_pct=math.nan,
            support_price=support_price,
            support_type=support_type,
        )
        return signals, None, rolling_signals

    trend_first3: tuple[str, str, float] | None = None
    fast_watch_added = False
    above8_added = False
    final_sell: dict | None = None

    for d in dates:
        day = minute_lookup.get((lot.symbol, d), pd.DataFrame())
        if day.empty:
            add_signal(
                signals,
                lot=lot,
                trade_date=d,
                time="",
                signal_type="SELL_REVIEW_DATA_GAP_HELD",
                code_action="REVIEW_HELD",
                human_action="REVIEW_HELD",
                reason=f"{d} 分钟线缺失，无法裁决精准卖点，保留待审。",
                price=math.nan,
                gain_pct=math.nan,
                support_price=support_price,
                support_type=support_type,
            )
            continue

        prev_close = day_prev_close(lot.symbol, d, trade_dates, daily_lookup)
        # Hard support breach: buy reason support line is broken after T+1.
        breach = day[day["low_price"] <= support_price * (1 - SUPPORT_BREAK_TOLERANCE)]
        if not breach.empty:
            r = breach.iloc[0]
            gain = safe_float(r["close_price"]) / lot.entry_price - 1
            final_sell = add_signal(
                signals,
                lot=lot,
                trade_date=d,
                time=r["trade_time"],
                signal_type="SELL_SUPPORT_REASON_BROKEN",
                code_action="SELL",
                human_action="SELL",
                reason=f"买入理由支撑线 {support_price:.4f} 被跌破，按止损铁律卖出。",
                price=safe_float(r["close_price"]),
                gain_pct=gain,
                support_price=support_price,
                support_type=support_type,
            )
            break

        first = day.iloc[0]
        if not math.isnan(prev_close):
            first_gain = safe_float(first["close_price"]) / prev_close - 1
            avg_first_vol = previous_first_minute_avg(lot.symbol, d, trade_dates, minute_lookup)
            vol_ratio = safe_float(first["volume"]) / avg_first_vol if avg_first_vol and not math.isnan(avg_first_vol) and avg_first_vol > 0 else math.nan
            if not math.isnan(vol_ratio) and vol_ratio >= 10 and 0.01 <= first_gain <= 0.02:
                final_sell = add_signal(
                    signals,
                    lot=lot,
                    trade_date=d,
                    time=first["trade_time"],
                    signal_type="SELL_OPEN_VOLUME_STALL",
                    code_action="SELL",
                    human_action="SELL",
                    reason=f"开盘一分钟量比 {vol_ratio:.2f}，涨幅 {pct(first_gain)}，放量滞涨，直接卖出。",
                    price=safe_float(first["close_price"]),
                    gain_pct=safe_float(first["close_price"]) / lot.entry_price - 1,
                    support_price=support_price,
                    support_type=support_type,
                    extra={"open_volume_ratio": f"{vol_ratio:.4f}", "open_gain_pct": f"{first_gain:.8f}"},
                )
                break

        row1040 = find_row_at_or_before(day, "10:40:00")
        if row1040 is not None and not math.isnan(prev_close):
            day_open = safe_float(day.iloc[0]["open_price"])
            morning = day[day["trade_time"] <= "10:40:00"]
            morning_high = safe_float(morning["high_price"].max())
            open_down = day_open < prev_close * (1 - OPEN_DOWN_TOLERANCE)
            if open_down and morning_high <= day_open * (1 + BREAKOUT_TOLERANCE):
                final_sell = add_signal(
                    signals,
                    lot=lot,
                    trade_date=d,
                    time=row1040["trade_time"],
                    signal_type="SELL_REBOUND_FAIL_OPEN",
                    code_action="SELL",
                    human_action="SELL",
                    reason=f"开盘下跌后反弹最高 {morning_high:.4f} 未有效突破开盘价 {day_open:.4f}，直接卖出。",
                    price=safe_float(row1040["close_price"]),
                    gain_pct=safe_float(row1040["close_price"]) / lot.entry_price - 1,
                    support_price=support_price,
                    support_type=support_type,
                )
                break
            if open_down:
                morning = morning.copy()
                morning["ma5m"] = morning["close_price"].rolling(5, min_periods=3).mean()
                seg1 = morning[(morning["trade_time"] >= "09:35:00") & (morning["trade_time"] <= "10:00:00")]
                seg2 = morning[(morning["trade_time"] > "10:00:00") & (morning["trade_time"] <= "10:40:00")]
                if not seg1.empty and not seg2.empty:
                    seg1_fail = safe_float(seg1["high_price"].max()) <= safe_float(seg1["ma5m"].max()) * (1 + BREAKOUT_TOLERANCE)
                    seg2_fail = safe_float(seg2["high_price"].max()) <= safe_float(seg2["ma5m"].max()) * (1 + BREAKOUT_TOLERANCE)
                    if seg1_fail and seg2_fail:
                        final_sell = add_signal(
                            signals,
                            lot=lot,
                            trade_date=d,
                            time=row1040["trade_time"],
                            signal_type="SELL_REBOUND_FAIL_MA_TWICE",
                            code_action="SELL",
                            human_action="SELL",
                            reason="开盘下跌后两段反弹均未有效突破 5 分钟均线，按精准卖点卖出。",
                            price=safe_float(row1040["close_price"]),
                            gain_pct=safe_float(row1040["close_price"]) / lot.entry_price - 1,
                            support_price=support_price,
                            support_type=support_type,
                        )
                        break

            high_ok, high_detail = high_rising_three_day(lot.symbol, d, day, trade_dates, daily_lookup)
            if not high_ok and high_detail != "THREE_DAY_SOURCE_GAP":
                final_sell = add_signal(
                    signals,
                    lot=lot,
                    trade_date=d,
                    time=row1040["trade_time"],
                    signal_type="SELL_THREE_DAY_HIGH_NOT_RISING",
                    code_action="SELL",
                    human_action="SELL",
                    reason=f"10:40 前三日高点未逐步抬高（{high_detail}），直接卖出。",
                    price=safe_float(row1040["close_price"]),
                    gain_pct=safe_float(row1040["close_price"]) / lot.entry_price - 1,
                    support_price=support_price,
                    support_type=support_type,
                )
                break

        # Trend take-profit state machine. It logs HOLD decisions as formal decisions too.
        for _, r in day.iterrows():
            gain_high = safe_float(r["high_price"]) / lot.entry_price - 1
            gain_low = safe_float(r["low_price"]) / lot.entry_price - 1
            gain_close = safe_float(r["close_price"]) / lot.entry_price - 1
            if trend_first3 is None and gain_high >= GAIN_3:
                trend_first3 = (d, r["trade_time"], safe_float(r["close_price"]))
                continue
            if trend_first3 is None:
                continue
            first3_date, first3_time, _ = trend_first3
            same_day_fast = d == first3_date and time_minutes(first3_time, r["trade_time"]) <= FAST_BREAKOUT_MAX_MINUTES
            if not fast_watch_added and gain_high >= GAIN_5 and same_day_fast:
                add_signal(
                    signals,
                    lot=lot,
                    trade_date=d,
                    time=r["trade_time"],
                    signal_type="TREND_FAST_BREAKOUT_5_WATCH",
                    code_action="HOLD_WATCH",
                    human_action="HOLD_WATCH",
                    reason="3% 以上后快速冲过 5%，按新版基版要求观望不卖。",
                    price=safe_float(r["close_price"]),
                    gain_pct=gain_close,
                    support_price=support_price,
                    support_type=support_type,
                    extra={"trend_first3_time": first3_time, "fast_breakout_minutes": f"{time_minutes(first3_time, r['trade_time']):.2f}"},
                )
                fast_watch_added = True
                continue
            if trend_first3 and not fast_watch_added and d == first3_date and time_minutes(first3_time, r["trade_time"]) > FAST_BREAKOUT_MAX_MINUTES and GAIN_3 <= gain_close < GAIN_5:
                final_sell = add_signal(
                    signals,
                    lot=lot,
                    trade_date=d,
                    time=r["trade_time"],
                    signal_type="SELL_TREND_3_TO_5_GRADUAL",
                    code_action="SELL",
                    human_action="SELL",
                    reason="上涨不是快速冲 5%，趋势性上涨在 3%-5% 区间，卖出兑现。",
                    price=safe_float(r["close_price"]),
                    gain_pct=gain_close,
                    support_price=support_price,
                    support_type=support_type,
                    extra={"trend_first3_time": first3_time},
                )
                break
            if fast_watch_added and not above8_added and gain_high >= GAIN_8:
                add_signal(
                    signals,
                    lot=lot,
                    trade_date=d,
                    time=r["trade_time"],
                    signal_type="TREND_ABOVE_8_HOLD",
                    code_action="HOLD_ABOVE_8",
                    human_action="HOLD_ABOVE_8",
                    reason="快速冲过 5% 后继续超过 8%，按新版基版要求强势持有不卖。",
                    price=safe_float(r["close_price"]),
                    gain_pct=gain_close,
                    support_price=support_price,
                    support_type=support_type,
                )
                above8_added = True
                continue
            if fast_watch_added and not above8_added and gain_low < GAIN_5 and gain_close >= GAIN_3:
                final_sell = add_signal(
                    signals,
                    lot=lot,
                    trade_date=d,
                    time=r["trade_time"],
                    signal_type="SELL_TREND_PULLBACK_3_TO_5",
                    code_action="SELL",
                    human_action="SELL",
                    reason="快速冲过 5% 后回落到 3%-5% 区间，按新版基版卖出。",
                    price=safe_float(r["close_price"]),
                    gain_pct=gain_close,
                    support_price=support_price,
                    support_type=support_type,
                )
                break
        if final_sell:
            break

    if final_sell is None:
        if any(s["human_decision_action"] in {"HOLD_WATCH", "HOLD_ABOVE_8"} for s in signals):
            last_date = dates[-1]
            day = minute_lookup.get((lot.symbol, last_date), pd.DataFrame())
            last_price = safe_float(day.iloc[-1]["close_price"]) if not day.empty else lot.entry_price
            add_signal(
                signals,
                lot=lot,
                trade_date=last_date,
                time=day.iloc[-1]["trade_time"] if not day.empty else "",
                signal_type="SELL_WINDOW_END_HOLD_BOUNDARY",
                code_action="REVIEW_HELD",
                human_action="REVIEW_HELD",
                reason="观察窗口结束仍未出现明确 V1 卖点，保留为窗口末待审边界，不强行卖出。",
                price=last_price,
                gain_pct=last_price / lot.entry_price - 1,
                support_price=support_price,
                support_type=support_type,
            )

    if allow_rolling and any(s["human_decision_action"] in {"HOLD_WATCH", "HOLD_ABOVE_8"} for s in signals):
        rolling_signal = find_rolling_low_buy(lot, dates, trade_dates, minute_lookup, daily_lookup, support_price, support_type)
        if rolling_signal:
            rolling_signals.append(rolling_signal)
        else:
            add_rolling_signal(
                rolling_signals,
                lot=lot,
                trade_date=dates[-1],
                time="",
                signal_type="ROLLING_LOW_BUY_REVIEW_HELD",
                human_action="REVIEW_HELD",
                reason="已出现趋势观察但观察窗口内未找到五日线附近止跌放量低吸确认，保留待审。",
                price=math.nan,
                ma5=math.nan,
                volume_ratio=math.nan,
            )

    return signals, final_sell, rolling_signals


def add_rolling_signal(rows: list[dict], *, lot: LotInput, trade_date: str, time: str, signal_type: str, human_action: str, reason: str, price: float, ma5: float, volume_ratio: float) -> dict:
    row = {
        "rolling_signal_id": f"ROLL-{RUN_ID}-{len(rows)+1:06d}",
        "run_id": RUN_ID,
        "source_lot_id": lot.source_lot_id,
        "case_id": lot.case_id,
        "candidate_id": lot.candidate_id,
        "symbol": lot.symbol,
        "entry_trade_date": lot.entry_trade_date,
        "rolling_trade_date": trade_date,
        "rolling_time": time,
        "signal_type": signal_type,
        "human_decision_action": human_action,
        "human_decision_reason_cn": reason,
        "decision_operator": "case_analysis.analyst",
        "decision_time": now_iso(),
        "rolling_price": "" if math.isnan(price) else f"{price:.4f}",
        "ma5_close": "" if math.isnan(ma5) else f"{ma5:.4f}",
        "near_ma5_pct": "" if math.isnan(price) or math.isnan(ma5) or ma5 == 0 else f"{abs(price - ma5) / ma5:.8f}",
        "volume_ratio_vs_prev20m": "" if math.isnan(volume_ratio) else f"{volume_ratio:.4f}",
        "chart_path": "",
        "chart_reason_rendered": "False",
        "storyboard_reason_rendered": "False",
    }
    rows.append(row)
    return row


def find_rolling_low_buy(lot: LotInput, dates: list[str], trade_dates: list[str], minute_lookup: dict[tuple[str, str], pd.DataFrame], daily_lookup: dict[tuple[str, str], dict], support_price: float, support_type: str) -> dict | None:
    for d in dates[1:]:
        day = minute_lookup.get((lot.symbol, d), pd.DataFrame())
        if day.empty:
            continue
        daily_row = daily_lookup.get((lot.symbol, d), {})
        ma5 = safe_float(daily_row.get("ma5_close"))
        if math.isnan(ma5) or ma5 <= 0:
            continue
        window = day[(day["trade_time"] >= ROLLING_START_TIME) & (day["trade_time"] <= ROLLING_END_TIME)].copy()
        if window.empty:
            continue
        window["prev_close"] = window["close_price"].shift(1)
        window["vol20"] = window["volume"].rolling(20, min_periods=5).mean().shift(1)
        candidates = window[
            (abs(window["close_price"] - ma5) / ma5 <= ROLLING_NEAR_MA5_PCT)
            & (window["close_price"] >= window["prev_close"])
            & (window["volume"] >= window["vol20"] * ROLLING_VOLUME_MULTIPLE)
        ]
        if not candidates.empty:
            r = candidates.iloc[0]
            vol_ratio = safe_float(r["volume"]) / safe_float(r["vol20"]) if safe_float(r["vol20"]) > 0 else math.nan
            return add_rolling_signal(
                [],
                lot=lot,
                trade_date=d,
                time=r["trade_time"],
                signal_type="ADD_ROLLING_LOW_BUY",
                human_action="BUY_ROLLING_LOW",
                reason=f"趋势观察后回到五日线附近，接近 MA5 {ma5:.4f}，分钟止跌且量能放大 {vol_ratio:.2f} 倍，按滚动渣男低吸一份仓。",
                price=safe_float(r["close_price"]),
                ma5=ma5,
                volume_ratio=vol_ratio,
            )[0] if False else {
                "rolling_signal_id": "",
                "run_id": RUN_ID,
                "source_lot_id": lot.source_lot_id,
                "case_id": lot.case_id,
                "candidate_id": lot.candidate_id,
                "symbol": lot.symbol,
                "entry_trade_date": lot.entry_trade_date,
                "rolling_trade_date": d,
                "rolling_time": r["trade_time"],
                "signal_type": "ADD_ROLLING_LOW_BUY",
                "human_decision_action": "BUY_ROLLING_LOW",
                "human_decision_reason_cn": f"趋势观察后回到五日线附近，接近 MA5 {ma5:.4f}，分钟止跌且量能放大 {vol_ratio:.2f} 倍，按滚动渣男低吸一份仓。",
                "decision_operator": "case_analysis.analyst",
                "decision_time": now_iso(),
                "rolling_price": f"{safe_float(r['close_price']):.4f}",
                "ma5_close": f"{ma5:.4f}",
                "near_ma5_pct": f"{abs(safe_float(r['close_price']) - ma5) / ma5:.8f}",
                "volume_ratio_vs_prev20m": f"{vol_ratio:.4f}",
                "chart_path": "",
                "chart_reason_rendered": "False",
                "storyboard_reason_rendered": "False",
            }
    return None


def draw_line_chart(day: pd.DataFrame, signal: dict, out_path: Path, title: str, reason: str, point_time: str, point_price: float, ma5: float | None = None) -> None:
    out_path.parent.mkdir(parents=True, exist_ok=True)
    w, h = 1500, 900
    left, top, right, bottom = 90, 90, 1040, 760
    img = Image.new("RGB", (w, h), "#fffdf7")
    draw = ImageDraw.Draw(img)
    draw.text((40, 24), title, fill="#111111", font=FONT_30)
    if day.empty:
        draw.text((100, 200), "分钟数据缺失，无法绘图。", fill="#b00020", font=FONT_24)
        img.save(out_path)
        return
    prices = day["close_price"].astype(float).tolist()
    lows = day["low_price"].astype(float).tolist()
    highs = day["high_price"].astype(float).tolist()
    y_min = min(lows + [point_price])
    y_max = max(highs + [point_price])
    entry_price = safe_float(signal.get("entry_price"))
    thresholds = []
    if not math.isnan(entry_price):
        thresholds = [entry_price, entry_price * (1 + GAIN_3), entry_price * (1 + GAIN_5), entry_price * (1 + GAIN_8)]
        y_min = min(y_min, min(thresholds))
        y_max = max(y_max, max(thresholds))
    if ma5 and not math.isnan(ma5):
        y_min = min(y_min, ma5)
        y_max = max(y_max, ma5)
    span = max(y_max - y_min, 0.01)
    y_min -= span * 0.08
    y_max += span * 0.08
    span = y_max - y_min

    def x_at(i: int) -> float:
        return left + i * (right - left) / max(len(prices) - 1, 1)

    def y_at(v: float) -> float:
        return bottom - (v - y_min) * (bottom - top) / span

    draw.rectangle((left, top, right, bottom), outline="#333333", width=2)
    for frac in [0, 0.25, 0.5, 0.75, 1.0]:
        y = top + frac * (bottom - top)
        draw.line((left, y, right, y), fill="#e5e5e5")
    points = [(x_at(i), y_at(p)) for i, p in enumerate(prices)]
    if len(points) > 1:
        draw.line(points, fill="#1f77b4", width=3)
    time_to_idx = {t: i for i, t in enumerate(day["trade_time"].tolist())}
    if point_time in time_to_idx:
        i = time_to_idx[point_time]
    else:
        i = min(range(len(day)), key=lambda k: abs(safe_float(day.iloc[k]["close_price"]) - point_price))
    px, py = x_at(i), y_at(point_price)
    draw.line((px, top, px, bottom), fill="#c00000", width=3)
    draw.ellipse((px - 7, py - 7, px + 7, py + 7), fill="#c00000")
    draw.text((px + 10, max(top + 5, py - 22)), f"{point_time} {point_price:.2f}", fill="#c00000", font=FONT_20)
    if not math.isnan(entry_price):
        for label, value, color in [
            ("买入价", entry_price, "#444444"),
            ("3%", entry_price * 1.03, "#2ca02c"),
            ("5%", entry_price * 1.05, "#ff7f0e"),
            ("8%", entry_price * 1.08, "#9467bd"),
        ]:
            y = y_at(value)
            draw.line((left, y, right, y), fill=color, width=2)
            draw.text((right + 8, y - 12), f"{label} {value:.2f}", fill=color, font=FONT_18)
    if ma5 and not math.isnan(ma5):
        y = y_at(ma5)
        draw.line((left, y, right, y), fill="#17becf", width=2)
        draw.text((right + 8, y - 12), f"日线MA5 {ma5:.2f}", fill="#008899", font=FONT_18)
    times = day["trade_time"].tolist()
    for t in ["09:30:00", "10:40:00", "14:40:00", "15:00:00"]:
        if t in time_to_idx:
            x = x_at(time_to_idx[t])
            draw.line((x, bottom, x, bottom + 10), fill="#333333")
            draw.text((x - 35, bottom + 14), t[:5], fill="#333333", font=FONT_18)
    side_x = 1080
    draw.text((side_x, 100), "人工裁决", fill="#111111", font=FONT_24)
    lines = [
        f"动作：{signal.get('human_decision_action', '')}",
        f"信号：{signal.get('signal_type', signal.get('rolling_signal_type', ''))}",
        f"股票：{signal.get('symbol', '')}",
        f"案例：{signal.get('case_id', '')}",
        f"触发：{signal.get('observation_trade_date', signal.get('rolling_trade_date', ''))} {point_time}",
        f"价格：{point_price:.4f}",
        "理由：",
    ]
    y = 145
    for line in lines:
        draw.text((side_x, y), line, fill="#111111", font=FONT_20)
        y += 34
    for part in wrap_cn(reason, 19):
        draw.text((side_x, y), part, fill="#111111", font=FONT_20)
        y += 32
    draw.text((side_x, 780), "audit_view：图用于人工复核，不反推当时决策。", fill="#666666", font=FONT_18)
    img.save(out_path)


def wrap_cn(text: str, width: int) -> list[str]:
    text = str(text)
    return [text[i : i + width] for i in range(0, len(text), width)] or [""]


def render_charts(signals: list[dict], rolling_rows: list[dict], minute_lookup: dict[tuple[str, str], pd.DataFrame], daily_lookup: dict[tuple[str, str], dict]) -> None:
    for sig in signals:
        date = sig["observation_trade_date"]
        day = minute_lookup.get((sig["symbol"], date), pd.DataFrame())
        price = safe_float(sig.get("action_price"), safe_float(sig.get("entry_price")))
        t = sig["candidate_time"] or (day.iloc[-1]["trade_time"] if not day.empty else "")
        out_rel = f"cases/{sig['case_id']}/img/v1_sell_signal_{sig['signal_id']}.png"
        draw_line_chart(
            day,
            sig,
            ROOT / out_rel,
            f"V1精准卖点/趋势裁决：{sig['case_id']} {sig['symbol']}",
            sig["human_decision_reason_cn"],
            t,
            price,
        )
        sig["chart_path"] = out_rel
        sig["chart_reason_rendered"] = "True"
        sig["storyboard_reason_rendered"] = "True"

    for i, row in enumerate(rolling_rows, 1):
        if not row.get("rolling_signal_id"):
            row["rolling_signal_id"] = f"ROLL-{RUN_ID}-{i:06d}"
        date = row["rolling_trade_date"]
        day = minute_lookup.get((row["symbol"], date), pd.DataFrame())
        price = safe_float(row.get("rolling_price"))
        ma5 = safe_float(row.get("ma5_close"))
        t = row["rolling_time"] or (day.iloc[-1]["trade_time"] if not day.empty else "")
        out_rel = f"cases/{row['case_id']}/img/v1_rolling_low_buy_{row['rolling_signal_id']}.png"
        signal_like = {
            "case_id": row["case_id"],
            "symbol": row["symbol"],
            "human_decision_action": row["human_decision_action"],
            "signal_type": row["signal_type"],
            "rolling_trade_date": date,
            "entry_price": row.get("rolling_price", ""),
        }
        draw_line_chart(
            day,
            signal_like,
            ROOT / out_rel,
            f"V1滚动渣男低吸裁决：{row['case_id']} {row['symbol']}",
            row["human_decision_reason_cn"],
            t,
            price if not math.isnan(price) else safe_float(day.iloc[-1]["close_price"], 0) if not day.empty else 0,
            ma5,
        )
        row["chart_path"] = out_rel
        row["chart_reason_rendered"] = "True"
        row["storyboard_reason_rendered"] = "True"


def make_order(order_id: str, lot: LotInput, action: str, trade_date: str, trade_time: str, price: float, position_delta_pct: float, reason: str, evidence: str, source_lot_id: str = "", exit_signal_type: str = "") -> dict:
    return {
        "order_id": order_id,
        "run_id": RUN_ID,
        "source_run_id": SOURCE_RUN_ID,
        "case_id": lot.case_id,
        "candidate_id": lot.candidate_id,
        "variant_id": lot.variant_id or "V1_STRICT_SELL_ROLLING",
        "symbol": lot.symbol,
        "trade_date": trade_date,
        "trade_time": trade_time,
        "action": action,
        "price": f"{price:.4f}",
        "position_delta_pct": f"{position_delta_pct:.8f}",
        "tranche_index": lot.tranche_index,
        "planned_tranche_count": MAX_TRANCHES_PER_CASE_SYMBOL,
        "decision_reason_cn": reason,
        "evidence_image_path": evidence,
        "t1_sellable_from_trade_date": lot.sellable_from_trade_date if action == "BUY" else "",
        "lookahead_violation_flag": "False",
        "source_lot_id": source_lot_id,
        "source_order_id": lot.source_order_id,
        "exit_signal_type": exit_signal_type,
    }


def build_ledgers(source_lots: list[LotInput], sell_signals: list[dict], final_sells: dict[str, dict], rolling_rows: list[dict], trade_dates: list[str], minute_lookup: dict[tuple[str, str], pd.DataFrame], daily_lookup: dict[tuple[str, str], dict]) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame, list[LotInput]]:
    orders: list[dict] = []
    lots: list[dict] = []
    new_lots: list[LotInput] = []
    order_seq = 1
    lot_seq = 1
    tranche_counts: Counter[tuple[str, str]] = Counter()

    for lot in source_lots:
        tranche_counts[(lot.case_id, lot.symbol)] = max(tranche_counts[(lot.case_id, lot.symbol)], lot.tranche_index)
        oid = f"ORD-{RUN_ID}-{order_seq:06d}"
        order_seq += 1
        orders.append(
            make_order(
                oid,
                lot,
                "BUY",
                lot.entry_trade_date,
                lot.entry_time,
                lot.entry_price,
                lot.position_pct,
                "沿用已审核旧包买入裁决，V1 只返修卖点和滚动低吸；旧买入图片作为来源证据。",
                lot.evidence_image_path,
            )
        )
        sell = final_sells.get(lot.source_lot_id)
        status = "V1_OPEN_OR_BOUNDARY_HELD"
        exit_date = exit_time = exit_price = ""
        lot_ret = account_ret = ""
        if sell and sell["human_decision_action"] == "SELL":
            price = safe_float(sell["action_price"])
            oid_sell = f"ORD-{RUN_ID}-{order_seq:06d}"
            order_seq += 1
            orders.append(
                make_order(
                    oid_sell,
                    lot,
                    "SELL",
                    sell["observation_trade_date"],
                    sell["candidate_time"],
                    price,
                    lot.position_pct,
                    sell["human_decision_reason_cn"],
                    sell["chart_path"],
                    source_lot_id=lot.source_lot_id,
                    exit_signal_type=sell["signal_type"],
                )
            )
            status = "CLOSED_BY_V1_SELL"
            exit_date = sell["observation_trade_date"]
            exit_time = sell["candidate_time"]
            exit_price = f"{price:.4f}"
            ret = price / lot.entry_price - 1
            lot_ret = f"{ret:.8f}"
            account_ret = f"{ret * lot.position_pct:.8f}"
        lots.append(
            {
                "strict_lot_id": f"LOT-{RUN_ID}-{lot_seq:06d}",
                "source_lot_id": lot.source_lot_id,
                "source_order_id": lot.source_order_id,
                "case_id": lot.case_id,
                "symbol": lot.symbol,
                "entry_trade_date": lot.entry_trade_date,
                "entry_time": lot.entry_time,
                "entry_price": f"{lot.entry_price:.4f}",
                "position_pct": f"{lot.position_pct:.8f}",
                "tranche_index": lot.tranche_index,
                "sellable_from_trade_date": lot.sellable_from_trade_date,
                "lot_status": status,
                "exit_trade_date": exit_date,
                "exit_time": exit_time,
                "exit_price": exit_price,
                "lot_return_pct": lot_ret,
                "account_return_contribution_pct": account_ret,
                "parent_lot_id": lot.parent_lot_id,
                "lot_source_type": "SOURCE_BUY" if not lot.is_rolling else "ROLLING_LOW_BUY",
            }
        )
        lot_seq += 1

    confirmed_rolls = [r for r in rolling_rows if r["human_decision_action"] == "BUY_ROLLING_LOW"]
    for row in confirmed_rolls:
        key = (row["case_id"], row["symbol"])
        if tranche_counts[key] >= MAX_TRANCHES_PER_CASE_SYMBOL:
            continue
        tranche_counts[key] += 1
        roll_price = safe_float(row["rolling_price"])
        if math.isnan(roll_price):
            continue
        sellable = next_trade_date(trade_dates, row["rolling_trade_date"])
        roll_lot = LotInput(
            source_lot_id=row["rolling_signal_id"],
            source_order_id=row["rolling_signal_id"],
            case_id=row["case_id"],
            symbol=row["symbol"],
            entry_trade_date=row["rolling_trade_date"],
            entry_time=row["rolling_time"],
            entry_price=roll_price,
            position_pct=POSITION_PCT_PER_TRANCHE,
            tranche_index=tranche_counts[key],
            sellable_from_trade_date=sellable or row["rolling_trade_date"],
            candidate_id=row["candidate_id"],
            variant_id="V1_ROLLING_LOW_BUY",
            decision_reason_cn=row["human_decision_reason_cn"],
            evidence_image_path=row["chart_path"],
            is_rolling=True,
            parent_lot_id=row["source_lot_id"],
        )
        new_lots.append(roll_lot)
        oid = f"ORD-{RUN_ID}-{order_seq:06d}"
        order_seq += 1
        orders.append(
            make_order(
                oid,
                roll_lot,
                "BUY",
                roll_lot.entry_trade_date,
                roll_lot.entry_time,
                roll_lot.entry_price,
                roll_lot.position_pct,
                roll_lot.decision_reason_cn,
                roll_lot.evidence_image_path,
            )
        )

    # Evaluate rolling lots once, without recursive additional rolling.
    rolling_sell_signals: list[dict] = []
    rolling_final_sells: dict[str, dict] = {}
    for roll_lot in new_lots:
        sigs, final_sell, _ = evaluate_lot(roll_lot, trade_dates, minute_lookup, daily_lookup, allow_rolling=False)
        rolling_sell_signals.extend(sigs)
        if final_sell:
            rolling_final_sells[roll_lot.source_lot_id] = final_sell
    assign_sell_signal_ids(rolling_sell_signals, len(sell_signals) + 1)
    sell_signals.extend(rolling_sell_signals)
    render_charts(rolling_sell_signals, [], minute_lookup, daily_lookup)
    for roll_lot in new_lots:
        sell = rolling_final_sells.get(roll_lot.source_lot_id)
        status = "V1_OPEN_OR_BOUNDARY_HELD"
        exit_date = exit_time = exit_price = ""
        lot_ret = account_ret = ""
        if sell and sell["human_decision_action"] == "SELL":
            price = safe_float(sell["action_price"])
            oid_sell = f"ORD-{RUN_ID}-{order_seq:06d}"
            order_seq += 1
            orders.append(
                make_order(
                    oid_sell,
                    roll_lot,
                    "SELL",
                    sell["observation_trade_date"],
                    sell["candidate_time"],
                    price,
                    roll_lot.position_pct,
                    sell["human_decision_reason_cn"],
                    sell["chart_path"],
                    source_lot_id=roll_lot.source_lot_id,
                    exit_signal_type=sell["signal_type"],
                )
            )
            status = "CLOSED_BY_V1_SELL"
            exit_date = sell["observation_trade_date"]
            exit_time = sell["candidate_time"]
            exit_price = f"{price:.4f}"
            ret = price / roll_lot.entry_price - 1
            lot_ret = f"{ret:.8f}"
            account_ret = f"{ret * roll_lot.position_pct:.8f}"
        lots.append(
            {
                "strict_lot_id": f"LOT-{RUN_ID}-{lot_seq:06d}",
                "source_lot_id": roll_lot.source_lot_id,
                "source_order_id": roll_lot.source_order_id,
                "case_id": roll_lot.case_id,
                "symbol": roll_lot.symbol,
                "entry_trade_date": roll_lot.entry_trade_date,
                "entry_time": roll_lot.entry_time,
                "entry_price": f"{roll_lot.entry_price:.4f}",
                "position_pct": f"{roll_lot.position_pct:.8f}",
                "tranche_index": roll_lot.tranche_index,
                "sellable_from_trade_date": roll_lot.sellable_from_trade_date,
                "lot_status": status,
                "exit_trade_date": exit_date,
                "exit_time": exit_time,
                "exit_price": exit_price,
                "lot_return_pct": lot_ret,
                "account_return_contribution_pct": account_ret,
                "parent_lot_id": roll_lot.parent_lot_id,
                "lot_source_type": "ROLLING_LOW_BUY",
            }
        )
        lot_seq += 1

    order_df = pd.DataFrame(orders)
    lot_df = pd.DataFrame(lots)
    account_df = build_account_ledger(order_df, lot_df)
    case_df = build_case_summary(lot_df)
    return order_df, lot_df, account_df, case_df, new_lots


def next_trade_date(trade_dates: list[str], date: str) -> str:
    if date not in trade_dates:
        return ""
    idx = trade_dates.index(date)
    return trade_dates[idx + 1] if idx + 1 < len(trade_dates) else ""


def build_account_ledger(order_df: pd.DataFrame, lot_df: pd.DataFrame) -> pd.DataFrame:
    lot_by_source = lot_df.set_index("source_lot_id").to_dict("index") if not lot_df.empty else {}
    rows = []
    cash = 1.0
    open_pos = 0.0
    realized = 0.0
    seq = 1
    df = order_df.copy()
    df["_dt"] = pd.to_datetime(df["trade_date"] + " " + df["trade_time"].replace("", "00:00:00"))
    for _, order in df.sort_values(["_dt", "order_id"]).iterrows():
        pos = safe_float(order["position_delta_pct"], 0.0)
        flow = 0.0
        if order["action"] == "BUY":
            cash -= pos
            open_pos += pos
            flow = -pos
        elif order["action"] == "SELL":
            source_lot_id = order.get("source_lot_id", "")
            lot = lot_by_source.get(source_lot_id, {})
            entry_price = safe_float(lot.get("entry_price"), safe_float(order["price"]))
            sell_price = safe_float(order["price"])
            ret = sell_price / entry_price - 1 if entry_price else 0.0
            release = pos * (1 + ret)
            cash += release
            open_pos -= pos
            realized += pos * ret
            flow = release
        rows.append(
            {
                "seq": seq,
                "order_id": order["order_id"],
                "case_id": order["case_id"],
                "symbol": order["symbol"],
                "trade_date": order["trade_date"],
                "trade_time": order["trade_time"],
                "action": order["action"],
                "cash_flow_pct": f"{flow:.8f}",
                "cash_pct_after_event": f"{cash:.8f}",
                "open_position_pct_after_event": f"{open_pos:.8f}",
                "realized_return_pct_after_event": f"{realized:.8f}",
                "nav_pct_after_event": f"{cash + open_pos:.8f}",
            }
        )
        seq += 1
    return pd.DataFrame(rows)


def build_case_summary(lot_df: pd.DataFrame) -> pd.DataFrame:
    rows = []
    for case_id, g in lot_df.groupby("case_id"):
        closed = g[g["lot_status"] == "CLOSED_BY_V1_SELL"]
        unresolved = g[g["lot_status"] != "CLOSED_BY_V1_SELL"]
        account_ret = pd.to_numeric(closed["account_return_contribution_pct"], errors="coerce").fillna(0).sum()
        rows.append(
            {
                "case_id": case_id,
                "buy_lot_count": len(g),
                "closed_lot_count": len(closed),
                "unresolved_lot_count": len(unresolved),
                "account_return_closed_lots": f"{account_ret:.8f}",
                "v1_primary_strict_closed_case_flag": 1 if len(g) > 0 and len(unresolved) == 0 else 0,
                "v1_case_success_flag": 1 if len(g) > 0 and len(unresolved) == 0 and account_ret > 0 else 0,
                "v1_return_scope": "V1_PRIMARY_STRICT_CLOSED_CASE" if len(g) > 0 and len(unresolved) == 0 else "V1_RETURN_HELD_BOUNDARY_TABLE",
                "v1_boundary_reason": "" if len(unresolved) == 0 else "存在 V1 未闭合 / 待审 lot，不进入 V1 主收益口径。",
            }
        )
    return pd.DataFrame(rows).sort_values("case_id")


def build_scope_and_boundary(case_df: pd.DataFrame, lot_df: pd.DataFrame, sell_signals: list[dict], rolling_rows: list[dict]) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    case_scope = case_df.copy()
    lot_scope = lot_df.copy()
    lot_scope["v1_lot_scope"] = lot_scope["lot_status"].map(lambda s: "V1_STRICT_CLOSED_LOT_RECALC_ONLY" if s == "CLOSED_BY_V1_SELL" else "V1_RETURN_HELD_BOUNDARY_TABLE")
    boundary_rows = []
    for _, row in case_scope[case_scope["v1_primary_strict_closed_case_flag"] == 0].iterrows():
        boundary_rows.append(
            {
                "boundary_id": f"BOUND-CASE-{row['case_id']}",
                "boundary_level": "CASE",
                "case_id": row["case_id"],
                "source_lot_id": "",
                "symbol": "",
                "boundary_category": "V1_UNRESOLVED_LOT_OR_REVIEW_HELD",
                "boundary_reason": row["v1_boundary_reason"],
            }
        )
    for _, row in lot_scope[lot_scope["lot_status"] != "CLOSED_BY_V1_SELL"].iterrows():
        boundary_rows.append(
            {
                "boundary_id": f"BOUND-LOT-{row['source_lot_id']}",
                "boundary_level": "LOT",
                "case_id": row["case_id"],
                "source_lot_id": row["source_lot_id"],
                "symbol": row["symbol"],
                "boundary_category": row["lot_status"],
                "boundary_reason": "V1 观察窗口内没有真实 SELL 或存在人工待审 / 数据缺口。",
            }
        )
    if not boundary_rows:
        boundary_rows.append(
            {
                "boundary_id": "BOUND-GLOBAL-MARKET-RISK-DATA-GAP",
                "boundary_level": "GLOBAL",
                "case_id": "",
                "source_lot_id": "",
                "symbol": "",
                "boundary_category": "MARKET_RISK_MINUTE_BREADTH_DATA_GAP_HELD",
                "boundary_reason": "本地数据源未提供开盘 10 分钟全 A 下跌家数分钟级广度；市场风险卖点以数据缺口边界保留。",
            }
        )
    else:
        boundary_rows.append(
            {
                "boundary_id": "BOUND-GLOBAL-MARKET-RISK-DATA-GAP",
                "boundary_level": "GLOBAL",
                "case_id": "",
                "source_lot_id": "",
                "symbol": "",
                "boundary_category": "MARKET_RISK_MINUTE_BREADTH_DATA_GAP_HELD",
                "boundary_reason": "本地数据源未提供开盘 10 分钟全 A 下跌家数分钟级广度；市场风险卖点以数据缺口边界保留。",
            }
        )
    return case_scope, lot_scope, pd.DataFrame(boundary_rows)


def build_sampling_index(sell_signals: list[dict], rolling_rows: list[dict]) -> pd.DataFrame:
    rows = []
    combined = []
    for s in sell_signals:
        combined.append(("SELL_SIGNAL", s["human_decision_action"], s["signal_id"], s["case_id"], s["symbol"], s["chart_path"], s["human_decision_reason_cn"]))
    for r in rolling_rows:
        combined.append(("ROLLING_SIGNAL", r["human_decision_action"], r["rolling_signal_id"], r["case_id"], r["symbol"], r["chart_path"], r["human_decision_reason_cn"]))
    by_action: dict[str, list[tuple]] = defaultdict(list)
    for item in combined:
        by_action[item[1]].append(item)
    target = max(math.ceil(len(combined) * AUDIT_SAMPLE_TARGET_RATIO), math.ceil(len(combined) * AUDIT_SAMPLE_MIN_RATIO))
    selected = []
    for action, items in sorted(by_action.items()):
        take = len(items) if len(items) <= 30 else max(1, math.ceil(len(items) * AUDIT_SAMPLE_TARGET_RATIO))
        selected.extend(items[:take])
    if len(selected) < target:
        remaining = [x for x in combined if x not in selected]
        selected.extend(remaining[: target - len(selected)])
    for i, item in enumerate(selected, 1):
        rows.append(
            {
                "sample_seq": i,
                "artifact_type": item[0],
                "human_decision_action": item[1],
                "signal_id": item[2],
                "case_id": item[3],
                "symbol": item[4],
                "chart_path": item[5],
                "human_decision_reason_cn": item[6],
                "sample_policy": "30% target / 20% minimum, full check when action bucket <= 30",
            }
        )
    return pd.DataFrame(rows)


def write_config() -> None:
    config = {
        "schema_version": "1.0",
        "run_id": RUN_ID,
        "task_id": TASK_ID,
        "design_id": DESIGN_ID,
        "design_audit_id": DESIGN_AUDIT_ID,
        "supplemental_design_audit_id": SUPP_DESIGN_AUDIT_ID,
        "source_run_id": SOURCE_RUN_ID,
        "source_execution_audit_id": SOURCE_EXEC_AUDIT_ID,
        "issue_id": ISSUE_ID,
        "generated_at": now_iso(),
        "scope": "V1 strict sellpoint and rolling-low-buy repair replay over the audited full baseline source package.",
        "source_package_policy": "Read-only. Do not overwrite V0 audited package.",
        "observation_trading_days": OBSERVATION_TRADING_DAYS,
        "trend_take_profit": {
            "gain_reference": "entry_price",
            "thresholds": {"trend_low": GAIN_3, "fast_cross": GAIN_5, "strong_hold": GAIN_8},
            "fast_breakout_max_minutes": FAST_BREAKOUT_MAX_MINUTES,
            "human_decision_required": True,
            "actions": ["SELL", "HOLD_WATCH", "HOLD_ABOVE_8", "REVIEW_HELD"],
        },
        "precise_sellpoints": {
            "support_break": {"support_price_source": "ENTRY_DAY_OPEN_OR_PRE_BUY_LOW", "break_tolerance": SUPPORT_BREAK_TOLERANCE},
            "rebound_fail_open": {"open_down_tolerance": OPEN_DOWN_TOLERANCE, "breakout_tolerance": BREAKOUT_TOLERANCE},
            "rebound_fail_ma_twice": {"ma_type": "5-minute rolling close MA", "segments": ["09:35-10:00", "10:00-10:40"]},
            "open_volume_stall": {"first_minute_volume_ratio_threshold": 10, "gain_range": [0.01, 0.02]},
            "three_day_high": {"decision_time": "10:40:00", "policy": "D-2 high -> D-1 high -> current day high before 10:40 should rise"},
            "market_risk": {"minute_breadth_required": True, "data_gap_status": "MARKET_RISK_MINUTE_BREADTH_DATA_GAP_HELD"},
        },
        "rolling_low_buy": {
            "enabled": True,
            "window": [ROLLING_START_TIME, ROLLING_END_TIME],
            "near_ma5_pct": ROLLING_NEAR_MA5_PCT,
            "volume_multiple_vs_prev20m": ROLLING_VOLUME_MULTIPLE,
            "position_pct_per_tranche": POSITION_PCT_PER_TRANCHE,
            "max_tranches_per_case_symbol": MAX_TRANCHES_PER_CASE_SYMBOL,
            "human_decision_actions": ["BUY_ROLLING_LOW", "REVIEW_HELD"],
        },
        "manual_review_sampling": {
            "minimum_ratio": AUDIT_SAMPLE_MIN_RATIO,
            "target_ratio": AUDIT_SAMPLE_TARGET_RATIO,
            "small_bucket_full_check_threshold": 30,
            "must_cover_actions": ["SELL", "HOLD_WATCH", "HOLD_ABOVE_8", "REVIEW_HELD", "BUY_ROLLING_LOW"],
        },
        "return_boundary": {
            "return_stat_ready": False,
            "v1_execution_review_required_before_any_v1_performance_claim": True,
        },
    }
    write_json(ROOT / "strict_sell_rolling_run_config.json", config)
    lines = [
        "# 无忌 V1 精准卖点与滚动渣男返修执行配置冻结",
        "",
        f"- run_id：`{RUN_ID}`",
        f"- source_run_id：`{SOURCE_RUN_ID}`（只读，不覆盖）",
        f"- design_audit_id：`{DESIGN_AUDIT_ID}`",
        f"- supplemental_design_audit_id：`{SUPP_DESIGN_AUDIT_ID}`",
        f"- 观察窗口：买入后 {OBSERVATION_TRADING_DAYS} 个交易日，T+1 后才允许 SELL。",
        f"- 趋势阈值：3% / 5% / 8%，快速冲 5% 的最长窗口为 {FAST_BREAKOUT_MAX_MINUTES} 分钟。",
        "- 3% / 5% / 8% 不是机械止盈；代码只产出候选和证据，裁决字段必须保留动作与中文理由。",
        "- 滚动低吸：10:40-14:40，靠近日线 MA5，分钟止跌并放量，单次一份仓。",
        f"- 人工裁决审核抽样：最低 {AUDIT_SAMPLE_MIN_RATIO:.0%}，原则 {AUDIT_SAMPLE_TARGET_RATIO:.0%}；小桶不超过 30 条时倾向全量检查。",
        "- RETURN_STAT_READY=false；V1 执行审核通过前不得引用 V1 业绩结论。",
        "",
    ]
    (ROOT / "strict_sell_rolling_run_config.md").write_text("\n".join(lines), encoding="utf-8")


def write_rule_mapping() -> None:
    rows = [
        ["买入理由跌破就是止损线", "SELL_SUPPORT_REASON_BROKEN", "entry_day open/pre-buy low support with tolerance", "strict_sell_signal_ledger.csv"],
        ["两次反弹破不了均线", "SELL_REBOUND_FAIL_MA_TWICE", "open down + two morning rebound segments below 5m MA", "strict_sell_signal_ledger.csv"],
        ["反弹连开盘价都破不了", "SELL_REBOUND_FAIL_OPEN", "open down + morning high below open", "strict_sell_signal_ledger.csv"],
        ["开盘一分钟放量滞涨", "SELL_OPEN_VOLUME_STALL", "first-minute volume ratio >=10 and gain 1%-2%", "strict_sell_signal_ledger.csv"],
        ["10:40前三日高点未逐步抬高", "SELL_THREE_DAY_HIGH_NOT_RISING", "D-2/D-1/current morning high not rising", "strict_sell_signal_ledger.csv"],
        ["3%-5%非快速趋势上涨", "SELL_TREND_3_TO_5_GRADUAL", "3% entered but no fast 5% cross within frozen window", "strict_sell_signal_ledger.csv"],
        ["快速冲过5%观望", "TREND_FAST_BREAKOUT_5_WATCH", "hold/watch decision with chart reason", "strict_sell_signal_ledger.csv"],
        ["超过8%不动", "TREND_ABOVE_8_HOLD", "hold above 8 decision with chart reason", "strict_sell_signal_ledger.csv"],
        ["回落到3%-5%卖出", "SELL_TREND_PULLBACK_3_TO_5", "fast 5 watch then pullback into 3%-5%", "strict_sell_signal_ledger.csv"],
        ["五日线附近止跌放量反复低吸", "ADD_ROLLING_LOW_BUY", "near daily MA5 and minute stop-fall volume confirmation", "rolling_low_buy_signal_ledger.csv"],
        ["市场开盘10分钟下杀", "MARKET_RISK_MINUTE_BREADTH_DATA_GAP_HELD", "minute breadth unavailable; retained as boundary", "strict_boundary_table.csv"],
    ]
    with (ROOT / "strict_rule_mapping.csv").open("w", encoding="utf-8-sig", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["source_note_semantic", "v1_rule_code", "frozen_execution_policy", "evidence_artifact"])
        writer.writerows(rows)


def write_boards(case_df: pd.DataFrame, sell_signals: list[dict], rolling_rows: list[dict]) -> None:
    signals_by_case = defaultdict(list)
    for s in sell_signals:
        signals_by_case[s["case_id"]].append(s)
    rolls_by_case = defaultdict(list)
    for r in rolling_rows:
        rolls_by_case[r["case_id"]].append(r)

    root_lines = [
        "# 无忌 V1 精准卖点与滚动渣男返修图片入口",
        "",
        f"- run_id：`{RUN_ID}`",
        f"- 来源包：`{SOURCE_RUN_ID}`",
        f"- 当前阶段：`V1_STRICT_SELL_ROLLING_SELF_CHECK_DONE_EXEC_REVIEW_PENDING`",
        "- 说明：本入口展示 V1 精准卖点、趋势 3%/5%/8% 人工裁决、滚动低吸候选与中文理由。",
        "- 边界：RETURN_STAT_READY=false；执行审核通过前不得引用 V1 业绩结论。",
        "",
        "## 案例入口",
        "",
    ]
    for _, row in case_df.sort_values("case_id").iterrows():
        root_lines.append(
            f"- [{row['case_id']}](cases/{row['case_id']}/case_image_board.md)：scope={row['v1_return_scope']}，闭合 lot={row['closed_lot_count']}，未闭合/待审 lot={row['unresolved_lot_count']}，收益贡献={row['account_return_closed_lots']}"
        )
    (ROOT / "case_image_board.md").write_text("\n".join(root_lines) + "\n", encoding="utf-8")

    for _, row in case_df.iterrows():
        case_id = row["case_id"]
        case_dir = ROOT / "cases" / case_id
        case_dir.mkdir(parents=True, exist_ok=True)
        lines = [
            f"# {case_id} V1 精准卖点与滚动低吸图板",
            "",
            f"- 当前收益口径：`{row['v1_return_scope']}`",
            f"- 主口径标记：`{row['v1_primary_strict_closed_case_flag']}`",
            f"- 中文边界原因：{row['v1_boundary_reason'] or '所有 V1 lot 均已真实 SELL 闭合。'}",
            f"- 闭合 lot：{row['closed_lot_count']}；未闭合/待审 lot：{row['unresolved_lot_count']}",
            "",
            "## 精准卖点 / 趋势裁决",
            "",
        ]
        for sig in signals_by_case.get(case_id, []):
            rel = Path(sig["chart_path"]).relative_to(f"cases/{case_id}").as_posix() if sig.get("chart_path") else ""
            lines.extend(
                [
                    f"### {sig['signal_type']} / {sig['human_decision_action']}",
                    "",
                    f"- 时间：{sig['observation_trade_date']} {sig['candidate_time']}",
                    f"- 理由：{sig['human_decision_reason_cn']}",
                    f"- 图：![{sig['signal_id']}]({rel})",
                    "",
                ]
            )
        lines.extend(["## 滚动低吸裁决", ""])
        for roll in rolls_by_case.get(case_id, []):
            rel = Path(roll["chart_path"]).relative_to(f"cases/{case_id}").as_posix() if roll.get("chart_path") else ""
            lines.extend(
                [
                    f"### {roll['signal_type']} / {roll['human_decision_action']}",
                    "",
                    f"- 时间：{roll['rolling_trade_date']} {roll['rolling_time']}",
                    f"- 理由：{roll['human_decision_reason_cn']}",
                    f"- 图：![{roll['rolling_signal_id']}]({rel})",
                    "",
                ]
            )
        lines.extend(
            [
                "## 账本追溯",
                "",
                "- 根订单账本：`../../strict_order_ledger.csv`",
                "- 根 lot 账本：`../../strict_position_lot_ledger.csv`",
                "- 根 case scope：`../../strict_return_scope_case.csv`",
                "- 根 boundary table：`../../strict_boundary_table.csv`",
                "",
            ]
        )
        text = "\n".join(lines) + "\n"
        (case_dir / "case_image_board.md").write_text(text, encoding="utf-8")
        (case_dir / "case_story_board.md").write_text(text.replace("图板", "故事板"), encoding="utf-8")


def audit_links() -> pd.DataFrame:
    rows = []
    pattern = re.compile(r"\[[^\]]*\]\(([^)]+)\)|!\[[^\]]*\]\(([^)]+)\)")
    for md in ROOT.rglob("*.md"):
        text = md.read_text(encoding="utf-8", errors="ignore")
        for match in pattern.finditer(text):
            target = match.group(1) or match.group(2)
            if not target or target.startswith(("http://", "https://", "#")):
                continue
            path = (md.parent / target).resolve()
            rows.append(
                {
                    "markdown_path": md.relative_to(ROOT).as_posix(),
                    "target": target,
                    "resolved_project_path": path.as_posix(),
                    "exists": path.exists(),
                }
            )
    return pd.DataFrame(rows)


def manifest() -> pd.DataFrame:
    rows = []
    for path in sorted(ROOT.rglob("*")):
        if path.is_file():
            if "__pycache__" in path.parts:
                continue
            rel = path.relative_to(ROOT).as_posix()
            rows.append({"path": rel, "size": path.stat().st_size, "sha256": sha256_file(path)})
    return pd.DataFrame(rows)


def clean_generated_outputs() -> None:
    if ROOT.name != RUN_ID:
        raise RuntimeError(f"Refusing to clean unexpected result root: {ROOT}")
    for child in ROOT.iterdir():
        if child.name == "tools":
            continue
        if child.is_dir():
            shutil.rmtree(child)
        else:
            child.unlink()


def assign_sell_signal_ids(signals: list[dict], start: int = 1) -> None:
    for i, sig in enumerate(signals, start):
        sig["signal_id"] = f"SIG-{RUN_ID}-{i:06d}"
        sig["chart_path"] = ""
        sig["chart_reason_rendered"] = "False"
        sig["storyboard_reason_rendered"] = "False"


def assign_rolling_signal_ids(rows: list[dict], start: int = 1) -> None:
    for i, row in enumerate(rows, start):
        row["rolling_signal_id"] = f"ROLL-{RUN_ID}-{i:06d}"
        row["chart_path"] = ""
        row["chart_reason_rendered"] = "False"
        row["storyboard_reason_rendered"] = "False"


def chart_audit(signals: list[dict], rolling_rows: list[dict]) -> pd.DataFrame:
    rows = []
    for s in signals:
        p = ROOT / s["chart_path"] if s.get("chart_path") else ROOT / "__missing__"
        rows.append({"artifact_type": "strict_sell_signal", "signal_id": s["signal_id"], "case_id": s["case_id"], "chart_path": s.get("chart_path", ""), "exists": p.exists()})
    for r in rolling_rows:
        p = ROOT / r["chart_path"] if r.get("chart_path") else ROOT / "__missing__"
        rows.append({"artifact_type": "rolling_low_buy_signal", "signal_id": r["rolling_signal_id"], "case_id": r["case_id"], "chart_path": r.get("chart_path", ""), "exists": p.exists()})
    return pd.DataFrame(rows)


def write_summary_and_self_check(order_df: pd.DataFrame, lot_df: pd.DataFrame, case_df: pd.DataFrame, sell_signals: list[dict], rolling_rows: list[dict], link_df: pd.DataFrame, chart_df: pd.DataFrame, manifest_df: pd.DataFrame, sampling_df: pd.DataFrame) -> None:
    action_counts = Counter([s["human_decision_action"] for s in sell_signals] + [r["human_decision_action"] for r in rolling_rows])
    signal_type_counts = Counter([s["signal_type"] for s in sell_signals] + [r["signal_type"] for r in rolling_rows])
    closed_cases = int(case_df["v1_primary_strict_closed_case_flag"].sum()) if not case_df.empty else 0
    positive_cases = int(case_df[(case_df["v1_primary_strict_closed_case_flag"] == 1) & (pd.to_numeric(case_df["account_return_closed_lots"], errors="coerce") > 0)].shape[0])
    contribution = float(pd.to_numeric(case_df[case_df["v1_primary_strict_closed_case_flag"] == 1]["account_return_closed_lots"], errors="coerce").fillna(0).sum()) if closed_cases else 0.0
    summary = {
        "schema_version": "1.0",
        "task_id": TASK_ID,
        "run_id": RUN_ID,
        "generated_at": now_iso(),
        "stage": "V1_STRICT_SELL_ROLLING_SELF_CHECK_DONE_EXEC_REVIEW_PENDING",
        "source_run_id": SOURCE_RUN_ID,
        "design_audit_id": DESIGN_AUDIT_ID,
        "supplemental_design_audit_id": SUPP_DESIGN_AUDIT_ID,
        "issue_id": ISSUE_ID,
        "return_stat_ready": False,
        "v1_execution_review_required": True,
        "scope": {
            "source_buy_lots": int((lot_df["lot_source_type"] == "SOURCE_BUY").sum()),
            "rolling_buy_lots": int((lot_df["lot_source_type"] == "ROLLING_LOW_BUY").sum()),
            "buy_orders_total": int((order_df["action"] == "BUY").sum()),
            "sell_orders": int((order_df["action"] == "SELL").sum()),
            "strict_lot_rows": int(len(lot_df)),
            "case_rows": int(len(case_df)),
            "v1_primary_strict_closed_cases": closed_cases,
            "v1_positive_primary_cases": positive_cases,
            "v1_primary_success_readout": positive_cases / closed_cases if closed_cases else None,
            "v1_primary_account_contribution_readout": contribution,
        },
        "manual_decision_action_counts": dict(action_counts),
        "signal_type_counts": dict(signal_type_counts),
        "artifacts": {
            "strict_sell_signal_ledger": "strict_sell_signal_ledger.csv",
            "rolling_low_buy_signal_ledger": "rolling_low_buy_signal_ledger.csv",
            "manual_review_sampling_index": "manual_review_sampling_index.csv",
            "case_image_board": "case_image_board.md",
            "manifest": "manifest.json",
        },
        "boundaries": [
            "V1 execution review is pending; do not cite V1 success, return, win rate, drawdown, or strategy validity before review passes.",
            "Human-decision sampling must be at least 20%, target 30%, and cover major action buckets.",
            "Minute-level market breadth for market-risk sellpoint is unavailable and retained as a boundary.",
        ],
    }
    write_json(ROOT / "summary.json", summary)
    (ROOT / "summary.md").write_text(
        "\n".join(
            [
                "# V1 精准卖点与滚动渣男返修执行包摘要",
                "",
                f"- run_id：`{RUN_ID}`",
                f"- 阶段：`{summary['stage']}`",
                f"- 来源包：`{SOURCE_RUN_ID}`",
                f"- V1 主口径严格闭合 case 候选读数：{closed_cases}",
                f"- V1 主口径正收益 case 候选读数：{positive_cases}",
                f"- V1 主口径成功率候选读数：{summary['scope']['v1_primary_success_readout']}",
                f"- V1 主口径账户贡献候选读数：{contribution:.8f}",
                "",
                "边界：执行审核通过前，上述仅为待审候选读数，不得引用为 V1 正式结论。",
                "",
            ]
        ),
        encoding="utf-8",
    )

    checks = []

    def check(name: str, passed: bool, detail: str) -> None:
        checks.append({"check_id": name, "status": "PASS" if passed else "FAIL", "detail": detail})

    check("CONFIG_FILES_EXIST", (ROOT / "strict_sell_rolling_run_config.json").exists() and (ROOT / "strict_sell_rolling_run_config.md").exists(), "strict sell rolling config frozen")
    check("SELL_SIGNAL_REQUIRED_FIELDS", all(s.get("human_decision_action") and s.get("human_decision_reason_cn") and s.get("decision_operator") for s in sell_signals), f"sell_signals={len(sell_signals)}")
    check("ROLLING_SIGNAL_REQUIRED_FIELDS", all(r.get("human_decision_action") and r.get("human_decision_reason_cn") for r in rolling_rows), f"rolling_signals={len(rolling_rows)}")
    check("SELL_SIGNAL_IDS_UNIQUE", len({s.get("signal_id") for s in sell_signals}) == len(sell_signals), f"sell_signal_ids={len(sell_signals)}")
    check("ROLLING_SIGNAL_IDS_UNIQUE", len({r.get("rolling_signal_id") for r in rolling_rows}) == len(rolling_rows), f"rolling_signal_ids={len(rolling_rows)}")
    check("CHART_REASONS_RENDERED", all(s.get("chart_reason_rendered") == "True" for s in sell_signals) and all(r.get("chart_reason_rendered") == "True" for r in rolling_rows), "all signal charts render reason")
    check("STORYBOARD_REASONS_RENDERED", all(s.get("storyboard_reason_rendered") == "True" for s in sell_signals) and all(r.get("storyboard_reason_rendered") == "True" for r in rolling_rows), "all story boards can show reason")
    check("NO_V0_V1_MIXED_OUTPUT", all(str(v).startswith("ORD-" + RUN_ID) for v in order_df["order_id"]), "strict orders use V1 run id")
    check("MARKET_RISK_BOUNDARY_RETAINED", (ROOT / "strict_boundary_table.csv").read_text(encoding="utf-8-sig").find("MARKET_RISK_MINUTE_BREADTH_DATA_GAP_HELD") >= 0, "market risk data gap retained")
    check("LINKS_REACHABLE", link_df.empty or bool(link_df["exists"].all()), f"links={len(link_df)}, missing={0 if link_df.empty else int((~link_df['exists']).sum())}")
    check("CHARTS_EXIST", chart_df.empty or bool(chart_df["exists"].all()), f"charts={len(chart_df)}, missing={0 if chart_df.empty else int((~chart_df['exists']).sum())}")
    check("MANUAL_REVIEW_SAMPLING_RATIO", len(sampling_df) >= math.ceil((len(sell_signals) + len(rolling_rows)) * AUDIT_SAMPLE_MIN_RATIO), f"samples={len(sampling_df)}, total={len(sell_signals)+len(rolling_rows)}")
    actions = set(sampling_df["human_decision_action"].tolist()) if not sampling_df.empty else set()
    must = {a for a in ["SELL", "HOLD_WATCH", "HOLD_ABOVE_8", "REVIEW_HELD", "BUY_ROLLING_LOW"] if action_counts.get(a, 0) > 0}
    check("MANUAL_REVIEW_SAMPLING_COVERS_ACTIONS", must.issubset(actions), f"required={sorted(must)}, sampled={sorted(actions)}")
    check("MANIFEST_PRELIMINARY_READY", not manifest_df.empty, f"manifest preliminary files={len(manifest_df)}")
    checks_df = pd.DataFrame(checks)
    checks_df.to_csv(ROOT / "self_check_items.csv", index=False, encoding="utf-8-sig")
    self_check = {
        "schema_version": "1.0",
        "run_id": RUN_ID,
        "generated_at": now_iso(),
        "status": "PASS_FOR_V1_EXECUTION_REVIEW_READY" if (checks_df["status"] == "PASS").all() else "FAIL",
        "pass_count": int((checks_df["status"] == "PASS").sum()),
        "fail_count": int((checks_df["status"] == "FAIL").sum()),
        "return_stat_ready": False,
        "items_path": "self_check_items.csv",
    }
    write_json(ROOT / "self_check.json", self_check)
    (ROOT / "self_check.md").write_text(
        f"# 自检\n\n- status：`{self_check['status']}`\n- PASS：{self_check['pass_count']}\n- FAIL：{self_check['fail_count']}\n",
        encoding="utf-8",
    )


def main() -> None:
    ROOT.mkdir(parents=True, exist_ok=True)
    clean_generated_outputs()
    (ROOT / "cases").mkdir(exist_ok=True)
    write_config()
    write_rule_mapping()
    order_src, selected, source_lots_df, lots = load_inputs()
    trade_dates = fetch_trade_calendar()
    date_symbols, symbols, min_date, max_date = build_fetch_maps(lots, trade_dates)
    minute, daily = fetch_market_data(date_symbols, symbols, min_date, max_date)
    minute_lookup = make_lookup(minute)
    daily_lookup = daily_lookup_map(daily)

    sell_signals: list[dict] = []
    rolling_rows: list[dict] = []
    final_sells: dict[str, dict] = {}
    for lot in lots:
        sigs, final_sell, rolls = evaluate_lot(lot, trade_dates, minute_lookup, daily_lookup, allow_rolling=True)
        sell_signals.extend(sigs)
        rolling_rows.extend(rolls)
        if final_sell:
            final_sells[lot.source_lot_id] = final_sell

    assign_sell_signal_ids(sell_signals)
    assign_rolling_signal_ids(rolling_rows)
    render_charts(sell_signals, rolling_rows, minute_lookup, daily_lookup)
    order_df, lot_df, account_df, case_df, new_lots = build_ledgers(lots, sell_signals, final_sells, rolling_rows, trade_dates, minute_lookup, daily_lookup)
    case_scope, lot_scope, boundary = build_scope_and_boundary(case_df, lot_df, sell_signals, rolling_rows)
    sampling = build_sampling_index(sell_signals, rolling_rows)

    pd.DataFrame(sell_signals).to_csv(ROOT / "strict_sell_signal_ledger.csv", index=False, encoding="utf-8-sig")
    pd.DataFrame(rolling_rows).to_csv(ROOT / "rolling_low_buy_signal_ledger.csv", index=False, encoding="utf-8-sig")
    order_df.to_csv(ROOT / "strict_order_ledger.csv", index=False, encoding="utf-8-sig")
    lot_df.to_csv(ROOT / "strict_position_lot_ledger.csv", index=False, encoding="utf-8-sig")
    account_df.to_csv(ROOT / "strict_daily_account_ledger.csv", index=False, encoding="utf-8-sig")
    case_df.to_csv(ROOT / "strict_case_summary.csv", index=False, encoding="utf-8-sig")
    case_scope.to_csv(ROOT / "strict_return_scope_case.csv", index=False, encoding="utf-8-sig")
    lot_scope.to_csv(ROOT / "strict_return_scope_lot.csv", index=False, encoding="utf-8-sig")
    boundary.to_csv(ROOT / "strict_boundary_table.csv", index=False, encoding="utf-8-sig")
    sampling.to_csv(ROOT / "manual_review_sampling_index.csv", index=False, encoding="utf-8-sig")

    write_boards(case_df, sell_signals, rolling_rows)
    chart_df = chart_audit(sell_signals, rolling_rows)
    chart_df.to_csv(ROOT / "chart_evidence_audit.csv", index=False, encoding="utf-8-sig")
    link_df = audit_links()
    link_df.to_csv(ROOT / "link_evidence_audit.csv", index=False, encoding="utf-8-sig")
    prelim_manifest = manifest()
    write_summary_and_self_check(order_df, lot_df, case_df, sell_signals, rolling_rows, link_df, chart_df, prelim_manifest, sampling)
    readme = [
        "# RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001",
        "",
        "本包用于执行 V1 精准卖点与滚动渣男返修。旧 V0 全量包只作为只读来源，本包不覆盖旧包。",
        "",
        "阅读入口：",
        "",
        "- `case_image_board.md`：人工图片第一入口。",
        "- `strict_sell_signal_ledger.csv`：精准卖点与 3% / 5% / 8% 趋势候选账本。",
        "- `rolling_low_buy_signal_ledger.csv`：滚动低吸候选账本。",
        "- `manual_review_sampling_index.csv`：审核员 20% 起、原则 30% 抽样入口。",
        "- `summary.md/json`：待审候选读数和边界。",
        "",
        "边界：V1 执行审核通过前，不得引用 V1 成功率、收益率、胜率、回撤或策略有效性结论。",
        "",
    ]
    (ROOT / "README.md").write_text("\n".join(readme), encoding="utf-8")
    final_manifest = manifest()
    final_manifest = final_manifest[~final_manifest["path"].isin(["manifest.csv", "manifest.json"])].reset_index(drop=True)
    final_manifest.to_csv(ROOT / "manifest.csv", index=False, encoding="utf-8-sig")
    write_json(ROOT / "manifest.json", {"schema_version": "1.0", "run_id": RUN_ID, "generated_at": now_iso(), "files": final_manifest.to_dict("records")})

    print(json.dumps({"run_id": RUN_ID, "sell_signals": len(sell_signals), "rolling_signals": len(rolling_rows), "orders": len(order_df), "lots": len(lot_df), "cases": len(case_df)}, ensure_ascii=False))


if __name__ == "__main__":
    main()