增加排序

backend/app/aggregator.py

@@ -5,6 +5,7 @@ from datetime import datetime
 
 from .binance import binance_client
 from .config import settings
+from .exceptions import BinanceRateLimitedError
 from .periods import to_ms
 
 logger = logging.getLogger(__name__)
@@ -20,6 +21,7 @@ class SymbolStats:
     rank: int = 0
     is_high_volume: bool = False
     is_high_change: bool = False
+    data_source: str = "klines"
 
     def to_dict(self) -> dict:
         d = asdict(self)
@@ -36,6 +38,16 @@ def format_volume(vol: float) -> str:
     return f"{vol:.0f}"
 
 
+def _finalize_top(stats: list[SymbolStats]) -> list[dict]:
+    stats.sort(key=lambda x: x.quote_volume, reverse=True)
+    top = stats[: settings.top_n]
+    for i, s in enumerate(top, 1):
+        s.rank = i
+        s.is_high_volume = s.quote_volume >= settings.volume_threshold
+        s.is_high_change = abs(s.price_change_pct) >= settings.change_threshold
+    return [s.to_dict() for s in top]
+
+
 def _aggregate_klines(klines: list, start_ms: int, end_ms: int) -> tuple[float, float, float]:
     quote_vol = 0.0
     open_price = 0.0
@@ -53,6 +65,31 @@ def _aggregate_klines(klines: list, start_ms: int, end_ms: int) -> tuple[float,
     return quote_vol, open_price, last_price
 
 
+async def aggregate_from_ticker24hr() -> list[dict]:
+    """仅 1 次 API 请求，使用滚动 24h 数据（今日刷新推荐）。"""
+    tickers = await binance_client.get_24hr_tickers()
+    stats: list[SymbolStats] = []
+    for t in tickers:
+        sym = t.get("symbol", "")
+        if not sym.endswith("USDT"):
+            continue
+        vol = float(t.get("quoteVolume", 0) or 0)
+        if vol <= 0:
+            continue
+        stats.append(
+            SymbolStats(
+                symbol=sym,
+                quote_volume=vol,
+                price_change_pct=float(t.get("priceChangePercent", 0) or 0),
+                open_price=float(t.get("openPrice", 0) or 0),
+                last_price=float(t.get("lastPrice", 0) or 0),
+                data_source="ticker24h",
+            )
+        )
+    logger.info("ticker24h mode: %d symbols, 1 API call", len(stats))
+    return _finalize_top(stats)
+
+
 async def _fetch_symbol_stats(
     symbol: str,
     start_ms: int,
@@ -79,14 +116,16 @@ async def _fetch_symbol_stats(
                 price_change_pct=pct,
                 open_price=open_price,
                 last_price=last_price,
+                data_source="klines",
             )
+        except BinanceRateLimitedError:
+            raise
         except Exception as e:
             logger.warning("Failed %s: %s", symbol, e)
             return None
 
 
 async def _pick_candidate_symbols(symbols: list[str]) -> list[str]:
-    """用 24h ticker 成交额预筛，避免对全市场并发拉 K 线触发 418 封禁。"""
     try:
         tickers = await binance_client.get_24hr_tickers()
         vol_map = {
@@ -97,18 +136,16 @@ async def _pick_candidate_symbols(symbols: list[str]) -> list[str]:
         ranked = sorted(symbols, key=lambda s: vol_map.get(s, 0.0), reverse=True)
         pool = min(settings.candidate_pool, len(ranked))
         picked = ranked[:pool]
-        logger.info(
-            "Candidate pool: %d / %d symbols (by 24h quoteVolume)",
-            len(picked),
-            len(symbols),
-        )
+        logger.info("Candidate pool: %d / %d symbols", len(picked), len(symbols))
         return picked
+    except BinanceRateLimitedError:
+        raise
     except Exception as e:
-        logger.warning("24hr ticker prescreen failed, using full list: %s", e)
-        return symbols
+        logger.warning("24hr prescreen failed: %s", e)
+        return symbols[: settings.candidate_pool]
 
 
-async def aggregate_period(
+async def aggregate_period_klines(
     start: datetime,
     end: datetime,
     use_live_prices: bool = False,
@@ -122,6 +159,8 @@ async def aggregate_period(
     if use_live_prices:
         try:
             prices = await binance_client.get_prices_batch(symbols)
+        except BinanceRateLimitedError:
+            raise
         except Exception as e:
             logger.warning("Batch prices failed: %s", e)
 
@@ -130,7 +169,7 @@ async def aggregate_period(
         _fetch_symbol_stats(s, start_ms, end_ms, prices, sem) for s in candidates
     ]
     logger.info(
-        "Aggregating period %s ~ %s (%d symbols, concurrency=%d)",
+        "klines mode: %s ~ %s, %d symbols, concurrency=%d",
         start.isoformat(),
         end.isoformat(),
         len(candidates),
@@ -138,21 +177,28 @@ async def aggregate_period(
     )
     results = await asyncio.gather(*tasks)
     stats = [r for r in results if r is not None and r.quote_volume > 0]
-    stats.sort(key=lambda x: x.quote_volume, reverse=True)
-    top = stats[: settings.top_n]
+    return _finalize_top(stats)
 
-    for i, s in enumerate(top, 1):
-        s.rank = i
-        s.is_high_volume = s.quote_volume >= settings.volume_threshold
-        s.is_high_change = abs(s.price_change_pct) >= settings.change_threshold
 
-    return [s.to_dict() for s in top]
+async def aggregate_period(
+    start: datetime,
+    end: datetime,
+    use_live_prices: bool = False,
+    mode: str | None = None,
+) -> list[dict]:
+    mode = mode or (
+        settings.today_data_mode if use_live_prices else settings.yesterday_data_mode
+    )
+    if mode == "ticker24h":
+        return await aggregate_from_ticker24hr()
+    return await aggregate_period_klines(start, end, use_live_prices)
 
 
 def enrich_snapshot_meta(
     items: list[dict],
     period_start: datetime,
     period_end: datetime,
+    data_mode: str = "",
 ) -> dict:
     return {
         "period_start": period_start.isoformat(),
@@ -161,5 +207,6 @@ def enrich_snapshot_meta(
         "top_n": settings.top_n,
         "volume_threshold": settings.volume_threshold,
         "change_threshold": settings.change_threshold,
+        "data_mode": data_mode,
         "items": items,
     }