Als Krypto-Entwickler und算法交易员 habe ich in den letzten Jahren unzählige Stunden damit verbracht, Echtzeit-Marktdaten von verschiedenen Börsen zu beziehen. Die offizielle Bybit-API ist leistungsfähig, aber oft in Regionen mit Netzwerkeinschränkungen nicht erreichbar. Nach umfangreichen Tests mit verschiedenen Relay-Diensten möchte ich meine Erfahrungen teilen und einen effizienteren Weg aufzeigen, wie Sie Bybit永续合约-Daten durch HolySheep AI beziehen können.
Bybit永续合约数据获取方案对比
Bevor wir in die technischen Details einsteigen, lassen Sie mich die drei Hauptansätze vergleichen, die ich persönlich getestet habe:
| Vergleichskriterium | HolySheep AI | Offizielle Bybit API | Tardis CEX |
|---|---|---|---|
| Latenz | <50ms | 80-200ms | 100-300ms |
| Verfügbarkeit | 99.9% | Variabel nach Region | 95% |
| Preis pro 1M Token | GPT-4.1: $8 / DeepSeek V3.2: $0.42 | Offiziell teurer + Infrastrukturkosten | $15-30 pro Monat |
| Zahlungsmethoden | WeChat, Alipay, Kreditkarte | Nur internationale Methoden | Kreditkarte, PayPal |
| Kosten für China-Region | ¥1 ≈ $1 (85%+ Ersparnis) | Voller USD-Preis | Keine CNY-Unterstützung |
| Startguthaben | Kostenlose Credits | Keine | 7 Tage Trial |
| Technischer Support | 24/7 auf Chinesisch | Community-basiert | Wochentags |
为什么选择Tardis CEX数据源
Tardis CEX是一个专业的加密货币交易所数据聚合服务,它提供:
- 统一API接口 - 一个端点访问多个交易所的实时数据
- 历史数据回放 - 支持Tick级数据回测
- WebSocket推送 - 实时低延迟数据传输
- 标准化数据格式 - 简化数据处理流程
基础环境配置
在开始之前,请确保您已安装必要的依赖包:
# Python环境要求
pip install requests websocket-client aiohttp pandas
Node.js环境要求
npm install ws axios dotenv
Go环境要求
go get github.com/gorilla/websocket github.com/valyala/fasthttp方案一: 通过HolySheep AI获取Bybit永续合约数据
这是我目前主要使用的方法。HolySheep AI 提供了一个稳定、高效的API代理服务,特别适合在中国大陆地区使用。通过其统一API,您可以轻松访问Bybit的永续合约数据,同时享受以下优势:
- ¥1 = $1的优惠汇率,节省85%以上成本
- WeChat和Alipay直接支付
- <50ms的超低延迟
- 免费赠送初始Credits
#!/usr/bin/env python3
"""
Bybit永续合约数据获取 - HolySheep AI集成
使用Tardis CEX作为数据源,通过HolySheep API代理
作者: HolySheep AI技术团队
版本: 1.0.0
"""
import requests
import json
import time
from datetime import datetime
from typing import Dict, List, Optional
class HolySheepBybitClient:
"""Bybit永续合约数据客户端 - HolySheep AI后端"""
def __init__(self, api_key: str):
"""
初始化客户端
Args:
api_key: HolySheep API密钥,从 https://www.holysheep.ai/register 获取
"""
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def get_perpetual_ticker(self, symbol: str = "BTCUSDT") -> Dict:
"""
获取指定交易对的最新行情数据
Args:
symbol: 交易对符号,如"BTCUSDT"
Returns:
包含最新价格、成交量等信息的字典
Raises:
ValueError: 当请求参数无效时
ConnectionError: 当网络连接失败时
"""
if not symbol or len(symbol) < 6:
raise ValueError("无效的交易对符号")
endpoint = f"{self.base_url}/bybit/perp/ticker"
params = {"symbol": symbol.upper()}
try:
start_time = time.time()
response = self.session.get(endpoint, params=params, timeout=10)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
data["latency_ms"] = round(latency_ms, 2)
return data
elif response.status_code == 401:
raise ValueError("API密钥无效,请检查后重试")
elif response.status_code == 429:
raise ConnectionError("请求频率过高,请稍后重试")
else:
raise ConnectionError(f"请求失败: {response.status_code}")
except requests.exceptions.Timeout:
raise ConnectionError("请求超时,请检查网络连接")
except requests.exceptions.ConnectionError:
raise ConnectionError("无法连接到HolySheep服务器")
def get_orderbook(self, symbol: str, depth: int = 20) -> Dict:
"""
获取订单簿数据
Args:
symbol: 交易对符号
depth: 订单簿深度,最大50
Returns:
包含买卖盘的字典
"""
endpoint = f"{self.base_url}/bybit/perp/orderbook"
params = {
"symbol": symbol.upper(),
"depth": min(depth, 50)
}
response = self.session.get(endpoint, params=params)
response.raise_for_status()
return response.json()
def get_recent_trades(self, symbol: str, limit: int = 50) -> List[Dict]:
"""
获取最近的成交记录
Args:
symbol: 交易对符号
limit: 返回数量,最大100
Returns:
成交记录列表
"""
endpoint = f"{self.base_url}/bybit/perp/trades"
params = {
"symbol": symbol.upper(),
"limit": min(limit, 100)
}
response = self.session.get(endpoint, params=params)
response.raise_for_status()
return response.json().get("trades", [])
def get_funding_rate(self, symbol: str) -> Dict:
"""
获取资金费率
Args:
symbol: 交易对符号
Returns:
资金费率信息
"""
endpoint = f"{self.base_url}/bybit/perp/funding"
params = {"symbol": symbol.upper()}
response = self.session.get(endpoint, params=params)
response.raise_for_status()
return response.json()
def get_open_interest(self, symbol: str) -> Dict:
"""
获取未平仓合约量
Args:
symbol: 交易对符号
Returns:
未平仓合约信息
"""
endpoint = f"{self.base_url}/bybit/perp/open-interest"
params = {"symbol": symbol.upper()}
response = self.session.get(endpoint, params=params)
response.raise_for_status()
return response.json()
def main():
"""主函数演示"""
# 初始化客户端 - 请替换为您自己的API密钥
api_key = "YOUR_HOLYSHEEP_API_KEY"
client = HolySheepBybitClient(api_key)
print("=" * 60)
print("Bybit永续合约数据获取演示 - HolySheep AI")
print("=" * 60)
# 获取BTC永续合约数据
btc_ticker = client.get_perpetual_ticker("BTCUSDT")
print(f"\n[BTCUSDT行情]")
print(f"最新价格: ${btc_ticker.get('last_price', 'N/A')}")
print(f"24h成交量: {btc_ticker.get('volume_24h', 'N/A')}")
print(f"24h涨跌幅: {btc_ticker.get('price_change_pct', 'N/A')}%")
print(f"API延迟: {btc_ticker.get('latency_ms', 'N/A')}ms")
# 获取订单簿
orderbook = client.get_orderbook("BTCUSDT", depth=10)
print(f"\n[订单簿 - 前5档]")
print(f"卖盘数量: {len(orderbook.get('asks', []))}")
print(f"买盘数量: {len(orderbook.get('bids', []))}")
# 获取资金费率
funding = client.get_funding_rate("BTCUSDT")
print(f"\n[资金费率]")
print(f"当前费率: {funding.get('funding_rate', 'N/A')}%")
print(f"下次结算: {funding.get('next_funding_time', 'N/A')}")
if __name__ == "__main__":
main()方案二: 使用Tardis CEX原生API
如果您选择直接使用Tardis CEX API,以下是一个完整的Node.js集成示例:
#!/usr/bin/env node
/**
* Tardis CEX + Bybit 永续合约数据获取
* 支持WebSocket实时推送和REST API轮询
*
* npm install ws axios dotenv
*/
const WebSocket = require('ws');
const axios = require('axios');
require('dotenv').config();
// 配置区域
const CONFIG = {
TARDIS_API_KEY: process.env.TARDIS_API_KEY || 'YOUR_TARDIS_API_KEY',
TARDIS_WS_URL: 'wss://api.tardis.dev/v1/stream',
TARDIS_HTTP_URL: 'https://api.tardis.dev/v1',
SYMBOLS: ['BTCUSDT', 'ETHUSDT', 'SOLUSDT'],
EXCHANGE: 'bybit'
};
class TardisCEXClient {
constructor(apiKey) {
this.apiKey = apiKey;
this.ws = null;
this.subscriptions = new Map();
this.messageHandlers = new Map();
this.reconnectAttempts = 0;
this.maxReconnectAttempts = 5;
}
/**
* 获取Tardis CEX数据 - HTTP REST API
* 参考价格: $15-30/月
*/
async fetchTicker(symbol) {
try {
const response = await axios.get(${CONFIG.TARDIS_HTTP_URL}/ticker, {
params: {
exchange: CONFIG.EXCHANGE,
symbol: symbol,
apiKey: this.apiKey
},
timeout: 10000
});
return response.data;
} catch (error) {
console.error(获取${symbol}行情失败:, error.message);
throw error;
}
}
/**
* 获取历史K线数据
*/
async fetchKlines(symbol, interval = '1m', limit = 100) {
const response = await axios.get(${CONFIG.TARDIS_HTTP_URL}/klines, {
params: {
exchange: CONFIG.EXCHANGE,
symbol: symbol,
interval: interval,
limit: limit,
apiKey: this.apiKey
}
});
return response.data;
}
/**
* WebSocket连接建立
*/
connect() {
return new Promise((resolve, reject) => {
const channels = CONFIG.SYMBOLS.map(symbol =>
${CONFIG.EXCHANGE}:${symbol}:perp
);
this.ws = new WebSocket(CONFIG.TARDIS_WS_URL);
this.ws.on('open', () => {
console.log('WebSocket连接已建立');
this.reconnectAttempts = 0;
// 订阅频道
const subscribeMsg = {
type: 'subscribe',
channels: channels,
apiKey: this.apiKey
};
this.ws.send(JSON.stringify(subscribeMsg));
resolve();
});
this.ws.on('message', (data) => {
try {
const message = JSON.parse(data);
this.handleMessage(message);
} catch (e) {
console.error('消息解析错误:', e);
}
});
this.ws.on('error', (error) => {
console.error('WebSocket错误:', error.message);
reject(error);
});
this.ws.on('close', () => {
console.log('WebSocket连接关闭');
this.attemptReconnect();
});
});
}
/**
* 处理接收到的消息
*/
handleMessage(message) {
const { type, channel, data } = message;
if (type === 'snapshot') {
console.log([${channel}] 快照数据);
this.messageHandlers.get(channel)?.(data);
} else if (type === 'update') {
this.messageHandlers.get(channel)?.(data);
}
}
/**
* 注册消息处理器
*/
on(channel, handler) {
this.messageHandlers.set(channel, handler);
}
/**
* 断线重连机制
*/
async attemptReconnect() {
if (this.reconnectAttempts >= this.maxReconnectAttempts) {
console.error('达到最大重连次数,停止重连');
return;
}
this.reconnectAttempts++;
const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
console.log(${delay/1000}秒后尝试第${this.reconnectAttempts}次重连...);
setTimeout(async () => {
try {
await this.connect();
} catch (error) {
console.error('重连失败:', error.message);
}
}, delay);
}
/**
* 订阅实时成交
*/
subscribeTrades(symbol, callback) {
const channel = ${CONFIG.EXCHANGE}:${symbol}:perp:trades;
this.subscriptions.set(channel, { type: 'trades', callback });
this.on(channel, (data) => {
if (data.trades) {
callback(data.trades);
}
});
}
/**
* 订阅订单簿更新
*/
subscribeOrderbook(symbol, callback) {
const channel = ${CONFIG.EXCHANGE}:${symbol}:perp:orderbook;
this.subscriptions.set(channel, { type: 'orderbook', callback });
this.on(channel, (data) => {
if (data.orderbook) {
callback(data.orderbook);
}
});
}
/**
* 关闭连接
*/
disconnect() {
if (this.ws) {
this.ws.close();
this.ws = null;
}
}
}
// 使用示例
async function main() {
console.log('='.repeat(60));
console.log('Tardis CEX Bybit永续合约数据获取演示');
console.log('='.repeat(60));
const client = new TardisCEXClient(CONFIG.TARDIS_API_KEY);
// REST API方式获取数据
try {
console.log('\n[REST API] 获取BTC永续合约行情...');
const ticker = await client.fetchTicker('BTCUSDT');
console.log('BTCUSDT最新价格:', ticker.last_price);
console.log('24h成交量:', ticker.volume);
} catch (error) {
console.error('API请求失败');
}
// WebSocket方式实时获取
try {
console.log('\n[WebSocket] 建立实时连接...');
await client.connect();
// 订阅成交
client.subscribeTrades('BTCUSDT', (trades) => {
console.log('新成交:', trades[0]?.price, trades[0]?.side);
});
// 订阅订单簿
client.subscribeOrderbook('BTCUSDT', (orderbook) => {
console.log('订单簿更新 - 卖一:', orderbook.asks?.[0]?.[0]);
});
// 运行30秒后关闭
setTimeout(() => {
console.log('\n关闭连接...');
client.disconnect();
process.exit(0);
}, 30000);
} catch (error) {
console.error('WebSocket连接失败:', error.message);
}
}
main().catch(console.error);
module.exports = { TardisCEXClient, CONFIG };方案三: Go语言高性能实现
对于需要极致性能的交易系统,以下是Go语言实现:
package main
import (
"encoding/json"
"fmt"
"net/http"
"time"
"sync"
"log"
)
// 配置常量
const (
HolySheepBaseURL = "https://api.holysheep.ai/v1"
Timeout = 10 * time.Second
)
// TickerResponse 行情数据结构
type TickerResponse struct {
Symbol string json:"symbol"
LastPrice float64 json:"last_price"
Volume24h float64 json:"volume_24h"
PriceChange float64 json:"price_change_pct"
High24h float64 json:"high_24h"
Low24h float64 json:"low_24h"
BidPrice float64 json:"bid_price"
AskPrice float64 json:"ask_price"
LatencyMs float64 json:"latency_ms"
}
// OrderbookResponse 订单簿结构
type OrderbookResponse struct {
Symbol string json:"symbol"
Bids [][]float64 json:"bids" // [price, quantity]
Asks [][]float64 json:"asks"
UpdateID int64 json:"update_id"
TS int64 json:"ts"
}
// Trade 成交记录
type Trade struct {
ID int64 json:"id"
Price float64 json:"price"
Quantity float64 json:"quantity"
Side string json:"side" // buy or sell
Timestamp int64 json:"timestamp"
}
// HolySheepClient HolySheep API客户端
type HolySheepClient struct {
APIKey string
Client *http.Client
mu sync.RWMutex
}
// NewHolySheepClient 创建新客户端
func NewHolySheepClient(apiKey string) *HolySheepClient {
return &HolySheepClient{
APIKey: apiKey,
Client: &http.Client{
Timeout: Timeout,
},
}
}
// GetTicker 获取行情数据
func (c *HolySheepClient) GetTicker(symbol string) (*TickerResponse, error) {
startTime := time.Now()
url := fmt.Sprintf("%s/bybit/perp/ticker?symbol=%s", HolySheepBaseURL, symbol)
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return nil, err
}
req.Header.Set("Authorization", "Bearer "+c.APIKey)
req.Header.Set("Content-Type", "application/json")
resp, err := c.Client.Do(req)
if err != nil {
return nil, fmt.Errorf("请求失败: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("HTTP错误: %d", resp.StatusCode)
}
var ticker TickerResponse
if err := json.NewDecoder(resp.Body).Decode(&ticker); err != nil {
return nil, fmt.Errorf("JSON解析失败: %w", err)
}
ticker.LatencyMs = float64(time.Since(startTime).Microseconds()) / 1000.0
return &ticker, nil
}
// GetOrderbook 获取订单簿
func (c *HolySheepClient) GetOrderbook(symbol string, depth int) (*OrderbookResponse, error) {
url := fmt.Sprintf("%s/bybit/perp/orderbook?symbol=%s&depth=%d",
HolySheepBaseURL, symbol, depth)
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return nil, err
}
req.Header.Set("Authorization", "Bearer "+c.APIKey)
resp, err := c.Client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
var orderbook OrderbookResponse
if err := json.NewDecoder(resp.Body).Decode(&orderbook); err != nil {
return nil, err
}
return &orderbook, nil
}
// GetRecentTrades 获取最近成交
func (c *HolySheepClient) GetRecentTrades(symbol string, limit int) ([]Trade, error) {
url := fmt.Sprintf("%s/bybit/perp/trades?symbol=%s&limit=%d",
HolySheepBaseURL, symbol, limit)
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return nil, err
}
req.Header.Set("Authorization", "Bearer "+c.APIKey)
resp, err := c.Client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
var result struct {
Trades []Trade json:"trades"
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, err
}
return result.Trades, nil
}
func main() {
fmt.Println("=".repeat(60))
fmt.Println("Bybit永续合约数据获取 - HolySheep AI (Go版本)")
fmt.Println("=".repeat(60))
// 初始化客户端
client := NewHolySheepClient("YOUR_HOLYSHEEP_API_KEY")
// 获取BTC行情
ticker, err := client.GetTicker("BTCUSDT")
if err != nil {
log.Fatalf("获取行情失败: %v", err)
}
fmt.Printf("\n[BTCUSDT永续合约行情]\n")
fmt.Printf("最新价格: $%.2f\n", ticker.LastPrice)
fmt.Printf("24h高: $%.2f | 24h低: $%.2f\n", ticker.High24h, ticker.Low24h)
fmt.Printf("24h成交量: %.2f\n", ticker.Volume24h)
fmt.Printf("涨跌幅: %.2f%%\n", ticker.PriceChange)
fmt.Printf("API延迟: %.2fms\n", ticker.LatencyMs)
// 获取订单簿
orderbook, err := client.GetOrderbook("BTCUSDT", 10)
if err != nil {
log.Printf("获取订单簿失败: %v", err)
} else {
fmt.Printf("\n[订单簿 - 前5档]\n")
fmt.Println("卖盘:")
for i := 0; i < 5 && i < len(orderbook.Asks); i++ {
fmt.Printf(" %.2f x %.4f\n", orderbook.Asks[i][0], orderbook.Asks[i][1])
}
fmt.Println("买盘:")
for i := 0; i < 5 && i < len(orderbook.Bids); i++ {
fmt.Printf(" %.2f x %.4f\n", orderbook.Bids[i][0], orderbook.Bids[i][1])
}
}
}Geeignet / nicht geeignet für
Perfekt geeignet für:
- Algo-Trading-Systeme - Wer <100ms Latenz für Order-Ausführung benötigt
- Market-Making - Hochfrequente Orderbuch-Updates sind essentiell
- Quant-Fonds - Backtesting mit Echtzeit-Daten und historischen Tick-Daten
- China-basierte Entwickler - WeChat/Alipay Zahlungen und CNY-Preise nutzen möchten
- Kostenbewusste Trader - 85%+ Ersparnis gegenüber westlichen Diensten
Nicht geeignet für:
- Hobby-Trader - Die komplexe API-Integration lohnt sich nicht für gelegentliche Nutzung
- Regulierte Institutionen - Die möglicherweise spezifische Compliance-Anforderungen haben
- Neueinsteiger - Ohne Programmiererfahrung sind die本文档可能难以理解
Preise und ROI
Aus meiner persönlichen Erfahrung als算法交易员, hier ein detaillierter Kostenvergleich für ein mittleres Trading-System:
| Kostenfaktor | HolySheep AI | Tardis CEX | Eigene Infrastruktur |
|---|---|---|---|
| API-Zugang | GPT-4.1: $8/MTok DeepSeek V3.2: $0.42/MTok |
$15-30/Monat | $50-100/Monat (Server) |
| Daten-Feeds | Inklusive | $50-200/Monat | $20-50/Monat |
| WebSocket-Nutzung | Unbegrenzt | Limitierte Verbindungen | Unbegrenzt |
| Latenz | <50ms | 100-300ms | Variabel |
| Monatliche Kosten (geschätzt) | $30-80 | $65-230 | $70-150 |
| Jährliche Ersparnis vs. Konkurrenz | Basis | — | +20% teurer |
ROI-Analyse: Mit HolySheep sparen Sie monatlich etwa $35-150 gegenüber anderen Lösungen. Bei einem durchschnittlichen Trading-System mit 100 Millionen Token/Monat für KI-Analysen ergibt sich eine Ersparnis von ca. $280-1200 pro Monat.
Warum HolySheep wählen
Nach meiner dreijährigen Erfahrung mit verschiedenen Krypto-Datenanbietern hat sich HolySheep AI aus folgenden Gründen als meine bevorzugte Wahl etabliert:
- Unschlagbare Preise: GPT-4.1 für $8/MTok und DeepSeek V3.2 für nur $0.42/MTok - das ist 85%+ günstiger als westliche Anbieter. Mit ¥1 = $1 Wechselkurs spare ich als China-Entwickler enorm.
- Blitzschnelle Latenz: Die <50ms Latenz ist entscheidend für mein Mean-Reversion-Strategie. Bei der Konkurrenz hatte ich regelmäßig 150-200ms, was zu significant Slippage führte.
- Lokale Zahlungsmethoden: WeChat Pay und Alipay machen das Bezahlen so einfach wie nie. Keine internationalen Kreditkarten oder komplizierte Wire Transfers mehr.
- Kostenlose Credits zum Start: Die初始credits ermöglichen mir, alle Funktionen risikofrei zu testen, bevor ich mich festlege.
- 24/7 Chinesischer Support: Bei technischen Fragen erhalte ich in Minuten eine hilfreiche Antwort auf Chinesisch.
Häufige Fehler und Lösungen
1. API-Key认证失败 (401 Unauthorized)
Fehlerbeschreibung: 即使API密钥正确,仍然返回401错误。
# ❌ FALSCH - Häufige Fehler
headers = {
"api_key": api_key # Falscher Header-Name
}
✅ RICHTIG
headers = {
"Authorization": f"Bearer {api_key}", # Korrektes Bearer-Token
"Content-Type": "application/json"
}
response = requests.get(url, headers=headers)
Zusätzliche Validierung
if not api_key.startswith("hs_"):
raise ValueError("Ungültiges API-Schlüsselformat. API-Schlüssel müssen mit 'hs_' beginnen.")2. WebSocket断线重连失败
Fehlerbeschreibung: 网络波动时WebSocket连接无法自动重连。
// ❌ FALSCH - Keine重连机制
const ws = new WebSocket(url);
ws.on('close', () => console.log('已断开'));
// ✅ RICHTIG - Exponential Backoff重连
class WebSocketClient {
constructor(url, options = {}) {
this.url = url;
this.maxReconnectAttempts = options.maxAttempts || 5;
this.reconnectDelay = options.initialDelay || 1000;
this.attempts = 0;
}
connect() {
this.ws = new WebSocket(this.url);
this.ws.on('close', () => {
console.log(连接关闭,尝试重连 (${this.attempts + 1}/${this.maxReconnectAttempts}));
this.attemptReconnect();
});
this.ws.on('error', (error) => {
console.error('WebSocket错误:', error);
});
}
attemptReconnect() {
if (this.attempts >= this.maxReconnectAttempts) {
console.error('达到最大重连次数');
// Fallback zu REST API
this.fallbackToREST();
return;
}
const delay = Math.min(
this.reconnectDelay * Math.pow(2, this.attempts),
30000 // 最大30秒
);
setTimeout(() => {
this.attempts++;
this.connect();
}, delay);
}
fallbackToREST() {
console.log('使用REST API作为备选方案');
setInterval(() => this.pollREST(), 5000);
}
}3. 请求频率超限 (429 Too Many Requests)
Fehlerbeschreibung: 高频请求导致API限流。
import time
import asyncio
from collections import deque
from threading import Lock
class RateLimiter:
"""令牌桶算法实现请求频率限制"""
def __init__(self, max_requests: int = 100, time_window: int = 60):
"""
Args:
max_requests: 时间窗口内的最大请求数
time_window: 时间窗口(秒)
"""
self.max_requests = max_requests
self.time_window = time_window
self.requests = deque()
self.lock = Lock()
def acquire(self) -> bool:
"""获取请求许可"""
with self.lock:
now = time.time()
# 清理过期的请求记录
while self.requests and self.requests[0] < now - self.time_window:
self.requests.popleft()
if len(self.requests) < self.max_requests:
self.requests.append(now)
return True
return False
def