背景 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))

一个参考场景 如果跑批量翻译任务，时段差异可以观察一下：

时段 响应时间 排队情况 平峰 几百毫秒 基本不排队 高峰 几秒 偶发排队 不同任务、并发量、网络环境都会影响，具体以你本地测试为准。

怎么用这种定价 按这种定价方式使用，openstarry只做实时中转，不对 Token 收取额外费用，使用者调用，按量计费模式下价格为各模型官方定价 0 加价。结合上面说到的时段调整，能让花在 API 上的钱花在更值得的地方。

几个调整方向 接 Celery 做生产级队列：把 schedule_task 接入 Celery，每天定时跑批 加重试机制：用 tenacity 重试 3 次，处理网络抖动 加费用估算：每次调用前先算 token 数 + 价格 你跑 DeepSeek 高峰遇到过排队吗？排了多久？评论区聊聊。