from __future__ import annotations

import argparse
import csv
import json
import math
from collections import Counter
from datetime import datetime
from pathlib import Path
from typing import Any

import pandas as pd

from retry_public_5m_proxy_baostock_per_request import PACKAGE_ROOT, cache_path, date_key, fnum, local_daily_high, time_hhmm


STRICT_ROOT = PACKAGE_ROOT.parent / "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001"
LOCAL_MINUTE_ROOT = Path("E:/quant/2023_front_m")
LOCAL_DAILY_ROOT = Path("E:/quant/a_share_daily_front_20230101_20260508_complete/daily")

LOT_LEDGER = STRICT_ROOT / "strict_position_lot_ledger.csv"
SELL_SIGNAL_LEDGER = STRICT_ROOT / "strict_sell_signal_ledger.csv"
COVERAGE_CSV = PACKAGE_ROOT / "open_volume_stall_scan_coverage.csv"
CANDIDATE_CSV = PACKAGE_ROOT / "open_volume_stall_independent_candidates.csv"
SUMMARY_JSON = PACKAGE_ROOT / "open_volume_stall_scan_summary.json"
SUMMARY_MD = PACKAGE_ROOT / "open_volume_stall_scan_summary.md"
PROGRESS_JSON = PACKAGE_ROOT / "open_volume_stall_scan_progress.json"


def safe_float(value: Any) -> float:
    try:
        if pd.isna(value) or str(value).strip() == "":
            return math.nan
        return float(value)
    except Exception:
        return math.nan


def market_code(symbol: str) -> tuple[str, str]:
    code, market = symbol.split(".")
    return code, market


def minute_path(symbol: str) -> Path:
    code, market = market_code(symbol)
    return LOCAL_MINUTE_ROOT / market / f"price_{code}.csv"


def daily_path(symbol: str) -> Path:
    return LOCAL_DAILY_ROOT / f"{symbol}.csv"


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


def load_daily(symbol: str, cache: dict[str, pd.DataFrame]) -> pd.DataFrame:
    if symbol in cache:
        return cache[symbol]
    path = daily_path(symbol)
    if not path.exists():
        df = pd.DataFrame()
    else:
        df = pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig")
        df["date_key"] = df["trade_date"].map(date_key)
        for col in ["open", "high", "low", "close", "preClose", "volume"]:
            if col in df.columns:
                df[f"{col}_num"] = pd.to_numeric(df[col], errors="coerce")
        df = df.sort_values("date_key")
    cache[symbol] = df
    return df


def load_minute(symbol: str, cache: dict[str, pd.DataFrame]) -> pd.DataFrame:
    if symbol in cache:
        return cache[symbol]
    path = minute_path(symbol)
    if not path.exists() or path.stat().st_size == 0:
        df = pd.DataFrame()
    else:
        df = pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig")
        df["date_key"] = df["timetag"].str.slice(0, 8)
        df["trade_time"] = df["timetag"].str.slice(9, 17)
        df = df.rename(columns={"volumn": "volume"})
        for col in ["open", "high", "low", "close", "volume", "amount"]:
            if col in df.columns:
                df[f"{col}_num"] = pd.to_numeric(df[col], errors="coerce")
        df = df.sort_values(["date_key", "trade_time"])
    cache[symbol] = df
    return df


def trade_dates_for_lot(daily: pd.DataFrame, start_date: str, end_date: str, max_open_days: int) -> list[str]:
    if daily.empty:
        return []
    start = date_key(start_date)
    end = date_key(end_date) if end_date else str(daily["date_key"].max())
    rows = daily[(daily["date_key"] >= start) & (daily["date_key"] <= end)]["date_key"].tolist()
    return rows[:max_open_days] if not end_date else rows


def previous_dates(daily: pd.DataFrame, d: str, n: int) -> list[str]:
    dates = daily[daily["date_key"] < d]["date_key"].tolist()
    return dates[-n:]


def daily_prev_close(daily: pd.DataFrame, d: str) -> float:
    prevs = previous_dates(daily, d, 1)
    if not prevs:
        return math.nan
    row = daily[daily["date_key"] == prevs[-1]]
    return safe_float(row.iloc[0].get("close_num", math.nan)) if not row.empty else math.nan


def scale_for_day(symbol: str, d: str, day_high: float, daily_cache: dict[str, pd.DataFrame]) -> tuple[float, str]:
    daily_high = local_daily_high(symbol, f"{d[:4]}-{d[4:6]}-{d[6:8]}", daily_cache)
    if daily_high is None or not day_high or math.isnan(day_high) or day_high <= 0:
        return 1.0, "RAW_PRICE_NO_SCALE"
    return daily_high / day_high, "SCALED_TO_LOCAL_DAILY_HIGH"


def exact_1m_metrics(symbol: str, d: str, daily: pd.DataFrame, minute: pd.DataFrame, daily_cache: dict[str, pd.DataFrame]) -> dict[str, Any]:
    if minute.empty:
        return {"scan_status": "NO_LOCAL_1M_FILE"}
    day = minute[minute["date_key"] == d]
    if day.empty:
        return {"scan_status": "NO_LOCAL_1M_DATE"}
    first = day.iloc[0]
    prev_close = daily_prev_close(daily, d)
    if math.isnan(prev_close) or prev_close <= 0:
        return {"scan_status": "NO_DAILY_PREV_CLOSE"}
    scale, source = scale_for_day(symbol, d, safe_float(day["high_num"].max()), daily_cache)
    first_close_raw = safe_float(first.get("close_num"))
    first_close = first_close_raw * scale if not math.isnan(first_close_raw) else math.nan
    first_gain = first_close / prev_close - 1 if not math.isnan(first_close) else math.nan
    vols = []
    for pd_key in previous_dates(daily, d, 5):
        prev_day = minute[minute["date_key"] == pd_key]
        if not prev_day.empty:
            vols.append(safe_float(prev_day.iloc[0].get("volume_num")))
    vols = [v for v in vols if not math.isnan(v) and v > 0]
    avg_first_vol = float(pd.Series(vols).mean()) if vols else math.nan
    first_vol = safe_float(first.get("volume_num"))
    vol_ratio = first_vol / avg_first_vol if avg_first_vol and not math.isnan(avg_first_vol) and avg_first_vol > 0 else math.nan
    hit = (not math.isnan(vol_ratio)) and vol_ratio >= 10 and (not math.isnan(first_gain)) and 0.01 <= first_gain <= 0.02
    return {
        "scan_status": "SCANNED_EXACT_1M",
        "scan_basis": "EXACT_1M",
        "price_source": source,
        "first_bar_time": first.get("trade_time", ""),
        "first_close_raw": first_close_raw,
        "first_close_scaled": first_close,
        "prev_close_daily": prev_close,
        "first_gain_pct": first_gain,
        "first_volume": first_vol,
        "avg_previous_first_volume": avg_first_vol,
        "open_volume_ratio": vol_ratio,
        "stall_hit": hit,
    }


def load_5m_day(symbol: str, d: str) -> pd.DataFrame:
    path = cache_path(symbol, f"{d[:4]}-{d[4:6]}-{d[6:8]}")
    if not path.exists() or path.stat().st_size == 0:
        return pd.DataFrame()
    df = pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig")
    if df.empty:
        return df
    df["hhmm"] = df["time"].map(time_hhmm)
    for col in ["open", "high", "low", "close", "volume", "amount"]:
        if col in df.columns:
            df[f"{col}_num"] = pd.to_numeric(df[col], errors="coerce")
    return df.sort_values("hhmm")


def proxy_5m_metrics(symbol: str, d: str, daily: pd.DataFrame, daily_cache: dict[str, pd.DataFrame]) -> dict[str, Any]:
    day = load_5m_day(symbol, d)
    if day.empty:
        return {"scan_status": "NO_5M_PROXY_CACHE"}
    prev_close = daily_prev_close(daily, d)
    if math.isnan(prev_close) or prev_close <= 0:
        return {"scan_status": "NO_DAILY_PREV_CLOSE"}
    first = day.iloc[0]
    scale, source = scale_for_day(symbol, d, safe_float(day["high_num"].max()), daily_cache)
    first_close_raw = safe_float(first.get("close_num"))
    first_close = first_close_raw * scale if not math.isnan(first_close_raw) else math.nan
    first_gain = first_close / prev_close - 1 if not math.isnan(first_close) else math.nan
    vols = []
    missing_prev = 0
    for pd_key in previous_dates(daily, d, 5):
        prev_day = load_5m_day(symbol, pd_key)
        if prev_day.empty:
            missing_prev += 1
        else:
            vols.append(safe_float(prev_day.iloc[0].get("volume_num")))
    vols = [v for v in vols if not math.isnan(v) and v > 0]
    if not vols:
        return {"scan_status": "NO_PROXY_PREVIOUS_5M_VOLUME", "scan_basis": "PROXY_5M", "missing_previous_5m_days": missing_prev}
    avg_first_vol = float(pd.Series(vols).mean())
    first_vol = safe_float(first.get("volume_num"))
    vol_ratio = first_vol / avg_first_vol if avg_first_vol > 0 else math.nan
    hit = (not math.isnan(vol_ratio)) and vol_ratio >= 10 and (not math.isnan(first_gain)) and 0.01 <= first_gain <= 0.02
    return {
        "scan_status": "SCANNED_PROXY_5M",
        "scan_basis": "PROXY_5M",
        "price_source": source,
        "first_bar_time": first.get("hhmm", ""),
        "first_close_raw": first_close_raw,
        "first_close_scaled": first_close,
        "prev_close_daily": prev_close,
        "first_gain_pct": first_gain,
        "first_volume": first_vol,
        "avg_previous_first_volume": avg_first_vol,
        "open_volume_ratio": vol_ratio,
        "stall_hit": hit,
        "missing_previous_5m_days": missing_prev,
    }


def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
    keys: list[str] = []
    seen: set[str] = set()
    for row in rows:
        for key in row:
            if key not in seen:
                seen.add(key)
                keys.append(key)
    with path.open("w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(rows)


def write_progress(payload: dict[str, Any]) -> None:
    PROGRESS_JSON.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--max-open-lot-days", type=int, default=10)
    parser.add_argument("--use-proxy-5m", action="store_true")
    args = parser.parse_args()

    lots = pd.read_csv(LOT_LEDGER, dtype=str, keep_default_na=False, encoding="utf-8-sig")
    signals = pd.read_csv(SELL_SIGNAL_LEDGER, dtype=str, keep_default_na=False, encoding="utf-8-sig")
    signal_by_id = signals.set_index("signal_id").to_dict("index") if "signal_id" in signals.columns else {}
    daily_cache: dict[str, pd.DataFrame] = {}
    coverage: list[dict[str, Any]] = []
    candidates: list[dict[str, Any]] = []

    grouped = list(lots.groupby("symbol", sort=True))
    for symbol_index, (symbol, symbol_lots) in enumerate(grouped, start=1):
        daily = load_daily(symbol, daily_cache)
        minute = load_minute(symbol, {})
        for lot in symbol_lots.to_dict("records"):
            dates = trade_dates_for_lot(
                daily,
                lot["sellable_from_trade_date"],
                lot.get("exit_trade_date", ""),
                args.max_open_lot_days,
            )
            exit_time = normalize_time(lot.get("exit_time", ""))
            exit_signal = signal_by_id.get(lot.get("exit_signal_id", ""), {})
            for d in dates:
                metrics = exact_1m_metrics(symbol, d, daily, minute, daily_cache)
                if metrics.get("scan_status", "").startswith("NO_") and args.use_proxy_5m:
                    metrics = proxy_5m_metrics(symbol, d, daily, daily_cache)
                first_time = normalize_time(str(metrics.get("first_bar_time", "")))
                if (
                    lot.get("exit_trade_date")
                    and date_key(lot["exit_trade_date"]) == d
                    and exit_time
                    and first_time
                    and exit_time <= first_time
                ):
                    metrics = {"scan_status": "SKIP_EXIT_BEFORE_OPEN_BAR", **metrics}

                base = {
                    "strict_lot_id": lot.get("strict_lot_id", ""),
                    "source_lot_id": lot.get("source_lot_id", ""),
                    "case_id": lot.get("case_id", ""),
                    "symbol": symbol,
                    "scan_trade_date": f"{d[:4]}-{d[4:6]}-{d[6:8]}",
                    "entry_trade_date": lot.get("entry_trade_date", ""),
                    "sellable_from_trade_date": lot.get("sellable_from_trade_date", ""),
                    "exit_trade_date": lot.get("exit_trade_date", ""),
                    "exit_time": lot.get("exit_time", ""),
                    "existing_exit_signal_id": lot.get("exit_signal_id", ""),
                    "existing_exit_signal_type": exit_signal.get("signal_type", ""),
                }
                row = {**base, **metrics}
                coverage.append(row)
                if str(row.get("stall_hit")) == "True":
                    candidates.append(row)
        if symbol_index % 50 == 0:
            write_progress(
                {
                    "status": "RUNNING",
                    "symbol_index": symbol_index,
                    "symbol_count": len(grouped),
                    "coverage_rows": len(coverage),
                    "candidate_rows": len(candidates),
                    "generated_at": datetime.now().isoformat(timespec="seconds"),
                }
            )

    write_csv(COVERAGE_CSV, coverage)
    write_csv(CANDIDATE_CSV, candidates)
    summary = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "lot_rows": len(lots),
        "coverage_rows": len(coverage),
        "candidate_rows": len(candidates),
        "coverage_status_counts": dict(Counter(str(row.get("scan_status", "")) for row in coverage)),
        "candidate_basis_counts": dict(Counter(str(row.get("scan_basis", "")) for row in candidates)),
        "threshold": "open_volume_ratio >= 10 and 0.01 <= first_gain_pct <= 0.02",
        "boundaries": [
            "EXACT_1M rows use local 1-minute files and daily previous close, with prices scaled to local daily high when possible.",
            "PROXY_5M rows use cached public 5-minute bars and are not strict opening-one-minute evidence.",
            "This is an independent scan; candidate rows indicate possible omissions that require chart/manual review before changing V1 ledgers.",
        ],
    }
    SUMMARY_JSON.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
    SUMMARY_MD.write_text(
        "\n".join(
            [
                "# Open Volume Stall Independent Scan",
                "",
                f"- generated_at: {summary['generated_at']}",
                f"- lot_rows: {summary['lot_rows']}",
                f"- coverage_rows: {summary['coverage_rows']}",
                f"- candidate_rows: {summary['candidate_rows']}",
                f"- coverage_status_counts: {json.dumps(summary['coverage_status_counts'], ensure_ascii=False)}",
                f"- candidate_basis_counts: {json.dumps(summary['candidate_basis_counts'], ensure_ascii=False)}",
                f"- threshold: {summary['threshold']}",
                "",
                "## Boundaries",
                *[f"- {item}" for item in summary["boundaries"]],
                "",
            ]
        ),
        encoding="utf-8",
    )
    write_progress({"status": "FINISHED", **summary})
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()