market_sync/app/tasks/task_longhubang.py

"""同步任务：龙虎榜聚合层 + 席位层（akshare 源）。

数据源：
  - 聚合层：`ak.stock_lhb_detail_em(start_date, end_date)` → `longhubang_daily` 表
  - 席位层：`ak.stock_lhb_stock_detail_em(symbol, date, flag)` → `longhubang_seat` 表

设计要点：
  1) 22:00 触发（mairui 21:30 跑 moneyflow；龙虎榜 19:00-21:00 陆续出齐，22:00 兜底）
  2) 默认"今天"增量；CLI 支持 --backfill YYYYMMDD 单日；--backfill-range YYYYMMDD YYYYMMDD 日期范围
  3) 聚合层一次性拉日期范围；席位层先抓聚合层取出 stock_code 集合再逐只抓（节省调用）
  4) 串行执行（每日调用 ~3 次聚合层 + ~200 次席位层；3-5 分钟内完成）
  5) akshare 对"未上榜"的股票会抛 `'NoneType' object is not subscriptable`
     → source 层 try/except 兜住，按"空数据"处理
  6) 数据是历史回填友好（全量 4 年 ≈ 25 分钟，2026-07-01 烟测通过）
"""
from __future__ import annotations

import time
from datetime import datetime, timedelta
from typing import Any

from app.core.db import ops as db_ops
from app.core.datasource.base import registry as ds_registry
from app.core.datasource.registry import is_source_ready
from app.core.sync.base import SyncTask
from app.core.sync.registry import mark_sync_blocked
from app.core.utils.logging import get_logger

logger = get_logger("sync.longhubang")


def _parse_yyyymmdd(s: str) -> str:
    """'YYYYMMDD' 或 'YYYY-MM-DD' → 'YYYY-MM-DD'。"""
    s = str(s).strip()
    if len(s) == 8 and s.isdigit():
        return f"{s[:4]}-{s[4:6]}-{s[6:8]}"
    return s


def _date_range(start: str, end: str) -> list[str]:
    """返回 start..end（含）期间所有日期字符串列表（YYYY-MM-DD）。"""
    s = datetime.strptime(start, "%Y-%m-%d")
    e = datetime.strptime(end, "%Y-%m-%d")
    if e < s:
        s, e = e, s
    out = []
    cur = s
    while cur <= e:
        out.append(cur.strftime("%Y-%m-%d"))
        cur += timedelta(days=1)
    return out


class SyncLonghubang(SyncTask):
    """龙虎榜聚合层 + 席位层同步。

    CLI 用法：
      python -m app.entrypoints.cli sync longhubang                  # 今天增量
      ... sync longhubang --backfill 20240614                        # 单日
      ... sync longhubang --backfill-range 20240101 20240614          # 日期范围
      ... sync longhubang --start 20240101 --end 20240614            # 等价于 backfill-range
    """
    dataset_id = "longhubang"

    def _run(
        self,
        *,
        trigger_source: str = "manual",
        backfill: str | None = None,
        backfill_range: tuple[str, str] | None = None,
        start: str | None = None,
        end: str | None = None,
        **kwargs,
    ) -> dict[str, Any]:
        src = ds_registry.get("datasource_akshare_lhb")
        if src is None:
            return {"status": "error", "message": "akshare_lhb 数据源未注册"}
        ok, reason = is_source_ready(src.key)
        if not ok:
            mark_sync_blocked(self.dataset_id, message=f"akshare_lhb 未就绪: {reason}")
            return {"status": "blocked", "message": f"akshare_lhb 未就绪: {reason}"}

        # ── 决定要拉哪几天 ──
        if backfill_range:
            s, e = _parse_yyyymmdd(backfill_range[0]), _parse_yyyymmdd(backfill_range[1])
        elif start and end:
            s, e = _parse_yyyymmdd(start), _parse_yyyymmdd(end)
        elif backfill:
            s = e = _parse_yyyymmdd(backfill)
        else:
            # 默认"今天"（盘后未收盘时 fallback 到昨天）
            from app.core.sync.base import _effective_sync_end
            s = e = _effective_sync_end()

        dates = _date_range(s, e)
        if not dates:
            return {"status": "error", "message": f"日期范围无效: {s}..{e}"}
        logger.info(f"[longhubang] 计划: {len(dates)} 天 ({s} ~ {e})")

        # ── 逐日同步 ──
        t0 = time.time()
        daily_ok = daily_fail = 0
        daily_total = seat_total = 0
        for i, d in enumerate(dates, 1):
            self._progress(
                message=f"longhubang 进度 {i}/{len(dates)} 当前 {d}",
                current=i, total=len(dates), current_step=d,
            )
            try:
                d_rows, s_rows = self._sync_one_day(src, d)
                daily_ok += 1
                daily_total += d_rows
                seat_total += s_rows
            except Exception as e:
                daily_fail += 1
                logger.warning(f"[longhubang {d}] 失败: {e}")

        elapsed = round(time.time() - t0, 1)
        msg = (
            f"龙虎榜 {len(dates)}天 daily_ok={daily_ok} daily_fail={daily_fail} "
            f"聚合层={daily_total}行 席位层={seat_total}行, {elapsed}s"
        )
        return {
            "status": "ok" if daily_fail == 0 else "warning",
            "message": msg,
            "ok": daily_ok, "fail": daily_fail,
            "daily_rows": daily_total, "seat_rows": seat_total,
            "elapsed_sec": elapsed,
        }

    # ── 单日同步（聚合 + 席位）──

    def _sync_one_day(self, src, trade_date: str) -> tuple[int, int]:
        """返回 (聚合层行数, 席位层行数)。"""
        # 1) 聚合层（1 调用/天）
        df_daily = src.fetch_longhubang_daily(trade_date, trade_date)
        daily_rows = 0
        stock_codes: list[str] = []
        if df_daily is not None and not df_daily.empty:
            rows = []
            for _, r in df_daily.iterrows():
                rows.append({
                    "stock_code": str(r.get("stock_code") or "").strip(),
                    "stock_name": str(r.get("stock_name") or "").strip(),
                    "trade_date": str(r.get("trade_date") or trade_date)[:10],
                    "rank_idx": self._to_int(r.get("rank_idx")),
                    "comment": self._to_str(r.get("comment")),
                    "close": self._to_float(r.get("close")),
                    "pct_chg": self._to_float(r.get("pct_chg")),
                    "lhb_net_buy": self._to_float(r.get("lhb_net_buy")),
                    "lhb_buy_amt": self._to_float(r.get("lhb_buy_amt")),
                    "lhb_sell_amt": self._to_float(r.get("lhb_sell_amt")),
                    "lhb_total_amt": self._to_float(r.get("lhb_total_amt")),
                    "market_total_amt": self._to_float(r.get("market_total_amt")),
                    "net_buy_ratio": self._to_float(r.get("net_buy_ratio")),
                    "total_amt_ratio": self._to_float(r.get("total_amt_ratio")),
                    "turnover_rate": self._to_float(r.get("turnover_rate")),
                    "float_mv": self._to_float(r.get("float_mv")),
                    "reason": self._to_str(r.get("reason")),
                    "post_1d_pct": self._to_float(r.get("post_1d_pct")),
                    "post_2d_pct": self._to_float(r.get("post_2d_pct")),
                    "post_5d_pct": self._to_float(r.get("post_5d_pct")),
                    "post_10d_pct": self._to_float(r.get("post_10d_pct")),
                })
            if rows:
                db_ops.upsert_longhubang_daily(rows)
                daily_rows = len(rows)
                # 取出 stock_code（去交易所前缀得 6 位，供席位层调用）
                for r in rows:
                    code6 = str(r["stock_code"]).replace("SH", "").replace("SZ", "").replace("BJ", "")
                    if len(code6) == 6 and code6.isdigit():
                        stock_codes.append(code6)
                logger.info(f"[longhubang {trade_date}] 聚合层 {daily_rows} 行, {len(stock_codes)} 只上榜")

        # 2) 席位层（每只票 2 调用：buy + sell）
        seat_rows_total = 0
        if stock_codes:
            seat_rows = []
            for code6 in stock_codes:
                for direction in ("buy", "sell"):
                    df_seat = src.fetch_longhubang_seat(code6, trade_date, direction)
                    if df_seat is None or df_seat.empty:
                        continue
                    for _, r in df_seat.iterrows():
                        # 字段映射（akshare 6 中文列名）
                        if direction == "buy":
                            buy_amt = self._to_float(r.get("买入金额"))
                            sell_amt = self._to_float(r.get("卖出金额"))
                        else:
                            buy_amt = self._to_float(r.get("买入金额"))
                            sell_amt = self._to_float(r.get("卖出金额"))
                        # 买方向的占比列叫"买入金额-占总成交比例"，卖方向叫"卖出金额-..."
                        amt_ratio = (
                            self._to_float(r.get("买入金额-占总成交比例"))
                            if direction == "buy"
                            else self._to_float(r.get("卖出金额-占总成交比例"))
                        )
                        # amount=主金额：买方向=买入金额，卖方向=卖出金额
                        amt = buy_amt if direction == "buy" else sell_amt
                        seat_rows.append({
                            "stock_code": f"{'SH' if code6.startswith(('5','6','9')) else 'SZ'}{code6}",
                            "trade_date": trade_date,
                            "direction": direction,
                            "rank_idx": self._to_int(r.get("序号")) or 0,
                            "branch_name": self._to_str(r.get("交易营业部名称")),
                            "buy_amount": buy_amt,
                            "sell_amount": sell_amt,
                            "amount": amt,
                            "net_amount": self._to_float(r.get("净额")),
                            "amount_ratio": amt_ratio,
                            "explanation": self._to_str(r.get("类型")),
                        })
            if seat_rows:
                db_ops.upsert_longhubang_seat(seat_rows)
                seat_rows_total = len(seat_rows)
                logger.info(f"[longhubang {trade_date}] 席位层 {seat_rows_total} 行")
        return daily_rows, seat_rows_total

    # ── 类型转换 helper ──

    @staticmethod
    def _to_float(v) -> float | None:
        if v is None:
            return None
        try:
            f = float(v)
            return f if f == f else None  # NaN → None
        except (TypeError, ValueError):
            return None

    @staticmethod
    def _to_int(v) -> int | None:
        if v is None:
            return None
        try:
            return int(v)
        except (TypeError, ValueError):
            return None

    @staticmethod
    def _to_str(v) -> str | None:
        if v is None:
            return None
        s = str(v).strip()
        return s if s else None