from __future__ import annotations

import hashlib
import json
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):
    for p in [
        Path("C:/Windows/Fonts/msyh.ttc"),
        Path("C:/Windows/Fonts/simhei.ttf"),
        Path("C:/Windows/Fonts/simsun.ttc"),
    ]:
        if p.exists():
            return ImageFont.truetype(str(p), size)
    return ImageFont.load_default()


FONT_TITLE = font(26)
FONT_MID = font(17)
FONT_SMALL = font(13)


def short_id(value: object) -> str:
    return str(value).replace(f"LOT-{RUN_ID}-", "LOT-")


def signal_label(signal_type: str) -> str:
    return {
        "STOP5_SELL_SIGNAL_CANDIDATE": "-5%止损候选",
        "TREND_TAKE_PROFIT_SIGNAL_CANDIDATE": "趋势止盈候选",
        "THREE_HIGH_NOT_RISING_SIGNAL_CANDIDATE": "三高不升候选",
        "HOLD_REVIEW_CANDIDATE": "继续持仓复核",
    }.get(signal_type, signal_type)


def wrap_text(text: object, max_chars: int = 22) -> list[str]:
    lines: list[str] = []
    for raw in str(text).splitlines():
        item = raw.strip()
        if not item:
            lines.append("")
            continue
        while len(item) > max_chars:
            lines.append(item[:max_chars])
            item = item[max_chars:]
        lines.append(item)
    return lines


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


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


def draw_daily_exit_chart(df: pd.DataFrame, lot: pd.Series, signal: dict, out_path: Path) -> None:
    w, h = 1500, 860
    img = Image.new("RGB", (w, h), "#fbfbf7")
    d = ImageDraw.Draw(img)
    d.rectangle([0, 0, w - 1, h - 1], outline="#cbd5e1")
    title = f"持仓/卖点日K复核：{lot.symbol}  入场 {lot.entry_trade_date}"
    d.text((32, 24), title, fill="#111827", font=FONT_TITLE)
    d.text((32, 58), "当前为卖点信号材料，不直接生成 SELL；真实卖出需 AI/人工账本裁决。", fill="#334155", font=FONT_MID)
    left, top, right, bottom = 80, 105, 1060, 590
    vol_top, vol_bottom = 640, 790
    note_left, note_top = 1090, 110
    d.rectangle([left, top, right, bottom], outline="#94a3b8")
    d.rectangle([left, vol_top, right, vol_bottom], outline="#94a3b8")
    if df.empty:
        d.text((left + 200, top + 180), "无日线数据", fill="#b91c1c", font=FONT_TITLE)
        price_low, price_high = 0, 1
    else:
        price_low = min(float(df.low_price.min()), float(lot.entry_price) * 0.95) * 0.98
        price_high = max(float(df.high_price.max()), float(lot.entry_price) * 1.05) * 1.02
        max_vol = max(float(df.volume.max()), 1.0)
        n = len(df)
        gap = (right - left) / max(n, 1)
        body_w = max(5, int(gap * 0.55))
        for i, row in df.reset_index(drop=True).iterrows():
            cx = int(left + gap * i + gap / 2)
            op, hi, lo, cl = [float(row[c]) for c in ["open_price", "high_price", "low_price", "close_price"]]
            color = "#dc2626" if cl >= op else "#16a34a"
            d.line([cx, y_price(lo, price_low, price_high, top, bottom), cx, y_price(hi, price_low, price_high, top, bottom)], fill=color, width=2)
            y1, y2 = y_price(op, price_low, price_high, top, bottom), y_price(cl, price_low, price_high, top, 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)
            date_s = str(row.trade_date)
            if date_s == str(lot.entry_trade_date):
                d.line([cx, top, cx, vol_bottom], fill="#2563eb", width=2)
                d.text((cx + 5, top + 8), "买入", fill="#2563eb", font=FONT_SMALL)
            if date_s == signal.get("signal_trade_date", ""):
                d.line([cx, top, cx, vol_bottom], fill="#b91c1c", width=2)
                d.text((cx + 5, top + 30), "信号", fill="#b91c1c", font=FONT_SMALL)
            if i % max(1, n // 6) == 0:
                d.text((cx - 28, vol_bottom + 8), date_s[5:], fill="#64748b", font=FONT_SMALL)
        for ref, label, color in [
            (float(lot.entry_price), "买入价", "#2563eb"),
            (float(lot.entry_price) * 0.95, "-5%止损", "#b91c1c"),
        ]:
            yy = y_price(ref, price_low, price_high, top, bottom)
            d.line([left, yy, right, yy], fill=color, width=2)
            d.text((right - 98, yy - 8), f"{label} {ref:.2f}", fill=color, font=FONT_SMALL)
    d.rounded_rectangle([note_left, note_top, 1460, 790], radius=8, outline="#334155", fill="#ffffff")
    d.text((note_left + 18, note_top + 18), "卖点信号", fill="#111827", font=FONT_TITLE)
    notes = [
        f"lot：{short_id(lot.trade_lot_id)}",
        f"入场：{lot.entry_trade_date} {lot.entry_time}",
        f"买入价：{float(lot.entry_price):.2f}",
        f"可卖日：{lot.sellable_from_trade_date}",
        f"信号：{signal_label(signal['signal_type'])}",
        f"信号日：{signal.get('signal_trade_date', '')}",
        "说明：",
        *wrap_text(signal["signal_note_cn"], 20),
        "",
        "本图只准备卖点材料，",
        "不写真实 SELL。",
    ]
    yy = note_top + 58
    for line in notes:
        d.text((note_left + 18, yy), line, fill="#334155", font=FONT_SMALL)
        yy += 24 if line else 12
    img.save(out_path)


def main() -> None:
    lots = pd.read_csv(ROOT / "position_lot_ledger.csv", encoding="utf-8-sig")
    if lots.empty:
        raise RuntimeError("No open lots found.")
    lots["entry_trade_date"] = pd.to_datetime(lots["entry_trade_date"])
    lots["sellable_from_trade_date"] = pd.to_datetime(lots["sellable_from_trade_date"])
    lots["entry_price"] = pd.to_numeric(lots["entry_price"], errors="coerce")
    symbols = sorted(lots.symbol.unique().tolist())
    min_date = (lots.entry_trade_date.min() - pd.Timedelta(days=90)).strftime("%Y-%m-%d")
    max_date = (lots.entry_trade_date.max() + pd.Timedelta(days=25)).strftime("%Y-%m-%d")
    sym_ph = ",".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 symbol IN ({sym_ph}) AND trade_date BETWEEN %s AND %s
            ORDER BY symbol, trade_date
            """,
            conn,
            params=[*symbols, min_date, max_date],
        )
        calendar = pd.read_sql(
            """
            SELECT calendar_date
            FROM a_share_trading_calendar
            WHERE is_trading_day=1 AND calendar_date BETWEEN %s AND %s
            ORDER BY calendar_date
            """,
            conn,
            params=[min_date, max_date],
        )
    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")
    calendar.calendar_date = pd.to_datetime(calendar.calendar_date)
    trade_dates = list(calendar.calendar_date.dt.strftime("%Y-%m-%d"))

    signal_rows = []
    manifest_rows = []
    for _, lot in lots.iterrows():
        entry = lot.entry_trade_date.strftime("%Y-%m-%d")
        sellable = lot.sellable_from_trade_date.strftime("%Y-%m-%d")
        idx = trade_dates.index(entry)
        obs_dates = trade_dates[idx : idx + 11]
        obs = daily[(daily.symbol == lot.symbol) & (daily.trade_date.dt.strftime("%Y-%m-%d").isin(obs_dates))].copy()
        sellable_obs = obs[obs.trade_date.dt.strftime("%Y-%m-%d") >= sellable].copy()
        signal = {
            "signal_type": "HOLD_REVIEW_CANDIDATE",
            "signal_trade_date": sellable if not sellable_obs.empty else "",
            "signal_note_cn": "观察窗口内未触发硬止损或趋势止盈代码信号，需人工继续复核卖点。",
            "proxy_or_substitute_used_flag": False,
        }
        if not sellable_obs.empty:
            stop = sellable_obs[sellable_obs.low_price.le(float(lot.entry_price) * 0.95)]
            trend = sellable_obs[
                sellable_obs.high_price.ge(float(lot.entry_price) * 1.05)
                & sellable_obs.close_price.ge(sellable_obs.open_price)
            ]
            three_high_signal = None
            if len(sellable_obs) >= 3:
                for i in range(2, len(sellable_obs)):
                    tri = sellable_obs.iloc[i - 2 : i + 1]
                    highs = list(tri.high_price)
                    if not (highs[0] < highs[1] < highs[2]):
                        three_high_signal = sellable_obs.iloc[i]
                        break
            if not stop.empty:
                r = stop.iloc[0]
                signal = {
                    "signal_type": "STOP5_SELL_SIGNAL_CANDIDATE",
                    "signal_trade_date": r.trade_date.strftime("%Y-%m-%d"),
                    "signal_note_cn": "卖出候选：可卖日后触及单票-5%硬止损线，需AI/人工确认真实SELL。",
                    "proxy_or_substitute_used_flag": False,
                }
            elif not trend.empty:
                r = trend.iloc[0]
                signal = {
                    "signal_type": "TREND_TAKE_PROFIT_SIGNAL_CANDIDATE",
                    "signal_trade_date": r.trade_date.strftime("%Y-%m-%d"),
                    "signal_note_cn": "卖出候选：可卖日后出现相对买入价5%以上趋势性上涨，需AI/人工确认是否按原文止盈卖出。",
                    "proxy_or_substitute_used_flag": False,
                }
            elif three_high_signal is not None:
                signal = {
                    "signal_type": "THREE_HIGH_NOT_RISING_SIGNAL_CANDIDATE",
                    "signal_trade_date": three_high_signal.trade_date.strftime("%Y-%m-%d"),
                    "signal_note_cn": "卖出候选：三日高点未逐步抬高，需AI/人工确认是否卖出。",
                    "proxy_or_substitute_used_flag": False,
                }
        case_dir = ROOT / "cases" / lot.case_id
        img_dir = case_dir / "img"
        img_dir.mkdir(parents=True, exist_ok=True)
        out_path = img_dir / f"05_exit_daily_signal_{lot.symbol.replace('.', '_')}_{entry}.png"
        window = daily[(daily.symbol == lot.symbol) & (daily.trade_date.dt.strftime("%Y-%m-%d") <= obs_dates[-1])].tail(80).copy()
        window["trade_date"] = window.trade_date.dt.strftime("%Y-%m-%d")
        lot_for_chart = lot.copy()
        lot_for_chart.entry_trade_date = entry
        lot_for_chart.sellable_from_trade_date = sellable
        draw_daily_exit_chart(window, lot_for_chart, signal, out_path)
        rel = out_path.relative_to(ROOT).as_posix()
        signal_rows.append(
            {
                "trade_lot_id": lot.trade_lot_id,
                "order_id": lot.order_id,
                "case_id": lot.case_id,
                "symbol": lot.symbol,
                "entry_trade_date": entry,
                "entry_time": lot.entry_time,
                "entry_price": f"{float(lot.entry_price):.4f}",
                "sellable_from_trade_date": sellable,
                "signal_type": signal["signal_type"],
                "signal_trade_date": signal.get("signal_trade_date", ""),
                "signal_note_cn": signal["signal_note_cn"],
                "manual_sell_decision": "PENDING_AI_MANUAL_DECISION",
                "decision_chart_path": rel,
                "t1_guard_passed_flag": signal.get("signal_trade_date", "") >= sellable if signal.get("signal_trade_date", "") else "",
                "proxy_or_substitute_used_flag": signal["proxy_or_substitute_used_flag"],
            }
        )
        manifest_rows.append(
            {
                "case_id": lot.case_id,
                "symbol": lot.symbol,
                "trade_date": signal.get("signal_trade_date", ""),
                "event_id": f"{lot.trade_lot_id}_exit_signal",
                "chart_role": "exit_daily_signal_review_view",
                "decision_time": f"{signal.get('signal_trade_date', '')} close",
                "path": rel,
                "sha256": sha256_file(out_path),
                "status": "PASS",
                "note": "卖点/持仓日K复核图；不自动生成SELL。",
            }
        )

    signals = pd.DataFrame(signal_rows)
    signals.to_csv(ROOT / "sell_signal_candidates.csv", index=False, encoding="utf-8-sig")
    root_manifest = pd.read_csv(ROOT / "image_manifest.csv", encoding="utf-8-sig")
    combined = pd.concat([root_manifest, pd.DataFrame(manifest_rows)], ignore_index=True)
    combined = combined.drop_duplicates(subset=["case_id", "symbol", "event_id", "chart_role"], keep="last")
    combined.to_csv(ROOT / "image_manifest.csv", index=False, encoding="utf-8-sig")

    for case_id, group in pd.DataFrame(manifest_rows).groupby("case_id"):
        case_dir = ROOT / "cases" / case_id
        board_path = case_dir / "case_image_board.md"
        existing = board_path.read_text(encoding="utf-8") if board_path.exists() else f"# {case_id} 图片审核板\n"
        existing = re.sub(r"\n## 5\. 卖点 / 持仓日K信号图\n[\s\S]*$", "", existing.rstrip())
        lines = [existing.rstrip(), "", "## 5. 卖点 / 持仓日K信号图", ""]
        for _, row in group.iterrows():
            rel = Path(row.path).relative_to(f"cases/{case_id}").as_posix()
            lines.extend([f"### {row.symbol}", "", f"![{row.symbol}]({rel})", "", "- 本图只提示卖点/持仓信号，不写真实 SELL。", ""])
        board_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
        combined[combined.case_id == case_id].to_csv(case_dir / "image_manifest.csv", index=False, encoding="utf-8-sig")

    summary = {
        "schema_version": "1.0",
        "run_id": RUN_ID,
        "generated_at": "2026-06-08T01:45:00+08:00",
        "stage": "SELL_SIGNAL_MATERIAL_READY",
        "signal_counts": signals.signal_type.value_counts().to_dict(),
        "signal_rows": len(signals),
        "exit_signal_images": len(manifest_rows),
        "artifacts": {
            "sell_signal_candidates.csv": {
                "size": (ROOT / "sell_signal_candidates.csv").stat().st_size,
                "sha256": sha256_file(ROOT / "sell_signal_candidates.csv"),
            },
            "image_manifest.csv": {
                "size": (ROOT / "image_manifest.csv").stat().st_size,
                "sha256": sha256_file(ROOT / "image_manifest.csv"),
            },
        },
        "boundary": "Signal material only; no SELL orders and no return statistics.",
        "next_step": "AI/manual sell decision from sell_signal_candidates.csv and images.",
    }
    (ROOT / "exit_review_generation_summary.json").write_text(
        json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    (ROOT / "exit_review_generation_summary.md").write_text(
        "\n".join(
            [
                "# exit_review_generation_summary",
                "",
                f"run_id：`{RUN_ID}`",
                "阶段：`SELL_SIGNAL_MATERIAL_READY`",
                "",
                f"- 卖点/持仓信号行：{len(signals)}",
                f"- 卖点/持仓日K图：{len(manifest_rows)}",
                "",
                "当前只生成卖点材料，不产生 SELL 或收益结论。",
                "",
            ]
        ),
        encoding="utf-8",
    )


if __name__ == "__main__":
    main()