背景
DeepSeek 在 deepseek.com 发布的文档中确定 DeepSeek API 相对高峰时间为:9:00–12:00 和 14:00–18:00 是相对高峰,其他时段是平峰。
了解了 DeepSeek 时段分布后,我们建议使用者调整调度方式:高峰时段的批量任务排到下一个平峰调用,平峰时段正常跑。
使用前提示:本文示例适用于按量计费(Token Plan)用户。Coding Plan 套餐按协议条款,不适用于本文场景。
下面是一个简单的 Python 实现思路,覆盖时段判断、排队、API 调用三部分。
代码
“”"
DeepSeek API 峰谷时段调度器
原理: 高峰排队慢, 自动把任务排到下一个平峰时段
环境: Python 3.8+, 需要 pip install requests
使用前确认 3 件事:
- 时区: datetime.now() 用本地时间, UTC 服务器需要调整
- API URL: 以 deepseek.com 当前文档为准
- API Key: 替换 your-d…-key, 第一次跑建议小请求测试
“”"
import time
from datetime import datetime
import requests
import json
========== 1. 配置区 ==========
DEEPSEEK_API_KEY = “***”
API_URL = “https://api.deepseek.com/v1/chat/completions”
MODEL = “deepseek-chat”
高峰时段定义 (24小时制)
PEAK_HOURS = [(9, 12), (14, 18)]
最多等待小时数
MAX_WAIT_HOURS = 6
请求参数
REQUEST_TIMEOUT = 30
========== 2. 时段判断 ==========
def is_peak_hour():
hour = datetime.now().hour
for start, end in PEAK_HOURS:
if start <= hour < end:
return True
return False
def next_offpeak_time():
now = datetime.now()
hour = now.hour
if hour < 9 or hour >= 18:
return now
if 9 <= hour < 12:
return now.replace(hour=12, minute=0, second=0)
if 14 <= hour < 18:
return now.replace(hour=18, minute=0, second=0)
return now
========== 3. 任务调度 ==========
def schedule_task(prompt, max_wait_hours=MAX_WAIT_HOURS):
if not is_peak_hour():
print(f"[{datetime.now()}] 平峰时段, 立即调用")
return call_api(prompt)
next_time = next_offpeak_time() wait_seconds = (next_time - datetime.now()).total_seconds() wait_hours = wait_seconds / 3600 if wait_hours > max_wait_hours: return { "status": "skipped", "reason": f"等待时间 {wait_hours:.1f} 小时超过上限 {max_wait_hours}" } print(f"[{datetime.now()}] 高峰时段, 排队 {wait_hours:.1f} 小时到 {next_time}") time.sleep(wait_seconds) return call_api(prompt)========== 4. API 调用 ==========
def call_api(prompt):
headers = {
“Authorization”: f"Bearer {DEEPSEEK_API_KEY}",
“Content-Type”: “application/json”
}
payload = {
“model”: MODEL,
“messages”: [{“role”: “user”, “content”: prompt}],
“max_tokens”: 1000
}
try: response = requests.post( API_URL, headers=headers, json=payload, timeout=REQUEST_TIMEOUT ) response.raise_for_status() result = response.json() return { "status": "success", "content": result["choices"][0]["message"]["content"], "timestamp": datetime.now().isoformat() } except Exception as e: return { "status": "error", "error": str(e), "timestamp": datetime.now().isoformat() }========== 5. 使用示例 ==========
ifname== “main”:
task = “把这句话翻译成英文: DeepSeek API 峰谷调度测试”
result = schedule_task(task) print(json.dumps(result, ensure_ascii=False, indent=2))一个参考场景
如果跑批量翻译任务,时段差异可以观察一下:
时段 响应时间 排队情况
平峰 几百毫秒 基本不排队
高峰 几秒 偶发排队
不同任务、并发量、网络环境都会影响,具体以你本地测试为准。
怎么用这种定价
按这种定价方式使用,只做实时中转,不对 Token 收取额外费用,使用者调用,按量计费模式下价格为各模型官方定价 0 加价。结合上面说到的时段调整,能让花在 API 上的钱花在更值得的地方。
几个调整方向
接 Celery 做生产级队列:把 schedule_task 接入 Celery,每天定时跑批
加重试机制:用 tenacity 重试 3 次,处理网络抖动
加费用估算:每次调用前先算 token 数 + 价格
你跑 DeepSeek 高峰遇到过排队吗?排了多久?评论区聊聊。