背景
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 件事:

1. 时区: datetime.now() 用本地时间, UTC 服务器需要调整
2. API URL: 以 deepseek.com 当前文档为准
3. 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. 使用示例 ==========

if name == “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 高峰遇到过排队吗？排了多久？评论区聊聊。