Token Usage Precision and Rate Limiting Strategies

Token Calculation Method

DeepSeek's billing is based on the actual number of tokens processed. One token is approximately equivalent to 0.75 English words or 0.5 Chinese characters:

from openai import OpenAI

client = OpenAI(
    api_key='your-key',
    base_url='https://api.deepseek.com'
)

# Each request returns usage information
response = client.chat.completions.create(
    model='deepseek-v4-flash',
    messages=[{"role": "user", "content": "Hello"}]
)

print(f"Prompt tokens: {response.usage.prompt_tokens}")      # Input consumption
print(f"Completion tokens: {response.usage.completion_tokens}")  # Output consumption
print(f"Total tokens: {response.usage.total_tokens}")

# In thinking mode, there are also reasoning_tokens
if hasattr(response.usage, 'reasoning_tokens'):
    print(f"Reasoning tokens: {response.usage.reasoning_tokens}")

Real-time Cost Estimation

# V4 Flash prices (RMB, per million tokens)
FLASH_INPUT_CACHE_HIT = 0.02
FLASH_INPUT_CACHE_MISS = 1.00
FLASH_OUTPUT = 2.00

def estimate_cost(prompt_tokens, completion_tokens, cache_hit=True):
    input_price = FLASH_INPUT_CACHE_HIT if cache_hit else FLASH_INPUT_CACHE_MISS
    input_cost = (prompt_tokens / 1_000_000) * input_price
    output_cost = (completion_tokens / 1_000_000) * FLASH_OUTPUT
    return input_cost + output_cost

# Example: 3000 input tokens + 500 output tokens
cost = estimate_cost(3000, 500, cache_hit=True)
print(f"Cost for this request: ¥{cost:.4f}")
# Output: Cost for this request: ¥0.0011

Rate Limits

Exceeding the rate limit returns a 429 status code, requiring backoff retry implementation.

Intelligent Retry Mechanism

import time
import random

def call_with_retry(func, max_retries=3):
    """API retry with exponential backoff"""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if '429' in str(e) or 'rate_limit' in str(e).lower():
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited, waiting {wait:.1f} seconds before retry...")
                time.sleep(wait)
            elif attempt == max_retries - 1:
                raise
            else:
                time.sleep(1)

# Usage
result = call_with_retry(lambda: client.chat.completions.create(
    model='deepseek-v4-flash',
    messages=[{"role": "user", "content": "Hello"}]
))

Concurrent Request Queue

import asyncio
from asyncio import Semaphore

class DeepSeekClient:
    def __init__(self, max_concurrent=10):
        self.semaphore = Semaphore(max_concurrent)
        self.client = OpenAI(
            api_key=os.environ['DEEPSEEK_API_KEY'],
            base_url='https://api.deepseek.com'
        )

    async def chat_async(self, messages, model='deepseek-v4-flash'):
        async with self.semaphore:
            loop = asyncio.get_event_loop()
            return await loop.run_in_executor(
                None,
                lambda: self.client.chat.completions.create(
                    model=model,
                    messages=messages
                )
            )

# Process 100 requests concurrently
client = DeepSeekClient(max_concurrent=10)
async def process_batch(queries):
    tasks = [client.chat_async([{"role": "user", "content": q}]) for q in queries]
    results = await asyncio.gather(*tasks)
    return results

Cost Monitoring Dashboard Ideas

In production environments, it is recommended to record the following metrics:

• Daily total token consumption (by model)
• Cache hit rate (higher means more cost savings)
• Average response latency
• Number of 429 rate limit occurrences
• Cost allocation by user/API Key