from __future__ import annotations

import argparse
import csv
import json
import math
import time
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
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,
    fetch_one_with_optional_timeout,
    fnum,
    local_daily_high,
    time_hhmm,
    to_baostock_code,
)


SOURCE_EVENTS = PACKAGE_ROOT / "minute_event_coverage_audit.csv"
FETCH_LEDGER = PACKAGE_ROOT / "minute_gap_5m_proxy_fetch_inventory.csv"
AUDIT_LEDGER = PACKAGE_ROOT / "minute_event_5m_proxy_audit.csv"
SUMMARY_JSON = PACKAGE_ROOT / "minute_event_5m_proxy_summary.json"
SUMMARY_MD = PACKAGE_ROOT / "minute_event_5m_proxy_summary.md"


def normalize_event_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 bar_time(value: str) -> str:
    text = str(value)
    if len(text) >= 12 and text[:8].isdigit():
        return f"{text[8:10]}:{text[10:12]}:00"
    return normalize_event_time(text)


def load_events() -> pd.DataFrame:
    events = pd.read_csv(SOURCE_EVENTS, dtype=str, keep_default_na=False, encoding="utf-8-sig")
    gaps = events[events["exact_time_found"] != "True"].copy()
    gaps["event_time_norm"] = gaps["event_time"].map(normalize_event_time)
    return gaps


def load_fetch_rows(gaps: pd.DataFrame) -> list[dict[str, str]]:
    existing: dict[tuple[str, str], dict[str, str]] = {}
    if FETCH_LEDGER.exists():
        with FETCH_LEDGER.open("r", newline="", encoding="utf-8-sig") as f:
            for row in csv.DictReader(f):
                existing[(row["symbol"], row["event_trade_date"])] = row

    rows: list[dict[str, str]] = []
    unique = (
        gaps[gaps["event_time_provided"] == "True"][["symbol", "event_trade_date"]]
        .drop_duplicates()
        .sort_values(["symbol", "event_trade_date"])
    )
    for row in unique.to_dict("records"):
        key = (row["symbol"], row["event_trade_date"])
        current = existing.get(key, {})
        path = cache_path(row["symbol"], row["event_trade_date"])
        if path.exists() and path.stat().st_size > 0:
            try:
                count = len(pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig"))
            except Exception:
                count = 0
            if count > 0:
                current = {
                    **current,
                    "fetch_status": "FETCH_OK",
                    "row_count": str(count),
                    "cache_path": str(path.relative_to(PACKAGE_ROOT)).replace("\\", "/"),
                    "error": "",
                }
        rows.append(
            {
                "symbol": row["symbol"],
                "event_trade_date": row["event_trade_date"],
                "baostock_code": to_baostock_code(row["symbol"]) or "",
                "fetch_status": current.get("fetch_status", "PENDING"),
                "row_count": current.get("row_count", "0"),
                "cache_path": current.get("cache_path", ""),
                "error": current.get("error", ""),
            }
        )
    return rows


def write_fetch_rows(rows: list[dict[str, str]]) -> None:
    keys = ["symbol", "event_trade_date", "baostock_code", "fetch_status", "row_count", "cache_path", "error"]
    with FETCH_LEDGER.open("w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(rows)


def update_fetch_rows(rows: list[dict[str, str]], args: argparse.Namespace) -> int:
    retry_statuses = {item.strip() for item in args.retry_status.split(",") if item.strip()}
    if retry_statuses:
        pending = [row for row in rows if row.get("fetch_status") in retry_statuses]
    else:
        pending = [row for row in rows if row.get("fetch_status") != "FETCH_OK"]
    pending = pending[: args.fetch_limit]
    attempted = 0

    def fetch_row(row: dict[str, str]) -> tuple[dict[str, str], tuple[str, str, str, str]]:
        return row, fetch_one_with_optional_timeout(row["symbol"], row["event_trade_date"], args.per_request_timeout)

    if not pending:
        return 0
    if args.workers > 1:
        with ThreadPoolExecutor(max_workers=args.workers) as pool:
            futures = [pool.submit(fetch_row, row) for row in pending]
            for future in as_completed(futures):
                row, result = future.result()
                status, cache, count, error = result
                row["fetch_status"] = status
                row["row_count"] = count
                row["cache_path"] = cache
                row["error"] = error
                row["baostock_code"] = to_baostock_code(row["symbol"]) or ""
                attempted += 1
                write_fetch_rows(rows)
                if args.sleep:
                    time.sleep(args.sleep)
    else:
        for row in pending:
            status, cache, count, error = fetch_one_with_optional_timeout(
                row["symbol"], row["event_trade_date"], args.per_request_timeout
            )
            row["fetch_status"] = status
            row["row_count"] = count
            row["cache_path"] = cache
            row["error"] = error
            row["baostock_code"] = to_baostock_code(row["symbol"]) or ""
            attempted += 1
            write_fetch_rows(rows)
            if args.sleep:
                time.sleep(args.sleep)
    return attempted


def load_5m_frame(path_text: str) -> pd.DataFrame:
    if not path_text:
        return pd.DataFrame()
    path = PACKAGE_ROOT / path_text
    if not path.exists():
        return pd.DataFrame()
    df = pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig")
    if df.empty:
        return df
    df["bar_time"] = df["time"].map(bar_time)
    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("bar_time")


def scaled_values(df: pd.DataFrame, symbol: str, trade_date: str, daily_cache: dict[str, pd.DataFrame]) -> tuple[float | None, float | None, str]:
    if df.empty or "high_num" not in df.columns:
        return None, None, "NO_5M_HIGH"
    raw_high = float(df["high_num"].max())
    local_high = local_daily_high(symbol, trade_date, daily_cache)
    if local_high is None or raw_high <= 0:
        return None, None, "RAW_5M_NO_LOCAL_DAILY_SCALE"
    return raw_high, local_high / raw_high, "BAOSTOCK_5M_SCALED_TO_LOCAL_DAILY_HIGH"


def choose_proxy_bar(df: pd.DataFrame, event_time: str) -> pd.Series | None:
    if df.empty or not event_time:
        return None
    rows = df[df["bar_time"] >= event_time]
    if rows.empty:
        rows = df.tail(1)
    return rows.iloc[0] if not rows.empty else None


def build_audit(gaps: pd.DataFrame, fetch_rows: list[dict[str, str]]) -> dict[str, Any]:
    fetch_by_key = {(row["symbol"], row["event_trade_date"]): row for row in fetch_rows}
    frame_cache: dict[str, pd.DataFrame] = {}
    daily_cache: dict[str, pd.DataFrame] = {}
    out: list[dict[str, Any]] = []

    for row in gaps.to_dict("records"):
        symbol = row["symbol"]
        trade_date = row["event_trade_date"]
        event_time = row.get("event_time_norm", "")
        fetch = fetch_by_key.get((symbol, trade_date), {})
        path_text = fetch.get("cache_path", "")
        df = frame_cache.get(path_text)
        if df is None:
            df = load_5m_frame(path_text)
            frame_cache[path_text] = df
        raw_day_high, scale, price_source = scaled_values(df, symbol, trade_date, daily_cache)
        proxy_bar = choose_proxy_bar(df, event_time) if row.get("event_time_provided") == "True" else None

        proxy_status = "NO_EVENT_TIME"
        values: dict[str, Any] = {}
        if row.get("event_time_provided") == "True":
            if fetch.get("fetch_status") != "FETCH_OK":
                proxy_status = "NO_5M_DATA"
            elif proxy_bar is None:
                proxy_status = "NO_5M_BAR"
            else:
                proxy_status = "PROXY_5M_BAR_FOUND"
                values = {
                    "proxy_5m_bar_time": proxy_bar.get("bar_time", ""),
                    "proxy_5m_raw_open": proxy_bar.get("open", ""),
                    "proxy_5m_raw_high": proxy_bar.get("high", ""),
                    "proxy_5m_raw_low": proxy_bar.get("low", ""),
                    "proxy_5m_raw_close": proxy_bar.get("close", ""),
                    "proxy_5m_volume": proxy_bar.get("volume", ""),
                }
                if scale is not None:
                    for col in ["open", "high", "low", "close"]:
                        num = fnum(proxy_bar.get(col, ""))
                        values[f"proxy_5m_scaled_{col}"] = "" if num is None else num * scale

        out.append(
            {
                "event_source": row.get("event_source", ""),
                "event_id": row.get("event_id", ""),
                "case_id": row.get("case_id", ""),
                "symbol": symbol,
                "event_trade_date": trade_date,
                "event_time": row.get("event_time", ""),
                "action_or_signal": row.get("action_or_signal", ""),
                "signal_type": row.get("signal_type", ""),
                "original_coverage_status": row.get("coverage_status", ""),
                "minute_file_status": row.get("minute_file_status", ""),
                "public_5m_fetch_status": fetch.get("fetch_status", ""),
                "public_5m_row_count": fetch.get("row_count", ""),
                "public_5m_cache_path": path_text,
                "proxy_status": proxy_status,
                "comparison_price_source": price_source,
                "public_5m_full_day_high_raw": raw_day_high if raw_day_high is not None else "",
                "public_5m_to_local_daily_scale": scale if scale is not None else "",
                **values,
            }
        )

    keys: list[str] = []
    seen: set[str] = set()
    for row in out:
        for key in row:
            if key not in seen:
                seen.add(key)
                keys.append(key)
    with AUDIT_LEDGER.open("w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(out)

    summary = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "input_gap_events": len(gaps),
        "gap_events_with_time": int((gaps["event_time_provided"] == "True").sum()),
        "unique_symbol_dates_requested": len(fetch_rows),
        "fetch_status_counts": dict(Counter(row.get("fetch_status", "") for row in fetch_rows)),
        "proxy_status_counts": dict(Counter(str(row["proxy_status"]) for row in out)),
        "boundaries": [
            "This is a 5-minute public K-line proxy for events whose original 1-minute evidence was missing or not exact.",
            "A proxy bar is the first 5-minute bar with period-end time >= the event time; it is not a precise 1-minute replacement.",
            "Raw Baostock front-adjusted prices are scaled to the local daily front-adjusted high when local daily data is available.",
            "NO_EVENT_TIME rows cannot be mapped to an exact proxy bar without a rule-specific timestamp.",
        ],
    }
    SUMMARY_JSON.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
    SUMMARY_MD.write_text(
        "\n".join(
            [
                "# Minute Gap 5m Proxy Summary",
                "",
                f"- generated_at: {summary['generated_at']}",
                f"- input_gap_events: {summary['input_gap_events']}",
                f"- gap_events_with_time: {summary['gap_events_with_time']}",
                f"- unique_symbol_dates_requested: {summary['unique_symbol_dates_requested']}",
                f"- fetch_status_counts: {json.dumps(summary['fetch_status_counts'], ensure_ascii=False)}",
                f"- proxy_status_counts: {json.dumps(summary['proxy_status_counts'], ensure_ascii=False)}",
                "",
                "## Boundaries",
                *[f"- {item}" for item in summary["boundaries"]],
                "",
            ]
        ),
        encoding="utf-8",
    )
    return summary


def write_manifest() -> None:
    rows = []
    for path in sorted(PACKAGE_ROOT.rglob("*")):
        if path.is_file():
            rows.append({"path": str(path.relative_to(PACKAGE_ROOT)).replace("\\", "/"), "bytes": path.stat().st_size})
    with (PACKAGE_ROOT / "manifest.csv").open("w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=["path", "bytes"])
        writer.writeheader()
        writer.writerows(rows)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--fetch-limit", type=int, default=100)
    parser.add_argument("--workers", type=int, default=2)
    parser.add_argument("--per-request-timeout", type=int, default=120)
    parser.add_argument("--sleep", type=float, default=0.0)
    parser.add_argument("--retry-status", default="")
    parser.add_argument("--rebuild-only", action="store_true")
    args = parser.parse_args()

    gaps = load_events()
    fetch_rows = load_fetch_rows(gaps)
    attempted = 0
    if not args.rebuild_only and args.fetch_limit != 0:
        attempted = update_fetch_rows(fetch_rows, args)
    else:
        write_fetch_rows(fetch_rows)
    summary = build_audit(gaps, fetch_rows)
    write_manifest()
    print(json.dumps({"attempted": attempted, **summary}, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()