diff --git a/app/sources/xueqiu.py b/app/sources/xueqiu.py index c3e36f1..882df3f 100644 --- a/app/sources/xueqiu.py +++ b/app/sources/xueqiu.py @@ -178,6 +178,83 @@ class XueqiuSource(DataSource): logger.exception("[kline %s] 雪球 kline 异常: %s", code6, e) return pd.DataFrame() + def fetch_kline_5min(self, code6: str, start: str, end: str) -> pd.DataFrame: + """5 分钟 K 线(雪球 period=5m)。返回标准列:bar_time, open, high, low, close, volume, amount。""" + if not self._read_token(): + return pd.DataFrame() + try: + ball = self._import_ball() + self._set_token_once() + self._wait_rps() + + symbol = code6_to_xueqiu(code6) + start_dt = datetime.strptime(start, "%Y-%m-%d") + end_dt = datetime.strptime(end, "%Y-%m-%d") + days_needed = max((end_dt - start_dt).days + 2, 1) + count = min(days_needed * 48, 5000) + + result = None + for attempt in range(3): + result = call_with_timeout( + ball.kline, symbol, period="5m", count=count, + timeout=20.0, on_timeout=None, + description=f"xueqiu.kline.5m {symbol}", + ) + if result and result.get("error_code") == 0: + break + if result is None and attempt < 2: + time.sleep(0.3 * (attempt + 1)) + continue + if result and result.get("error_code") != 0: + time.sleep(0.3 * (attempt + 1)) + if not result or result.get("error_code") != 0: + logger.warning( + "[kline5m %s] 雪球 5m 失败: error_code=%s desc=%s", + code6, + result.get("error_code") if result else "None", + result.get("error_description") if result else "None", + ) + return pd.DataFrame() + data = result.get("data", {}) + columns = data.get("column", []) + items = data.get("item", []) + if not columns or not items: + return pd.DataFrame() + idx = {c: i for i, c in enumerate(columns)} + needed = {"timestamp": idx.get("timestamp"), "volume": idx.get("volume"), + "open": idx.get("open"), "high": idx.get("high"), + "low": idx.get("low"), "close": idx.get("close")} + if any(v is None for v in needed.values()): + return pd.DataFrame() + records = [] + for row in items: + try: + ts = row[needed["timestamp"]] / 1000 + records.append({ + "bar_time": datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S"), + "open": float(row[needed["open"]]), + "high": float(row[needed["high"]]), + "low": float(row[needed["low"]]), + "close": float(row[needed["close"]]), + "volume": float(row[needed["volume"]]), + "amount": 0.0, + }) + except (IndexError, ValueError, TypeError, OSError): + continue + if not records: + return pd.DataFrame() + df = pd.DataFrame(records) + df["bar_time"] = pd.to_datetime(df["bar_time"], errors="coerce") + df = df.dropna(subset=["bar_time"]).sort_values("bar_time").reset_index(drop=True) + if start: + df = df[df["bar_time"] >= pd.Timestamp(start)] + if end: + df = df[df["bar_time"] < pd.Timestamp(end) + pd.Timedelta(days=1)] + return df + except Exception as e: + logger.exception("[kline5m %s] 雪球 5m 异常: %s", code6, e) + return pd.DataFrame() + def fetch_share_snapshot(self, code6: str) -> Optional[dict[str, Any]]: """单只股票的最新股本快照。 diff --git a/app/tasks/task_kline_5min.py b/app/tasks/task_kline_5min.py index f86e8bb..d645232 100644 --- a/app/tasks/task_kline_5min.py +++ b/app/tasks/task_kline_5min.py @@ -40,7 +40,7 @@ INCREMENT_OVERLAP_DAYS = 2 # 2026-07-08 教训: 数据源层有 20s timeout,task 层再硬兜底, # 防 SDK 升级 / 网络层 bug 让单只股票卡住(7月7日 4.4只/秒 后突然 0 持续 24h) # 7月9日 mairui/雪球 间歇性 "服务器连接失败" + Broken pipe,拉慢。给 120s 容忍。 -FETCH_HARD_TIMEOUT = 120.0 # 5min K 一只拉多年,单只允许更久 +FETCH_HARD_TIMEOUT = 600.0 # 5min K 全市场 5200+ 只,按 RPS 预估约 17min,给 10min 硬超时 class SyncKline5Min(SyncTask):