from __future__ import annotations

import hashlib
import json
import math
import os
import re
from collections import Counter
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-STRICT-NOTE-FULL-RERUN-20260614-001"
ROOT = Path(__file__).resolve().parents[1]
LOCAL_DB_INDEX = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md")
TZ = timezone(timedelta(hours=8))

OBSERVATION_TRADING_DAYS = 10
NEAR_MA5_PCT = 0.015
ROLLING_VOLUME_RATIO = 1.5
ROLLING_START_TIME = "10:40:00"
ROLLING_END_TIME = "14:40:00"


def now_iso() -> str:
    return datetime.now(TZ).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 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) >= 3:
        return f"{int(parts[0]):02d}:{int(parts[1]):02d}:{int(float(parts[2])):02d}"
    if len(parts) == 2:
        return f"{int(parts[0]):02d}:{int(parts[1]):02d}:00"
    return text


def read_mysql_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"密码：`([^`]+)`", text)
    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=240,
        write_timeout=240,
    )


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_22 = font(22)
FONT_28 = font(28)


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 window_dates(trade_dates: list[str], entry_date: str, start_date: str) -> list[str]:
    if entry_date not in trade_dates or start_date not in trade_dates:
        return []
    start = trade_dates.index(start_date)
    end = min(len(trade_dates) - 1, trade_dates.index(entry_date) + OBSERVATION_TRADING_DAYS)
    return trade_dates[start : end + 1]


def chunked(values: list, size: int):
    for i in range(0, len(values), size):
        yield values[i : i + size]


def fetch_market(signals: pd.DataFrame, trade_dates: list[str]) -> tuple[dict[tuple[str, str], pd.DataFrame], dict[tuple[str, str], dict]]:
    minute_pairs: set[tuple[str, str]] = set()
    all_daily_dates: set[str] = set()
    symbols = sorted(signals["symbol"].unique())
    for row in signals.itertuples(index=False):
        start_date = row.observation_trade_date if row.observation_trade_date else row.entry_trade_date
        dates = window_dates(trade_dates, row.entry_trade_date, start_date)
        for d in dates:
            minute_pairs.add((row.symbol, d))
            all_daily_dates.add(d)
        if row.entry_trade_date in trade_dates:
            idx = trade_dates.index(row.entry_trade_date)
            for d in trade_dates[max(0, idx - 5) : min(len(trade_dates), idx + OBSERVATION_TRADING_DAYS + 1)]:
                all_daily_dates.add(d)
    if not minute_pairs:
        return {}, {}

    minute_parts = []
    with get_conn() as conn:
        pair_list = sorted(minute_pairs, key=lambda x: (x[1], x[0]))
        for pairs in chunked(pair_list, 600):
            placeholders = ",".join(["(%s,%s)"] * len(pairs))
            params = []
            for symbol, trade_date in pairs:
                params.extend([trade_date, symbol])
            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, symbol) IN ({placeholders})
                    ORDER BY symbol, trade_date, trade_time
                    """,
                    conn,
                    params=params,
                )
            )
        ph = ",".join(["%s"] * len(symbols))
        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=[*symbols, min(all_daily_dates), max(all_daily_dates)],
        )

    minute = pd.concat(minute_parts, ignore_index=True) if minute_parts else pd.DataFrame()
    minute_lookup = {}
    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")
        for (symbol, trade_date), group in minute.groupby(["symbol", "trade_date"]):
            minute_lookup[(symbol, trade_date)] = group.sort_values("trade_time").reset_index(drop=True)

    daily_lookup = {}
    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())
        daily_lookup = {(r["symbol"], r["trade_date"]): r.to_dict() for _, r in daily.iterrows()}
    return minute_lookup, daily_lookup


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


def in_rolling_time(time_text: str) -> bool:
    return ROLLING_START_TIME <= time_text <= ROLLING_END_TIME


def rolling_candidate(row, minute_lookup: dict[tuple[str, str], pd.DataFrame], daily_lookup: dict[tuple[str, str], dict], trade_dates: list[str]) -> dict:
    start_date = row.observation_trade_date if row.observation_trade_date else row.entry_trade_date
    dates = window_dates(trade_dates, row.entry_trade_date, start_date)
    base = {
        "rolling_signal_id": f"ROLL-CAND-{row.lot_id}",
        "source_sell_signal_id": row.sell_signal_id,
        "lot_id": row.lot_id,
        "open_order_id": row.open_order_id,
        "case_id": row.case_id,
        "candidate_id": row.candidate_id,
        "symbol": row.symbol,
        "entry_trade_date": row.entry_trade_date,
        "rolling_trade_date": "",
        "rolling_time": "",
        "signal_type": "ROLLING_LOW_BUY_REVIEW_HELD",
        "code_suggested_action": "REVIEW_HELD",
        "code_suggested_reason_cn": "趋势观察后未找到五日线附近止跌放量的滚动低吸确认点，保留待人工复核。",
        "rolling_price": "",
        "ma5_close": "",
        "near_ma5_pct": "",
        "volume_ratio_vs_prev20m": "",
        "review_input_chart_path": "",
        "review_input_chart_sha256": "",
    }
    for d in dates:
        day = minute_lookup.get((row.symbol, d), pd.DataFrame())
        daily = daily_lookup.get((row.symbol, d), {})
        ma5 = safe_float(daily.get("ma5_close"))
        if day.empty or math.isnan(ma5):
            continue
        day = day.copy()
        day["prev20_volume_avg"] = day["volume"].rolling(20, min_periods=5).mean().shift(1)
        for r in day.itertuples(index=False):
            if not in_rolling_time(r.trade_time):
                continue
            price = safe_float(r.close_price)
            near = abs(price - ma5) / ma5 if ma5 else math.nan
            vol_avg = safe_float(getattr(r, "prev20_volume_avg", math.nan))
            vol_ratio = safe_float(r.volume) / vol_avg if vol_avg and not math.isnan(vol_avg) else math.nan
            if not math.isnan(near) and not math.isnan(vol_ratio) and near <= NEAR_MA5_PCT and vol_ratio >= ROLLING_VOLUME_RATIO:
                base.update(
                    {
                        "rolling_trade_date": d,
                        "rolling_time": r.trade_time,
                        "signal_type": "ADD_ROLLING_LOW_BUY",
                        "code_suggested_action": "BUY_ROLLING_LOW",
                        "code_suggested_reason_cn": f"趋势观察后回到五日线附近，距 MA5 {near:.2%}，分钟量能相对前 20 分钟均量放大 {vol_ratio:.2f} 倍，代码建议进入滚动低吸人工确认。",
                        "rolling_price": f"{price:.4f}",
                        "ma5_close": f"{ma5:.4f}",
                        "near_ma5_pct": f"{near:.8f}",
                        "volume_ratio_vs_prev20m": f"{vol_ratio:.4f}",
                    }
                )
                return base
    return base


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


def draw_rolling_chart(signal: dict, minute_lookup: dict[tuple[str, str], pd.DataFrame], daily_lookup: dict[tuple[str, str], dict]) -> None:
    d = signal["rolling_trade_date"] or signal["entry_trade_date"]
    day = minute_lookup.get((signal["symbol"], d), pd.DataFrame())
    out = ROOT / "charts" / "rolling_low_review" / signal["case_id"] / f"{signal['rolling_signal_id']}.png"
    out.parent.mkdir(parents=True, exist_ok=True)
    img = Image.new("RGB", (1500, 900), "#fbfaf6")
    draw = ImageDraw.Draw(img)
    draw.text((40, 24), f"严格版滚动低吸候选：{signal['case_id']} {signal['symbol']} {d}", fill="#111111", font=FONT_28)
    left, top, right, bottom = 70, 110, 1020, 620
    draw.rectangle((left, top, right, bottom), outline="#cccccc")
    prices = [safe_float(v) for v in day["close_price"].tolist()] if not day.empty else [safe_float(signal.get("rolling_price"), 1.0)]
    ma5 = safe_float(signal.get("ma5_close"))
    levels = prices + ([] if math.isnan(ma5) else [ma5])
    lo, hi = min(levels), max(levels)
    pad = max((hi - lo) * 0.08, 0.01)
    lo -= pad
    hi += pad

    def x_at(i: int) -> float:
        return left if len(prices) <= 1 else left + i * (right - left) / (len(prices) - 1)

    def y_at(price: float) -> float:
        return bottom - (price - lo) * (bottom - top) / (hi - lo)

    if prices and not day.empty:
        pts = [(x_at(i), y_at(p)) for i, p in enumerate(prices)]
        draw.line(pts, fill="#1f77b4", width=2)
    if not math.isnan(ma5):
        y = y_at(ma5)
        draw.line((left, y, right, y), fill="#8c564b", width=2)
        draw.text((right + 8, y - 10), f"日线MA5 {ma5:.2f}", fill="#8c564b", font=FONT_18)
    if not day.empty and signal["rolling_time"] in set(day["trade_time"].tolist()):
        idx = day.index[day["trade_time"].eq(signal["rolling_time"])][0]
        price = safe_float(day.loc[idx, "close_price"])
        x, y = x_at(int(idx)), y_at(price)
        draw.ellipse((x - 7, y - 7, x + 7, y + 7), fill="#d62728")
        draw.line((x, top, x, bottom), fill="#d62728", width=2)
        draw.text((x + 8, y - 28), f"低吸候选 {price:.2f}", fill="#d62728", font=FONT_18)

    side_x = 1060
    draw.text((side_x, 110), "外部人工裁决待填", fill="#111111", font=FONT_22)
    info = [
        f"代码建议：{signal['code_suggested_action']}",
        f"信号：{signal['signal_type']}",
        f"时间：{signal['rolling_trade_date']} {signal['rolling_time']}",
        f"MA5：{signal['ma5_close']}",
        "理由：",
    ]
    y = 150
    for line in info:
        draw.text((side_x, y), line, fill="#111111", font=FONT_18)
        y += 30
    for part in wrap(signal["code_suggested_reason_cn"], 18):
        draw.text((side_x, y), part, fill="#111111", font=FONT_18)
        y += 28
    draw.text((40, 825), "audit_view：本图只用于外部人工/AI人工确认滚动低吸，不生成最终 BUY；不买也必须写清理由。", fill="#666666", font=FONT_18)
    img.save(out)
    rel = out.relative_to(ROOT).as_posix()
    signal["review_input_chart_path"] = rel
    signal["review_input_chart_sha256"] = sha256_file(out)


def build_manifest() -> pd.DataFrame:
    rows = []
    for path in sorted(ROOT.rglob("*")):
        if path.is_file() and path.name not in {"manifest.csv", "manifest.json"}:
            rows.append({"path": path.relative_to(ROOT).as_posix(), "size": path.stat().st_size, "sha256": sha256_file(path)})
    return pd.DataFrame(rows)


def main() -> None:
    sell_df = pd.read_csv(ROOT / "strict_note_sell_rolling_review_candidate_ledger.csv", encoding="utf-8-sig")
    hold_df = sell_df[sell_df["code_suggested_action"].isin(["HOLD_WATCH", "HOLD_ABOVE_8"])].copy()
    trade_dates = fetch_trade_calendar()
    minute_lookup, daily_lookup = fetch_market(hold_df, trade_dates) if not hold_df.empty else ({}, {})

    rolling_rows = []
    for row in hold_df.itertuples(index=False):
        sig = rolling_candidate(row, minute_lookup, daily_lookup, trade_dates)
        draw_rolling_chart(sig, minute_lookup, daily_lookup)
        rolling_rows.append(sig)
    rolling_df = pd.DataFrame(rolling_rows)
    rolling_df.to_csv(ROOT / "strict_note_rolling_low_review_candidate_ledger.csv", index=False, encoding="utf-8-sig")

    sell_template = sell_df.copy()
    sell_template.insert(0, "artifact_type", "SELL_SIGNAL")
    sell_template["signal_id"] = sell_template["sell_signal_id"]
    sell_template["rolling_signal_id"] = ""

    rolling_template = rolling_df.copy()
    rolling_template.insert(0, "artifact_type", "ROLLING_LOW_SIGNAL")
    rolling_template["signal_id"] = ""
    if not rolling_template.empty:
        rolling_template["sell_signal_id"] = ""

    common_cols = [
        "artifact_type",
        "signal_id",
        "rolling_signal_id",
        "sell_signal_id",
        "lot_id",
        "open_order_id",
        "case_id",
        "candidate_id",
        "symbol",
        "entry_trade_date",
        "signal_type",
        "code_suggested_action",
        "code_suggested_reason_cn",
        "review_input_chart_path",
        "review_input_chart_sha256",
    ]
    template = pd.concat([sell_template, rolling_template], ignore_index=True, sort=False)
    for col in common_cols:
        if col not in template.columns:
            template[col] = ""
    template = template[common_cols].copy()
    template["external_decision_id"] = [f"EXT-SELLROLL-STRICT-NOTE-{i:06d}" for i in range(1, len(template) + 1)]
    for col in [
        "human_decision_action",
        "human_decision_reason_cn",
        "decision_operator",
        "decision_time",
        "decision_source",
        "accept_code_suggestion_flag",
        "reviewer_notes",
    ]:
        template[col] = ""
    template.to_csv(ROOT / "manual_sell_rolling_decision_external_template.csv", index=False, encoding="utf-8-sig")

    chart_rows = []
    for df, artifact, id_col in [(sell_df, "SELL_SIGNAL", "sell_signal_id"), (rolling_df, "ROLLING_LOW_SIGNAL", "rolling_signal_id")]:
        for r in df.itertuples(index=False):
            chart_rows.append(
                {
                    "artifact_type": artifact,
                    "signal_id": getattr(r, id_col),
                    "case_id": r.case_id,
                    "symbol": r.symbol,
                    "review_input_chart_path": r.review_input_chart_path,
                    "review_input_chart_sha256": r.review_input_chart_sha256,
                    "exists": (ROOT / r.review_input_chart_path).exists(),
                }
            )
    chart_df = pd.DataFrame(chart_rows)
    chart_df.to_csv(ROOT / "sell_rolling_chart_evidence_audit.csv", index=False, encoding="utf-8-sig")

    generated_at = now_iso()
    sell_counts = Counter(sell_df["code_suggested_action"])
    rolling_counts = Counter(rolling_df["code_suggested_action"]) if not rolling_df.empty else Counter()
    checks = [
        ("STRICT_SELL_SIGNAL_SCOPE_IS_424", len(sell_df) == 424, f"sell_signals={len(sell_df)}"),
        ("ROLLING_CANDIDATES_COVER_HOLD_SIGNALS", len(rolling_df) == len(hold_df), f"rolling={len(rolling_df)}, hold_signals={len(hold_df)}"),
        ("UNIFIED_TEMPLATE_COVERS_SELL_AND_ROLLING", len(template) == len(sell_df) + len(rolling_df), f"template={len(template)}"),
        ("TEMPLATE_FINAL_HUMAN_FIELDS_BLANK", template["human_decision_action"].astype(str).str.strip().eq("").all(), "human fields blank"),
        ("CHARTS_EXIST", bool(chart_df["exists"].all()), f"charts={len(chart_df)}, missing={int((~chart_df['exists']).sum())}"),
        ("NO_STRICT_PERFORMANCE_READOUT", True, "prep package only; no return/success/win-rate generated"),
    ]
    self_items = pd.DataFrame([{"item": k, "status": "PASS" if ok else "FAIL", "detail": detail} for k, ok, detail in checks])
    self_items.to_csv(ROOT / "sell_rolling_review_prep_self_check_items.csv", index=False, encoding="utf-8-sig")
    status = "PASS_FOR_SELL_ROLLING_MANUAL_REVIEW_PREP_READY" if self_items["status"].eq("PASS").all() else "FAIL"
    (ROOT / "sell_rolling_review_prep_self_check.json").write_text(
        json.dumps({"run_id": RUN_ID, "generated_at": generated_at, "stage": status, "pass_count": int(self_items["status"].eq("PASS").sum()), "fail_count": int(self_items["status"].eq("FAIL").sum())}, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    summary = {
        "run_id": RUN_ID,
        "generated_at": generated_at,
        "stage": status,
        "strict_buy_lots_input": 424,
        "sell_review_candidates": int(len(sell_df)),
        "rolling_low_review_candidates": int(len(rolling_df)),
        "manual_decision_template_rows": int(len(template)),
        "sell_code_suggested_action_counts": dict(sell_counts),
        "rolling_code_suggested_action_counts": dict(rolling_counts),
        "boundary": "Prep only: final sell/hold/rolling actions require external manual decision source and execution review.",
    }
    (ROOT / "sell_rolling_review_prep_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
    (ROOT / "sell_rolling_review_prep_summary.md").write_text(
        "# Strict Note Sell/Rolling Manual Review Prep\n\n"
        f"- generated_at: {generated_at}\n"
        "- strict BUY lots input: 424\n"
        f"- sell review candidates: {len(sell_df)}\n"
        f"- rolling low review candidates: {len(rolling_df)}\n"
        f"- manual decision template rows: {len(template)}\n"
        f"- sell code suggested actions: {dict(sell_counts)}\n"
        f"- rolling code suggested actions: {dict(rolling_counts)}\n\n"
        "Boundary: this is a manual-review prep package only. It does not generate final SELL orders, rolling BUY orders, return, success rate, win rate, drawdown, or strategy-effectiveness conclusions.\n",
        encoding="utf-8",
    )
    manifest = build_manifest()
    manifest.to_csv(ROOT / "manifest.csv", index=False, encoding="utf-8-sig")
    (ROOT / "manifest.json").write_text(json.dumps({"run_id": RUN_ID, "generated_at": generated_at, "file_count": int(len(manifest)), "files": manifest.to_dict("records")}, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(summary, ensure_ascii=False))


if __name__ == "__main__":
    main()