"""大盘行情总览（主页展示）——两层结构。

- live 层：腾讯免费实时行情（qt.gtimg.cn，一次 GET 拿 A股/美股指数现价/涨跌幅）+
  东财 push2（日韩指数，腾讯 s_ 前缀不覆盖，实测 s_jp*/s_kr* 均无符号），
  进程内缓存 30s（成功）/ 15s 负缓存（失败，防止接口抖动持续拖慢请求）。
- EOD 层：tushare 收盘数据 —— A 股指数 pro.index_daily、全球指数 pro.index_global
  （45 日 spark 走势）+ 两市统计/成交额历史 pro.daily_info。收盘数据一天一变，
  进程内新鲜期 4h + Redis 兜底 24h，过期走 SWR（先返旧值，后台刷新，永不阻塞用户）。
- merge：实时价覆盖 close/change/pct_chg（realtime=True），拿不到实时值的指数回退
  收盘口径；今日为交易日且 EOD 尚未含今日时，用腾讯全市口径成交额追加盘中 bar。

口径说明（daily_info 板块行，实测 2026-09）：
- 沪市 SH_MARKET = 主板A + 科创板(SH_STAR) + B股，不含基金(SH_FUND)；
  旧口径 SH_A 漏科创板（日均 ~2500 亿），是成交额偏小的根因。
- 深市 SZ_MARKET = 主板 + 创业板，全部为股票。
- 任何单个来源失败只是跳过（errors 里注明），全部失败才抛 MarketOverviewError。
"""
from __future__ import annotations

import asyncio
import time
from datetime import date, datetime, timedelta

import httpx
import pandas as pd

from .. import cache
from ..config import settings
from .sync_utils import d8_iso, f_clean

# (tushare代码, 名称, 地区, 腾讯符号) —— 展示顺序即列表顺序
# 首页聚焦中美 + 日韩（其余国际指数在 /indexes 国际指数页）；标普500 腾讯符号是 s_usINX（不是 s_usSPX）
MARKET_INDEXES: list[tuple[str, str, str, str]] = [
    ("000001.SH", "上证指数", "cn", "s_sh000001"),
    ("399001.SZ", "深证成指", "cn", "s_sz399001"),
    ("399006.SZ", "创业板指", "cn", "s_sz399006"),
    ("000688.SH", "科创50", "cn", "s_sh000688"),
    ("DJI", "道琼斯", "us", "s_usDJI"),
    ("IXIC", "纳斯达克", "us", "s_usIXIC"),
    ("SPX", "标普500", "us", "s_usINX"),
    ("N225", "日经225", "apac", ""),   # 日韩实时走东财（_APAC_SECIDS），腾讯符号留空
    ("KS11", "韩国KOSPI", "apac", ""),
]

# 深证综指：不展示，仅取其 f[7]（深市全市成交额，万元）
_TENCENT_SZ_TOTAL = "s_sz399106"
_TENCENT_MAP = {ts_code: sym for ts_code, _, _, sym in MARKET_INDEXES if sym}
_TENCENT_URL = "http://qt.gtimg.cn/q=" + ",".join([*_TENCENT_MAP.values(), _TENCENT_SZ_TOTAL])

# 日韩指数实时源：东财 push2 ulist 一次 GET（主站 + 延迟镜像双 host）。腾讯 s_ 前缀不
# 覆盖日韩（实测均返回 pv_none_match）；新浪 int_kospi 为空且 int_* 行情明显滞后
# （实测 DJI 差价 ~6000 点），故不用。
_APAC_SECIDS = {"N225": "100.N225", "KS11": "100.KS11"}
_EM_HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                             "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"}  # 同 etf_provider
_EM_HOSTS = ["https://push2.eastmoney.com", "https://push2delay.eastmoney.com"]
_EM_PATH = ("/api/qt/ulist.np/get?fltt=2&invt=2&fields=f2,f3,f4,f12,f14&secids="
            + ",".join(_APAC_SECIDS.values()))
# 指数 code -> live 行情键（em: 前缀与腾讯符号隔离），merge 时查表
_LIVE_SYM_MAP: dict[str, str] = {**_TENCENT_MAP, **{c: f"em:{c}" for c in _APAC_SECIDS}}

_SPARK_DAYS = 45        # 迷你走势取最近 45 个交易日收盘
_HISTORY_DAYS = 150     # 日历日窗口（约 100 个交易日，够取 spark）
_CALL_INTERVAL = 0.12   # 顺序调用间隔（秒），对 tushare 控频
_EOD_KEY = "market_overview:eod:v4"   # v4：新增日韩指数（N225/KS11），与 v3 隔离
_AMOUNT_HIST_CAL_DAYS = 190  # 成交额历史的日历日窗口（≈128 交易日）
_AMOUNT_HIST_BARS = 120      # 输出的柱数（取尾部）
_LIVE_FAIL_TTL = 15.0        # 腾讯 live 层失败负缓存（秒）
_EM_FAIL_TTL = 90.0          # 东财日韩失败负缓存（秒）：长于 live 缓存 30s，
                             # 持续故障时最多每 90s 才有一次超时等待，其余刷新直接跳过


class MarketOverviewError(RuntimeError):
    """所有指数都拉不到（token/网络故障）——接口层转 503。"""


def _get_pro():
    if not settings.tushare_token:
        raise MarketOverviewError("未配置 TUSHARE_TOKEN，无法获取大盘行情（backend/.env）")
    # 走统一入口：15000 积分档 token 只认 quicksync 镜像（直连 api.tushare.pro 会 40101）
    from .tushare_provider import get_pro

    return get_pro()


# ======================= live 层：腾讯实时行情 =======================

_live: dict = {"at": 0.0, "quotes": None}   # 进程内缓存（monotonic 时钟）
_live_error: str | None = None               # 最近一次腾讯实时拉取失败的原因（成功后清空）
_live_em_error: str | None = None            # 最近一次东财日韩实时拉取失败的原因（成功后清空）
_em_state: dict = {"at": 0.0, "ok": None, "host": 0}  # 东财日韩负缓存 + 最近成功 host 索引


async def _fetch_live_http() -> dict[str, dict]:
    """一次 GET 拿全部符号。响应 GBK，行如 v_s_sh000001="1~上证指数~000001~3930.12~-11.97~-0.30~537286161~93825519~~";

    字段序：f[1]名称 / f[3]现价 / f[4]涨跌 / f[5]涨跌% / f[7]成交额。
    单位陷阱：仅 s_sh000001 与 s_sz399106 的 f[7] 是「万元、全市口径」，可算两市成交额；
    港股行的 f[7] 是手数、美股行非人民币金额，s_sz399001（深证成指）是成分股口径——都不能用。
    """
    async with httpx.AsyncClient(timeout=settings.tencent_quote_timeout) as client:
        resp = await client.get(_TENCENT_URL)
        resp.raise_for_status()
        text = resp.content.decode("gbk", errors="replace")  # 响应头 charset 不可靠，显式解码

    quotes: dict[str, dict] = {}
    for line in text.splitlines():
        if "=" not in line:
            continue
        head, _, body = line.partition("=")
        sym = head.strip().removeprefix("v_")
        fields = body.strip().strip(';"').split("~")
        if not sym or len(fields) < 8:
            continue

        def _num(i: int) -> float | None:
            try:
                return float(fields[i])
            except (TypeError, ValueError):
                return None

        quotes[sym] = {
            "name": fields[1],
            "price": _num(3),
            "change": _num(4),
            "pct": _num(5),
            "amount_wan": _num(7),
        }
    if not quotes:
        raise RuntimeError("腾讯行情响应为空或无法解析")
    return quotes


def _parse_em_ulist(payload: dict) -> dict[str, dict]:
    """ulist JSON -> {em:code: {name, price, change, pct, amount_wan}}。fltt=2 下 f2/f3/f4
    直接是浮点；未开盘等场景字段是 "-"，解析为 None。"""
    diff = (payload.get("data") or {}).get("diff") or []
    quotes: dict[str, dict] = {}
    for row in diff:
        code = str(row.get("f12") or "")
        if code not in _APAC_SECIDS:
            continue

        def _num(v) -> float | None:
            try:
                return float(v)
            except (TypeError, ValueError):
                return None

        quotes[f"em:{code}"] = {
            "name": row.get("f14"),
            "price": _num(row.get("f2")),
            "change": _num(row.get("f4")),
            "pct": _num(row.get("f3")),
            "amount_wan": None,  # 日韩成交额口径不同且未使用，不取
        }
    if not quotes:
        raise RuntimeError("东财日韩行情响应为空或无法解析")
    return quotes


async def _fetch_live_apac_http() -> dict[str, dict]:
    """日韩指数实时（东财 ulist 一次 GET，主站/延迟镜像按序重试，粘住最近成功的 host）。

    坑（同 etf_provider 实测）：push2 主站短连发几次会直接断连，push2delay 镜像稳；
    本机 v2rayN 系统代理(127.0.0.1:10808)对 push2 的 https CONNECT 隧道断连（curl -x
    同样失败，腾讯 http:// 却正常）——必须带浏览器 UA 且 trust_env=False 直连。"""
    start = _em_state.get("host", 0)
    last_err: Exception | None = None
    for i in range(len(_EM_HOSTS)):
        url = _EM_HOSTS[(start + i) % len(_EM_HOSTS)] + _EM_PATH
        try:
            async with httpx.AsyncClient(
                timeout=settings.tencent_quote_timeout, headers=_EM_HEADERS, trust_env=False,
            ) as client:
                resp = await client.get(url)
                resp.raise_for_status()
                payload = resp.json()
            quotes = _parse_em_ulist(payload)
            _em_state["host"] = (start + i) % len(_EM_HOSTS)  # 粘住成功 host
            return quotes
        except Exception as e:  # noqa: BLE001 —— 换镜像整重来
            last_err = e
    raise RuntimeError(f"东财日韩行情双镜像均失败: {str(last_err)[:60]}")


async def _fetch_live() -> dict[str, dict] | None:
    """实时行情（进程内缓存）；腾讯与东财各自独立降级，全失败返回 None，上层降级 EOD。
    东财有独立负缓存（_EM_FAIL_TTL）：持续故障时后续刷新直接跳过，不吃超时等待。"""
    global _live_error, _live_em_error
    now = time.monotonic()
    age = now - _live["at"]
    if _live["quotes"] is not None and age < settings.market_live_ttl:
        return _live["quotes"]
    if _live["quotes"] is None and _live["at"] > 0 and age < _LIVE_FAIL_TTL:
        return None  # 负缓存：刚失败过，短时间内不再打行情接口

    quotes: dict[str, dict] = {}
    try:
        quotes.update(await _fetch_live_http())
        _live_error = None
    except Exception as e:  # noqa: BLE001 —— 实时层是锦上添花，失败不拖垮整包
        _live_error = f"实时行情: {str(e)[:60]}"
    if _em_state["ok"] is False and now - _em_state["at"] < _EM_FAIL_TTL:
        pass  # 东财刚失败过：负缓存期内跳过（_live_em_error 保留上次原因）
    else:
        try:
            quotes.update(await _fetch_live_apac_http())
            _live_em_error = None
            _em_state.update(at=now, ok=True)
        except Exception as e:  # noqa: BLE001
            _live_em_error = f"日韩实时行情: {str(e)[:60]}"
            _em_state.update(at=now, ok=False)

    if not quotes:
        _live.update(at=now, quotes=None)
        return None
    _live.update(at=now, quotes=quotes)
    return quotes


# ======================= EOD 层：tushare 收盘数据 =======================

def _fetch_index_sync(pro, ts_code: str) -> pd.DataFrame:
    start = (datetime.now() - timedelta(days=_HISTORY_DAYS)).strftime("%Y%m%d")
    if "." in ts_code:  # A 股指数（000001.SH 形式）
        return pro.index_daily(ts_code=ts_code, start_date=start)
    return pro.index_global(ts_code=ts_code, start_date=start)


def _quote_from_df(df: pd.DataFrame) -> dict | None:
    """DataFrame -> {close, change, pct_chg, trade_date, spark, spark_dates}（旧 -> 新）。"""
    if df is None or df.empty:
        return None
    df = df.sort_values("trade_date")
    tail = df.tail(_SPARK_DAYS)
    last = df.iloc[-1]
    return {
        "close": f_clean(last["close"]),
        "change": f_clean(last.get("change")),
        "pct_chg": f_clean(last.get("pct_chg")),
        "trade_date": d8_iso(last["trade_date"]),
        "spark": [round(float(c), 4) for c in tail["close"]],
        "spark_dates": [str(d) for d in tail["trade_date"]],
    }


def _fetch_stats_sync(pro) -> dict | None:
    """两市市值/成交统计：沪 SH_MARKET（主板A+科创+B，不含基金）+ 深 SZ_MARKET（全部股票）。

    口径与本地 candles 全市场 sum(amount) 吻合（candles 另含北交所，约 +70 亿）。
    """
    start = (datetime.now() - timedelta(days=14)).strftime("%Y%m%d")
    sh = pro.daily_info(exchange="SH", start_date=start)
    sz = pro.daily_info(exchange="SZ", start_date=start)
    if sh is None or sh.empty or sz is None or sz.empty:
        return None

    def _board(df: pd.DataFrame, code: str):
        sub = df[df["ts_code"] == code].sort_values("trade_date")  # 接口不保证有序
        return sub.iloc[-1] if not sub.empty else None

    sh_m, sz_m = _board(sh, "SH_MARKET"), _board(sz, "SZ_MARKET")
    if sh_m is None or sz_m is None:
        return None
    # 两边各自取最新，日期不一致时以较旧一天为准凑齐口径（罕见，通常同日）
    d = min(d8_iso(sh_m["trade_date"]), d8_iso(sz_m["trade_date"]))

    def _sum(col: str) -> float | None:
        a, b = f_clean(sh_m.get(col)), f_clean(sz_m.get(col))
        return None if a is None or b is None else round(a + b, 2)

    return {
        "trade_date": d,
        "total_mv": _sum("total_mv"),
        "float_mv": _sum("float_mv"),
        "amount": _sum("amount"),
        "turnover": f_clean(sh_m.get("tr")),  # 换手率仅沪市有，展示口径注明沪市
    }


def _fetch_amount_history_sync(pro) -> list[dict]:
    """两市成交额历史：daily_info 范围查询一次拉多日（SH 6 个月实测 0.09s），
    沪 SH_MARKET + 深 SZ_MARKET 按日对齐相加（接口原生亿元），升序取尾部 120 根。"""
    start = (datetime.now() - timedelta(days=_AMOUNT_HIST_CAL_DAYS)).strftime("%Y%m%d")
    sh = pro.daily_info(exchange="SH", start_date=start)
    sz = pro.daily_info(exchange="SZ", start_date=start)
    if sh is None or sh.empty or sz is None or sz.empty:
        return []
    sh_m = sh[sh["ts_code"] == "SH_MARKET"].set_index("trade_date")["amount"]
    sz_m = sz[sz["ts_code"] == "SZ_MARKET"].set_index("trade_date")["amount"]
    common = sh_m.index.intersection(sz_m.index)  # 内连接：两市都有数据的交易日
    if len(common) == 0:
        return []
    total = (sh_m[common] + sz_m[common]).sort_index()
    return [{"date": d8_iso(d), "amount": round(float(v), 2)} for d, v in total.tail(_AMOUNT_HIST_BARS).items()]


# ---- EOD 的 SWR（stale-while-revalidate）：新鲜期内直返；过期先返旧值后台刷新 ----

_eod_state: dict = {"payload": None}  # 进程内新鲜/陈旧兜底（payload 自带 fetched_ts 墙钟）
_eod_refreshing = False               # 后台刷新防重入标志
_eod_refresh_error: str | None = None  # 最近一次后台刷新失败的原因
_bg_tasks: set[asyncio.Task] = set()   # 持引用防 GC


async def _refresh_eod() -> dict:
    """拉全量 EOD（9 指数 + 统计 + 成交额历史，顺序控频），写进程内 state + Redis。"""
    pro = await asyncio.to_thread(_get_pro)

    indexes: list[dict] = []
    errors: list[str] = []
    for ts_code, name, region, _sym in MARKET_INDEXES:
        try:
            df = await asyncio.to_thread(_fetch_index_sync, pro, ts_code)
            q = _quote_from_df(df)
            if q is None:
                raise MarketOverviewError("无数据")
            indexes.append({"code": ts_code, "name": name, "region": region, **q})
        except Exception as e:  # noqa: BLE001 —— 单个指数失败不拖垮整包
            errors.append(f"{name}: {str(e)[:60]}")
        await asyncio.sleep(_CALL_INTERVAL)

    if not indexes:
        raise MarketOverviewError("大盘行情全部拉取失败: " + "; ".join(errors)[:200])

    stats: dict | None = None
    try:
        await asyncio.sleep(_CALL_INTERVAL)
        stats = await asyncio.to_thread(_fetch_stats_sync, pro)
    except Exception as e:  # noqa: BLE001 —— 统计缺失时指数照常展示
        errors.append(f"两市统计: {str(e)[:60]}")

    amount_history: list[dict] = []
    try:
        await asyncio.sleep(_CALL_INTERVAL)
        amount_history = await asyncio.to_thread(_fetch_amount_history_sync, pro)
    except Exception as e:  # noqa: BLE001 —— 历史图缺数据时其余照常
        errors.append(f"成交额历史: {str(e)[:60]}")

    payload = {
        "fetched_at": datetime.now().isoformat(),
        "fetched_ts": time.time(),  # 墙钟（epoch float）：跨进程（Redis）判新鲜度用
        "indexes": indexes,
        "stats": stats,
        "amount_history": amount_history,
        "errors": errors,
    }
    _eod_state["payload"] = payload
    await cache.cache_set(_EOD_KEY, payload, ttl=settings.market_eod_redis_ttl)
    return payload


async def _refresh_eod_wrapped() -> None:
    """后台刷新的主体：失败静默保留旧值并记录原因（下次请求并入 errors 便于排查）。"""
    global _eod_refresh_error, _eod_refreshing
    try:
        await _refresh_eod()
        _eod_refresh_error = None
    except Exception as e:  # noqa: BLE001
        _eod_refresh_error = f"EOD后台刷新: {str(e)[:60]}"
    finally:
        _eod_refreshing = False


def _spawn_eod_refresh() -> None:
    global _eod_refreshing
    if _eod_refreshing:
        return
    _eod_refreshing = True
    task = asyncio.create_task(_refresh_eod_wrapped())
    _bg_tasks.add(task)
    task.add_done_callback(_bg_tasks.discard)


async def _get_eod() -> dict:
    """读 EOD：内存新鲜直返（0 RTT）→ Redis 回填 → 有旧值先返 + SWR 后台刷新 → 真冷启动同步拉。"""
    p = _eod_state["payload"]
    if p is not None and time.time() - p["fetched_ts"] < settings.market_eod_fresh_ttl:
        return p
    if p is None:
        cached = await cache.cache_get(_EOD_KEY)
        if cached:
            p = cached
            _eod_state["payload"] = p
    if p is not None:
        _spawn_eod_refresh()  # 陈旧但可用：立即返回，后台拉新
        return p
    return await _refresh_eod()  # 首次访问：同步等（~3.5s，与旧行为一致）


# ======================= merge：实时叠加收盘 =======================

async def fetch_overview(is_trading_day: bool | None = None) -> dict:
    """聚合 live + EOD。实时价覆盖 close/change/pct_chg（realtime=True），今日实时成交额
    （腾讯全市口径）在 EOD 尚未含今日时追加为盘中 bar。响应不再整包缓存：两层各有
    进程内缓存，合并是 O(10) 操作，热路径 0 外部 RTT。"""
    eod = await _get_eod()
    live = await _fetch_live()
    today_iso = date.today().isoformat()

    indexes: list[dict] = []
    for it in eod["indexes"]:
        out = dict(it)
        sym = _LIVE_SYM_MAP.get(it["code"])
        q = live.get(sym) if (live and sym) else None
        if q and q.get("price") is not None:
            # spark 永远来自 EOD（末点是上一收盘点，与实时价并存是已知的装饰性差异，不改历史序列）
            out.update(close=q["price"], change=q["change"], pct_chg=q["pct"],
                       trade_date=today_iso, realtime=True)
        else:
            out["realtime"] = False
        indexes.append(out)

    stats = dict(eod["stats"]) if eod.get("stats") else None
    history = list(eod.get("amount_history") or [])

    # 今日实时两市成交额：沪深全市口径（万元->亿）。EOD 已含今日、非交易日、金额缺失时不追加。
    if live and stats and stats.get("trade_date") != today_iso:
        if is_trading_day is None:
            is_trading_day = datetime.now().weekday() < 5  # 日历判定不可用时的降级启发式
        if is_trading_day:
            sh_amt = (live.get("s_sh000001") or {}).get("amount_wan")
            sz_amt = (live.get(_TENCENT_SZ_TOTAL) or {}).get("amount_wan")
            if sh_amt is not None and sz_amt is not None:
                amt = round((sh_amt + sz_amt) / 10000, 2)
                stats["amount_today"] = amt
                history.append({"date": today_iso, "amount": amt, "intraday": True})

    errors = list(eod.get("errors") or [])
    if _eod_refresh_error:
        errors.append(_eod_refresh_error)
    if _live_error:
        errors.append(_live_error)
    if _live_em_error:
        errors.append(_live_em_error)

    return {
        "updated_at": datetime.now().isoformat(),
        "indexes": indexes,
        "stats": stats,
        "amount_history": history,
        "errors": errors,
    }