API reference

Everything below is exported from the package root: offpeak.run, offpeak.Job, and so on.

Running work

offpeak.run(jobs, deadline, *, venues=None, fallback='sync', poll_interval=None, risk_buffer=None)

Run jobs against deadline on the cheapest supporting venue.

Submits each job to its venue's batch tier, polls until everything lands, and — if the batch has not completed by the time the remaining window shrinks to risk_buffer seconds — cancels and re-runs the stragglers synchronously at list price so the deadline is met (fallback="sync", the default; fallback="none" reports them failed instead).

Returns one :class:Result per job, in input order, each with a :class:Receipt.

Provider failures never escape: if a venue raises while submitting, polling or running the sync fallback, the affected jobs are rescued through the fallback where the deadline still allows it and otherwise come back as failed :class:Result objects carrying the provider's message. Exceptions out of run() are reserved for programming errors — a bad deadline, or a model no configured venue supports.

Source code in src/offpeak/client.py

def run(
    jobs: Job | list[Job],
    deadline: object,
    *,
    venues: list[Venue] | None = None,
    fallback: str = "sync",
    poll_interval: float | None = None,
    risk_buffer: float | None = None,
) -> list[Result]:
    """Run *jobs* against *deadline* on the cheapest supporting venue.

    Submits each job to its venue's batch tier, polls until everything lands,
    and — if the batch has not completed by the time the remaining window
    shrinks to ``risk_buffer`` seconds — cancels and re-runs the stragglers
    synchronously at list price so the deadline is met (``fallback="sync"``,
    the default; ``fallback="none"`` reports them failed instead).

    Returns one :class:`Result` per job, in input order, each with a
    :class:`Receipt`.

    Provider failures never escape: if a venue raises while submitting, polling
    or running the sync fallback, the affected jobs are rescued through the
    fallback where the deadline still allows it and otherwise come back as
    failed :class:`Result` objects carrying the provider's message. Exceptions
    out of ``run()`` are reserved for programming errors — a bad deadline, or a
    model no configured venue supports.
    """
    job_list = [jobs] if isinstance(jobs, Job) else list(jobs)
    if not job_list:
        return []
    resolved = parse_deadline(deadline)
    window = seconds_until(resolved)
    if risk_buffer is None:
        risk_buffer = max(60.0, min(600.0, 0.15 * window))
    venue_list = venues if venues is not None else default_venues()

    groups: dict[str, tuple[Venue, list[Job]]] = {}
    for j in job_list:
        venue = _pick_venue(j.model, venue_list)
        groups.setdefault(venue.name, (venue, []))[1].append(j)

    submitted_at = datetime.now().astimezone()
    pending: dict[str, str] = {}  # venue name -> batch handle
    venue_errors: dict[str, str] = {}  # venue name -> why its batch path died
    for name, (venue, group_jobs) in groups.items():
        try:
            pending[name] = venue.submit(group_jobs)
        except Exception as exc:  # noqa: BLE001 — the provider failed, not us
            venue_errors[name] = f"submit failed: {exc}"
            continue
        for j in group_jobs:
            j.status = Status.SUBMITTED

    collected: dict[str, Result] = {}
    fell_back: set[str] = set()

    while pending:
        for name in list(pending):
            venue = groups[name][0]
            try:
                state = venue.status(pending[name])
                if state.status == "completed":
                    collected.update(venue.collect(pending[name]))
                    del pending[name]
                elif state.status in ("failed", "cancelled"):
                    venue_errors[name] = f"batch {state.status}"
                    del pending[name]  # jobs surface below as fallback or errors
            except Exception as exc:  # noqa: BLE001 — the provider failed, not us
                venue_errors[name] = f"batch polling failed: {exc}"
                del pending[name]

        remaining = seconds_until(resolved)
        if not pending and not _missing(groups, collected):
            break
        if remaining <= risk_buffer or not pending:
            break
        time.sleep(
            poll_interval
            if poll_interval is not None
            else min(30.0, max(2.0, remaining / 50.0))
        )

    # Settle stragglers outside the poll loop. A venue whose submit failed never
    # got a handle, so it never entered `pending` — running the fallback inside
    # the loop would skip exactly the jobs that most need rescuing.
    for name in list(pending):
        _cancel(groups[name][0], pending.pop(name))
    stragglers = _missing(groups, collected)
    if stragglers and fallback == "sync" and seconds_until(resolved) > 0:
        for j in stragglers:
            name = _venue_of(j, groups)
            result = _run_sync(groups[name][0], j)
            if result.ok:
                fell_back.add(j.id)
            elif name in venue_errors:
                result.error = f"{venue_errors[name]}; sync fallback failed: {result.error}"
            collected[j.id] = result

    completed_at = datetime.now().astimezone()
    results: list[Result] = []
    for j in job_list:
        result = collected.get(j.id)
        if result is None:
            reason = venue_errors.get(
                _venue_of(j, groups), "not returned by venue before the deadline"
            )
            result = Result(job=j, error=reason)
            j.status = Status.FAILED
        else:
            result.job = j
            j.status = (
                Status.FELL_BACK
                if j.id in fell_back
                else (Status.SUCCEEDED if result.ok else Status.FAILED)
            )
        input_tokens, output_tokens = _usage_tokens(result.raw)
        result.receipt = Receipt(
            venue=groups[_venue_of(j, groups)][0].name,
            model=j.model,
            deadline=resolved,
            submitted_at=submitted_at,
            completed_at=completed_at if result.error is None else None,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            fell_back=j.id in fell_back,
        )
        results.append(result)
    return results

offpeak.quote

The free quote — what a deadline is worth, before you spend anything.

quote() prices a job list against the bundled price sheet and returns what each venue's batch tier would save versus running the same tokens synchronously at list. It makes no API calls: no submission, no token-counting round trip, no key required. It is arithmetic against published numbers, which is the same thing a receipt is — just before the trade instead of after.

Token counts come from the job where the job knows them and are estimated where it does not. Every quote says which, per figure, in :attr:Quote.basis: a number you cannot trace back to its source is not a quote.

Output size is the one figure a pre-trade quote cannot know. Left alone, an unknown output is priced at zero and the whole quote is marked a FLOOR — understated on purpose, and saying so. A caller who does know roughly what the model will write can say so and get a usable number instead, per job with metadata={"expected_output_tokens": n} or across the run with quote(..., assumed_output_ratio=r). Those quotes are marked EST. The assumption is always the caller's, never the library's: nothing here invents an output size on your behalf.

VenueQuote dataclass

What one venue's batch tier is worth for the jobs routed to it.

Source code in src/offpeak/quote.py

@dataclass
class VenueQuote:
    """What one venue's batch tier is worth for the jobs routed to it."""

    venue: str
    jobs: int = 0
    input_tokens: int = 0
    output_tokens: int = 0
    list_usd: float = 0.0
    batch_usd: float = 0.0
    unpriced: int = 0
    unknown_output: int = 0
    assumed_output: int = 0

    @property
    def spread_usd(self) -> float:
        return self.list_usd - self.batch_usd

    @property
    def spread_pct(self) -> float:
        return 0.0 if not self.list_usd else 100.0 * self.spread_usd / self.list_usd

Quote dataclass

A pre-trade quote. No API calls were made to produce this.

Source code in src/offpeak/quote.py

@dataclass
class Quote:
    """A pre-trade quote. No API calls were made to produce this."""

    deadline: datetime
    window_seconds: float
    by_venue: dict[str, VenueQuote] = field(default_factory=dict)
    basis: dict[str, str] = field(default_factory=dict)

    @property
    def jobs(self) -> int:
        return sum(v.jobs for v in self.by_venue.values())

    @property
    def input_tokens(self) -> int:
        return sum(v.input_tokens for v in self.by_venue.values())

    @property
    def output_tokens(self) -> int:
        return sum(v.output_tokens for v in self.by_venue.values())

    @property
    def list_usd(self) -> float:
        return sum(v.list_usd for v in self.by_venue.values())

    @property
    def batch_usd(self) -> float:
        return sum(v.batch_usd for v in self.by_venue.values())

    @property
    def spread_usd(self) -> float:
        return self.list_usd - self.batch_usd

    @property
    def spread_pct(self) -> float:
        return 0.0 if not self.list_usd else 100.0 * self.spread_usd / self.list_usd

    @property
    def unpriced(self) -> int:
        return sum(v.unpriced for v in self.by_venue.values())

    @property
    def unknown_output(self) -> int:
        return sum(v.unknown_output for v in self.by_venue.values())

    @property
    def assumed_output(self) -> int:
        return sum(v.assumed_output for v in self.by_venue.values())

    @property
    def is_floor(self) -> bool:
        """True when some job's output tokens were unknown and priced at zero.

        Output is the expensive side on every model on the sheet, so a quote
        that silently omits it reads far cheaper than the bill. Such a quote is
        a floor, and says so.
        """
        return self.unknown_output > 0

    @property
    def is_estimated(self) -> bool:
        """True when some job's output size was assumed rather than known.

        Distinct from :attr:`is_floor`. A floor is understated by construction —
        output priced at zero. An estimate is priced on an assumption the caller
        supplied, so it can land either side of the bill. Both are marked on the
        card; neither is silent.
        """
        return self.assumed_output > 0

    @property
    def within_batch_window(self) -> bool:
        """Whether the deadline clears the venues' published completion window."""
        return self.window_seconds >= BATCH_COMPLETION_WINDOW_S

    def __str__(self) -> str:
        lines = [
            "OFFPEAK QUOTE " + "─" * 33,
            f"jobs      {self.jobs} across {len(self.by_venue)} venue(s)",
            f"deadline  {self.deadline:%Y-%m-%d %H:%M %Z} ({self.window_seconds / 3600:.1f}h out)",
            f"tokens    {self.input_tokens:,} in · {self.output_tokens:,} out",
            "",
        ]
        for name in sorted(self.by_venue):
            v = self.by_venue[name]
            lines.append(
                f"  {name:<16} {v.jobs:>5} job(s)  list ${format_usd(v.list_usd)}"
                f"  batch ${format_usd(v.batch_usd)}"
                f"  save ${format_usd(v.spread_usd)} ({v.spread_pct:.1f}%)"
            )
        lines += [
            "",
            f"list      ${format_usd(self.list_usd)}   (run now, synchronously)",
            f"batch     ${format_usd(self.batch_usd)}   (run by the deadline)",
            f"save      ${format_usd(self.spread_usd)} ({self.spread_pct:.1f}%)",
        ]
        if not self.within_batch_window:
            lines.append(
                f"risk      deadline is inside the {BATCH_COMPLETION_WINDOW_S // 3600}h batch "
                "window — the SLA rests on the sync fallback, which pays list"
            )
        if self.is_floor:
            lines.append(
                f"FLOOR     {self.unknown_output} job(s) gave no output-token signal; their "
                "output is priced at zero"
            )
            lines.append(
                "          output costs more than input on every model here — "
                "pass max_tokens or metadata to quote it properly"
            )
        if self.is_estimated:
            lines.append(
                f"EST       {self.assumed_output} job(s) priced on an assumed output size, "
                "not a measured one"
            )
            lines.append(
                "          the assumption is yours; the bill moves with what the "
                "model actually writes"
            )
        if self.unpriced:
            lines.append(f"note      {self.unpriced} job(s) had no price sheet entry")
        lines += [
            f"basis     {'; '.join(f'{k} {v}' for k, v in sorted(self.basis.items()))}",
            f"prices    snapshot {PRICE_SHEET_DATE} — estimate only, not a bill",
            "─" * 47,
        ]
        return "\n".join(lines)

is_floor property

True when some job's output tokens were unknown and priced at zero.

Output is the expensive side on every model on the sheet, so a quote that silently omits it reads far cheaper than the bill. Such a quote is a floor, and says so.

is_estimated property

True when some job's output size was assumed rather than known.

Distinct from :attr:is_floor. A floor is understated by construction — output priced at zero. An estimate is priced on an assumption the caller supplied, so it can land either side of the bill. Both are marked on the card; neither is silent.

within_batch_window property

Whether the deadline clears the venues' published completion window.

estimate_tokens(j, *, assumed_output_ratio=None)

(input, output, input_basis, output_basis) for one job.

Input: an explicit count on job.metadata wins, else a chars/4 estimate.

Output, in order — a count, then the caller's own expectation, then a ceiling, then the run-wide ratio if one was opted into, then nothing:

1. metadata["output_tokens"] — a count someone measured.
2. metadata["expected_output_tokens"] — what the caller expects this job to write. More specific than a ceiling set for safety, so it outranks one, and labeled an assumption either way.
3. params["max_tokens"] — an upper bound, priced as one.
4. assumed_output_ratio × the input tokens, when the caller passed one.
5. Nothing: zero, labeled unknown, which is what makes a quote a floor.

Each figure reports its own provenance so a quote never launders an estimate into a fact.

Source code in src/offpeak/quote.py

def estimate_tokens(
    j: Job, *, assumed_output_ratio: float | None = None
) -> tuple[int, int, str, str]:
    """(input, output, input_basis, output_basis) for one job.

    Input: an explicit count on ``job.metadata`` wins, else a chars/4 estimate.

    Output, in order — a count, then the caller's own expectation, then a
    ceiling, then the run-wide ratio if one was opted into, then nothing:

    1. ``metadata["output_tokens"]`` — a count someone measured.
    2. ``metadata["expected_output_tokens"]`` — what the caller expects this job
       to write. More specific than a ceiling set for safety, so it outranks
       one, and labeled an assumption either way.
    3. ``params["max_tokens"]`` — an upper bound, priced as one.
    4. *assumed_output_ratio* × the input tokens, when the caller passed one.
    5. Nothing: zero, labeled unknown, which is what makes a quote a floor.

    Each figure reports its own provenance so a quote never launders an
    estimate into a fact.
    """
    meta = j.metadata or {}

    if isinstance(meta.get("input_tokens"), int):
        input_tokens, input_basis = int(meta["input_tokens"]), "explicit"
    else:
        chars = sum(_text_len(m.get("content")) for m in j.messages)
        input_tokens = max(1, math.ceil(chars / CHARS_PER_TOKEN))
        input_basis = f"estimated (chars/{CHARS_PER_TOKEN})"

    if isinstance(meta.get("output_tokens"), int):
        output_tokens, output_basis = int(meta["output_tokens"]), "explicit"
    elif isinstance(meta.get("expected_output_tokens"), int):
        output_tokens = int(meta["expected_output_tokens"])
        output_basis = "assumed (expected_output_tokens)"
    elif isinstance(j.params.get("max_tokens"), int):
        output_tokens, output_basis = int(j.params["max_tokens"]), "ceiling (max_tokens)"
    elif assumed_output_ratio is not None:
        output_tokens = max(0, round(input_tokens * assumed_output_ratio))
        output_basis = f"assumed (ratio {assumed_output_ratio:g} x input)"
    else:
        output_tokens, output_basis = 0, "unknown (no max_tokens, none given)"

    return input_tokens, output_tokens, input_basis, output_basis

quote(jobs, deadline, *, venues=None, assumed_output_ratio=None)

Price jobs against deadline without calling any provider.

Routes each job to the venue that would run it, then settles list versus batch cost from the bundled price sheet.

assumed_output_ratio is an explicit opt-in: for jobs that carry no output signal at all, assume they write ratio x their input tokens. 0.25 suits summarization; a long-form generator writes more than it reads and wants a ratio above 1. Without it, such jobs price at zero output and the quote is a FLOOR — the library does not guess on your behalf. With it, the quote is marked EST and :attr:Quote.is_estimated is true. Per-job expectations (metadata={"expected_output_tokens": n}) take precedence and are marked the same way.

Raises ValueError for a deadline in the past, a model no venue supports, or a non-positive ratio — the same programming errors :func:offpeak.run reserves exceptions for.

Source code in src/offpeak/quote.py

def quote(
    jobs: Job | list[Job],
    deadline: object,
    *,
    venues: list[Venue] | None = None,
    assumed_output_ratio: float | None = None,
) -> Quote:
    """Price *jobs* against *deadline* without calling any provider.

    Routes each job to the venue that would run it, then settles list versus
    batch cost from the bundled price sheet.

    *assumed_output_ratio* is an explicit opt-in: for jobs that carry no output
    signal at all, assume they write ``ratio x`` their input tokens. ``0.25``
    suits summarization; a long-form generator writes more than it reads and
    wants a ratio above 1. Without it, such jobs price at zero output and the
    quote is a ``FLOOR`` — the library does not guess on your behalf. With it,
    the quote is marked ``EST`` and :attr:`Quote.is_estimated` is true. Per-job
    expectations (``metadata={"expected_output_tokens": n}``) take precedence
    and are marked the same way.

    Raises ``ValueError`` for a deadline in the past, a model no venue supports,
    or a non-positive ratio — the same programming errors :func:`offpeak.run`
    reserves exceptions for.
    """
    if assumed_output_ratio is not None and assumed_output_ratio <= 0:
        raise ValueError(
            f"assumed_output_ratio must be positive, got {assumed_output_ratio!r} "
            "(omit it to price unknown output at zero and get a FLOOR quote)"
        )
    job_list = [jobs] if isinstance(jobs, Job) else list(jobs)
    resolved = parse_deadline(deadline)
    q = Quote(deadline=resolved, window_seconds=seconds_until(resolved))
    if not job_list:
        return q

    venue_list = venues if venues is not None else default_venues()
    bases: dict[str, set[str]] = {"input": set(), "output": set()}

    for j in job_list:
        venue = _pick_venue(j.model, venue_list)
        vq = q.by_venue.setdefault(venue.name, VenueQuote(venue=venue.name))
        input_tokens, output_tokens, input_basis, output_basis = estimate_tokens(
            j, assumed_output_ratio=assumed_output_ratio
        )
        bases["input"].add(input_basis)
        bases["output"].add(output_basis)

        vq.jobs += 1
        vq.input_tokens += input_tokens
        vq.output_tokens += output_tokens
        if output_basis.startswith("unknown"):
            vq.unknown_output += 1
        elif output_basis.startswith("assumed"):
            vq.assumed_output += 1

        price = get_price(j.model)
        if price is None:
            vq.unpriced += 1
            continue
        list_usd = (input_tokens * price[0] + output_tokens * price[1]) / 1_000_000
        vq.list_usd += list_usd
        vq.batch_usd += list_usd * BATCH_DISCOUNT

    q.basis = {k: ", ".join(sorted(v)) for k, v in bases.items() if v}
    return q

offpeak.receipt(results)

Settle a run: aggregate per-job receipts into one :class:Settlement.

Source code in src/offpeak/client.py

def receipt(results: list[Result]) -> Settlement:
    """Settle a run: aggregate per-job receipts into one :class:`Settlement`."""
    settlement = Settlement()
    for result in results:
        settlement.total += 1
        if result.ok:
            settlement.ok += 1
        else:
            settlement.failed += 1
        r = result.receipt
        if r is None:
            continue
        settlement.sla_met += int(r.sla_met)
        settlement.fell_back += int(r.fell_back)
        settlement.input_tokens += r.input_tokens
        settlement.output_tokens += r.output_tokens
        settlement.by_venue[r.venue] = settlement.by_venue.get(r.venue, 0) + 1
        if r.list_usd is None or r.paid_usd is None:
            settlement.unpriced += 1
        else:
            settlement.list_usd += r.list_usd
            settlement.paid_usd += r.paid_usd
            if r.fell_back:
                # The spread this job would have captured had the batch held.
                settlement.left_on_table_usd += r.list_usd * (1 - BATCH_DISCOUNT)
    return settlement

offpeak.job

Job, Result, and Receipt — the unit of deferred work and its settlement.

Job dataclass

A venue-agnostic chat-completion job.

Source code in src/offpeak/job.py

@dataclass
class Job:
    """A venue-agnostic chat-completion job."""

    model: str
    messages: list[dict]
    params: dict = field(default_factory=dict)
    id: str = field(default_factory=lambda: f"job_{uuid.uuid4().hex[:12]}")
    metadata: dict = field(default_factory=dict)
    status: Status = Status.QUEUED

Receipt dataclass

Per-job settlement: what ran where, when, and what the hour was worth.

Source code in src/offpeak/job.py

@dataclass
class Receipt:
    """Per-job settlement: what ran where, when, and what the hour was worth."""

    venue: str
    model: str
    deadline: datetime
    submitted_at: datetime
    completed_at: datetime | None = None
    input_tokens: int = 0
    output_tokens: int = 0
    fell_back: bool = False

    @property
    def sla_met(self) -> bool:
        return self.completed_at is not None and self.completed_at <= self.deadline

    @property
    def list_usd(self) -> float | None:
        """What the job would have cost run synchronously at list price."""
        return list_cost_usd(self.model, self.input_tokens, self.output_tokens)

    @property
    def paid_usd(self) -> float | None:
        """What the job cost on the venue it actually ran on."""
        if self.fell_back:
            return self.list_usd
        return batch_cost_usd(self.model, self.input_tokens, self.output_tokens)

    @property
    def spread_usd(self) -> float | None:
        """Captured spread: list minus paid."""
        if self.list_usd is None or self.paid_usd is None:
            return None
        return self.list_usd - self.paid_usd

    def __str__(self) -> str:
        """One line, in money you can actually read.

        The float properties above stay floats — this is the rendering, so a
        sub-cent job reports what it cost instead of $0.00.
        """
        where = f"{self.venue} {self.model}"
        if self.fell_back:
            where += " (sync fallback)"
        return (
            f"{where}: {self.input_tokens:,} in · {self.output_tokens:,} out · "
            f"list ${format_usd(self.list_usd)} · paid ${format_usd(self.paid_usd)} · "
            f"captured ${format_usd(self.spread_usd)}"
        )

list_usd property

What the job would have cost run synchronously at list price.

paid_usd property

What the job cost on the venue it actually ran on.

spread_usd property

Captured spread: list minus paid.

__str__()

One line, in money you can actually read.

The float properties above stay floats — this is the rendering, so a sub-cent job reports what it cost instead of $0.00.

Source code in src/offpeak/job.py

def __str__(self) -> str:
    """One line, in money you can actually read.

    The float properties above stay floats — this is the rendering, so a
    sub-cent job reports what it cost instead of $0.00.
    """
    where = f"{self.venue} {self.model}"
    if self.fell_back:
        where += " (sync fallback)"
    return (
        f"{where}: {self.input_tokens:,} in · {self.output_tokens:,} out · "
        f"list ${format_usd(self.list_usd)} · paid ${format_usd(self.paid_usd)} · "
        f"captured ${format_usd(self.spread_usd)}"
    )

Result dataclass

The outcome of one job.

Source code in src/offpeak/job.py

@dataclass
class Result:
    """The outcome of one job."""

    job: Job
    text: str | None = None
    raw: object = None
    error: str | None = None
    receipt: Receipt | None = None

    @property
    def ok(self) -> bool:
        return self.error is None and self.text is not None

job(model, input=None, *, system=None, metadata=None, **params)

Build a :class:Job.

input may be a plain prompt string or a full messages list. Extra keyword arguments (temperature, max_tokens, ...) are passed through to the venue.

Source code in src/offpeak/job.py

def job(
    model: str,
    input: str | list[dict] | None = None,
    *,
    system: str | None = None,
    metadata: dict | None = None,
    **params: object,
) -> Job:
    """Build a :class:`Job`.

    ``input`` may be a plain prompt string or a full ``messages`` list.
    Extra keyword arguments (``temperature``, ``max_tokens``, ...) are passed
    through to the venue.
    """
    if input is None:
        raise ValueError("job() requires an input (a prompt string or a messages list)")
    if isinstance(input, str):
        messages = [{"role": "user", "content": input}]
    else:
        messages = list(input)
    if system is not None:
        messages = [{"role": "system", "content": system}, *messages]
    return Job(model=model, messages=messages, params=dict(params), metadata=metadata or {})

Types

offpeak.Job dataclass

A venue-agnostic chat-completion job.

Source code in src/offpeak/job.py

@dataclass
class Job:
    """A venue-agnostic chat-completion job."""

    model: str
    messages: list[dict]
    params: dict = field(default_factory=dict)
    id: str = field(default_factory=lambda: f"job_{uuid.uuid4().hex[:12]}")
    metadata: dict = field(default_factory=dict)
    status: Status = Status.QUEUED

offpeak.Result dataclass

The outcome of one job.

Source code in src/offpeak/job.py

@dataclass
class Result:
    """The outcome of one job."""

    job: Job
    text: str | None = None
    raw: object = None
    error: str | None = None
    receipt: Receipt | None = None

    @property
    def ok(self) -> bool:
        return self.error is None and self.text is not None

offpeak.Receipt dataclass

Per-job settlement: what ran where, when, and what the hour was worth.

Source code in src/offpeak/job.py

@dataclass
class Receipt:
    """Per-job settlement: what ran where, when, and what the hour was worth."""

    venue: str
    model: str
    deadline: datetime
    submitted_at: datetime
    completed_at: datetime | None = None
    input_tokens: int = 0
    output_tokens: int = 0
    fell_back: bool = False

    @property
    def sla_met(self) -> bool:
        return self.completed_at is not None and self.completed_at <= self.deadline

    @property
    def list_usd(self) -> float | None:
        """What the job would have cost run synchronously at list price."""
        return list_cost_usd(self.model, self.input_tokens, self.output_tokens)

    @property
    def paid_usd(self) -> float | None:
        """What the job cost on the venue it actually ran on."""
        if self.fell_back:
            return self.list_usd
        return batch_cost_usd(self.model, self.input_tokens, self.output_tokens)

    @property
    def spread_usd(self) -> float | None:
        """Captured spread: list minus paid."""
        if self.list_usd is None or self.paid_usd is None:
            return None
        return self.list_usd - self.paid_usd

    def __str__(self) -> str:
        """One line, in money you can actually read.

        The float properties above stay floats — this is the rendering, so a
        sub-cent job reports what it cost instead of $0.00.
        """
        where = f"{self.venue} {self.model}"
        if self.fell_back:
            where += " (sync fallback)"
        return (
            f"{where}: {self.input_tokens:,} in · {self.output_tokens:,} out · "
            f"list ${format_usd(self.list_usd)} · paid ${format_usd(self.paid_usd)} · "
            f"captured ${format_usd(self.spread_usd)}"
        )

list_usd property

What the job would have cost run synchronously at list price.

paid_usd property

What the job cost on the venue it actually ran on.

spread_usd property

Captured spread: list minus paid.

__str__()

One line, in money you can actually read.

The float properties above stay floats — this is the rendering, so a sub-cent job reports what it cost instead of $0.00.

Source code in src/offpeak/job.py

def __str__(self) -> str:
    """One line, in money you can actually read.

    The float properties above stay floats — this is the rendering, so a
    sub-cent job reports what it cost instead of $0.00.
    """
    where = f"{self.venue} {self.model}"
    if self.fell_back:
        where += " (sync fallback)"
    return (
        f"{where}: {self.input_tokens:,} in · {self.output_tokens:,} out · "
        f"list ${format_usd(self.list_usd)} · paid ${format_usd(self.paid_usd)} · "
        f"captured ${format_usd(self.spread_usd)}"
    )

offpeak.Status

Bases: str, Enum

Source code in src/offpeak/job.py

class Status(str, Enum):
    QUEUED = "queued"
    SUBMITTED = "submitted"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    FELL_BACK = "fell_back"  # completed, but via the sync fallback (list price)

offpeak.Settlement dataclass

Aggregate receipt across a run.

Source code in src/offpeak/client.py

@dataclass
class Settlement:
    """Aggregate receipt across a run."""

    total: int = 0
    ok: int = 0
    sla_met: int = 0
    fell_back: int = 0
    failed: int = 0
    input_tokens: int = 0
    output_tokens: int = 0
    list_usd: float = 0.0
    paid_usd: float = 0.0
    left_on_table_usd: float = 0.0
    unpriced: int = 0
    by_venue: dict = field(default_factory=dict)

    @property
    def captured_usd(self) -> float:
        return self.list_usd - self.paid_usd

    @property
    def captured_pct(self) -> float:
        return 0.0 if not self.list_usd else 100.0 * self.captured_usd / self.list_usd

    def __str__(self) -> str:
        venues = " · ".join(f"{k} {v}" for k, v in sorted(self.by_venue.items()))
        lines = [
            "OFFPEAK SETTLEMENT " + "─" * 28,
            f"jobs      {self.total} ({self.ok} ok, {self.fell_back} sync fallback, "
            f"{self.failed} failed)",
            f"sla       {self.sla_met}/{self.total} met",
            f"venues    {venues or '—'}",
            f"tokens    {self.input_tokens:,} in · {self.output_tokens:,} out",
            f"list      ${_usd(self.list_usd)}",
            f"paid      ${_usd(self.paid_usd)}",
            f"captured  ${_usd(self.captured_usd)} ({self.captured_pct:.1f}%)",
            f"prices    snapshot {PRICE_SHEET_DATE} — override via offpeak.prices",
        ]
        if self.fell_back:
            lines.append(
                f"left      ${_usd(self.left_on_table_usd)} on the table "
                f"({self.fell_back} job(s) missed the batch tier)"
            )
        if self.unpriced:
            lines.append(f"note      {self.unpriced} job(s) had no price sheet entry")
        lines.append("─" * 47)
        return "\n".join(lines)

offpeak.Quote dataclass

A pre-trade quote. No API calls were made to produce this.

Source code in src/offpeak/quote.py

@dataclass
class Quote:
    """A pre-trade quote. No API calls were made to produce this."""

    deadline: datetime
    window_seconds: float
    by_venue: dict[str, VenueQuote] = field(default_factory=dict)
    basis: dict[str, str] = field(default_factory=dict)

    @property
    def jobs(self) -> int:
        return sum(v.jobs for v in self.by_venue.values())

    @property
    def input_tokens(self) -> int:
        return sum(v.input_tokens for v in self.by_venue.values())

    @property
    def output_tokens(self) -> int:
        return sum(v.output_tokens for v in self.by_venue.values())

    @property
    def list_usd(self) -> float:
        return sum(v.list_usd for v in self.by_venue.values())

    @property
    def batch_usd(self) -> float:
        return sum(v.batch_usd for v in self.by_venue.values())

    @property
    def spread_usd(self) -> float:
        return self.list_usd - self.batch_usd

    @property
    def spread_pct(self) -> float:
        return 0.0 if not self.list_usd else 100.0 * self.spread_usd / self.list_usd

    @property
    def unpriced(self) -> int:
        return sum(v.unpriced for v in self.by_venue.values())

    @property
    def unknown_output(self) -> int:
        return sum(v.unknown_output for v in self.by_venue.values())

    @property
    def assumed_output(self) -> int:
        return sum(v.assumed_output for v in self.by_venue.values())

    @property
    def is_floor(self) -> bool:
        """True when some job's output tokens were unknown and priced at zero.

        Output is the expensive side on every model on the sheet, so a quote
        that silently omits it reads far cheaper than the bill. Such a quote is
        a floor, and says so.
        """
        return self.unknown_output > 0

    @property
    def is_estimated(self) -> bool:
        """True when some job's output size was assumed rather than known.

        Distinct from :attr:`is_floor`. A floor is understated by construction —
        output priced at zero. An estimate is priced on an assumption the caller
        supplied, so it can land either side of the bill. Both are marked on the
        card; neither is silent.
        """
        return self.assumed_output > 0

    @property
    def within_batch_window(self) -> bool:
        """Whether the deadline clears the venues' published completion window."""
        return self.window_seconds >= BATCH_COMPLETION_WINDOW_S

    def __str__(self) -> str:
        lines = [
            "OFFPEAK QUOTE " + "─" * 33,
            f"jobs      {self.jobs} across {len(self.by_venue)} venue(s)",
            f"deadline  {self.deadline:%Y-%m-%d %H:%M %Z} ({self.window_seconds / 3600:.1f}h out)",
            f"tokens    {self.input_tokens:,} in · {self.output_tokens:,} out",
            "",
        ]
        for name in sorted(self.by_venue):
            v = self.by_venue[name]
            lines.append(
                f"  {name:<16} {v.jobs:>5} job(s)  list ${format_usd(v.list_usd)}"
                f"  batch ${format_usd(v.batch_usd)}"
                f"  save ${format_usd(v.spread_usd)} ({v.spread_pct:.1f}%)"
            )
        lines += [
            "",
            f"list      ${format_usd(self.list_usd)}   (run now, synchronously)",
            f"batch     ${format_usd(self.batch_usd)}   (run by the deadline)",
            f"save      ${format_usd(self.spread_usd)} ({self.spread_pct:.1f}%)",
        ]
        if not self.within_batch_window:
            lines.append(
                f"risk      deadline is inside the {BATCH_COMPLETION_WINDOW_S // 3600}h batch "
                "window — the SLA rests on the sync fallback, which pays list"
            )
        if self.is_floor:
            lines.append(
                f"FLOOR     {self.unknown_output} job(s) gave no output-token signal; their "
                "output is priced at zero"
            )
            lines.append(
                "          output costs more than input on every model here — "
                "pass max_tokens or metadata to quote it properly"
            )
        if self.is_estimated:
            lines.append(
                f"EST       {self.assumed_output} job(s) priced on an assumed output size, "
                "not a measured one"
            )
            lines.append(
                "          the assumption is yours; the bill moves with what the "
                "model actually writes"
            )
        if self.unpriced:
            lines.append(f"note      {self.unpriced} job(s) had no price sheet entry")
        lines += [
            f"basis     {'; '.join(f'{k} {v}' for k, v in sorted(self.basis.items()))}",
            f"prices    snapshot {PRICE_SHEET_DATE} — estimate only, not a bill",
            "─" * 47,
        ]
        return "\n".join(lines)

is_floor property

True when some job's output tokens were unknown and priced at zero.

Output is the expensive side on every model on the sheet, so a quote that silently omits it reads far cheaper than the bill. Such a quote is a floor, and says so.

is_estimated property

True when some job's output size was assumed rather than known.

Distinct from :attr:is_floor. A floor is understated by construction — output priced at zero. An estimate is priced on an assumption the caller supplied, so it can land either side of the bill. Both are marked on the card; neither is silent.

within_batch_window property

Whether the deadline clears the venues' published completion window.

offpeak.VenueQuote dataclass

What one venue's batch tier is worth for the jobs routed to it.

Source code in src/offpeak/quote.py

@dataclass
class VenueQuote:
    """What one venue's batch tier is worth for the jobs routed to it."""

    venue: str
    jobs: int = 0
    input_tokens: int = 0
    output_tokens: int = 0
    list_usd: float = 0.0
    batch_usd: float = 0.0
    unpriced: int = 0
    unknown_output: int = 0
    assumed_output: int = 0

    @property
    def spread_usd(self) -> float:
        return self.list_usd - self.batch_usd

    @property
    def spread_pct(self) -> float:
        return 0.0 if not self.list_usd else 100.0 * self.spread_usd / self.list_usd

Deadlines

offpeak.parse_deadline(value, *, now=None)

Resolve value to an aware datetime.

Raises ValueError if the form is unrecognized or the resolved deadline is not in the future, and TypeError for unsupported types.

Source code in src/offpeak/deadline.py

def parse_deadline(value: object, *, now: datetime | None = None) -> datetime:
    """Resolve *value* to an aware datetime.

    Raises ``ValueError`` if the form is unrecognized or the resolved deadline
    is not in the future, and ``TypeError`` for unsupported types.
    """
    if now is None:
        now = _local_now()
    elif now.tzinfo is None:
        now = now.astimezone()
    deadline = _parse(value, now)
    if deadline <= now:
        raise ValueError(
            f"deadline {deadline.isoformat()} is not in the future (now: {now.isoformat()})"
        )
    return deadline

offpeak.seconds_until(deadline, *, now=None)

Seconds remaining until deadline (negative if it has passed).

Source code in src/offpeak/deadline.py

def seconds_until(deadline: datetime, *, now: datetime | None = None) -> float:
    """Seconds remaining until *deadline* (negative if it has passed)."""
    if now is None:
        now = _local_now()
    return (deadline - now).total_seconds()

Venues

offpeak.Venue

Bases: ABC

A place deferred work can execute, plus a synchronous escape hatch.

Source code in src/offpeak/venues/base.py

class Venue(ABC):
    """A place deferred work can execute, plus a synchronous escape hatch."""

    name: str = "venue"

    @abstractmethod
    def supports(self, model: str) -> bool:
        """Whether this venue can run *model*."""

    @abstractmethod
    def submit(self, jobs: list[Job]) -> str:
        """Submit *jobs* as one batch; return an opaque batch handle."""

    @abstractmethod
    def status(self, handle: str) -> BatchState:
        """Poll a batch's progress."""

    @abstractmethod
    def collect(self, handle: str) -> dict[str, Result]:
        """Fetch results for a finished batch, keyed by job id."""

    @abstractmethod
    def cancel(self, handle: str) -> None:
        """Best-effort cancel of an in-flight batch."""

    @abstractmethod
    def run_sync(self, job: Job) -> Result:
        """Run one job synchronously at list price (the SLA fallback path)."""

supports(model) abstractmethod

Whether this venue can run model.

Source code in src/offpeak/venues/base.py

@abstractmethod
def supports(self, model: str) -> bool:
    """Whether this venue can run *model*."""

submit(jobs) abstractmethod

Submit jobs as one batch; return an opaque batch handle.

Source code in src/offpeak/venues/base.py

@abstractmethod
def submit(self, jobs: list[Job]) -> str:
    """Submit *jobs* as one batch; return an opaque batch handle."""

status(handle) abstractmethod

Poll a batch's progress.

Source code in src/offpeak/venues/base.py

@abstractmethod
def status(self, handle: str) -> BatchState:
    """Poll a batch's progress."""

collect(handle) abstractmethod

Fetch results for a finished batch, keyed by job id.

Source code in src/offpeak/venues/base.py

@abstractmethod
def collect(self, handle: str) -> dict[str, Result]:
    """Fetch results for a finished batch, keyed by job id."""

cancel(handle) abstractmethod

Best-effort cancel of an in-flight batch.

Source code in src/offpeak/venues/base.py

@abstractmethod
def cancel(self, handle: str) -> None:
    """Best-effort cancel of an in-flight batch."""

run_sync(job) abstractmethod

Run one job synchronously at list price (the SLA fallback path).

Source code in src/offpeak/venues/base.py

@abstractmethod
def run_sync(self, job: Job) -> Result:
    """Run one job synchronously at list price (the SLA fallback path)."""

offpeak.BatchState dataclass

A venue batch's progress.

Source code in src/offpeak/venues/base.py

@dataclass
class BatchState:
    """A venue batch's progress."""

    status: str  # "in_progress" | "completed" | "failed" | "cancelled"
    completed: int = 0
    failed: int = 0
    total: int = 0

    @property
    def done(self) -> bool:
        return self.status in ("completed", "failed", "cancelled")

offpeak.default_venues()

Provider batch tiers, tried in order. SDKs import lazily on first use.

Source code in src/offpeak/client.py

def default_venues() -> list[Venue]:
    """Provider batch tiers, tried in order. SDKs import lazily on first use."""
    from .venues.anthropic_batch import AnthropicBatch
    from .venues.openai_batch import OpenAIBatch

    return [AnthropicBatch(), OpenAIBatch()]

Prices

offpeak.prices

List-price sheet and batch discounts, for receipts.

Receipts are arithmetic against public price sheets — no estimates. The prices below are a bundled snapshot (see PRICE_SHEET_DATE); providers change prices, so verify against their published sheets and override at runtime with :func:register_price where they have moved. Costs for unknown models resolve to None rather than a guess.

Batch tiers at OpenAI, Anthropic, and Google are publicly priced at 50% of list, which is what :data:BATCH_DISCOUNT encodes. OpenAI's flex tier prices identically to its batch tier on the gpt-5.6 family, and its fast tier at twice list — the same model, priced for urgency. Fast is stored rather than derived (:func:get_fast_price), because unlike batch it is not a discount rule but its own published row; :func:urgency_spread divides the two so the price of an hour is a computed number and not a claim in prose.

Some list prices are promotional and will step up on a published date. Those carry a :class:PromoNote in :data:PROMO_NOTES — the date and the post-promo list — so a quote or a docs page can flag the decay instead of reading a temporary number as permanent.

Corrections

2026-08-21 — the OpenAI block through 0.2.0 held that provider's batch sheet in the standard-price table (gpt-5.6-sol 2.50/15.00, terra 1.00/6.00, luna 0.10/0.60). The published short-context standard rates are 4.00/20.00, 2.00/12.00 and 0.20/1.20; the batch rows are 2.00/10.00, 1.00/6.00 and 0.10/0.60. Receipts for OpenAI models in 0.1.1–0.2.0 therefore understated both the list cost they compared against and the batch price actually billed — the wrong sheet derived $1.25/$7.50 for a batched sol job against a true $2.00 / $10.00. Anthropic's block was unaffected.

PromoNote dataclass

A list price that is promotional, and what it decays to.

A promotional rate is a real price today and a wrong one later. Carrying the step-up here keeps the sheet honest in both directions: receipts settle at the price actually charged, while a quote or a docs page can say — from data rather than prose — that the number has an expiry and what replaces it.

Source code in src/offpeak/prices.py

@dataclass(frozen=True)
class PromoNote:
    """A list price that is promotional, and what it decays to.

    A promotional rate is a real price today and a wrong one later. Carrying the
    step-up here keeps the sheet honest in both directions: receipts settle at
    the price actually charged, while a quote or a docs page can say — from data
    rather than prose — that the number has an expiry and what replaces it.
    """

    #: The date the provider guarantees the promo through (ISO 8601). It may run
    #: longer: the sheet says "at least through" this date, never "until".
    through: str
    #: (input, output) USD per 1M tokens once the promo lapses.
    post_promo: tuple[float, float]
    #: Where the claim is checkable.
    source: str
    #: The provider's own wording, verbatim.
    note: str

register_price(model, input_per_m, output_per_m)

Set or override the list price for model (USD per 1M tokens).

Source code in src/offpeak/prices.py

def register_price(model: str, input_per_m: float, output_per_m: float) -> None:
    """Set or override the list price for *model* (USD per 1M tokens)."""
    _PRICES[model] = (float(input_per_m), float(output_per_m))

get_price(model)

Standard (synchronous) list price for model, USD per 1M tokens.

Source code in src/offpeak/prices.py

def get_price(model: str) -> tuple[float, float] | None:
    """Standard (synchronous) list price for *model*, USD per 1M tokens."""
    return _lookup(_PRICES, model)

get_fast_price(model)

Fast-tier price for model, USD per 1M tokens.

None where the venue publishes no fast tier — which is everywhere except OpenAI's gpt-5.6 family today. Unlike batch, fast is not a discount rule applied to list: it is its own published row, so it is stored, not derived.

Source code in src/offpeak/prices.py

def get_fast_price(model: str) -> tuple[float, float] | None:
    """Fast-tier price for *model*, USD per 1M tokens.

    ``None`` where the venue publishes no fast tier — which is everywhere
    except OpenAI's gpt-5.6 family today. Unlike batch, fast is not a discount
    rule applied to list: it is its own published row, so it is stored, not
    derived.
    """
    return _lookup(_FAST_PRICES, model)

get_promo_note(model)

The :class:PromoNote for model, if its list price is promotional.

None means "no published promotion", which is also what a model registered at runtime with :func:register_price returns — an override is a price we were told, not a price we can date.

Source code in src/offpeak/prices.py

def get_promo_note(model: str) -> PromoNote | None:
    """The :class:`PromoNote` for *model*, if its list price is promotional.

    ``None`` means "no published promotion", which is also what a model
    registered at runtime with :func:`register_price` returns — an override is
    a price we were told, not a price we can date.
    """
    return _lookup(PROMO_NOTES, model)

promo_decay(model)

Multiple the (input, output) price steps up by when the promo lapses.

(1.25, 1.5) on gpt-5.6-sol: $4/$20 today, $5/$30 after. None where the price is not promotional or the model is off the sheet.

Source code in src/offpeak/prices.py

def promo_decay(model: str) -> tuple[float, float] | None:
    """Multiple the (input, output) price steps up by when the promo lapses.

    ``(1.25, 1.5)`` on gpt-5.6-sol: $4/$20 today, $5/$30 after. ``None`` where
    the price is not promotional or the model is off the sheet.
    """
    note = get_promo_note(model)
    price = get_price(model)
    if note is None or price is None or not price[0] or not price[1]:
        return None
    return (note.post_promo[0] / price[0], note.post_promo[1] / price[1])

fast_cost_usd(model, input_tokens, output_tokens)

What the same tokens cost on the venue's fast tier, where it has one.

Source code in src/offpeak/prices.py

def fast_cost_usd(model: str, input_tokens: int, output_tokens: int) -> float | None:
    """What the same tokens cost on the venue's fast tier, where it has one."""
    price = get_fast_price(model)
    if price is None:
        return None
    return (input_tokens * price[0] + output_tokens * price[1]) / 1_000_000

urgency_spread(model)

How much the same model costs at its most urgent published tier over its most patient one: fast ÷ batch.

This is the intra-venue price of an hour with the model held constant — one provider, one model, two deadlines. On gpt-5.6-sol that is $8/$40 per 1M against $2/$10, a 4x spread.

Both legs are checked and the lower is returned, so the figure can never overstate what a venue publishes. None where the venue prices no fast tier for the model, or the model is off the sheet.

Source code in src/offpeak/prices.py

def urgency_spread(model: str) -> float | None:
    """How much the same model costs at its most urgent published tier over its
    most patient one: fast ÷ batch.

    This is the intra-venue price of an hour with the model held constant — one
    provider, one model, two deadlines. On gpt-5.6-sol that is $8/$40 per 1M
    against $2/$10, a **4x** spread.

    Both legs are checked and the *lower* is returned, so the figure can never
    overstate what a venue publishes. ``None`` where the venue prices no fast
    tier for the model, or the model is off the sheet.
    """
    fast = get_fast_price(model)
    standard = get_price(model)
    if fast is None or standard is None:
        return None
    legs = [
        fast[i] / (standard[i] * BATCH_DISCOUNT)
        for i in (0, 1)
        if standard[i] * BATCH_DISCOUNT
    ]
    return min(legs) if legs else None

format_usd(amount)

Money for humans: 2dp once there are cents to show, more significant digits below that so a sub-cent job does not settle as a column of $0.00.

None (an unpriced model) renders as an em dash, never as zero — a price we do not know is not a price of nothing.

Source code in src/offpeak/prices.py

def format_usd(amount: float | None) -> str:
    """Money for humans: 2dp once there are cents to show, more significant
    digits below that so a sub-cent job does not settle as a column of $0.00.

    ``None`` (an unpriced model) renders as an em dash, never as zero — a price
    we do not know is not a price of nothing.
    """
    if amount is None:
        return "—"
    if amount == 0:
        return "0.00"
    if abs(amount) >= 0.005:
        return f"{amount:,.2f}"
    return f"{amount:,.{-math.floor(math.log10(abs(amount))) + 2}f}"