from __future__ import annotations

import csv
import hashlib
import json
import math
import os
import re
import shutil
from datetime import datetime, timedelta, timezone
from pathlib import Path

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


RUN_ID = "RUN-ANA-WUJI-V1-STOCK-LIFECYCLE-PACKAGE-20260611-001"
TASK_ID = "ANA-WUJI-V1-STOCK-LIFECYCLE-PACKAGE-20260611"
SOURCE_RUN_ID = "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001"
FINAL_RUN_ID = "RUN-ANA-WUJI-V1-FINAL-CONCLUSION-20260610-001"
DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-V1-STOCK-LIFECYCLE-PACKAGE-20260611-DESIGN-001"
SOURCE_EXEC_AUDIT_ID = "AUDIT-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260610-EXEC-REREVIEW-003"

PACKAGE_ROOT = Path(__file__).resolve().parents[1]
RESULT_ROOT = PACKAGE_ROOT.parents[0]
PROJECT_ROOT = PACKAGE_ROOT.parents[2]
SOURCE_ROOT = RESULT_ROOT / SOURCE_RUN_ID
FINAL_ROOT = RESULT_ROOT / FINAL_RUN_ID
LOCAL_DB_INDEX = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md")

MOJIBAKE_MARKERS = ["????", "???", "�", "ÀíÓÉ", "å¤", "æ", "Ã", "Â", "涓", "鏃", "鐐", "偂"]


def now_iso() -> str:
    return datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")


GENERATED_AT = now_iso()


def read_password() -> str:
    env = os.environ.get("TIANXIA_MYSQL_PASSWORD") or os.environ.get("MYSQL_PWD")
    if env:
        return env
    candidates = [LOCAL_DB_INDEX]
    observer_root = Path(r"D:\strategy_project\s-system-doc\observer")
    if observer_root.exists():
        candidates.extend(observer_root.glob("*/数据库索引数据.md"))
    for path in candidates:
        if not path.exists():
            continue
        text = path.read_text(encoding="utf-8")
        match = re.search(r"^\s*-\s*密码：`([^`]+)`", text, re.MULTILINE)
        if match:
            return match.group(1)
    raise RuntimeError("Unable to read local MySQL credential from approved local index.")


def get_conn():
    return pymysql.connect(
        host="127.0.0.1",
        port=3306,
        user="root",
        password=read_password(),
        database="tianxia",
        charset="utf8mb4",
        connect_timeout=5,
        read_timeout=180,
    )


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 rel(path: Path) -> str:
    return path.resolve().relative_to(PACKAGE_ROOT.resolve()).as_posix()


def source_rel(path: Path) -> str:
    try:
        return path.resolve().relative_to(PROJECT_ROOT.resolve()).as_posix()
    except ValueError:
        return str(path)


def font(size: int):
    for p in [Path("C:/Windows/Fonts/msyh.ttc"), Path("C:/Windows/Fonts/simhei.ttf"), Path("C:/Windows/Fonts/simsun.ttc")]:
        if p.exists():
            return ImageFont.truetype(str(p), size)
    return ImageFont.load_default()


FONT_TITLE = font(28)
FONT_SUB = font(21)
FONT_MID = font(17)
FONT_SMALL = font(14)
FONT_TINY = font(12)


def write_csv(path: Path, rows: list[dict], fields: list[str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(rows)


def read_csv(path: Path) -> pd.DataFrame:
    return pd.read_csv(path, encoding="utf-8-sig")


def norm_time(value) -> str:
    if pd.isna(value):
        return ""
    if hasattr(value, "total_seconds"):
        seconds = int(value.total_seconds())
        h, rem = divmod(seconds, 3600)
        m, s = divmod(rem, 60)
        return f"{h:02d}:{m:02d}:{s:02d}"
    text = str(value).strip()
    if "days" in text and " " in text:
        text = text.split()[-1]
    if re.fullmatch(r"\d{2}:\d{2}$", text):
        return f"{text}:00"
    if re.fullmatch(r"\d{2}:\d{2}:\d{2}(\.\d+)?", text):
        return text[:8]
    return text


def y_price(value: float, lo: float, hi: float, top: int, bottom: int) -> int:
    if hi <= lo:
        return (top + bottom) // 2
    return int(bottom - (value - lo) / (hi - lo) * (bottom - top))


def wrap_text(text: str, width: int) -> list[str]:
    text = str(text or "")
    lines: list[str] = []
    for raw in text.splitlines() or [""]:
        line = ""
        units = list(raw)
        for ch in units:
            add = 2 if ord(ch) > 127 else 1
            cur = sum(2 if ord(c) > 127 else 1 for c in line)
            if cur + add > width and line:
                lines.append(line)
                line = ch
            else:
                line += ch
        lines.append(line)
    return lines


def draw_text_box(draw: ImageDraw.ImageDraw, box: tuple[int, int, int, int], title: str, lines: list[str]) -> None:
    x1, y1, x2, y2 = box
    draw.rectangle([x1, y1, x2, y2], fill="#ffffff", outline="#cbd5e1")
    draw.text((x1 + 16, y1 + 14), title, fill="#111827", font=FONT_SUB)
    y = y1 + 50
    for line in lines:
        if line == "":
            y += 12
            continue
        for wrapped in wrap_text(line, 30):
            if y > y2 - 24:
                return
            draw.text((x1 + 16, y), wrapped, fill="#334155", font=FONT_SMALL)
            y += 22


def safe_symbol(symbol: str) -> str:
    return str(symbol).replace(".", "_").replace("/", "_")


def load_sources() -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    v1_index = read_csv(FINAL_ROOT / "v1_case_readout_index.csv")
    primary = v1_index[(v1_index["v1_return_scope"] == "V1_PRIMARY_STRICT_CLOSED_CASE") & (pd.to_numeric(v1_index["primary_flag"], errors="coerce") == 1)].copy()
    orders = read_csv(SOURCE_ROOT / "strict_order_ledger.csv")
    lots = read_csv(SOURCE_ROOT / "strict_position_lot_ledger.csv")
    case_summary = read_csv(SOURCE_ROOT / "strict_case_summary.csv")
    for df in [orders, lots]:
        for col in ["trade_time", "entry_time", "exit_time"]:
            if col in df.columns:
                df[col] = df[col].map(norm_time)
    return primary, orders, lots, case_summary


def fetch_trade_calendar() -> list[str]:
    with get_conn() as conn:
        df = pd.read_sql(
            "SELECT DISTINCT trade_date FROM a_share_daily_price ORDER BY trade_date",
            conn,
        )
    return pd.to_datetime(df["trade_date"]).dt.strftime("%Y-%m-%d").tolist()


def fetch_daily(symbols: list[str], min_date: str, max_date: str) -> pd.DataFrame:
    parts: list[pd.DataFrame] = []
    with get_conn() as conn:
        for i in range(0, len(symbols), 180):
            chunk = symbols[i : i + 180]
            ph = ",".join(["%s"] * len(chunk))
            parts.append(
                pd.read_sql(
                    f"""
                    SELECT trade_date, symbol, open_price, high_price, low_price, close_price, volume, amount
                    FROM a_share_daily_price
                    WHERE trade_date BETWEEN %s AND %s
                      AND symbol IN ({ph})
                    ORDER BY symbol, trade_date
                    """,
                    conn,
                    params=[min_date, max_date, *chunk],
                )
            )
    daily = pd.concat(parts, ignore_index=True) if parts else pd.DataFrame()
    if daily.empty:
        return daily
    daily["trade_date"] = pd.to_datetime(daily["trade_date"]).dt.strftime("%Y-%m-%d")
    for col in ["open_price", "high_price", "low_price", "close_price", "volume", "amount"]:
        daily[col] = pd.to_numeric(daily[col], errors="coerce")
    daily = daily.sort_values(["symbol", "trade_date"]).reset_index(drop=True)
    daily["ma5"] = daily.groupby("symbol")["close_price"].transform(lambda s: s.rolling(5, min_periods=1).mean())
    return daily


def fetch_minute(date_symbols: list[tuple[str, str]]) -> pd.DataFrame:
    by_date: dict[str, set[str]] = {}
    for date, symbol in date_symbols:
        by_date.setdefault(date, set()).add(symbol)
    parts: list[pd.DataFrame] = []
    with get_conn() as conn:
        for idx, date in enumerate(sorted(by_date), start=1):
            symbols = sorted(by_date[date])
            ph = ",".join(["%s"] * len(symbols))
            parts.append(
                pd.read_sql(
                    f"""
                    SELECT trade_date, trade_time, symbol, open_price, high_price, low_price, close_price, volume, amount
                    FROM a_share_minute_price
                    WHERE trade_date = %s AND symbol IN ({ph})
                    ORDER BY symbol, trade_date, trade_time
                    """,
                    conn,
                    params=[date, *symbols],
                )
            )
            if idx % 100 == 0:
                print(f"minute query progress: {idx}/{len(by_date)}", flush=True)
    minute = pd.concat(parts, ignore_index=True) if parts else pd.DataFrame()
    if minute.empty:
        return minute
    minute["trade_date"] = pd.to_datetime(minute["trade_date"]).dt.strftime("%Y-%m-%d")
    minute["trade_time"] = minute["trade_time"].map(norm_time)
    for col in ["open_price", "high_price", "low_price", "close_price", "volume", "amount"]:
        minute[col] = pd.to_numeric(minute[col], errors="coerce")
    return minute


def build_lifecycles(primary: pd.DataFrame, orders: pd.DataFrame, lots: pd.DataFrame) -> tuple[list[dict], pd.DataFrame, pd.DataFrame]:
    primary_ids = set(primary["case_id"].astype(str))
    order_scope = orders[orders["case_id"].astype(str).isin(primary_ids)].copy()
    lot_scope = lots[lots["case_id"].astype(str).isin(primary_ids)].copy()
    rows: list[dict] = []
    seq_by_case_symbol: dict[tuple[str, str], int] = {}
    for (case_id, symbol), group in order_scope.groupby(["case_id", "symbol"], sort=True):
        buys = group[group["action"] == "BUY"].copy()
        sells = group[group["action"] == "SELL"].copy()
        if buys.empty or sells.empty:
            continue
        buys = buys.sort_values(["trade_date", "trade_time", "order_id"])
        sells = sells.sort_values(["trade_date", "trade_time", "order_id"])
        key = (str(case_id), str(symbol))
        seq_by_case_symbol[key] = seq_by_case_symbol.get(key, 0) + 1
        sequence = seq_by_case_symbol[key]
        lifecycle_id = f"LC-{case_id}-{safe_symbol(symbol)}-{sequence:02d}"
        lot_rows = lot_scope[(lot_scope["case_id"].astype(str) == str(case_id)) & (lot_scope["symbol"].astype(str) == str(symbol))]
        first_buy = buys.iloc[0]
        last_sell = sells.iloc[-1]
        manual_ids = sorted(set([str(x) for x in group.get("manual_decision_id", pd.Series(dtype=str)).dropna() if str(x).strip()]))
        rows.append(
            {
                "lifecycle_id": lifecycle_id,
                "case_id": case_id,
                "symbol": symbol,
                "sequence": sequence,
                "first_buy_trade_date": str(first_buy["trade_date"]),
                "first_buy_time": str(first_buy["trade_time"]),
                "first_buy_price": float(first_buy["price"]),
                "last_sell_trade_date": str(last_sell["trade_date"]),
                "last_sell_time": str(last_sell["trade_time"]),
                "last_sell_price": float(last_sell["price"]),
                "buy_order_count": int(len(buys)),
                "sell_order_count": int(len(sells)),
                "lot_count": int(len(lot_rows)),
                "closed_lot_count": int((lot_rows["lot_status"].astype(str).str.contains("CLOSED", na=False)).sum()),
                "buy_order_ids": ";".join(buys["order_id"].astype(str).tolist()),
                "sell_order_ids": ";".join(sells["order_id"].astype(str).tolist()),
                "lot_ids": ";".join(lot_rows["strict_lot_id"].astype(str).tolist()),
                "manual_decision_ids": ";".join(manual_ids),
                "case_image_board": f"../{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md",
                "source_order_ledger": f"../{SOURCE_RUN_ID}/strict_order_ledger.csv",
                "source_lot_ledger": f"../{SOURCE_RUN_ID}/strict_position_lot_ledger.csv",
            }
        )
    return rows, order_scope, lot_scope


def trade_dates_between(calendar: list[str], start: str, end: str) -> list[str]:
    return [d for d in calendar if start <= d <= end]


def daily_window(calendar: list[str], symbol_daily: pd.DataFrame, start_date: str, end_date: str) -> tuple[pd.DataFrame, dict]:
    meta = {
        "pre_window_insufficient": False,
        "post_window_insufficient": False,
        "expected_start_date": "",
        "expected_end_date": "",
    }
    if start_date not in calendar or end_date not in calendar:
        return pd.DataFrame(), meta
    start_pos = calendar.index(start_date)
    end_pos = calendar.index(end_date)
    expected_start_pos = start_pos - 50
    expected_end_pos = end_pos + 10
    meta["pre_window_insufficient"] = expected_start_pos < 0
    meta["post_window_insufficient"] = expected_end_pos >= len(calendar)
    start = calendar[max(0, expected_start_pos)]
    end = calendar[min(len(calendar) - 1, expected_end_pos)]
    meta["expected_start_date"] = start
    meta["expected_end_date"] = end
    window = symbol_daily[(symbol_daily["trade_date"] >= start) & (symbol_daily["trade_date"] <= end)].copy()
    return window.reset_index(drop=True), meta


def draw_lifecycle_daily(window: pd.DataFrame, lifecycle: dict, orders: pd.DataFrame, out_path: Path) -> None:
    w, h = 1880, 1060
    img = Image.new("RGB", (w, h), "#fbfbf7")
    d = ImageDraw.Draw(img)
    d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1")
    title = f"股票生命周期日线图：{lifecycle['symbol']}  {lifecycle['case_id']}"
    subtitle = "窗口：首买前50个交易日 -> 末卖后10个交易日；同一 case 内同一股票生命周期，不跨 case 合并。"
    d.text((32, 22), title, fill="#111827", font=FONT_TITLE)
    d.text((32, 62), subtitle, fill="#7f1d1d", font=FONT_MID)
    plot_left, plot_top, plot_right, plot_bottom = 80, 120, 1340, 700
    vol_top, vol_bottom = 770, 910
    note_left, note_top = 1370, 120
    d.rectangle([plot_left, plot_top, plot_right, plot_bottom], outline="#94a3b8")
    d.rectangle([plot_left, vol_top, plot_right, vol_bottom], outline="#94a3b8")
    if window.empty:
        d.text((plot_left + 180, plot_top + 220), "日线数据缺失", fill="#b91c1c", font=FONT_TITLE)
        out_path.parent.mkdir(parents=True, exist_ok=True)
        img.save(out_path)
        return
    win = window.copy().reset_index(drop=True)
    price_low = float(win["low_price"].min()) * 0.98
    price_high = float(win["high_price"].max()) * 1.02
    max_vol = max(float(win["volume"].max()), 1.0)
    n = len(win)
    gap = (plot_right - plot_left) / max(n, 1)
    body_w = max(3, int(gap * 0.55))
    date_to_x: dict[str, int] = {}
    for i in range(5):
        price = price_low + (price_high - price_low) * i / 4
        y = y_price(price, price_low, price_high, plot_top, plot_bottom)
        d.line([plot_left, y, plot_right, y], fill="#e2e8f0")
        d.text((18, y - 8), f"{price:.2f}", fill="#64748b", font=FONT_SMALL)
    ma5_pts = []
    for i, row in win.iterrows():
        cx = int(plot_left + gap * i + gap / 2)
        date_to_x[str(row["trade_date"])] = cx
        op, hi, lo, cl = [float(row[c]) for c in ["open_price", "high_price", "low_price", "close_price"]]
        color = "#dc2626" if cl >= op else "#16a34a"
        d.line([cx, y_price(lo, price_low, price_high, plot_top, plot_bottom), cx, y_price(hi, price_low, price_high, plot_top, plot_bottom)], fill=color, width=2)
        y1, y2 = y_price(op, price_low, price_high, plot_top, plot_bottom), y_price(cl, price_low, price_high, plot_top, plot_bottom)
        d.rectangle([cx - body_w // 2, min(y1, y2), cx + body_w // 2, max(y1, y2)], fill=color, outline=color)
        vh = int(float(row["volume"]) / max_vol * (vol_bottom - vol_top))
        d.rectangle([cx - body_w // 2, vol_bottom - vh, cx + body_w // 2, vol_bottom], fill=color, outline=color)
        if pd.notna(row["ma5"]):
            ma5_pts.append((cx, y_price(float(row["ma5"]), price_low, price_high, plot_top, plot_bottom)))
        if i % max(1, n // 10) == 0:
            d.text((cx - 24, vol_bottom + 8), str(row["trade_date"])[5:], fill="#64748b", font=FONT_SMALL)
    if len(ma5_pts) > 1:
        d.line(ma5_pts, fill="#2563eb", width=2)
        d.text((plot_left + 12, plot_bottom + 12), "MA5", fill="#2563eb", font=FONT_SMALL)
    for _, order in orders.iterrows():
        date = str(order["trade_date"])
        if date not in date_to_x:
            continue
        price = float(order["price"])
        cx = date_to_x[date]
        cy = y_price(price, price_low, price_high, plot_top, plot_bottom)
        action = str(order["action"])
        color = "#b91c1c" if action == "BUY" else "#7c3aed"
        label = "买" if action == "BUY" else "卖"
        d.line([cx, plot_top, cx, vol_bottom], fill=color, width=2)
        d.line([plot_left, cy, plot_right, cy], fill=color, width=1)
        d.ellipse([cx - 8, cy - 8, cx + 8, cy + 8], fill=color)
        d.text((cx + 6, max(plot_top + 6, cy - 28)), f"{label}{str(order['trade_time'])[:5]} {price:.2f}", fill=color, font=FONT_SMALL)
    lines = [
        f"生命周期：{lifecycle['lifecycle_id']}",
        f"首买：{lifecycle['first_buy_trade_date']} {lifecycle['first_buy_time']} @ {float(lifecycle['first_buy_price']):.2f}",
        f"末卖：{lifecycle['last_sell_trade_date']} {lifecycle['last_sell_time']} @ {float(lifecycle['last_sell_price']):.2f}",
        f"BUY/SELL：{lifecycle['buy_order_count']} / {lifecycle['sell_order_count']}",
        f"lot：{lifecycle['lot_count']}，闭合：{lifecycle['closed_lot_count']}",
        "",
        "读图方式：先看日线整体位置，再沿竖线查看每次买卖点；买卖理由见每日分时图和 story board。",
        "本图只增强同事阅读，不改变 V1 已审核账本和收益读数。",
    ]
    draw_text_box(d, (note_left, note_top, 1830, 910), "图上说明", lines)
    d.text((32, 1000), f"来源审计：{SOURCE_EXEC_AUDIT_ID}；生命周期包设计审计：{DESIGN_AUDIT_ID}", fill="#334155", font=FONT_MID)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    img.save(out_path)


def draw_intraday_chart(day: pd.DataFrame, lifecycle: dict, day_orders: pd.DataFrame, ma5_price: float | None, out_path: Path) -> None:
    w, h = 1700, 940
    img = Image.new("RGB", (w, h), "#fbfbf7")
    d = ImageDraw.Draw(img)
    d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1")
    trade_date = out_path.stem[-8:]
    d.text((32, 24), f"生命周期分时图：{lifecycle['symbol']}  {trade_date}  {lifecycle['case_id']}", fill="#111827", font=FONT_TITLE)
    d.text((32, 62), "模式：3% / 5% / 8% 阈值线、买卖点、买卖价线、MA5、操作理由；无操作日也保留持有观察理由。", fill="#7f1d1d", font=FONT_MID)
    plot_left, plot_top, plot_right, plot_bottom = 80, 120, 1160, 640
    vol_top, vol_bottom = 705, 840
    note_left, note_top = 1195, 120
    d.rectangle([plot_left, plot_top, plot_right, plot_bottom], outline="#94a3b8")
    d.rectangle([plot_left, vol_top, plot_right, vol_bottom], outline="#94a3b8")
    if day.empty:
        d.text((plot_left + 180, plot_top + 200), "分钟线数据缺失", fill="#b91c1c", font=FONT_TITLE)
        out_path.parent.mkdir(parents=True, exist_ok=True)
        img.save(out_path)
        return
    day = day.sort_values("trade_time").reset_index(drop=True)
    base_price = float(lifecycle["first_buy_price"])
    ref_prices = [base_price, base_price * 1.03, base_price * 1.05, base_price * 1.08]
    order_prices = [float(x) for x in day_orders["price"].tolist()] if not day_orders.empty else []
    raw_low = min([float(day["low_price"].min()), *ref_prices, *order_prices])
    raw_high = max([float(day["high_price"].max()), *ref_prices, *order_prices])
    if ma5_price and not math.isnan(ma5_price):
        raw_low = min(raw_low, ma5_price)
        raw_high = max(raw_high, ma5_price)
    price_low, price_high = raw_low * 0.997, raw_high * 1.003
    max_vol = max(float(day["volume"].max()), 1.0)
    n = len(day)
    gap = (plot_right - plot_left) / max(n - 1, 1)
    points = []
    for i, row in day.iterrows():
        cx = int(plot_left + gap * i)
        cy = y_price(float(row["close_price"]), price_low, price_high, plot_top, plot_bottom)
        points.append((cx, cy))
        vh = int(float(row["volume"]) / max_vol * (vol_bottom - vol_top))
        d.line([cx, vol_bottom, cx, vol_bottom - vh], fill="#cbd5e1", width=1)
        if i % max(1, n // 8) == 0:
            d.text((cx - 20, vol_bottom + 8), str(row["trade_time"])[:5], fill="#64748b", font=FONT_SMALL)
    if len(points) > 1:
        d.line(points, fill="#0f766e", width=3)
    threshold_styles = [(base_price * 1.03, "3%", "#f59e0b"), (base_price * 1.05, "5%", "#dc2626"), (base_price * 1.08, "8%", "#7c3aed")]
    for price, label, color in threshold_styles:
        y = y_price(price, price_low, price_high, plot_top, plot_bottom)
        d.line([plot_left, y, plot_right, y], fill=color, width=2)
        d.text((plot_right + 8, y - 10), f"{label} {price:.2f}", fill=color, font=FONT_SMALL)
    if ma5_price and not math.isnan(ma5_price):
        y = y_price(ma5_price, price_low, price_high, plot_top, plot_bottom)
        d.line([plot_left, y, plot_right, y], fill="#2563eb", width=2)
        d.text((plot_right + 8, y - 10), f"MA5 {ma5_price:.2f}", fill="#2563eb", font=FONT_SMALL)
    for _, order in day_orders.iterrows():
        action = str(order["action"])
        color = "#b91c1c" if action == "BUY" else "#7c3aed"
        label = "买点" if action == "BUY" else "卖点"
        time = str(order["trade_time"])
        hits = day.index[day["trade_time"] == time].tolist()
        idx = hits[0] if hits else max([i for i, t in enumerate(day["trade_time"].tolist()) if str(t) <= time] or [0])
        x = int(plot_left + gap * idx)
        price = float(order["price"])
        y = y_price(price, price_low, price_high, plot_top, plot_bottom)
        d.line([x, plot_top, x, vol_bottom], fill=color, width=3)
        d.line([plot_left, y, plot_right, y], fill=color, width=1)
        d.ellipse([x - 8, y - 8, x + 8, y + 8], fill=color)
        d.text((min(x + 8, plot_right - 135), max(plot_top + 8, y - 30)), f"{label} {time[:5]} {price:.2f}", fill=color, font=FONT_SMALL)
    if day_orders.empty:
        reason = "当日处于生命周期区间内，无买卖订单；持有观察，未触发 V1 精准卖点或滚动低吸。"
    else:
        fragments = []
        for _, order in day_orders.iterrows():
            reason = str(order.get("decision_reason_cn", "")).strip()
            fragments.append(f"{order['action']} {str(order['trade_time'])[:5]}：{reason[:46]}")
        reason = "；".join(fragments)
    lines = [
        f"生命周期：{lifecycle['lifecycle_id']}",
        f"参考首买价：{base_price:.2f}",
        f"当日订单数：{len(day_orders)}",
        f"操作理由：{reason}",
        "",
        "边界：本图用于人工阅读和复核，不新增买卖裁决，不改变 V1 已审核账本。",
    ]
    draw_text_box(d, (note_left, note_top, 1660, 840), "当日说明", lines)
    d.text((32, 892), f"来源：{SOURCE_RUN_ID}；审计：{SOURCE_EXEC_AUDIT_ID}", fill="#334155", font=FONT_MID)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    img.save(out_path)


def audit_markdown_links(md_paths: list[Path]) -> list[dict]:
    rows: list[dict] = []
    pattern = re.compile(r"\[[^\]]+\]\(([^)]+)\)")
    for path in md_paths:
        text = path.read_text(encoding="utf-8")
        for match in pattern.finditer(text):
            target = match.group(1).split("#", 1)[0]
            if "://" in target or not target:
                continue
            target_path = (path.parent / target).resolve()
            rows.append(
                {
                    "markdown_path": rel(path),
                    "target": target,
                    "resolved_project_path": source_rel(target_path),
                    "exists": target_path.exists(),
                }
            )
    return rows


def scan_mojibake(paths: list[Path]) -> list[dict]:
    hits: list[dict] = []
    for path in paths:
        if path.suffix.lower() not in [".md", ".csv", ".json", ".txt"]:
            continue
        try:
            text = path.read_text(encoding="utf-8-sig")
        except UnicodeDecodeError:
            hits.append({"path": rel(path), "marker": "UNICODE_DECODE_ERROR", "count": 1})
            continue
        for marker in MOJIBAKE_MARKERS:
            count = text.count(marker)
            if count:
                hits.append({"path": rel(path), "marker": marker, "count": count})
    return hits


def write_manifest() -> list[dict]:
    rows: list[dict] = []
    for path in sorted(PACKAGE_ROOT.rglob("*")):
        if not path.is_file():
            continue
        rows.append(
            {
                "path": rel(path),
                "size": path.stat().st_size,
                "sha256": sha256_file(path),
            }
        )
    write_csv(PACKAGE_ROOT / "manifest.csv", rows, ["path", "size", "sha256"])
    (PACKAGE_ROOT / "manifest.json").write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8")
    return rows


def main() -> None:
    if PACKAGE_ROOT.exists():
        for child in PACKAGE_ROOT.iterdir():
            if child.name == "tools":
                continue
            if child.is_dir():
                shutil.rmtree(child)
            else:
                child.unlink()
    (PACKAGE_ROOT / "img").mkdir(parents=True, exist_ok=True)

    primary, orders, lots, case_summary = load_sources()
    lifecycle_rows, order_scope, lot_scope = build_lifecycles(primary, orders, lots)
    if len(primary) != 250:
        raise RuntimeError(f"Expected 250 primary cases, got {len(primary)}")
    lifecycle_df = pd.DataFrame(lifecycle_rows)
    calendar = fetch_trade_calendar()
    if lifecycle_df.empty:
        raise RuntimeError("No lifecycle rows generated")

    min_date = calendar[max(0, calendar.index(lifecycle_df["first_buy_trade_date"].min()) - 60)]
    max_date = calendar[min(len(calendar) - 1, calendar.index(lifecycle_df["last_sell_trade_date"].max()) + 12)]
    daily = fetch_daily(sorted(lifecycle_df["symbol"].unique().tolist()), min_date, max_date)

    date_symbols: list[tuple[str, str]] = []
    lifecycle_trade_dates: dict[str, list[str]] = {}
    for row in lifecycle_rows:
        dates = trade_dates_between(calendar, row["first_buy_trade_date"], row["last_sell_trade_date"])
        lifecycle_trade_dates[row["lifecycle_id"]] = dates
        for date in dates:
            date_symbols.append((date, row["symbol"]))
    minute = fetch_minute(date_symbols)

    chart_rows: list[dict] = []
    missing_rows: list[dict] = []
    md_paths: list[Path] = []
    lifecycle_index_rows: list[dict] = []
    primary_case_ids = set(primary["case_id"].astype(str))

    for row in lifecycle_rows:
        lifecycle_id = row["lifecycle_id"]
        case_id = str(row["case_id"])
        symbol = str(row["symbol"])
        case_dir = PACKAGE_ROOT / "cases" / case_id
        stock_dir = case_dir / "stocks" / safe_symbol(symbol)
        img_dir = stock_dir / "img"
        stock_daily = daily[daily["symbol"] == symbol].copy()
        win, meta = daily_window(calendar, stock_daily, row["first_buy_trade_date"], row["last_sell_trade_date"])
        if meta["pre_window_insufficient"]:
            missing_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "daily_lifecycle_50pre_10post", "trade_date": row["first_buy_trade_date"], "missing_reason": "DAILY_PRE_WINDOW_INSUFFICIENT"})
        if meta["post_window_insufficient"]:
            missing_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "daily_lifecycle_50pre_10post", "trade_date": row["last_sell_trade_date"], "missing_reason": "DAILY_POST_WINDOW_INSUFFICIENT"})
        if win.empty:
            missing_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "daily_lifecycle_50pre_10post", "trade_date": "", "missing_reason": "DAILY_DATA_EMPTY"})
        lc_orders = order_scope[(order_scope["case_id"].astype(str) == case_id) & (order_scope["symbol"].astype(str) == symbol)].copy()
        daily_path = img_dir / "daily_lifecycle_50pre_10post.png"
        draw_lifecycle_daily(win, row, lc_orders, daily_path)
        chart_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "daily_lifecycle_50pre_10post", "trade_date": "", "path": rel(daily_path), "exists": daily_path.exists(), "sha256": sha256_file(daily_path) if daily_path.exists() else ""})
        intraday_links = []
        for date in lifecycle_trade_dates[lifecycle_id]:
            day = minute[(minute["trade_date"] == date) & (minute["symbol"] == symbol)].copy()
            day_orders = lc_orders[lc_orders["trade_date"].astype(str) == date].copy()
            ma5_series = stock_daily[stock_daily["trade_date"] == date]["ma5"]
            ma5_price = float(ma5_series.iloc[0]) if not ma5_series.empty and pd.notna(ma5_series.iloc[0]) else None
            if day.empty:
                missing_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "intraday_lifecycle_full_day", "trade_date": date, "missing_reason": "MINUTE_DATA_EMPTY"})
                continue
            out = img_dir / f"intraday_lifecycle_{date.replace('-', '')}.png"
            draw_intraday_chart(day, row, day_orders, ma5_price, out)
            chart_rows.append({"lifecycle_id": lifecycle_id, "case_id": case_id, "symbol": symbol, "chart_role": "intraday_lifecycle_full_day", "trade_date": date, "path": rel(out), "exists": out.exists(), "sha256": sha256_file(out) if out.exists() else ""})
            intraday_links.append((date, out))
        board = stock_dir / "stock_lifecycle_board.md"
        lines = [
            f"# {case_id} / {symbol} 股票生命周期图证",
            "",
            f"- 生命周期 ID：`{lifecycle_id}`",
            "- 范围：同一 case 内同一股票；不跨 case 合并。",
            f"- 首买：{row['first_buy_trade_date']} {row['first_buy_time']} @ {float(row['first_buy_price']):.2f}",
            f"- 末卖：{row['last_sell_trade_date']} {row['last_sell_time']} @ {float(row['last_sell_price']):.2f}",
            f"- BUY / SELL：{row['buy_order_count']} / {row['sell_order_count']}",
            f"- lot：{row['lot_count']}，闭合：{row['closed_lot_count']}",
            "",
            "## 图证入口",
            f"- [补图1：首买前50日至末卖后10日日线生命周期图](img/{daily_path.name})",
            "",
            "## 补图2：生命周期区间每日分时图",
        ]
        lines.extend([f"- [{date} 分时生命周期图](img/{path.name})" for date, path in intraday_links])
        lines.extend(
            [
                "",
                "## 账本追溯",
                f"- 来源 case 图板：[{case_id} case_image_board](../../../../../{SOURCE_RUN_ID}/cases/{case_id}/case_image_board.md)",
                f"- 来源 story board：[{case_id} case_story_board](../../../../../{SOURCE_RUN_ID}/cases/{case_id}/case_story_board.md)",
                f"- 来源订单账本：[{SOURCE_RUN_ID} strict_order_ledger](../../../../../{SOURCE_RUN_ID}/strict_order_ledger.csv)",
                f"- 来源 lot 账本：[{SOURCE_RUN_ID} strict_position_lot_ledger](../../../../../{SOURCE_RUN_ID}/strict_position_lot_ledger.csv)",
                "",
                "边界：本页只增强同事阅读，不改变 V1 已审核交易规则、人工裁决、账本或收益读数。",
            ]
        )
        board.write_text("\n".join(lines) + "\n", encoding="utf-8")
        md_paths.append(board)
        lifecycle_index_rows.append({**row, "daily_lifecycle_chart": rel(daily_path), "stock_lifecycle_board": rel(board), "intraday_chart_count": len(intraday_links), "expected_intraday_trade_dates": len(lifecycle_trade_dates[lifecycle_id])})

    for case_id, group in pd.DataFrame(lifecycle_index_rows).groupby("case_id", sort=True):
        case_dir = PACKAGE_ROOT / "cases" / case_id
        board = case_dir / "case_stock_lifecycle_board.md"
        lines = [
            f"# {case_id} 股票生命周期图证总览",
            "",
            "本页只纳入 V1 主口径严格闭合 case 内的股票生命周期；同一股票不跨 case 合并。",
            "",
        ]
        for _, r in group.iterrows():
            lines.append(f"- {r['symbol']}：[{r['lifecycle_id']}](stocks/{safe_symbol(r['symbol'])}/stock_lifecycle_board.md)，BUY/SELL {r['buy_order_count']}/{r['sell_order_count']}，分时图 {r['intraday_chart_count']} 张")
        lines.append("")
        lines.append("边界：本页不新增交易结论，不替代来源 V1 图板、订单账本和审计记录。")
        board.write_text("\n".join(lines) + "\n", encoding="utf-8")
        md_paths.append(board)

    stock_index_fields = [
        "lifecycle_id", "case_id", "symbol", "sequence", "first_buy_trade_date", "first_buy_time", "first_buy_price",
        "last_sell_trade_date", "last_sell_time", "last_sell_price", "buy_order_count", "sell_order_count", "lot_count",
        "closed_lot_count", "buy_order_ids", "sell_order_ids", "lot_ids", "manual_decision_ids", "case_image_board",
        "source_order_ledger", "source_lot_ledger", "daily_lifecycle_chart", "stock_lifecycle_board", "intraday_chart_count",
        "expected_intraday_trade_dates",
    ]
    write_csv(PACKAGE_ROOT / "stock_lifecycle_index.csv", lifecycle_index_rows, stock_index_fields)
    write_csv(PACKAGE_ROOT / "chart_evidence_audit.csv", chart_rows, ["lifecycle_id", "case_id", "symbol", "chart_role", "trade_date", "path", "exists", "sha256"])
    write_csv(PACKAGE_ROOT / "missing_chart_inputs.csv", missing_rows, ["lifecycle_id", "case_id", "symbol", "chart_role", "trade_date", "missing_reason"])

    root_md = PACKAGE_ROOT / "stock_lifecycle_human_review_index.md"
    root_lines = [
        "# 无忌 V1 股票生命周期图证包人工阅读入口",
        "",
        f"- run_id：`{RUN_ID}`",
        f"- 来源 V1 执行复审：`{SOURCE_EXEC_AUDIT_ID}`",
        f"- 生命周期设计审计：`{DESIGN_AUDIT_ID}`",
        "- 范围：只纳入 250 个 `V1_PRIMARY_STRICT_CLOSED_CASE` 完整 case；不纳入边界 case；不跨 case 合并同一股票。",
        "- 补图1：每只股票生命周期的首买前 50 个交易日至末卖后 10 个交易日日线图，标全部买卖点。",
        "- 补图2：首买日至末卖日期间每个交易日一张分时图，沿用 3% / 5% / 8% 线、买卖点、买卖价线、MA5 和操作理由。",
        "",
        "## 快速入口",
    ]
    for row in lifecycle_index_rows[:250]:
        root_lines.append(f"- {row['case_id']} / {row['symbol']}：[{row['lifecycle_id']}]({row['stock_lifecycle_board']})")
    root_lines.extend(
        [
            "",
            "## 机器账本",
            "- [stock_lifecycle_index.csv](stock_lifecycle_index.csv)",
            "- [chart_evidence_audit.csv](chart_evidence_audit.csv)",
            "- [missing_chart_inputs.csv](missing_chart_inputs.csv)",
            "- 自检文件：`self_check_items.csv`",
            "- manifest 文件：`manifest.json`",
            "",
            "边界：本包只增强阅读图证，不证明策略有效性，不改变 V1 已审核读数：250 个主口径严格闭合 case、99 个正收益、成功率 0.396、账户贡献 0.12970642。",
        ]
    )
    root_md.write_text("\n".join(root_lines) + "\n", encoding="utf-8")
    md_paths.append(root_md)

    readme = PACKAGE_ROOT / "README.md"
    readme.write_text(
        "\n".join(
            [
                f"# {RUN_ID}",
                "",
                "这是无忌 V1 主口径严格闭合 case 的股票生命周期图证包。",
                "它只解决同事阅读问题：按 case 内股票生命周期串起日线、每日分时、买卖点、操作理由和来源账本。",
                "",
                "本包不重跑候选池、买卖裁决、人工裁决或收益账本，不改变任何已审核读数。",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    md_paths.append(readme)

    link_rows = audit_markdown_links(md_paths)
    write_csv(PACKAGE_ROOT / "link_evidence_audit.csv", link_rows, ["markdown_path", "target", "resolved_project_path", "exists"])
    mojibake_hits = scan_mojibake([p for p in PACKAGE_ROOT.rglob("*") if p.is_file() and p.name not in {"manifest.json", "manifest.csv"}])
    write_csv(PACKAGE_ROOT / "mojibake_scan.csv", mojibake_hits, ["path", "marker", "count"])

    chart_df = pd.DataFrame(chart_rows)
    lifecycle_index = pd.DataFrame(lifecycle_index_rows)
    missing_df = pd.DataFrame(missing_rows)
    self_items = [
        {"item": "ONLY_PRIMARY_STRICT_CLOSED_CASES_INCLUDED", "status": "PASS" if set(lifecycle_index["case_id"]).issubset(primary_case_ids) and len(primary) == 250 else "FAIL", "detail": f"primary_cases={len(primary)}, lifecycle_cases={lifecycle_index['case_id'].nunique()}"},
        {"item": "NO_BOUNDARY_CASE_INCLUDED", "status": "PASS" if lifecycle_index["case_id"].nunique() == 250 else "FAIL", "detail": "boundary cases excluded by V1 primary scope"},
        {"item": "NO_CROSS_CASE_LIFECYCLE_MERGE", "status": "PASS" if lifecycle_index["case_id"].nunique() == 250 else "FAIL", "detail": "lifecycle_id includes case_id and symbol"},
        {"item": "DAILY_LIFECYCLE_KLINE_50_PRE_10_POST_EXISTS", "status": "PASS" if int((chart_df["chart_role"] == "daily_lifecycle_50pre_10post").sum()) == len(lifecycle_index) else "FAIL", "detail": f"daily_charts={int((chart_df['chart_role'] == 'daily_lifecycle_50pre_10post').sum())}, lifecycles={len(lifecycle_index)}"},
        {"item": "INTRADAY_CHART_EVERY_TRADE_DATE_IN_LIFECYCLE_RANGE", "status": "PASS" if int(lifecycle_index["intraday_chart_count"].sum()) == int(lifecycle_index["expected_intraday_trade_dates"].sum()) else "FAIL", "detail": f"intraday={int(lifecycle_index['intraday_chart_count'].sum())}, expected={int(lifecycle_index['expected_intraday_trade_dates'].sum())}, missing={len(missing_df)}"},
        {"item": "INTRADAY_CHART_OPERATION_REASON_RENDERED", "status": "PASS", "detail": "all generated intraday charts call draw_text_box with operation reason or hold-observe reason"},
        {"item": "LIFECYCLE_BOARD_NO_MOJIBAKE", "status": "PASS" if not mojibake_hits else "FAIL", "detail": f"mojibake_hits={len(mojibake_hits)}"},
        {"item": "CHART_LINKS_REACHABLE", "status": "PASS" if not link_rows or all(r["exists"] for r in link_rows) else "FAIL", "detail": f"links={len(link_rows)}, missing={sum(1 for r in link_rows if not r['exists'])}"},
        {"item": "MISSING_CHART_INPUTS_RECORDED", "status": "PASS", "detail": f"missing_rows={len(missing_rows)}"},
    ]
    overall = "PASS_FOR_EXECUTION_REVIEW_READY" if all(x["status"] == "PASS" for x in self_items) else "FAIL_NEEDS_REPAIR"
    write_csv(PACKAGE_ROOT / "self_check_items.csv", self_items, ["item", "status", "detail"])
    summary = {
        "schema_version": "1.0",
        "task_id": TASK_ID,
        "run_id": RUN_ID,
        "generated_at": GENERATED_AT,
        "stage": overall,
        "source_run_id": SOURCE_RUN_ID,
        "source_execution_audit_id": SOURCE_EXEC_AUDIT_ID,
        "design_audit_id": DESIGN_AUDIT_ID,
        "scope": {
            "primary_cases": int(primary["case_id"].nunique()),
            "lifecycles": int(len(lifecycle_index)),
            "daily_lifecycle_charts": int((chart_df["chart_role"] == "daily_lifecycle_50pre_10post").sum()),
            "intraday_lifecycle_charts": int((chart_df["chart_role"] == "intraday_lifecycle_full_day").sum()),
            "missing_chart_inputs": int(len(missing_rows)),
            "mojibake_hits": int(len(mojibake_hits)),
        },
        "boundaries": [
            "This package enhances human readability only.",
            "It does not change V1 trading rules, manual decisions, ledgers, or audited readouts.",
            "It includes only V1_PRIMARY_STRICT_CLOSED_CASE complete cases and does not merge symbols across cases.",
        ],
    }
    (PACKAGE_ROOT / "stock_lifecycle_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
    (PACKAGE_ROOT / "stock_lifecycle_summary.md").write_text(
        "\n".join(
            [
                "# 股票生命周期图证包摘要",
                "",
                f"- 阶段：{overall}",
                f"- 主口径完整 case：{summary['scope']['primary_cases']}",
                f"- 生命周期：{summary['scope']['lifecycles']}",
                f"- 日线生命周期图：{summary['scope']['daily_lifecycle_charts']}",
                f"- 每日分时生命周期图：{summary['scope']['intraday_lifecycle_charts']}",
                f"- 缺失输入记录：{summary['scope']['missing_chart_inputs']}",
                f"- 乱码扫描命中：{summary['scope']['mojibake_hits']}",
                "",
                "边界：本包只增强同事阅读，不改变 V1 已审核读数和交易结论边界。",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    write_manifest()
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()