from __future__ import annotations

import hashlib
import json
import math
import os
import re
from pathlib import Path

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


RUN_ID = "RUN-ANA-WUJI-BASELINE-PILOT-20260607-001"
ROOT = Path(__file__).resolve().parents[1]
LOCAL_DB_INDEX = Path(
    r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md"
)


def read_password() -> str:
    env = os.environ.get("TIANXIA_MYSQL_PASSWORD") or os.environ.get("MYSQL_PWD")
    if env:
        return env
    text = LOCAL_DB_INDEX.read_text(encoding="utf-8")
    match = re.search(r"^\s*-\s*密码：`([^`]+)`", text, re.MULTILINE)
    if not match:
        raise RuntimeError("Unable to read local MySQL credential from approved local index.")
    return match.group(1)


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


def sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


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


FONT_TITLE = font(28)
FONT_MID = font(18)
FONT_SMALL = font(14)


def gate_label(raw: str) -> str:
    return "开仓闸门打开" if raw == "MKT_GATE_OPEN_PREV_DAY_UP_3000" else "开仓闸门关闭"


def status_label(raw: str) -> str:
    return "候选通过" if raw == "PASS" else "前高量能风险待复核"


def bool_label(raw) -> str:
    return "是" if str(raw).lower() == "true" else "否"


def draw_wrapped(draw: ImageDraw.ImageDraw, xy: tuple[int, int], text: str, max_chars: int, fill: str, used_font, line_gap: int = 8) -> int:
    x, y = xy
    chunks = []
    current = ""
    for ch in text:
        current += ch
        if len(current) >= max_chars:
            chunks.append(current)
            current = ""
    if current:
        chunks.append(current)
    for chunk in chunks:
        draw.text((x, y), chunk, fill=fill, font=used_font)
        y += used_font.size + line_gap if hasattr(used_font, "size") else 24
    return y


def price_y(value: float, low: float, high: float, top: int, bottom: int) -> int:
    if high <= low:
        return (top + bottom) // 2
    return bottom - int((value - low) / (high - low) * (bottom - top))


def draw_daily_chart(window: pd.DataFrame, cand: dict, out_path: Path) -> None:
    w, h = 1500, 920
    img = Image.new("RGB", (w, h), "#fbfbf7")
    d = ImageDraw.Draw(img)
    d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1")

    title = f"选股日K图：{cand['symbol']}  入场日 {cand['entry_trade_date']}"
    subtitle = (
        f"信号日 {cand['signal_trade_date']}｜排名 {cand['candidate_rank']}｜"
        f"量比 {float(cand['volume_ratio']):.2f}｜上影 {float(cand['upper_shadow_pct']):.2f}%｜"
        f"闸门 {gate_label(cand['market_gate_status'])}"
    )
    d.text((32, 24), title, fill="#111827", font=FONT_TITLE)
    d.text((32, 62), subtitle, fill="#334155", font=FONT_MID)

    plot_left, plot_top, plot_right, plot_bottom = 80, 115, 1060, 610
    vol_top, vol_bottom = 660, 820
    note_left, note_top = 1090, 118
    d.rectangle([plot_left, plot_top, plot_right, plot_bottom], outline="#94a3b8")
    d.rectangle([plot_left, vol_top, plot_right, vol_bottom], outline="#94a3b8")

    window = window.copy().reset_index(drop=True)
    price_low = float(window["low_price"].min()) * 0.98
    price_high = float(window["high_price"].max()) * 1.02
    max_vol = max(float(window["volume"].max()), 1.0)
    n = len(window)
    gap = (plot_right - plot_left) / max(n, 1)
    body_w = max(3, int(gap * 0.55))

    # Grid and price labels.
    for i in range(5):
        p = price_low + (price_high - price_low) * i / 4
        y = price_y(p, price_low, price_high, plot_top, plot_bottom)
        d.line([plot_left, y, plot_right, y], fill="#e2e8f0")
        d.text((18, y - 9), f"{p:.2f}", fill="#64748b", font=FONT_SMALL)

    ma_colors = {"ma5": "#2563eb", "ma20": "#f59e0b", "ma60": "#7c3aed"}
    ma_points: dict[str, list[tuple[int, int]]] = {k: [] for k in ma_colors}
    signal_date = str(cand["signal_trade_date"])
    last_limit_date = str(cand.get("last_limitup_date") or "")

    for i, row in window.iterrows():
        cx = int(plot_left + gap * i + gap / 2)
        op = float(row["open_price"])
        hi = float(row["high_price"])
        lo = float(row["low_price"])
        cl = float(row["close_price"])
        color = "#dc2626" if cl >= op else "#16a34a"
        d.line([cx, price_y(lo, price_low, price_high, plot_top, plot_bottom), cx, price_y(hi, price_low, price_high, plot_top, plot_bottom)], fill=color, width=2)
        y1 = price_y(op, price_low, price_high, plot_top, plot_bottom)
        y2 = price_y(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)
        trade_date = str(row["trade_date"])
        if trade_date == signal_date:
            d.line([cx, plot_top, cx, vol_bottom], fill="#0f172a", width=2)
            d.text((cx - 32, plot_top - 24), "信号日", fill="#0f172a", font=FONT_SMALL)
        if trade_date == last_limit_date:
            d.ellipse([cx - 8, price_y(hi, price_low, price_high, plot_top, plot_bottom) - 22, cx + 8, price_y(hi, price_low, price_high, plot_top, plot_bottom) - 6], fill="#ef4444")
            d.text((cx - 28, price_y(hi, price_low, price_high, plot_top, plot_bottom) - 48), "涨停记忆", fill="#ef4444", font=FONT_SMALL)
        if i % max(1, n // 8) == 0:
            d.text((cx - 28, vol_bottom + 8), trade_date[5:], fill="#64748b", font=FONT_SMALL)
        for ma in ma_colors:
            if pd.notna(row[ma]):
                ma_points[ma].append((cx, price_y(float(row[ma]), price_low, price_high, plot_top, plot_bottom)))

    for ma, pts in ma_points.items():
        if len(pts) > 1:
            d.line(pts, fill=ma_colors[ma], width=2)
    legend_x = plot_left + 8
    for ma, color in ma_colors.items():
        d.text((legend_x, plot_bottom + 12), ma.upper(), fill=color, font=FONT_SMALL)
        legend_x += 70

    d.rounded_rectangle([note_left, note_top, 1460, 820], radius=8, outline="#334155", fill="#ffffff")
    d.text((note_left + 18, note_top + 18), "候选判读", fill="#111827", font=FONT_TITLE)
    notes = [
        f"候选状态：{status_label(cand['candidate_status'])}",
        f"市场闸门：{gate_label(cand['market_gate_status'])}",
        f"上涨家数：{cand.get('up_count', '')}，下跌家数：{cand.get('down_count', '')}",
        f"近30日涨停记忆：{cand.get('last_limitup_date', '')}",
        f"量比：{float(cand['volume_ratio']):.2f}（阈值 1.70）",
        f"长上影：{float(cand['upper_shadow_pct']):.2f}%",
        f"触及前高：{bool_label(cand['touch_prev_high_flag'])}",
        f"前高量能通过：{bool_label(cand['prev_high_volume_pass_flag'])}",
        f"60日横盘标记：{bool_label(cand['flat60_flag'])}",
        "",
        "图片口径：decision_view",
        "只展示信号日及以前日线。",
        "本图只证明进入观察，",
        "不代表已经买入。",
    ]
    yy = note_top + 64
    for line in notes:
        if line:
            yy = draw_wrapped(d, (note_left + 18, yy), line, 24, "#334155", FONT_SMALL, line_gap=5)
        else:
            yy += 16

    footer = "无忌 baseline：候选池先看最近涨停记忆、放量、长上影、前高量能；买入还需后续 1 分钟 K 图和人工确认。"
    d.text((32, 870), footer, fill="#334155", font=FONT_MID)
    img.save(out_path)


def select_candidates(case_index: pd.DataFrame, candidate_ledger: pd.DataFrame) -> pd.DataFrame:
    selected_rows = []
    for _, case in case_index.iterrows():
        rows = candidate_ledger[candidate_ledger["entry_trade_date"] == case["entry_trade_date"]].copy()
        if rows.empty:
            continue
        if case["selection_bucket"] == "PREV_HIGH_REVIEW_RISK":
            review = rows[rows["candidate_status"] != "PASS"].sort_values("candidate_rank").head(2)
            strict = rows[rows["candidate_status"] == "PASS"].sort_values("candidate_rank").head(3)
            chosen = pd.concat([strict, review], ignore_index=True).sort_values("candidate_rank").head(5)
        else:
            strict = rows[rows["candidate_status"] == "PASS"].sort_values("candidate_rank").head(5)
            chosen = strict if len(strict) >= 5 else rows.sort_values("candidate_rank").head(5)
        chosen = chosen.copy()
        chosen["case_id"] = case["case_id"]
        chosen["case_status"] = case["case_status"]
        chosen["selection_bucket"] = case["selection_bucket"]
        selected_rows.append(chosen)
    return pd.concat(selected_rows, ignore_index=True) if selected_rows else pd.DataFrame()


def main() -> None:
    candidate_ledger = pd.read_csv(ROOT / "candidate_ledger.csv", encoding="utf-8-sig")
    case_index = pd.read_csv(ROOT / "case_index.csv", encoding="utf-8-sig")
    selected = select_candidates(case_index, candidate_ledger)
    selected.to_csv(ROOT / "selected_candidate_ledger.csv", index=False, encoding="utf-8-sig")
    if selected.empty:
        raise RuntimeError("No selected candidates available for image generation.")

    selected["signal_trade_date"] = pd.to_datetime(selected["signal_trade_date"])
    selected["entry_trade_date"] = pd.to_datetime(selected["entry_trade_date"])
    symbols = sorted(selected["symbol"].unique().tolist())
    min_date = (selected["signal_trade_date"].min() - pd.Timedelta(days=220)).strftime("%Y-%m-%d")
    max_date = selected["signal_trade_date"].max().strftime("%Y-%m-%d")
    placeholders = ",".join(["%s"] * len(symbols))

    with get_conn() as conn:
        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 trade_date BETWEEN %s AND %s
              AND symbol IN ({placeholders})
            ORDER BY symbol, trade_date
            """,
            conn,
            params=[min_date, max_date, *symbols],
        )
    daily["trade_date"] = pd.to_datetime(daily["trade_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)
    grouped = daily.groupby("symbol", group_keys=False)
    daily["ma5"] = grouped["close_price"].transform(lambda s: s.rolling(5, min_periods=1).mean())
    daily["ma20"] = grouped["close_price"].transform(lambda s: s.rolling(20, min_periods=1).mean())
    daily["ma60"] = grouped["close_price"].transform(lambda s: s.rolling(60, min_periods=1).mean())

    manifest_rows = []
    case_board_links = []
    for _, case in case_index.iterrows():
        case_id = case["case_id"]
        case_dir = ROOT / "cases" / case_id
        img_dir = case_dir / "img"
        img_dir.mkdir(parents=True, exist_ok=True)
        case_candidates = selected[selected["case_id"] == case_id].copy()
        case_candidates.to_csv(case_dir / "candidate_ledger.csv", index=False, encoding="utf-8-sig")
        case_manifest = []
        board_lines = [
            f"# {case_id} 图片审核板",
            "",
            f"入场日：`{case['entry_trade_date']}`",
            f"信号日：`{case['signal_trade_date']}`",
            f"样本分层：`{case['selection_bucket']}`",
            f"当前状态：`{case['case_status']}`",
            "",
            "本板当前只完成选股日 K 图阶段；买入 1 分钟 K 图、持仓图、卖出图和结果汇总图尚未生成。",
            "",
            "## 1. 市场闸门",
            "",
            f"- 闸门状态：`{case['market_gate_status']}`",
            f"- 上涨家数：`{case['up_count']}`",
            f"- 下跌家数：`{case['down_count']}`",
            "",
            "## 2. 选股证据图",
            "",
        ]
        story_lines = [
            f"# {case_id} 文字追溯板",
            "",
            "## 当前阶段",
            "",
            "`CANDIDATE_DAILY_IMAGE_PACKAGE_READY`",
            "",
            "当前只完成候选池和选股日 K 图；这些图只表示进入观察，不代表买入或收益结论。",
            "",
        ]
        for _, cand in case_candidates.iterrows():
            signal_date = pd.Timestamp(cand["signal_trade_date"])
            hist = daily[(daily["symbol"] == cand["symbol"]) & (daily["trade_date"] <= signal_date)].tail(100).copy()
            hist["trade_date"] = hist["trade_date"].dt.strftime("%Y-%m-%d")
            file_name = f"01_candidate_daily_100d_{cand['symbol'].replace('.', '_')}_{signal_date.strftime('%Y%m%d')}.png"
            out_path = img_dir / file_name
            cand_dict = cand.to_dict()
            cand_dict["signal_trade_date"] = signal_date.strftime("%Y-%m-%d")
            cand_dict["entry_trade_date"] = pd.Timestamp(cand["entry_trade_date"]).strftime("%Y-%m-%d")
            draw_daily_chart(hist, cand_dict, out_path)
            rel = out_path.relative_to(case_dir).as_posix()
            root_rel = out_path.relative_to(ROOT).as_posix()
            row = {
                "case_id": case_id,
                "symbol": cand["symbol"],
                "trade_date": cand_dict["signal_trade_date"],
                "event_id": cand["candidate_id"],
                "chart_role": "candidate_daily_100d_decision_view",
                "decision_time": f"{cand_dict['signal_trade_date']} close",
                "path": root_rel,
                "sha256": sha256_file(out_path),
                "status": "PASS",
                "note": "选股日K图，只展示信号日及以前数据；不代表买入。",
            }
            manifest_rows.append(row)
            case_manifest.append(row)
            board_lines.extend(
                [
                    f"### {cand['symbol']} / 候选排名 {cand['candidate_rank']}",
                    "",
                    f"![{cand['symbol']}]({rel})",
                    "",
                    f"- 候选状态：`{cand['candidate_status']}`",
                    f"- 量比：`{float(cand['volume_ratio']):.2f}`；长上影：`{float(cand['upper_shadow_pct']):.2f}%`",
                    f"- 前高量能通过：`{cand['prev_high_volume_pass_flag']}`",
                    "",
                ]
            )
        (case_dir / "image_manifest.csv").write_text(
            pd.DataFrame(case_manifest).to_csv(index=False), encoding="utf-8-sig"
        )
        (case_dir / "case_image_board.md").write_text("\n".join(board_lines) + "\n", encoding="utf-8")
        (case_dir / "case_story_board.md").write_text("\n".join(story_lines) + "\n", encoding="utf-8")
        case_files = [
            "candidate_ledger.csv",
            "image_manifest.csv",
            "case_image_board.md",
            "case_story_board.md",
        ]
        case_manifest_json = {
            "case_id": case_id,
            "run_id": RUN_ID,
            "stage": "CANDIDATE_DAILY_IMAGE_PACKAGE_READY",
            "files": [
                {
                    "path": name,
                    "size": (case_dir / name).stat().st_size,
                    "sha256": sha256_file(case_dir / name),
                }
                for name in case_files
            ],
            "image_count": len(case_manifest),
        }
        (case_dir / "manifest.json").write_text(
            json.dumps(case_manifest_json, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )
        case_board_links.append(f"- [{case_id}](cases/{case_id}/case_image_board.md)")

    image_manifest = pd.DataFrame(manifest_rows)
    image_manifest.to_csv(ROOT / "image_manifest.csv", index=False, encoding="utf-8-sig")
    root_board = [
        "# RUN 图片审核入口",
        "",
        f"run_id：`{RUN_ID}`",
        "阶段：`CANDIDATE_DAILY_IMAGE_PACKAGE_READY`",
        "",
        "当前只完成候选池和选股日 K 图；买卖分时图、订单账本和收益复算尚未生成。",
        "",
        "## 案例入口",
        "",
        *case_board_links,
        "",
    ]
    (ROOT / "case_image_board.md").write_text("\n".join(root_board), encoding="utf-8")
    (ROOT / "case_story_board.md").write_text(
        "# RUN 文字追溯入口\n\n当前只完成候选池和选股日 K 图。后续进入买点审核后补充分时图、订单、持仓和收益复算。\n",
        encoding="utf-8",
    )
    summary = {
        "schema_version": "1.0",
        "run_id": RUN_ID,
        "generated_at": "2026-06-08T00:50:00+08:00",
        "stage": "CANDIDATE_DAILY_IMAGE_PACKAGE_READY",
        "case_count": int(case_index["case_id"].nunique()),
        "selected_candidate_rows": int(len(selected)),
        "image_count": int(len(image_manifest)),
        "artifacts": {
            "selected_candidate_ledger.csv": {
                "size": (ROOT / "selected_candidate_ledger.csv").stat().st_size,
                "sha256": sha256_file(ROOT / "selected_candidate_ledger.csv"),
            },
            "image_manifest.csv": {
                "size": (ROOT / "image_manifest.csv").stat().st_size,
                "sha256": sha256_file(ROOT / "image_manifest.csv"),
            },
            "case_image_board.md": {
                "size": (ROOT / "case_image_board.md").stat().st_size,
                "sha256": sha256_file(ROOT / "case_image_board.md"),
            },
        },
        "next_step": "Generate entry 1-minute decision views and preliminary decision_log for selected pilot cases.",
    }
    (ROOT / "candidate_image_generation_summary.json").write_text(
        json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
    )
    (ROOT / "candidate_image_generation_summary.md").write_text(
        "\n".join(
            [
                "# candidate_image_generation_summary",
                "",
                f"run_id：`{RUN_ID}`",
                "阶段：`CANDIDATE_DAILY_IMAGE_PACKAGE_READY`",
                "",
                f"- 案例数：{summary['case_count']}",
                f"- 选中候选行：{summary['selected_candidate_rows']}",
                f"- 生成选股日 K 图：{summary['image_count']}",
                "",
                "当前图片只用于选股审核，不产生买入或收益结论。",
                "",
            ]
        ),
        encoding="utf-8",
    )


if __name__ == "__main__":
    main()