"""ETF 全市场数据同步（列表走东财快照，K 线走 Tushare fund_daily）。

分工：
- 东财 push2 clist：全市场 ETF 名单 + 总市值/流通市值/换手率（quicksync 镜像上
  fund_etf_basic 不存在，规模字段无替代源）-> etf_basic；
- Tushare fund_daily（quicksync 可用，与股票 daily 同源同控频）-> candles
  不复权底座，单位换算与股票一致（vol 手->份 ×100、amount 千元->元 ×1000）。

同步策略（全串行，Tushare 按分钟限频，并发无意义）：
- 逐日模式：交易日历里尚无任何 ETF 日线的日期，一天一调用拿全市场基金日线
  （过滤到 etf_basic 符号），日常增量通常只有当天 1 次调用；
- 逐只模式：无任何缓存的 ETF（新上市/历史缺口）按 ts_code 全量拉取，每次 1 调用；
  full=true 时对所有 ETF 重拉（修数/回补 amount 用）。

- 进程内后台任务（与 screener.market_sync 同款模式），前端轮询 /api/etf/sync/status；
- 复权：quicksync 无 fund_adj_factor，ETF 暂无因子，qfq/hfq 切换时按无因子
  原样返回（api._adjust_bars 的既有语义）。
"""
from __future__ import annotations

import asyncio
import time
from datetime import datetime, timedelta

from sqlalchemy import delete, func, select, text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from .. import cache
from ..config import settings
from . import etf_provider
from .sync_utils import call_retry, get_pro_lazy, utcnow

# 进程内单例任务状态（uvicorn 单进程场景够用）
_state: dict = {
    "running": False,
    "step": None,
    "total": 0,          # 本次需处理的单元数（缺失交易日 + 需拉取的 ETF 只数）
    "done": 0,
    "error": None,
    "started_at": None,
    "finished_at": None,
}
_task: asyncio.Task | None = None
_lock = asyncio.Lock()

_BATCH = 3000       # upsert 分批行数（asyncpg 单语句参数上限 32766，10 列/行）
# fund_daily 返回全市场基金 ~2100 行，一天一批远小于上限


def _parse_d(s: str) -> datetime:
    return datetime.strptime(str(s), "%Y%m%d")


async def _sync_spot(session: AsyncSession) -> int:
    """快照 -> etf_basic（upsert + 删除已退市），返回列表行数。"""
    from ..models import EtfBasic

    async with etf_provider.new_client() as client:
        rows = await etf_provider.fetch_etf_spot(client)
    now = utcnow()
    stmt = pg_insert(EtfBasic).values([{**r, "updated_at": now} for r in rows])
    stmt = stmt.on_conflict_do_update(
        index_elements=["ts_code"],
        set_={
            "name": stmt.excluded.name,
            "exchange": stmt.excluded.exchange,
            "total_mv": stmt.excluded.total_mv,
            "circ_mv": stmt.excluded.circ_mv,
            "turnover_rate": stmt.excluded.turnover_rate,
            "updated_at": stmt.excluded.updated_at,
        },
    )
    await session.execute(stmt)
    # 快照外的 ETF 已退市（东财列表不再返回）
    await session.execute(delete(EtfBasic).where(EtfBasic.ts_code.not_in({r["ts_code"] for r in rows})))
    await session.commit()
    return len(rows)


def _fetch_day_sync(pro, d: str) -> list[dict]:
    """拉某交易日全市场场内基金日线（fund_daily；未生成的日期返回空）。"""
    time.sleep(settings.screener_sync_interval)
    df = call_retry(pro.fund_daily, trade_date=d)
    if df is None or df.empty:
        return []
    rows = []
    for _, r in df.iterrows():
        amt = r.get("amount")
        rows.append({
            "ts": _parse_d(d),
            "ts_code": r["ts_code"],
            "open": float(r["open"]), "high": float(r["high"]),
            "low": float(r["low"]), "close": float(r["close"]),
            "vol": float(r["vol"]),        # 手
            "amount": (float(amt) if amt is not None and amt == amt else None),  # 千元
        })
    return rows


def _fetch_symbol_sync(pro, ts_code: str, start: str | None, end: str | None) -> list[dict]:
    """按 ts_code 增量/全量拉单只 ETF 日线（start=None 即上市以来全量）。"""
    time.sleep(settings.screener_sync_interval)
    df = call_retry(pro.fund_daily, ts_code=ts_code, start_date=start, end_date=end)
    if df is None or df.empty:
        return []
    df = df.sort_values("trade_date")
    rows = []
    for _, r in df.iterrows():
        amt = r.get("amount")
        rows.append({
            "ts": _parse_d(r["trade_date"]),
            "ts_code": ts_code,
            "open": float(r["open"]), "high": float(r["high"]),
            "low": float(r["low"]), "close": float(r["close"]),
            "vol": float(r["vol"]),
            "amount": (float(amt) if amt is not None and amt == amt else None),
        })
    return rows


def _upsert_candles_stmt(batch: list[dict]):
    from ..models import Candle

    stmt = pg_insert(Candle).values(batch)
    return stmt.on_conflict_do_update(
        index_elements=["symbol", "timeframe", "ts"],
        set_={"open": stmt.excluded.open, "high": stmt.excluded.high,
              "low": stmt.excluded.low, "close": stmt.excluded.close,
              "volume": stmt.excluded.volume,
              # 增量源缺额时保留库内旧值（与股票底座同语义）
              "amount": func.coalesce(stmt.excluded.amount, Candle.amount)},
    )


def _day_batch(rows: list[dict], sym_set: set[str], d_str: str) -> list[dict]:
    """某日 fund_daily 行 -> candles 批（过滤到 etf_basic 符号；手->份、千元->元）。"""
    return [
        {"symbol": r["ts_code"].split(".")[0], "timeframe": "1d",
         "ts": _parse_d(d_str),
         "open": r["open"], "high": r["high"], "low": r["low"], "close": r["close"],
         "volume": r["vol"] * 100.0,
         "amount": (r["amount"] * 1000.0) if r["amount"] is not None else None,
         "turnover": None}
        for r in rows if r["ts_code"].split(".")[0] in sym_set
    ]


def _symbol_batch(symbol: str, rows: list[dict]) -> list[dict]:
    return [
        {"symbol": symbol, "timeframe": "1d", "ts": r["ts"],
         "open": r["open"], "high": r["high"], "low": r["low"], "close": r["close"],
         "volume": r["vol"] * 100.0,
         "amount": (r["amount"] * 1000.0) if r["amount"] is not None else None,
         "turnover": None}
        for r in rows
    ]


async def _write_batches(session: AsyncSession, batch: list[dict]) -> None:
    for i in range(0, len(batch), _BATCH):
        await session.execute(_upsert_candles_stmt(batch[i : i + _BATCH]))
    await session.commit()


async def _backfill_list_dates(session: AsyncSession, dates: dict[str, str]) -> None:
    """首根 K 线日回填 list_date（仅空缺处）：一条 UPDATE ... FROM (VALUES)。"""
    pairs = list(dates.items())
    for i in range(0, len(pairs), 1000):
        chunk = pairs[i : i + 1000]
        vals = ", ".join(f"(:s{j}, :d{j})" for j in range(len(chunk)))
        params = {f"s{j}": s for j, (s, _) in enumerate(chunk)}
        params.update({f"d{j}": d for j, (_, d) in enumerate(chunk)})
        await session.execute(text(
            f"UPDATE etf_basic e SET list_date = v.d FROM (VALUES {vals}) AS v(symbol, d) "
            "WHERE e.symbol = v.symbol AND e.list_date IS NULL"
        ), params)


async def _run_sync(full: bool) -> None:
    from ..db import async_session
    from ..models import Candle, EtfBasic, TradeCalendar

    try:
        pro = await asyncio.to_thread(get_pro_lazy)

        # 1) 快照 -> etf_basic
        _state["step"] = "正在拉取 ETF 列表"
        async with async_session() as session:
            n_list = await _sync_spot(session)
            etfs = (await session.execute(
                select(EtfBasic.ts_code, EtfBasic.symbol, EtfBasic.exchange, EtfBasic.list_date)
            )).all()
            # 零缓存 ETF（新上市/历史缺口，需逐只全量拉）：
            # NOT EXISTS 走 (symbol,timeframe,ts) 索引探测，1514 次 ms 级；
            # 比对「全 ETF 符号 GROUP BY max(ts)」（扫数百万索引行）便宜得多
            have_any = set((await session.execute(text(
                "SELECT e.symbol FROM etf_basic e WHERE EXISTS (SELECT 1 FROM candles c "
                "WHERE c.symbol = e.symbol AND c.timeframe = '1d')"
            ))).scalars())
            # 已落库的 ETF 交易日（YYYYMMDD）：逐日模式的跳过依据。
            # 只看近 40 天（索引范围扫）——更早的历史缺口由 full 全量重拉兜底，
            # 全表 distinct 对千万行 candles 表要几十秒，不能每次同步都付
            fresh = not have_any or full
            have_dates: set[str] = set() if fresh else {
                r.strftime("%Y%m%d") for r in (await session.execute(
                    select(func.distinct(func.date(Candle.ts))).where(
                        Candle.timeframe == "1d",
                        Candle.symbol.in_(select(EtfBasic.symbol)),
                        Candle.ts >= datetime.now() - timedelta(days=40),
                    )
                )).scalars() if r is not None
            }
            cal = (await session.execute(
                select(TradeCalendar.trade_date).order_by(TradeCalendar.trade_date.desc())
            )).scalars().all()

        sym_set = {e.symbol for e in etfs}
        today = datetime.now().strftime("%Y%m%d")
        since40 = (datetime.now() - timedelta(days=40)).strftime("%Y%m%d")

        # 2) 任务编排：逐日模式补近窗缺失交易日（fresh 库改走逐只全量）；
        #    逐只模式拉零缓存 ETF，full=true 全量重拉
        todo_dates = [] if fresh else [
            d for d in cal if since40 <= d <= today and d not in have_dates
        ]
        per_symbol = list(etfs) if fresh else [
            e for e in etfs if e.symbol not in have_any
        ]
        _state["total"] = len(todo_dates) + len(per_symbol)
        _state["done"] = 0
        fails: list[str] = []
        new_list_dates: dict[str, str] = {}
        written_rows = 0  # 实际写入的 K 线行数（决定是否作废 candles 相关缓存）

        # 3) 逐日模式：一天一调用（交易日历空时跳过——由逐只模式兜底）
        for d in sorted(todo_dates):
            _state["step"] = f"正在同步 {d} 日线（{_state['done'] + 1}/{_state['total']}）"
            try:
                rows = await asyncio.to_thread(_fetch_day_sync, pro, d)
                if rows:
                    batch = _day_batch(rows, sym_set, d)
                    if batch:
                        written_rows += len(batch)
                        async with async_session() as session:
                            await _write_batches(session, batch)
            except Exception as ex:  # noqa: BLE001 —— 单日失败不拖垮整体
                fails.append(f"{d}: {str(ex)[:80]}")
            _state["done"] += 1

        # 4) 逐只模式：零缓存 ETF 全量拉取（start=None 即上市以来；full 同理）
        for ts_code, symbol, _exch, has_list_date in per_symbol:
            _state["step"] = f"正在同步 ETF 日线 {symbol}（{_state['done'] + 1}/{_state['total']}）"
            try:
                rows = await asyncio.to_thread(_fetch_symbol_sync, pro, ts_code, None, None)
            except Exception as ex:  # noqa: BLE001
                fails.append(f"{ts_code}: {str(ex)[:80]}")
                _state["done"] += 1
                continue
            if rows:
                # 全量拉取的首根 = 真实上市日
                if not has_list_date:
                    new_list_dates[symbol] = rows[0]["ts"].strftime("%Y%m%d")
                batch = _symbol_batch(symbol, rows)
                written_rows += len(batch)
                try:
                    async with async_session() as session:
                        await _write_batches(session, batch)
                except Exception as ex:  # noqa: BLE001
                    fails.append(f"{ts_code}: {str(ex)[:80]}")
            _state["done"] += 1

        # 5) list_date 回填（一次 SQL）
        if new_list_dates:
            async with async_session() as session:
                await _backfill_list_dates(session, new_list_dates)
                await session.commit()

        # candles 已更新：作废旧 K 线预览缓存（只在真的写了行时——空跑不作废，
        # 免得每次同步都触发一轮 >10s 的统计重聚合）；etf 版本号作废 ETF 列表缓存
        if written_rows:
            await cache.bump_version("candles")
        await cache.bump_version("etf")

        parts = [f"同步完成（{n_list} 只 ETF"]
        if todo_dates:
            parts.append(f"{len(todo_dates)} 个交易日")
        if per_symbol:
            parts.append(f"{len(per_symbol)} 只逐只补数")
        _state["step"] = "，".join(parts) + ")" + (f"，{len(fails)} 项失败" if fails else "")
        if fails:
            _state["error"] = "部分失败：" + "；".join(fails[:3]) + ("…" if len(fails) > 3 else "")
    except Exception as e:  # noqa: BLE001
        _state["error"] = f"同步失败：{str(e)[:300]}"
        _state["step"] = "同步失败"
    finally:
        _state["running"] = False
        _state["finished_at"] = datetime.now()


async def start_sync(full: bool = False) -> dict:
    """幂等启动后台同步；已在跑则直接返回当前状态。full=true 所有 ETF 全量重拉。"""
    global _task
    async with _lock:
        if _state["running"] and _task and not _task.done():
            return dict(_state)
        _state.update({
            "running": True, "step": "准备同步", "total": 0, "done": 0,
            "error": None, "started_at": datetime.now(), "finished_at": None,
        })
        _task = asyncio.create_task(_run_sync(full))
    return dict(_state)


async def get_status(session: AsyncSession) -> dict:
    """任务状态 + DB 实况。行情侧只做单符号 max(ts) 索引探测（candles 是千万行表，
    全表聚合 >10s，绝不能落在轮询热路径上），取最早上市且已有日线的一只当「数据更新至」。"""
    from ..models import Candle, EtfBasic

    etfs = int(await session.scalar(select(func.count()).select_from(EtfBasic)) or 0)
    # 探测样本：优先最早上市（历史最长最稳）且已有日线的一只（EXISTS 走索引，ms 级）
    probe = await session.scalar(
        text("""
            SELECT e.symbol FROM etf_basic e
            WHERE EXISTS (SELECT 1 FROM candles c
                          WHERE c.symbol = e.symbol AND c.timeframe = '1d')
            ORDER BY e.list_date NULLS LAST, e.symbol LIMIT 1
        """)
    )
    last = None
    if probe:
        last = await session.scalar(
            select(func.max(Candle.ts)).where(Candle.symbol == probe, Candle.timeframe == "1d")
        )
    status = dict(_state)
    status.update({
        "stats": {"etfs": etfs},
        "last_trade_date": last,
        "last_synced_at": _state.get("finished_at") or _state.get("started_at"),
        "ready": etfs > 0 and last is not None,
    })
    return status