DEV Community

Dakota Huang
Dakota Huang

Posted on

Quota Forensics: A Token Ledger for Free Model Endpoints

Your free model endpoint has a quota. Nobody tells you when it runs out. One request works. The next returns 429. No warning. No countdown.

The data exists. It hides in the response. The usage object reports token counts. Rate-limit headers report remaining capacity. Most integrations throw both away.

This tutorial builds a token ledger. It meters every call. It projects when the quota dies. Four stages. One hour. Each stage ends with a verification step.

Disclosure: This article was prepared as part of MonkeyCode's product outreach. I used MonkeyCode's free model access and free server option for testing. The pipeline works with any OpenAI-compatible endpoint.

Stage 1: Capture the usage object

Every OpenAI-compatible response carries a usage object.

{
  "usage": {
    "prompt_tokens": 42,
    "completion_tokens": 87,
    "total_tokens": 129
  }
}
Enter fullscreen mode Exit fullscreen mode

Some gateways add headers too. X-RateLimit-Remaining and X-RateLimit-Reset are common. Not every endpoint sends them. Parse both. Use what exists.

The wrapper below returns text and metadata together. Never discard the metadata.

import json
import time
import urllib.request

def call_and_measure(base_url, api_key, model, messages):
    body = json.dumps({
        "model": model,
        "messages": messages,
        "temperature": 0
    }).encode()
    req = urllib.request.Request(
        base_url + "/chat/completions",
        body,
        {"Content-Type": "application/json",
         "Authorization": "Bearer " + api_key}
    )
    start = time.monotonic()
    with urllib.request.urlopen(req, timeout=60) as resp:
        data = json.load(resp)
        headers = dict(resp.headers)
    elapsed_ms = (time.monotonic() - start) * 1000
    usage = data.get("usage", {})
    return {
        "text": data["choices"][0]["message"]["content"],
        "usage": usage,
        "headers": headers,
        "latency_ms": round(elapsed_ms, 1),
        "ts": time.time()
    }
Enter fullscreen mode Exit fullscreen mode

Verify Stage 1: Run one call. Print the returned dict. Confirm usage.total_tokens is present. Confirm the headers dict exists, even if empty.

Stage 2: Persist to SQLite

A dict in memory is not a ledger. A ledger survives restarts. SQLite ships with Python. Zero dependencies.

Create one table. Keep it narrow.

CREATE TABLE IF NOT EXISTS calls (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    ts REAL NOT NULL,
    model TEXT NOT NULL,
    prompt_tokens INTEGER NOT NULL,
    completion_tokens INTEGER NOT NULL,
    total_tokens INTEGER NOT NULL,
    latency_ms REAL NOT NULL,
    rate_limit_remaining TEXT,
    rate_limit_reset TEXT
);
Enter fullscreen mode Exit fullscreen mode

Insert one row per call. Use a transaction. Free servers die mid-write. Transactions keep the ledger intact.

Header casing varies between gateways. Normalize keys before reading them.

import sqlite3

def header_value(headers, name):
    lower = name.lower()
    for k, v in headers.items():
        if k.lower() == lower:
            return v
    return None

def insert_call(db_path, row):
    with sqlite3.connect(db_path) as conn:
        conn.execute(
            """INSERT INTO calls
               (ts, model, prompt_tokens, completion_tokens,
                total_tokens, latency_ms, rate_limit_remaining,
                rate_limit_reset)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
            (row["ts"], row["model"],
             row["usage"].get("prompt_tokens", 0),
             row["usage"].get("completion_tokens", 0),
             row["usage"].get("total_tokens", 0),
             row["latency_ms"],
             header_value(row["headers"], "X-RateLimit-Remaining"),
             header_value(row["headers"], "X-RateLimit-Reset"))
        )
Enter fullscreen mode Exit fullscreen mode

Verify Stage 2: Make three calls. Query the table. Confirm three rows. Confirm the token columns match the printed usage objects.

Stage 3: Project the wall

A ledger without a forecast is a diary. The forecast answers one question: when does the quota die?

The projection is linear. Take the last hour of calls. Compute tokens per hour. Divide the remaining quota by that rate.

Set your quota in a config file. The script cannot guess it. Different free tiers have different limits.

import json
import sqlite3
import time

QUOTA = json.load(open("quota.json"))
# {"monthly_tokens": 1000000, "reset_day": 1}

def project_exhaustion(db_path):
    with sqlite3.connect(db_path) as conn:
        row = conn.execute(
            """SELECT MIN(ts), MAX(ts),
                      SUM(total_tokens)
               FROM calls
               WHERE ts > ?""",
            (time.time() - 3600,)
        ).fetchone()
    start_ts, end_ts, tokens = row
    if not tokens or end_ts == start_ts:
        return None
    hours = (end_ts - start_ts) / 3600
    tokens_per_hour = tokens / hours
    remaining = QUOTA["monthly_tokens"]
    hours_left = remaining / tokens_per_hour
    return time.time() + hours_left * 3600
Enter fullscreen mode Exit fullscreen mode

Linear projection is not the only option.

Method Good for Weakness
Last-hour rate Steady traffic Misses batch spikes
Seven-day average Weekly cycles Slow to react
Manual review Small teams Not automated

Start with the last-hour rate. Upgrade when the shape of your traffic proves otherwise.

Verify Stage 3: Run the projection. Confirm the output moves after a burst of calls. Confirm it recovers after a quiet hour.

Stage 4: Alert before the wall

Projections drift. Alerts catch the drift. Check the projection hourly. Alert when the wall moves closer than 24 hours.

Write a marker file. It is the simplest alert that survives a free server reboot.

def check_wall(db_path, alert_path, threshold_hours=24):
    eta = project_exhaustion(db_path)
    if eta is None:
        return
    hours_left = (eta - time.time()) / 3600
    if hours_left < threshold_hours:
        with open(alert_path, "w") as f:
            f.write(json.dumps({"eta": eta, "hours_left": hours_left}))
    else:
        import os
        if os.path.exists(alert_path):
            os.remove(alert_path)
Enter fullscreen mode Exit fullscreen mode

Wire it to cron. Every thirty minutes is enough. Free servers kill long-running processes. Cron restarts the check.

*/30 * * * * cd /opt/token-ledger && python check.py >> check.log 2>&1
Enter fullscreen mode Exit fullscreen mode

Verify Stage 4: Temporarily set threshold_hours to 9999. Confirm the marker file appears. Revert. Confirm it clears.

Limitations

The usage object is only as honest as the provider. Some endpoints omit it. Some report only prompt tokens. The ledger records zeros. The projection degrades into a guess.

Rate-limit headers are advisory. Their format changes between gateways. Treat them as hints, not contracts.

Linear projection assumes steady consumption. A batch job on Monday breaks the model. The alert fires late. Review the ledger weekly. Look at the shape, not just the number.

Who should not use this

Teams with irregular traffic should not trust the linear forecast. Add a weighted average or a manual review step first.

Teams that need per-feature cost attribution need more than one table. Add a feature column. Group by it. The ledger is a foundation, not a full cost system.

The takeaway

The free tier is a black box. The response headers are the key. Meter every call. Project the wall. Alert before it hits.

The pipeline is about forty lines. It runs on a free server. It turns a surprise 429 into a scheduled event.

MonkeyCode's free model access and free server option gave me the infrastructure for this. The ledger is the part you build. Build it before the wall finds you.

Top comments (0)