"""同步任务：龙虎榜聚合层 + 席位层（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