Diagnostic LLM APIs
scheduled jobs pay full price for work the Batch API halves
Nothing here is broken. No request failed, no row is missing, and no alert should have fired. The nightly enrichment job fires forty thousand completions between 02:00 and 02:20, finishes cleanly, and does it again the next night. It has no user waiting on it and no latency requirement of any kind, and it is being billed at the interactive rate because the synchronous endpoint is what the SDK example used. This is not a bug report. It is an invoice roughly twice the size it needs to be.
With an organization admin key, read GET /v1/organization/usage/completions?start_time={now-7d}&bucket_width=1h&limit=168&group_by=batch&group_by=project_id&group_by=model. Each result carries a batch boolean beside input_tokens, output_tokens and num_model_requests.
If every result has batch: false, the Batch API is unused. That on its own is not a finding — interactive traffic belongs on the synchronous endpoint. The finding is batch-shaped traffic inside the synchronous half: a project whose requests are concentrated into a handful of hours a day rather than spread across them, which is a scheduled job wearing interactive pricing.
Then price it. GET /v1/organization/costs?start_time=…&group_by=line_item reports batch and non-batch as distinct line items, and batch is priced at half. The saving is half the synchronous spend of the jobs you can actually move.
The problem in plain words
The cost is invisible because it is a discount not taken rather than a charge incurred. There is no line on the invoice labelled "paid twice as much as necessary", no error, no degraded response and no metric that moves. The only artefact is a total that is larger than a counterfactual nobody computed. That is a category of problem that survives indefinitely, because every review of it concludes that the system is working.
It is also nobody's decision. Nothing chose the synchronous endpoint over the asynchronous one: the SDK's create() is synchronous, every example in the docs is synchronous, and the first prototype was necessarily synchronous because it was a person typing into a terminal. The nightly job was built by copying the prototype. Latency insensitivity is a property of the workload that the code has no way to declare and the API has no way to infer.
Why it happens
Batch is priced at half, on both input and output. That is the entire trade, and it is a real one: you give up latency guarantees and accept a completion window of up to 24 hours, and the same tokens cost half as much. For work with nobody waiting, the thing you gave up has no value.
The API cannot tell a scheduled job from user traffic. Request by request, forty thousand completions from a cron job look exactly like forty thousand completions from people. There is no field that says "this could have waited". The only signal is aggregate shape, which is why this check reads hourly buckets rather than totals.
The shape is the whole detection. Interactive traffic follows human hours: a broad curve with a floor. Scheduled work is a spike — most of the week's requests inside a few percent of the hours. Concentration is measurable from num_model_requests per bucket without knowing anything about what the requests contained.
The usage endpoints need an admin key, and only OpenAI counts requests. Everything under /v1/organization/* rejects a project key outright. And the request count that makes this check possible is an OpenAI field: Anthropic's messages usage report returns token sums per bucket with no request-count member at all, so the same shape analysis on that side has to be done on tokens and is correspondingly blunter.
Not every clustered job can move. A job with a downstream deadline four hours later cannot accept a 24 hour window, and one that feeds a user-visible dashboard by 09:00 might not either. The script reports what is eligible by shape; whether it can actually move is a fact about your schedule that no endpoint knows.
The fix, as a flow
Nothing here failed, so there is no error to trace. The evidence is the shape of the traffic in hourly buckets: a week of requests folded per workload, and the share of them that lands in the busiest few hours. A schedule spikes. An audience does not.
How to fix it
Get an organization admin key, provisioned read-only
/v1/organization/usage/* and /v1/organization/costs both reject project keys. Use an sk-admin- key with read scopes. This script only ever issues GETs, so read-only is all it wants.
Pull a week of hourly buckets, grouped by batch
bucket_width=1h with limit=168 over seven days, grouping by batch, project_id and model. The batch boolean is non-null only because you grouped by it. Follow next_page to the end.
Measure concentration, not just the batch share
For each project and model, take the synchronous num_model_requests per hour and compute what share of the week lands in the busiest ten percent of hours. Above about seventy percent, that is a schedule rather than an audience.
Price it from the cost report, not from a table you typed in
GET /v1/organization/costs?start_time=…&bucket_width=1d&group_by=line_item&group_by=project_id. Batch and non-batch appear as distinct line_item strings, so the synchronous spend is a filter rather than an estimate, and the saving is half of it. Hardcoded per-token prices go stale; the cost report does not.
Move one job, then read the same window again
Upload a JSONL of requests to /v1/files with purpose="batch", create the batch with a 24 hour completion window, and handle the two result files. A week later the same query should show a batch: true population that did not exist before. Then read the reconciliation note, because asynchronous work fails differently.
How to check it worked
Re-run after the first job has moved. The traffic that was batch-shaped should now be reported as already batched.
python3 openai_batch_discount_audit.py --days 7
# already-batched proj_night / gpt-5.6-terra 94% of requests already go through the Batch API
# 6 workload(s), 0 batch shaped
The full code
Two GETs against the organization endpoints, no writes, and an admin key that should be provisioned read-only. Four pure functions carry the note: the accumulator, which has to keep the hourly buckets aligned per workload or the shape measurement is meaningless; the concentration measure, which is the actual detection; the classifier, which keeps “too little traffic to say” and “already batched” as answers rather than folding them into a pass; and the saving, which is deliberately half of the measured spend rather than a per-token price table that would go stale by the time you read this.
"""Report synchronous OpenAI traffic that is shaped like batch work.
Read only. Two GET requests against the organization endpoints and nothing
else. Those endpoints reject project keys, so this needs an organization admin
key (sk-admin-), which can and should be provisioned read-only.
This is a cost note, not a failure note. Nothing found here is broken: the
finding is latency-insensitive work paying interactive prices, and the repair
is a change to how a job submits its requests, printed for you to run.
"""
import argparse
import logging
import math
import os
import sys
import time
import requests
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("openai_batch_discount_audit")
API = "https://api.openai.com/v1"
# The Batch API is priced at half the synchronous rate on both input and output
# tokens, in exchange for a completion window of up to 24 hours.
DISCOUNT = 0.50
def accumulate(buckets):
"""Fold usage buckets into one row per project and model. Pure.
The hourly request counts have to stay aligned across the whole window, so
each row carries a list as long as the bucket list with zeros where that
workload was idle. Compacting out the idle hours would make every workload
look concentrated, which is exactly the thing being measured.
"""
buckets = list(buckets or [])
rows = {}
for index, bucket in enumerate(buckets):
for result in bucket.get("results") or []:
project = str(result.get("project_id") or "unknown")
model = str(result.get("model") or "unknown")
key = "%s / %s" % (project, model)
row = rows.get(key)
if row is None:
row = {"key": key, "project_id": project, "model": model,
"sync_requests": 0, "batch_requests": 0,
"sync_input": 0, "sync_output": 0,
"hourly": [0] * len(buckets)}
rows[key] = row
requests_made = int(result.get("num_model_requests") or 0)
if result.get("batch") is True:
row["batch_requests"] += requests_made
else:
row["sync_requests"] += requests_made
row["sync_input"] += int(result.get("input_tokens") or 0)
row["sync_output"] += int(result.get("output_tokens") or 0)
row["hourly"][index] += requests_made
return rows
def concentration(hourly, top_fraction=0.10):
"""Share of requests inside the busiest slice of the window. Pure.
Returns a float between 0 and 1, or None when there is nothing to measure.
A scheduled job puts most of its week into a handful of hours; an audience
does not, however uneven its day looks.
"""
counts = [int(c or 0) for c in (hourly or [])]
total = sum(counts)
if not counts or total <= 0:
return None
top = max(1, int(math.ceil(len(counts) * top_fraction)))
return sum(sorted(counts, reverse=True)[:top]) / float(total)
def verdict(row, min_requests=1000, threshold=0.70, top_fraction=0.10):
"""Classify one workload's week. Pure. Returns (state, detail).
"interactive" and "already-batched" are answers, not failures to detect
something: synchronous is the correct endpoint for traffic with a person
waiting on it, and this script says so rather than staying silent.
"""
sync = int(row.get("sync_requests") or 0)
batched = int(row.get("batch_requests") or 0)
total = sync + batched
if total < min_requests:
return ("too-little-traffic",
"%d request(s) in the window, which is too few to say anything "
"about the shape" % total)
share = sync / float(total)
if share < 0.20:
return ("already-batched",
"%.0f%% of %d request(s) already go through the Batch API"
% (100 * (1 - share), total))
spike = concentration(row.get("hourly"), top_fraction)
if spike is None:
return ("unmeasurable",
"%d synchronous request(s) and no per bucket counts to spread "
"them over, so the shape cannot be measured" % sync)
if spike >= threshold:
return ("batch-shaped",
"%.0f%% of %d synchronous request(s) land in the busiest %.0f%% "
"of hours. That is a schedule, not an audience, and it is paying "
"interactive prices." % (spike * 100, sync, top_fraction * 100))
return ("interactive",
"%d synchronous request(s), %.0f%% of them in the busiest %.0f%% of "
"hours. Spread out like traffic with someone waiting on it, so the "
"synchronous endpoint is the right one."
% (sync, spike * 100, top_fraction * 100))
def sync_cost(buckets, project_id=None):
"""Non-batch dollars in the cost report, optionally for one project. Pure.
Batch and non-batch appear as distinct line_item strings, so the split is a
substring test and nothing more clever than that. Reading the money from the
cost report rather than from a per-token price table is deliberate: the
table goes stale, the report does not.
"""
total = 0.0
for bucket in buckets or []:
for result in bucket.get("results") or []:
if project_id and str(result.get("project_id") or "") != project_id:
continue
if "batch" in str(result.get("line_item") or "").lower():
continue
try:
total += float((result.get("amount") or {}).get("value") or 0.0)
except (TypeError, ValueError):
continue
return round(total, 2)
def saving(sync_cost_usd, discount=DISCOUNT):
"""What the same spend would have been worth at batch prices. Pure.
Not a promise: it is the value of the discount on money already spent, and
it says nothing about whether the job can accept a 24 hour window. That
part is a fact about your schedule and no endpoint knows it.
"""
if sync_cost_usd is None:
return None
try:
return round(max(0.0, float(sync_cost_usd)) * discount, 2)
except (TypeError, ValueError):
return None
def get(session, path, params):
r = session.get(API + path, params=params, timeout=90)
if r.status_code in (401, 403):
raise SystemExit("%d from OpenAI: /v1/organization/* needs an "
"organization admin key (sk-admin-), not a project key"
% r.status_code)
r.raise_for_status()
return r.json()
def pages(session, path, params, max_pages=40):
"""Walk a usage or cost report, which paginates on an opaque page cursor."""
params = dict(params)
for _ in range(max_pages):
page = get(session, path, params)
for bucket in page.get("data") or []:
yield bucket
if not page.get("has_more") or not page.get("next_page"):
return
params = dict(params)
params["page"] = page["next_page"]
def main():
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--days", type=int, default=7,
help="days of hourly buckets to read (default 7)")
ap.add_argument("--min-requests", type=int, default=1000,
help="ignore workloads below this many requests (default 1000)")
ap.add_argument("--threshold", type=float, default=0.70,
help="share of requests in the busiest hours above which a "
"workload is called batch shaped (default 0.70)")
ap.add_argument("--top-fraction", type=float, default=0.10,
help="the busiest share of buckets to measure against "
"(default 0.10)")
ap.add_argument("--show-all", action="store_true",
help="also print workloads that are correctly synchronous")
args = ap.parse_args()
key = os.environ.get("OPENAI_ADMIN_KEY") or os.environ.get("OPENAI_API_KEY")
if not key:
log.error("set OPENAI_ADMIN_KEY (an organization admin key, read-only "
"scopes are enough)")
return 2
session = requests.Session()
session.headers.update({"Authorization": "Bearer " + key})
start = int(time.time()) - args.days * 86400
usage = list(pages(session, "/organization/usage/completions", {
"start_time": start,
"bucket_width": "1h",
"limit": 168,
"group_by": ["batch", "project_id", "model"],
}))
costs = list(pages(session, "/organization/costs", {
"start_time": start,
"bucket_width": "1d",
"limit": 31,
"group_by": ["line_item", "project_id"],
}))
rows = accumulate(usage)
if not rows:
log.info("no completions usage in the last %d day(s) for this "
"organization", args.days)
return 0
found = 0
for key_name in sorted(rows):
row = rows[key_name]
state, detail = verdict(row, args.min_requests, args.threshold,
args.top_fraction)
line = "%-17s %s %s" % (state, key_name, detail)
if state == "batch-shaped":
found += 1
log.warning(line)
spend = sync_cost(costs, row["project_id"])
worth = saving(spend)
log.warning(" cost: $%.2f of synchronous spend on project %s over "
"%d day(s); about $%.2f of that is the batch discount "
"you are not taking", spend, row["project_id"],
args.days, worth)
log.warning(" repair: upload the requests as a .jsonl to /v1/files "
"with purpose=batch, create a batch with a 24h "
"completion window, and read both result files. The "
"trade is half price for no latency guarantee.")
elif state in ("interactive", "already-batched", "too-little-traffic"):
if args.show_all:
log.info(line)
else:
log.warning(line)
log.info("%d workload(s), %d batch shaped", len(rows), found)
return 1 if found else 0
if __name__ == "__main__":
sys.exit(main())
/**
* Report synchronous OpenAI traffic that is shaped like batch work.
*
* Read only. Two GET requests against the organization endpoints and nothing
* else. Those endpoints reject project keys, so this needs an organization
* admin key (sk-admin-), which can and should be provisioned read-only.
*
* This is a cost note, not a failure note. Nothing found here is broken.
*/
const API = 'https://api.openai.com/v1';
// The Batch API is priced at half the synchronous rate on both input and output
// tokens, in exchange for a completion window of up to 24 hours.
const DISCOUNT = 0.50;
/**
* Fold usage buckets into one row per project and model. Pure. The hourly
* request counts stay aligned across the whole window, with zeros where a
* workload was idle: compacting the idle hours out would make every workload
* look concentrated, which is the thing being measured.
*/
export function accumulate(buckets) {
const list = buckets ?? [];
const rows = new Map();
list.forEach((bucket, index) => {
for (const result of bucket.results ?? []) {
const project = String(result.project_id ?? 'unknown');
const model = String(result.model ?? 'unknown');
const key = `${project} / ${model}`;
let row = rows.get(key);
if (!row) {
row = {
key,
project_id: project,
model,
sync_requests: 0,
batch_requests: 0,
sync_input: 0,
sync_output: 0,
hourly: new Array(list.length).fill(0),
};
rows.set(key, row);
}
const made = Number(result.num_model_requests ?? 0) || 0;
if (result.batch === true) {
row.batch_requests += made;
} else {
row.sync_requests += made;
row.sync_input += Number(result.input_tokens ?? 0) || 0;
row.sync_output += Number(result.output_tokens ?? 0) || 0;
row.hourly[index] += made;
}
}
});
return rows;
}
/**
* Share of requests inside the busiest slice of the window. Pure. Returns a
* number between 0 and 1, or null when there is nothing to measure.
*/
export function concentration(hourly, topFraction = 0.10) {
const counts = (hourly ?? []).map((c) => Number(c) || 0);
const total = counts.reduce((a, b) => a + b, 0);
if (counts.length === 0 || total <= 0) return null;
const top = Math.max(1, Math.ceil(counts.length * topFraction));
const busiest = [...counts].sort((a, b) => b - a).slice(0, top);
return busiest.reduce((a, b) => a + b, 0) / total;
}
/**
* Classify one workload's week. Pure. Returns [state, detail]. "interactive"
* and "already-batched" are answers rather than failures to detect something:
* synchronous is correct for traffic with a person waiting on it.
*/
export function verdict(row, minRequests = 1000, threshold = 0.70,
topFraction = 0.10) {
const sync = Number(row.sync_requests ?? 0) || 0;
const batched = Number(row.batch_requests ?? 0) || 0;
const total = sync + batched;
if (total < minRequests) {
return ['too-little-traffic',
`${total} request(s) in the window, which is too few to say anything ` +
'about the shape'];
}
const share = sync / total;
if (share < 0.20) {
return ['already-batched',
`${Math.round(100 * (1 - share))}% of ${total} request(s) already go ` +
'through the Batch API'];
}
const spike = concentration(row.hourly, topFraction);
if (spike === null) {
return ['unmeasurable',
`${sync} synchronous request(s) and no per bucket counts to spread them ` +
'over, so the shape cannot be measured'];
}
const pct = Math.round(spike * 100);
const slice = Math.round(topFraction * 100);
if (spike >= threshold) {
return ['batch-shaped',
`${pct}% of ${sync} synchronous request(s) land in the busiest ` +
`${slice}% of hours. That is a schedule, not an audience, and it is ` +
'paying interactive prices.'];
}
return ['interactive',
`${sync} synchronous request(s), ${pct}% of them in the busiest ${slice}% ` +
'of hours. Spread out like traffic with someone waiting on it, so the ' +
'synchronous endpoint is the right one.'];
}
/**
* Non-batch dollars in the cost report, optionally for one project. Pure.
* Batch and non-batch appear as distinct line_item strings, so the split is a
* substring test and nothing more clever than that.
*/
export function syncCost(buckets, projectId = null) {
let total = 0;
for (const bucket of buckets ?? []) {
for (const result of bucket.results ?? []) {
if (projectId && String(result.project_id ?? '') !== projectId) continue;
if (String(result.line_item ?? '').toLowerCase().includes('batch')) continue;
total += Number(result.amount?.value ?? 0) || 0;
}
}
return Math.round(total * 100) / 100;
}
/**
* What the same spend would have been worth at batch prices. Pure. Not a
* promise: it says nothing about whether the job can accept a 24 hour window.
*/
export function saving(syncCostUsd, discount = DISCOUNT) {
if (syncCostUsd === null || syncCostUsd === undefined) return null;
const value = Number(syncCostUsd);
if (!Number.isFinite(value)) return null;
return Math.round(Math.max(0, value) * discount * 100) / 100;
}
async function get(key, path, params) {
const url = new URL(API + path);
for (const [k, v] of Object.entries(params)) {
if (Array.isArray(v)) v.forEach((one) => url.searchParams.append(k, String(one)));
else if (v !== undefined && v !== null) url.searchParams.set(k, String(v));
}
const res = await fetch(url, { headers: { Authorization: `Bearer ${key}` } });
if (res.status === 401 || res.status === 403) {
throw new Error(`${res.status} from OpenAI: /v1/organization/* needs an ` +
'organization admin key (sk-admin-), not a project key');
}
if (!res.ok) throw new Error(`${res.status} from ${path}`);
return res.json();
}
async function pages(key, path, params, maxPages = 40) {
const out = [];
let query = { ...params };
for (let i = 0; i < maxPages; i += 1) {
const page = await get(key, path, query);
out.push(...(page.data ?? []));
if (!page.has_more || !page.next_page) break;
query = { ...params, page: page.next_page };
}
return out;
}
async function main() {
const key = process.env.OPENAI_ADMIN_KEY ?? process.env.OPENAI_API_KEY;
if (!key) {
console.error('set OPENAI_ADMIN_KEY (an organization admin key, read-only ' +
'scopes are enough)');
process.exitCode = 2;
return;
}
const days = Number(process.env.DAYS ?? 7);
const minRequests = Number(process.env.MIN_REQUESTS ?? 1000);
const threshold = Number(process.env.THRESHOLD ?? 0.70);
const topFraction = Number(process.env.TOP_FRACTION ?? 0.10);
const showAll = process.argv.includes('--show-all');
const start = Math.floor(Date.now() / 1000) - days * 86400;
const usage = await pages(key, '/organization/usage/completions', {
start_time: start,
bucket_width: '1h',
limit: 168,
group_by: ['batch', 'project_id', 'model'],
});
const costs = await pages(key, '/organization/costs', {
start_time: start,
bucket_width: '1d',
limit: 31,
group_by: ['line_item', 'project_id'],
});
const rows = accumulate(usage);
if (rows.size === 0) {
console.log(`no completions usage in the last ${days} day(s) for this ` +
'organization');
return;
}
let found = 0;
for (const name of [...rows.keys()].sort()) {
const row = rows.get(name);
const [state, detail] = verdict(row, minRequests, threshold, topFraction);
const line = `${state.padEnd(17)} ${name} ${detail}`;
if (state === 'batch-shaped') {
found += 1;
console.warn(line);
const spend = syncCost(costs, row.project_id);
console.warn(` cost: $${spend.toFixed(2)} of synchronous spend on ` +
`project ${row.project_id} over ${days} day(s); about ` +
`$${saving(spend).toFixed(2)} of that is the batch discount you are ` +
'not taking');
console.warn(' repair: upload the requests as a .jsonl to /v1/files ' +
'with purpose=batch, create a batch with a 24h completion window, and ' +
'read both result files. The trade is half price for no latency ' +
'guarantee.');
} else if (['interactive', 'already-batched', 'too-little-traffic'].includes(state)) {
if (showAll) console.log(line);
} else {
console.warn(line);
}
}
console.log(`${rows.size} workload(s), ${found} batch shaped`);
process.exitCode = found ? 1 : 0;
}
// Only run when invoked directly, so importing this module from the test file
// does not fire main() and fail on the missing key.
if (import.meta.url === `file://${process.argv[1]}`) {
main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}
Add a test
No clock in these tests, because the finding is a shape rather than a moment. What they pin down instead is that idle hours are counted: a workload with one busy hour and nineteen empty ones is only concentrated if the empty ones are in the denominator, and an accumulator that drops them turns every workload into a finding. The rest hold apart the three answers that are not findings — already batched, genuinely interactive, and too small to judge.
from openai_batch_discount_audit import (accumulate, concentration, saving,
sync_cost, verdict)
def bucket(*results):
return {"start_time": 0, "end_time": 3600, "results": list(results)}
def result(project="proj_a", model="gpt-5.6-terra", batch=False, made=0,
input_tokens=0, output_tokens=0):
return {"project_id": project, "model": model, "batch": batch,
"num_model_requests": made, "input_tokens": input_tokens,
"output_tokens": output_tokens}
def test_idle_hours_stay_in_the_denominator():
# Four buckets, one of them busy. If the empty hours were dropped the
# workload would look perfectly flat instead of perfectly spiky.
buckets = [bucket(result(made=0)), bucket(result(made=4000)),
bucket(), bucket(result(made=0))]
rows = accumulate(buckets)
row = rows["proj_a / gpt-5.6-terra"]
assert row["hourly"] == [0, 4000, 0, 0]
assert row["sync_requests"] == 4000
def test_batch_and_synchronous_traffic_are_kept_apart():
rows = accumulate([bucket(result(made=100, input_tokens=50, batch=False),
result(made=900, batch=True))])
row = rows["proj_a / gpt-5.6-terra"]
assert row["sync_requests"] == 100
assert row["batch_requests"] == 900
assert row["sync_input"] == 50
assert row["hourly"] == [100]
def test_concentration_separates_a_schedule_from_an_audience():
spiky = [0] * 18 + [4000, 1000]
assert concentration(spiky, 0.10) == 1.0
assert concentration([250] * 20, 0.10) == 0.1
assert concentration([], 0.10) is None
assert concentration([0, 0, 0], 0.10) is None
def test_a_nightly_job_on_the_synchronous_endpoint_is_the_finding():
row = {"sync_requests": 5000, "batch_requests": 0,
"hourly": [0] * 18 + [4000, 1000]}
state, detail = verdict(row)
assert state == "batch-shaped"
assert "100% of 5000 synchronous request(s)" in detail
assert "paying interactive prices" in detail
def test_spread_out_traffic_is_correctly_synchronous():
row = {"sync_requests": 5000, "batch_requests": 0, "hourly": [250] * 20}
state, detail = verdict(row)
assert state == "interactive"
assert "right one" in detail
def test_the_three_answers_that_are_not_findings():
assert verdict({"sync_requests": 10, "batch_requests": 0,
"hourly": [10]})[0] == "too-little-traffic"
assert verdict({"sync_requests": 100, "batch_requests": 9900,
"hourly": [100]})[0] == "already-batched"
assert verdict({"sync_requests": 5000, "batch_requests": 0,
"hourly": []})[0] == "unmeasurable"
def test_the_money_comes_from_the_cost_report_not_a_price_table():
costs = [{"results": [
{"project_id": "proj_a", "line_item": "gpt-5.6-terra, input",
"amount": {"value": 300.0, "currency": "usd"}},
{"project_id": "proj_a", "line_item": "gpt-5.6-terra, batch input",
"amount": {"value": 40.0, "currency": "usd"}},
{"project_id": "proj_b", "line_item": "gpt-5.6-terra, input",
"amount": {"value": 99.0, "currency": "usd"}},
]}]
assert sync_cost(costs, "proj_a") == 300.0
assert sync_cost(costs) == 399.0
assert saving(300.0) == 150.0
assert saving(0) == 0.0
assert saving(None) is None
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { accumulate, concentration, saving, syncCost, verdict }
from './openai-batch-discount-audit.mjs';
function bucket(...results) {
return { start_time: 0, end_time: 3600, results };
}
function result({ project = 'proj_a', model = 'gpt-5.6-terra', batch = false,
made = 0, inputTokens = 0, outputTokens = 0 } = {}) {
return {
project_id: project,
model,
batch,
num_model_requests: made,
input_tokens: inputTokens,
output_tokens: outputTokens,
};
}
test('idle hours stay in the denominator', () => {
const buckets = [bucket(result({ made: 0 })), bucket(result({ made: 4000 })),
bucket(), bucket(result({ made: 0 }))];
const row = accumulate(buckets).get('proj_a / gpt-5.6-terra');
assert.deepEqual(row.hourly, [0, 4000, 0, 0]);
assert.equal(row.sync_requests, 4000);
});
test('batch and synchronous traffic are kept apart', () => {
const rows = accumulate([bucket(
result({ made: 100, inputTokens: 50, batch: false }),
result({ made: 900, batch: true }))]);
const row = rows.get('proj_a / gpt-5.6-terra');
assert.equal(row.sync_requests, 100);
assert.equal(row.batch_requests, 900);
assert.equal(row.sync_input, 50);
assert.deepEqual(row.hourly, [100]);
});
test('concentration separates a schedule from an audience', () => {
const spiky = [...new Array(18).fill(0), 4000, 1000];
assert.equal(concentration(spiky, 0.10), 1.0);
assert.equal(concentration(new Array(20).fill(250), 0.10), 0.1);
assert.equal(concentration([], 0.10), null);
assert.equal(concentration([0, 0, 0], 0.10), null);
});
test('a nightly job on the synchronous endpoint is the finding', () => {
const row = { sync_requests: 5000, batch_requests: 0,
hourly: [...new Array(18).fill(0), 4000, 1000] };
const [state, detail] = verdict(row);
assert.equal(state, 'batch-shaped');
assert.match(detail, /100% of 5000 synchronous request\(s\)/);
assert.match(detail, /paying interactive prices/);
});
test('spread out traffic is correctly synchronous', () => {
const row = { sync_requests: 5000, batch_requests: 0,
hourly: new Array(20).fill(250) };
const [state, detail] = verdict(row);
assert.equal(state, 'interactive');
assert.match(detail, /right one/);
});
test('the three answers that are not findings', () => {
assert.equal(verdict({ sync_requests: 10, batch_requests: 0, hourly: [10] })[0],
'too-little-traffic');
assert.equal(verdict({ sync_requests: 100, batch_requests: 9900, hourly: [100] })[0],
'already-batched');
assert.equal(verdict({ sync_requests: 5000, batch_requests: 0, hourly: [] })[0],
'unmeasurable');
});
test('the money comes from the cost report not a price table', () => {
const costs = [{ results: [
{ project_id: 'proj_a', line_item: 'gpt-5.6-terra, input',
amount: { value: 300.0, currency: 'usd' } },
{ project_id: 'proj_a', line_item: 'gpt-5.6-terra, batch input',
amount: { value: 40.0, currency: 'usd' } },
{ project_id: 'proj_b', line_item: 'gpt-5.6-terra, input',
amount: { value: 99.0, currency: 'usd' } },
] }];
assert.equal(syncCost(costs, 'proj_a'), 300.0);
assert.equal(syncCost(costs), 399.0);
assert.equal(saving(300.0), 150.0);
assert.equal(saving(0), 0);
assert.equal(saving(null), null);
});
FAQ
Is this a bug, exactly?
No, and the note is written that way on purpose. Nothing failed, no row is missing and no alert should have fired. It is a cost finding: work with nobody waiting on it is being billed at the price of work with somebody waiting on it. The other three notes in this cluster are failures; this one is an invoice.
How much cheaper is the Batch API?
Half, on both input and output tokens, in exchange for a completion window of up to 24 hours and no latency guarantee. Prompt caching and batch are separate discounts on different axes, so a job can take both: cache the stable prefix and submit the requests as a batch.
How do I know a clustered job could actually wait?
You do not, from the API. Concentration proves the traffic is scheduled rather than interactive, which is a necessary condition and not a sufficient one. A backfill can usually wait a day; a job that feeds a dashboard by 09:00 might have four hours, not twenty-four. The script reports what is eligible by shape and leaves the schedule question to you.
Why hourly buckets rather than the daily ones?
Because the whole detection is the shape within a day. Daily buckets flatten a twenty-minute spike and a twenty-four-hour plateau into the same number. With bucket_width=1h and limit=168 you get a week of hours in one call, which is enough resolution to see a cron job and cheap enough to run weekly.
Does the same check work on Anthropic?
Only in a blunter form. Claude's Message Batches API carries the same 50% discount, but the messages usage report returns token sums per bucket with no request-count field at all, so you cannot measure request concentration on that side — you would have to infer the shape from input tokens per hour, which mixes traffic volume with prompt size and is a weaker signal.
Related field notes
- A batch that reads completed with failed rows inside it
- A batch the 24 hour window closed on
- Output tokens, not input, are what the bill is made of
Sources
Every figure in this note is traced to one of these. Prices are list rates and change — check them for your own region before acting.
- Usage: completions — OpenAI API reference
- Costs — OpenAI API reference
- Batch — OpenAI API reference
- Batch API guide — OpenAI developer docs
If your setup is misbehaving in a way this note does not cover, message me on LinkedIn with what you are seeing.