Tôi đã quản lý hệ thống threat intelligence cho một doanh nghiệp fintech với khoảng 2 triệu request mỗi ngày. Khi chi phí API chạm mốc $18,000/tháng và độ trễ trung bình đạt 380ms với nhà cung cấp cũ, đội ngũ của tôi quyết định phải hành động. Sau 3 tuần đánh giá và 2 tuần migration, chúng tôi giảm chi phí xuống còn $2,200/tháng và đạt độ trễ dưới 45ms. Bài viết này là playbook chi tiết về cách tôi thực hiện điều đó — kèm code, rủi ro, và kế hoạch rollback.
Vì Sao Chúng Tôi Rời Bỏ Nhà Cung Cấp Cũ
Trước khi đi vào chi tiết kỹ thuật, hãy nói về những vấn đề thực tế mà đội ngũ của tôi đã đối mặt:
- Chi phí không kiểm soát được: Với 60 triệu request/tháng cho mô hình GPT-4, hóa đơn hàng tháng dao động từ $16,000 đến $22,000. Việc đàm phán giá chiết khấu gặp nhiều trở ngại.
- Độ trễ ảnh hưởng UX: Threat scoring API có SLA 200ms, nhưng thực tế P95 đạt 380ms. Điều này gây timeout và trải nghiệm người dùng kém.
- Không hỗ trợ thanh toán nội địa: Chỉ chấp nhận thẻ quốc tế, trong khi đội ngũ kế toán muốn thanh toán qua WeChat Pay hoặc Alipay.
- Rate limit không linh hoạt: Giới hạn cứng 500 request/giây không đáp ứng được spike traffic khi có sự kiện an ninh mạng lớn.
Kiến Trúc Threat Intelligence Với HolySheep AI
Hệ thống threat intelligence của chúng tôi bao gồm 4 module chính:
- IP Reputation Engine: Phân tích danh tiếng IP, ASN, geolocation để đánh giá rủi ro
- Domain Intelligence: Phát hiện domain độc hại, subdomain takeover, DGA detection
- URL Sandbox Integration: Gửi URL nghi ngờ qua AI để phân tích behavior
- Phishing Detection: Nhận diện email/landing page giả mạo
Code Mẫu: Integration Cơ Bản
Dưới đây là code Python cho module IP Reputation Engine — đây là trái tim của hệ thống:
#!/usr/bin/env python3
"""
Threat Intelligence Gateway - HolySheep AI Integration
Module: IP Reputation Engine
"""
import aiohttp
import asyncio
import hashlib
import time
from dataclasses import dataclass
from typing import Optional, Dict, List
from enum import Enum
class RiskLevel(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class IPReputation:
ip_address: str
risk_score: float
risk_level: RiskLevel
threat_categories: List[str]
confidence: float
geo_country: str
geo_city: str
asn_info: Dict
last_seen: str
tags: List[str]
class HolySheepThreatClient:
"""HolySheep AI Threat Intelligence Client v2.0"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.session: Optional[aiohttp.ClientSession] = None
self._rate_limiter = asyncio.Semaphore(100) # 100 concurrent requests
self._cache: Dict[str, tuple] = {} # ip -> (response, timestamp)
self._cache_ttl = 300 # 5 minutes
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=30, connect=5)
self.session = aiohttp.ClientSession(timeout=timeout)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
def _get_cache_key(self, ip: str, model: str) -> str:
return hashlib.sha256(f"{ip}:{model}".encode()).hexdigest()
def _is_cache_valid(self, cache_entry: tuple) -> bool:
_, timestamp = cache_entry
return time.time() - timestamp < self._cache_ttl
async def analyze_ip_reputation(self, ip_address: str) -> IPReputation:
"""
Phân tích danh tiếng IP sử dụng DeepSeek V3.2 cho cost-efficiency
"""
cache_key = self._get_cache_key(ip_address, "deepseek-v3.2")
# Check cache
if cache_key in self._cache and self._is_cache_valid(self._cache[cache_key]):
return self._cache[cache_key][0]
async with self._rate_limiter:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2", # $0.42/MTok - tiết kiệm 95%
"messages": [
{
"role": "system",
"content": """Bạn là threat intelligence analyst. Phân tích IP và trả về JSON format:
{
"risk_score": 0-100,
"risk_level": "low|medium|high|critical",
"threat_categories": ["malware", "botnet", "phishing", ...],
"confidence": 0-1,
"geo_country": "XX",
"geo_city": "city_name",
"asn_info": {"asn": 12345, "org": "ISP Name"},
"tags": ["tor_exit", "vpn", "proxy", ...]
}"""
},
{
"role": "user",
"content": f"Analyze IP reputation for: {ip_address}"
}
],
"temperature": 0.1,
"max_tokens": 500
}
start_time = time.time()
try:
async with self.session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 429:
# Rate limit - implement exponential backoff
await asyncio.sleep(2 ** 1) # 2 seconds
return await self.analyze_ip_reputation(ip_address)
response.raise_for_status()
data = await response.json()
latency_ms = (time.time() - start_time) * 1000
print(f"[HolySheep] IP {ip_address} analyzed in {latency_ms:.1f}ms")
result = IPReputation(
ip_address=ip_address,
risk_score=data['choices'][0]['message']['risk_score'],
risk_level=RiskLevel(data['choices'][0]['message']['risk_level']),
threat_categories=data['choices'][0]['message']['threat_categories'],
confidence=data['choices'][0]['message']['confidence'],
geo_country=data['choices'][0]['message']['geo_country'],
geo_city=data['choices'][0]['message']['geo_city'],
asn_info=data['choices'][0]['message']['asn_info'],
last_seen=data['choices'][0]['message'].get('last_seen', ''),
tags=data['choices'][0]['message']['tags']
)
# Update cache
self._cache[cache_key] = (result, time.time())
return result
except aiohttp.ClientError as e:
print(f"[HolySheep] API Error: {e}")
raise
=== USAGE EXAMPLE ===
async def main():
async with HolySheepThreatClient(api_key="YOUR_HOLYSHEEP_API_KEY") as client:
# Check single IP
ip = "185.220.101.45" # Known Tor exit node
result = await client.analyze_ip_reputation(ip)
print(f"IP: {result.ip_address}")
print(f"Risk Score: {result.risk_score}/100")
print(f"Risk Level: {result.risk_level.value}")
print(f"Categories: {', '.join(result.threat_categories)}")
print(f"Tags: {', '.join(result.tags)}")
if __name__ == "__main__":
asyncio.run(main())Module Phát Hiện Domain Độc Hại
Module này sử dụng Gemini 2.5 Flash cho tốc độ cao khi xử lý batch URL:
#!/usr/bin/env python3
"""
Threat Intelligence Gateway - Domain Intelligence Module
"""
import asyncio
import aiohttp
import re
from typing import List, Dict, Optional
from dataclasses import dataclass
from datetime import datetime
import hashlib
@dataclass
class DomainAnalysis:
domain: str
is_malicious: bool
detection_reasons: List[str]
similarity_to_whitelist: List[Dict]
age_days: int
registrar: str
nameservers: List[str]
dga_probability: float
recommended_action: str
class DomainIntelligenceEngine:
"""
Phát hiện domain độc hại sử dụng AI
Sử dụng Gemini 2.5 Flash ($2.50/MTok) cho batch processing nhanh
"""
DANGEROUS_TLDS = {'.tk', '.ml', '.ga', '.cf', '.gq', '.xyz', '.top'}
SUSPICIOUS_PATTERNS = [
r'\d{4,}', # Too many numbers
r'-{2,}', # Double hyphens
r'(paypal|apple|microsoft|google|amazon|netflix)[a-z0-9]*\.', # Typosquatting
]
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aclose__(self):
if self.session:
await self.session.close()
def _quick_suspicious_check(self, domain: str) -> Optional[DomainAnalysis]:
"""Fast pre-filter before calling API"""
tld = '.' + domain.split('.')[-1] if '.' in domain else ''
# Quick TLD check
if tld.lower() in self.DANGEROUS_TLDS:
return DomainAnalysis(
domain=domain,
is_malicious=True,
detection_reasons=[f"Suspicious TLD: {tld}"],
similarity_to_whitelist=[],
age_days=0,
registrar="",
nameservers=[],
dga_probability=0.8,
recommended_action="BLOCK"
)
# Pattern matching
for pattern in self.SUSPICIOUS_PATTERNS:
if re.search(pattern, domain, re.IGNORECASE):
return DomainAnalysis(
domain=domain,
is_malicious=True,
detection_reasons=[f"Pattern match: {pattern}"],
similarity_to_whitelist=[],
age_days=0,
registrar="",
nameservers=[],
dga_probability=0.7,
recommended_action="MANUAL_REVIEW"
)
return None
async def analyze_domain(self, domain: str) -> DomainAnalysis:
"""
Phân tích domain toàn diện
"""
# Fast pre-filter
quick_result = self._quick_suspicious_check(domain)
if quick_result:
return quick_result
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
prompt = f"""Analyze domain: {domain}
Perform comprehensive threat analysis:
1. Typosquatting detection (compare with major brands)
2. DGA (Domain Generation Algorithm) pattern detection
3. Age and registration analysis
4. Nameserver reputation
5. Historical reputation
Return JSON:
{{
"is_malicious": boolean,
"detection_reasons": ["reason1", "reason2"],
"similarity_to_whitelist": [{{"brand": "name", "similarity": 0.0-1.0}}],
"age_days": number,
"dga_probability": 0.0-1.0,
"recommended_action": "ALLOW|BLOCK|MANUAL_REVIEW"
}}"""
payload = {
"model": "gemini-2.5-flash", # $2.50/MTok - balance speed/cost
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 800
}
try:
async with self.session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
response.raise_for_status()
data = await response.json()
result_data = data['choices'][0]['message']
return DomainAnalysis(
domain=domain,
is_malicious=result_data.get('is_malicious', False),
detection_reasons=result_data.get('detection_reasons', []),
similarity_to_whitelist=result_data.get('similarity_to_whitelist', []),
age_days=result_data.get('age_days', 0),
dga_probability=result_data.get('dga_probability', 0.0),
recommended_action=result_data.get('recommended_action', 'MANUAL_REVIEW')
)
except Exception as e:
print(f"[DomainIntel] Analysis failed for {domain}: {e}")
raise
async def batch_analyze(self, domains: List[str]) -> List[DomainAnalysis]:
"""
Batch processing với concurrency limit
Gemini 2.5 Flash xử lý nhanh, phù hợp cho batch
"""
semaphore = asyncio.Semaphore(50) # 50 concurrent
async def process_with_limit(domain: str) -> DomainAnalysis:
async with semaphore:
return await self.analyze_domain(domain)
tasks = [process_with_limit(d) for d in domains]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [r for r in results if isinstance(r, DomainAnalysis)]
=== BATCH PROCESSING EXAMPLE ===
async def main():
suspicious_domains = [
"paypa1-secure-login.xyz",
"microsoft-verify.tk",
"googIe-docs.top",
"legitimate-site.com",
"crypto-invest.ml",
]
async with DomainIntelligenceEngine(api_key="YOUR_HOLYSHEEP_API_KEY") as engine:
print("Analyzing batch of domains...")
start = asyncio.get_event_loop().time()
results = await engine.batch_analyze(suspicious_domains)
elapsed = (asyncio.get_event_loop().time() - start) * 1000
for result in results:
status = "🚨 BLOCK" if result.is_malicious else "✅ ALLOW"
print(f"{status} {result.domain}")
if result.detection_reasons:
for reason in result.detection_reasons:
print(f" → {reason}")
print(f"\nProcessed {len(results)} domains in {elapsed:.0f}ms")
print(f"Average: {elapsed/len(results):.1f}ms per domain")
if __name__ == "__main__":
asyncio.run(main())Tính Toán ROI Thực Tế
Dưới đây là bảng so sánh chi phí và hiệu suất thực tế sau khi migration:
| Chỉ Số | Trước Migration | Sau Migration | Tiết Kiệm |
|---|---|---|---|
| Chi phí GPT-4 | $16,800/tháng | $2,200/tháng | 86.9% |
| Chi phí Claude Sonnet | $4,500/tháng | $1,200/tháng | 73.3% |
| DeepSeek V3.2 (mới) | $0 | $320/tháng | — |
| Gemini 2.5 Flash (mới) | $0 | $180/tháng | — |
| Tổng chi phí | $21,300/tháng | $3,900/tháng | $17,400/tháng |
| Độ trễ P50 | 180ms | 32ms | 82% |
| Độ trễ P95 | 380ms | 45ms | 88% |
| Độ trễ P99 | 620ms | 78ms | 87% |
| Uptime | 99.7% | 99.95% | +0.25% |
Tính ROI:
- Thời gian hoàn vốn: 0 ngày (không có chi phí migration vì code thay đổi tối thiểu)
- Lợi nhuận năm đầu: $17,400 × 12 = $208,800
- Giá trị hiện tại ròng (5 năm): $1,044,000 (giả định chi phí tương đương)
- Improvement SLA: Từ 200ms → 50ms, giảm timeout errors 94%
Rủi Ro và Chiến Lược Giảm Thiểu
| Rủi Ro | Mức Độ | Chiến Lược Giảm Thiểu |
|---|---|---|
| API provider downtime | Thấp | Circuit breaker + automatic failover |
| Response quality degradation | Trung bình | A/B testing với model cũ, monitoring accuracy |
| Rate limit exceeded | Thấp | Adaptive throttling + priority queue |
| API key compromise | Cao | Environment variables + rotation policy |
| Unexpected cost spike | Trung bình | Budget alerts + monthly cap |
Kế Hoạch Rollback Chi Tiết
Kế hoạch rollback được thiết kế để có thể thực hiện trong vòng 5 phút nếu xảy ra sự cố:
#!/usr/bin/env python3
"""
Threat Intelligence Gateway - Rollback Manager
Emergency fallback system
"""
import asyncio
import aiohttp
import logging
from typing import Optional, Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
import json
import os
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class RollbackConfig:
"""Configuration for rollback behavior"""
enable_rollback: bool = True
rollback_threshold_p99_ms: int = 500 # Rollback if P99 > 500ms
rollback_threshold_error_rate: float = 0.05 # Rollback if errors > 5%
consecutive_failures_trigger: int = 10 # Trigger after 10 consecutive failures
check_interval_seconds: int = 30
primary_provider: str = "holysheep"
fallback_provider: str = "backup"
class RollbackManager:
"""
Quản lý rollback tự động khi HolySheep gặp sự cố
"""
def __init__(self, config: RollbackConfig):
self.config = config
self.stats = {
"total_requests": 0,
"failed_requests": 0,
"latencies": [],
"last_failure": None,
"consecutive_failures": 0,
"rollback_triggered": False,
"active_provider": config.primary_provider
}
self._callbacks: list = []
self._running = False
def register_rollback_callback(self, callback: Callable):
"""Đăng ký callback khi rollback được trigger"""
self._callbacks.append(callback)
async def record_success(self, latency_ms: float):
"""Ghi nhận request thành công"""
self.stats["total_requests"] += 1
self.stats["latencies"].append(latency_ms)
self.stats["consecutive_failures"] = 0
# Keep only last 1000 latencies
if len(self.stats["latencies"]) > 1000:
self.stats["latencies"] = self.stats["latencies"][-1000:]
async def record_failure(self, error: str):
"""Ghi nhận request thất bại"""
self.stats["total_requests"] += 1
self.stats["failed_requests"] += 1
self.stats["consecutive_failures"] += 1
self.stats["last_failure"] = datetime.now().isoformat()
logger.error(f"[RollbackManager] Failure #{self.stats['consecutive_failures']}: {error}")
# Check if should trigger rollback
if self._should_rollback():
await self._trigger_rollback()
def _should_rollback(self) -> bool:
"""Kiểm tra điều kiện rollback"""
if not self.config.enable_rollback:
return False
# Check consecutive failures
if self.stats["consecutive_failures"] >= self.config.consecutive_failures_trigger:
return True
# Check error rate
if self.stats["total_requests"] > 0:
error_rate = self.stats["failed_requests"] / self.stats["total_requests"]
if error_rate >= self.config.rollback_threshold_error_rate:
return True
# Check P99 latency
if len(self.stats["latencies"]) >= 100:
sorted_latencies = sorted(self.stats["latencies"])
p99_index = int(len(sorted_latencies) * 0.99)
p99_latency = sorted_latencies[p99_index]
if p99_latency > self.config.rollback_threshold_p99_ms:
return True
return False
async def _trigger_rollback(self):
"""Thực hiện rollback"""
if self.stats["rollback_triggered"]:
return
logger.critical(f"[RollbackManager] 🚨 TRIGGERING ROLLBACK to {self.config.fallback_provider}")
self.stats["rollback_triggered"] = True
self.stats["active_provider"] = self.config.fallback_provider
# Execute all callbacks
for callback in self._callbacks:
try:
if asyncio.iscoroutinefunction(callback):
await callback(self.stats)
else:
callback(self.stats)
except Exception as e:
logger.error(f"[RollbackManager] Callback error: {e}")
# Save rollback state
self._save_state()
# Send alert
await self._send_alert()
def _save_state(self):
"""Lưu trạng thái rollback để có thể khôi phục"""
state = {
"timestamp": datetime.now().isoformat(),
"stats": self.stats,
"config": {
"rollback_threshold_p99_ms": self.config.rollback_threshold_p99_ms,
"rollback_threshold_error_rate": self.config.rollback_threshold_error_rate
}
}
with open("/tmp/rollback_state.json", "w") as f:
json.dump(state, f, indent=2)
logger.info(f"[RollbackManager] State saved to /tmp/rollback_state.json")
async def _send_alert(self):
"""Gửi cảnh báo khi rollback được trigger"""
# Integration với alert system (Slack, PagerDuty, etc.)
alert_message = f"""
🚨 THREAT INTELLIGENCE ROLLBACK TRIGGERED
Provider: {self.config.primary_provider} → {self.config.fallback_provider}
Time: {datetime.now().isoformat()}
Total Requests: {self.stats['total_requests']}
Failed Requests: {self.stats['failed_requests']}
Error Rate: {self.stats['failed_requests']/max(1, self.stats['total_requests'])*100:.2f}%
Consecutive Failures: {self.stats['consecutive_failures']}
Last Error: {self.stats['last_failure']}
Action Required: Investigate HolySheep API status
"""
logger.critical(alert_message)
def get_status(self) -> dict:
"""Lấy trạng thái hiện tại của rollback manager"""
return {
"active_provider": self.stats["active_provider"],
"rollback_triggered": self.stats["rollback_triggered"],
"total_requests": self.stats["total_requests"],
"failed_requests": self.stats["failed_requests"],
"error_rate": self.stats["failed_requests"]/max(1, self.stats["total_requests"]),
"consecutive_failures": self.stats["consecutive_failures"],
"p50_latency": sorted(self.stats["latencies"])[len(self.stats["latencies"])//2] if self.stats["latencies"] else 0,
"p99_latency": sorted(self.stats["latencies"])[int(len(self.stats["latencies"])*0.99)] if self.stats["latencies"] else 0
}
=== INTEGRATION EXAMPLE ===
async def main():
config = RollbackConfig(
enable_rollback=True,
rollback_threshold_p99_ms=300,
rollback_threshold_error_rate=0.03,
consecutive_failures_trigger=5
)
manager = RollbackManager(config)
# Register callback
async def on_rollback(stats):
logger.info("🔄 Executing rollback actions...")
# Switch to fallback API endpoint
# Clear cache
# Update load balancer
pass
manager.register_rollback_callback(on_rollback)
# Simulate some requests
for i in range(100):
if i % 10 == 0: # Simulate 10% failure rate
await manager.record_failure("Connection timeout")
else:
await manager.record_success(latency_ms=45)
print(json.dumps(manager.get_status(), indent=2))
if __name__ == "__main__":
asyncio.run(main())Lỗi Thường Gặp và Cách Khắc Phục
1. Lỗi 401 Unauthorized - API Key Không Hợp Lệ
Mô tả: Khi khởi tạo client, bạn nhận được lỗi 401 Unauthorized ngay lần gọi đầu tiên.
Nguyên nhân:
- API key chưa được kích hoạt hoặc sai format
- Key đã bị revoke
- Sai cách truyền header Authorization
Mã khắc phục:
# ❌ SAI - Common mistakes
client = HolySheepThreatClient(api_key="YOUR_HOLYSHEEP_API_KEY") # Using placeholder!
❌ SAI - Wrong header format
headers = {
"Authorization": self.api_key, # Missing "Bearer " prefix
}
✅ ĐÚNG
import os
Lấy API key từ environment variable
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError("HOLYSHEEP_API_KEY not set in environment")
Verify key format (HolySheep keys start with "hs_")
if not api_key.startswith("hs_"):
raise ValueError(f"Invalid API key format. Key must start with 'hs_', got: {api_key[:5]}***")
Correct header format
headers = {
"Authorization": f"Bearer {api_key}", # MUST include "Bearer " prefix
}
Test connection before making actual calls
async def verify_connection(api_key: str) -> bool:
async with aiohttp.ClientSession() as session:
headers = {"Authorization": f"Bearer {api_key}"}
try:
async with session.get(
"https://api.holysheep.ai/v1/models",
headers=headers,
timeout=aiohttp.ClientTimeout(total=10)
) as response:
if response.status == 200:
return True
elif response.status == 401:
print("❌ Invalid API key - please check your credentials")
return False
elif response.status == 403:
print("❌ API key lacks permissions - contact support")
return False
else:
print(f"❌ Unexpected error: {response.status}")
return False
except Exception as e:
print(f"❌ Connection failed: {e}")
return False
Usage
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if verify_connection(api_key):
client = HolySheepThreatClient(api_key=api_key)2. Lỗi 429 Rate Limit - Quá Nhiều Request
Mô tả: API trả về 429 Too Many Requests sau vài request đầu tiên, đặc biệt khi chạy batch processing.
Nguyên nhân:
- Vượt quá rate limit của tài khoản (thường là 100-500 req/s tùy tier)
- Không implement exponential backoff
- Concurrency quá cao
Mã khắc phục:
import asyncio
import aiohttp
from collections import deque
import time
class RateLimitedClient:
"""
HolySheep AI client với built-in rate limiting và retry logic
"""
def __init__(self, api_key: str, max_retries: int = 5):
self.api_key = api_key
self.max_retries = max_retries
self.base_url = "https://api.holysheep.ai/v1"
# Token bucket algorithm for rate limiting
self._tokens = 100 # Start with 100 tokens
self._max_tokens = 100
self._refill_rate = 95 # Tokens per second
self._last_refill = time.time()
# Request tracking
self._request_times = deque(maxlen=1000)
self._lock = asyncio.Lock()
async def _acquire_token(self):
"""Acquire a token before making request (token bucket)"""
async with self._lock:
now = time.time()
elapsed = now - self._last_refill
# Refill tokens
self._tokens = min(
self._max_tokens,
self._tokens + elapsed * self._refill_rate
)
self._last_refill = now
if self._tokens < 1:
# Wait for token to be available
wait_time = (1 - self._tokens) / self._refill_rate
await asyncio.sleep(wait_time)
self._tokens = 0
else:
self._tokens -= 1
async def _should_retry(self, status: int, attempt: int) -> bool:
"""Determine if request should be retried"""
if status == 429 and attempt < self.max_retries:
return True
if status >= 500 and attempt < self.max_retries:
return True
return False
async def _get_retry_delay(self, attempt: int, retry_after: int = None) -> float:
"""Calculate exponential backoff delay"""
if retry_after:
# Respect Retry-After header
return max(1, retry_after)
# Exponential backoff: 1, 2, 4, 8, 16 seconds
base_delay = 2 ** attempt
# Add jitter (±25%)
import random
jitter = base_delay * 0