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.

run() blocks for as long as the batch takes. When the calling process cannot stay alive that long — a laptop, a CI step, a serverless function — use :func:submit and :func:collect and keep the :class:Ticket between them; run() is exactly collect(submit(...)).

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.

    ``run()`` blocks for as long as the batch takes. When the calling process
    cannot stay alive that long — a laptop, a CI step, a serverless function —
    use :func:`submit` and :func:`collect` and keep the :class:`Ticket` between
    them; ``run()`` is exactly ``collect(submit(...))``.
    """
    job_list = [jobs] if isinstance(jobs, Job) else list(jobs)
    if not job_list:
        return []
    venue_list = venues if venues is not None else default_venues()
    ticket = submit(job_list, deadline, venues=venue_list, risk_buffer=risk_buffer)
    results = collect(
        ticket, venues=venue_list, fallback=fallback, wait=True, poll_interval=poll_interval
    )
    assert results is not None  # wait=True always settles
    return results

offpeak.submit(jobs, deadline, *, venues=None, risk_buffer=None)

Submit jobs to their venues' batch tiers and return immediately.

The returned :class:Ticket is the run's whole state; keep it (save()) and finish with :func:collect — in this process or another. Raises only for programming errors (a bad or past deadline, a model no venue supports); a venue that fails at submit is recorded on the ticket and its jobs are rescued by the fallback at collect time.

Source code in src/offpeak/ticket.py

def submit(
    jobs: Job | list[Job],
    deadline: object,
    *,
    venues: list[Venue] | None = None,
    risk_buffer: float | None = None,
) -> Ticket:
    """Submit *jobs* to their venues' batch tiers and return immediately.

    The returned :class:`Ticket` is the run's whole state; keep it (``save()``)
    and finish with :func:`collect` — in this process or another. Raises only
    for programming errors (a bad or past deadline, a model no venue supports);
    a venue that fails at submit is recorded on the ticket and its jobs are
    rescued by the fallback at collect time.
    """
    job_list = [jobs] if isinstance(jobs, Job) else list(jobs)
    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()

    ticket = Ticket(
        jobs=job_list,
        deadline=resolved,
        submitted_at=datetime.now().astimezone(),
        risk_buffer=risk_buffer,
    )
    if not job_list:
        return ticket

    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)
        ticket.assignment[j.id] = venue.name

    for name, (venue, group_jobs) in groups.items():
        try:
            ticket.batches[name] = venue.submit(group_jobs)
        except Exception as exc:  # noqa: BLE001 — the provider failed, not us
            ticket.venue_errors[name] = f"submit failed: {exc}"
            continue
        for j in group_jobs:
            j.status = Status.SUBMITTED
    return ticket

offpeak.collect(ticket, *, venues=None, fallback='sync', wait=True, poll_interval=None)

Finish a submitted run.

With wait=True (default) this is the back half of :func:run: poll until every batch lands or the remaining window shrinks to the ticket's risk buffer, then cancel stragglers and rescue them synchronously (fallback="sync") or report them failed (fallback="none"). Returns one :class:Result per job, in input order.

With wait=False it does one sweep and returns the results only if the run can be settled now — everything landed, or the deadline is close enough that the buffer rule fires. Otherwise it returns None with the ticket updated (anything that landed is kept on it); call again later.

Pass the same venues you submitted with. Venues are matched by name; a venue missing from this process fails its jobs with a clear message.

Source code in src/offpeak/ticket.py

def collect(
    ticket: Ticket,
    *,
    venues: list[Venue] | None = None,
    fallback: str = "sync",
    wait: bool = True,
    poll_interval: float | None = None,
) -> list[Result] | None:
    """Finish a submitted run.

    With ``wait=True`` (default) this is the back half of :func:`run`: poll
    until every batch lands or the remaining window shrinks to the ticket's
    risk buffer, then cancel stragglers and rescue them synchronously
    (``fallback="sync"``) or report them failed (``fallback="none"``). Returns
    one :class:`Result` per job, in input order.

    With ``wait=False`` it does one sweep and returns the results only if the
    run can be settled now — everything landed, or the deadline is close
    enough that the buffer rule fires. Otherwise it returns ``None`` with the
    ticket updated (anything that landed is kept on it); call again later.

    Pass the same ``venues`` you submitted with. Venues are matched by name;
    a venue missing from this process fails its jobs with a clear message.
    """
    if not ticket.jobs:
        return []
    by_name = _venues_by_name(ticket, venues)

    while True:
        _sweep(ticket, by_name)
        remaining = ticket.remaining
        if not ticket.pending and not ticket._missing():
            break
        if remaining <= ticket.risk_buffer or not ticket.pending:
            break
        if not wait:
            return None
        time.sleep(
            poll_interval if poll_interval is not None else min(30.0, max(2.0, remaining / 50.0))
        )
    return _settle(ticket, by_name, fallback)

offpeak.status(ticket, *, venues=None)

One look at every open batch on ticket, keyed by venue name.

Read-only: nothing is collected, cancelled or rescued. A venue that errors while being polled reports a failed state carrying the message.

Source code in src/offpeak/ticket.py

def status(ticket: Ticket, *, venues: list[Venue] | None = None) -> dict[str, BatchState]:
    """One look at every open batch on *ticket*, keyed by venue name.

    Read-only: nothing is collected, cancelled or rescued. A venue that
    errors while being polled reports a ``failed`` state carrying the message.
    """
    by_name = _venues_by_name(ticket, venues)
    out: dict[str, BatchState] = {}
    for name, handle in ticket.batches.items():
        venue = by_name.get(name)
        if venue is None:
            out[name] = BatchState(status="failed", raw_status="venue not configured")
            continue
        try:
            out[name] = venue.status(handle)
        except Exception as exc:  # noqa: BLE001 — the provider failed, not us
            out[name] = BatchState(status="failed", raw_status=f"status failed: {exc}")
    return out

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 {_prices.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,
                # less whatever it captured anyway — a clock-priced fallback
                # that ran off-peak paid half and left nothing on the table.
                spread = r.spread_usd or 0.0
                settlement.left_on_table_usd += max(0.0, r.list_usd * (1 - BATCH_DISCOUNT) - spread)
    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
    #: What the venue says this job paid, as a fraction of list — set only by a
    #: venue whose price is decided per request rather than per tier. A batch
    #: venue leaves it ``None`` and the tier rule prices the job: batch if it
    #: landed, list if it fell back. A clock-priced venue (DeepSeek) stamps
    #: 0.5 or 1.0 on each request as it is made, and that stamp outranks the
    #: rule — a fallback that happened to run off-peak paid half, and a hold
    #: that drained into a peak block paid list, whatever the path was called.
    paid_fraction: float | None = None

    @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.paid_fraction is not None:
            listed = self.list_usd
            return None if listed is None else listed * self.paid_fraction
        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)"
        if self.paid_fraction is not None:
            where += f" (paid {self.paid_fraction:g}x list)"
        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)"
    if self.paid_fraction is not None:
        where += f" (paid {self.paid_fraction:g}x list)"
    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.Ticket dataclass

Everything needed to finish a run that was started somewhere else.

Produced by :func:submit, consumed by :func:collect. Serialisable with :meth:to_dict / :meth:from_dict (plain JSON types only), and :meth:save / :meth:load for the one-file case.

Source code in src/offpeak/ticket.py

@dataclass
class Ticket:
    """Everything needed to finish a run that was started somewhere else.

    Produced by :func:`submit`, consumed by :func:`collect`. Serialisable with
    :meth:`to_dict` / :meth:`from_dict` (plain JSON types only), and
    :meth:`save` / :meth:`load` for the one-file case.
    """

    jobs: list[Job]
    deadline: datetime
    submitted_at: datetime
    risk_buffer: float
    #: job id -> venue name (every job has one, even if its submit failed)
    assignment: dict[str, str] = field(default_factory=dict)
    #: venue name -> provider batch handle, for batches still open
    batches: dict[str, str] = field(default_factory=dict)
    #: venue name -> why its batch path died, if it did
    venue_errors: dict[str, str] = field(default_factory=dict)
    #: job id -> result already fetched (a partial collect survives a restart)
    collected: dict[str, Result] = field(default_factory=dict)
    #: job ids rescued by the sync fallback
    fell_back: set[str] = field(default_factory=set)
    version: int = TICKET_VERSION

    # -- state -------------------------------------------------------------

    @property
    def pending(self) -> bool:
        """True while any batch is still open at a venue."""
        return bool(self.batches)

    @property
    def remaining(self) -> float:
        """Seconds until the deadline (negative once it has passed)."""
        return seconds_until(self.deadline)

    def _missing(self) -> list[Job]:
        return [j for j in self.jobs if j.id not in self.collected]

    # -- serialisation ----------------------------------------------------

    def to_dict(self) -> dict:
        return {
            "version": self.version,
            "deadline": self.deadline.isoformat(),
            "submitted_at": self.submitted_at.isoformat(),
            "risk_buffer": self.risk_buffer,
            "jobs": [
                {
                    "id": j.id,
                    "model": j.model,
                    "messages": j.messages,
                    "params": j.params,
                    "metadata": j.metadata,
                    "status": j.status.value,
                }
                for j in self.jobs
            ],
            "assignment": dict(self.assignment),
            "batches": dict(self.batches),
            "venue_errors": dict(self.venue_errors),
            "collected": {
                jid: {"text": r.text, "raw": r.raw, "error": r.error}
                for jid, r in self.collected.items()
            },
            "fell_back": sorted(self.fell_back),
        }

    @classmethod
    def from_dict(cls, data: dict) -> Ticket:
        version = int(data.get("version", 0))
        if version != TICKET_VERSION:
            raise ValueError(f"unsupported ticket version {version} (want {TICKET_VERSION})")
        jobs = [
            Job(
                model=d["model"],
                messages=list(d["messages"]),
                params=dict(d.get("params") or {}),
                id=d["id"],
                metadata=dict(d.get("metadata") or {}),
                status=Status(d.get("status", Status.SUBMITTED.value)),
            )
            for d in data["jobs"]
        ]
        by_id = {j.id: j for j in jobs}
        collected = {
            jid: Result(job=by_id[jid], text=r.get("text"), raw=r.get("raw"), error=r.get("error"))
            for jid, r in (data.get("collected") or {}).items()
            if jid in by_id
        }
        return cls(
            jobs=jobs,
            deadline=datetime.fromisoformat(data["deadline"]),
            submitted_at=datetime.fromisoformat(data["submitted_at"]),
            risk_buffer=float(data["risk_buffer"]),
            assignment=dict(data.get("assignment") or {}),
            batches=dict(data.get("batches") or {}),
            venue_errors=dict(data.get("venue_errors") or {}),
            collected=collected,
            fell_back=set(data.get("fell_back") or []),
            version=version,
        )

    def to_json(self, **kwargs: object) -> str:
        kwargs.setdefault("indent", 2)
        return json.dumps(self.to_dict(), **kwargs)  # type: ignore[arg-type]

    @classmethod
    def from_json(cls, text: str) -> Ticket:
        return cls.from_dict(json.loads(text))

    def save(self, path: str | Path) -> Path:
        """Write the ticket as JSON. Overwrites."""
        p = Path(path)
        p.write_text(self.to_json())
        return p

    @classmethod
    def load(cls, path: str | Path) -> Ticket:
        return cls.from_json(Path(path).read_text())

    def __str__(self) -> str:
        state = "pending" if self.pending else "settled" if not self._missing() else "open"
        left = self.remaining
        when = f"{left / 3600:.1f}h left" if left > 0 else f"{-left / 60:.0f}m past"
        venues = " · ".join(f"{k} {v}" for k, v in sorted(self.batches.items())) or "—"
        return (
            f"OFFPEAK TICKET {state} · {len(self.jobs)} job(s) · "
            f"{len(self.collected)} collected · deadline {self.deadline:%Y-%m-%d %H:%M %Z} "
            f"({when}) · batches {venues}"
        )

pending property

True while any batch is still open at a venue.

remaining property

Seconds until the deadline (negative once it has passed).

save(path)

Write the ticket as JSON. Overwrites.

Source code in src/offpeak/ticket.py

def save(self, path: str | Path) -> Path:
    """Write the ticket as JSON. Overwrites."""
    p = Path(path)
    p.write_text(self.to_json())
    return p

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
    #: What the venue says this job paid, as a fraction of list — set only by a
    #: venue whose price is decided per request rather than per tier. A batch
    #: venue leaves it ``None`` and the tier rule prices the job: batch if it
    #: landed, list if it fell back. A clock-priced venue (DeepSeek) stamps
    #: 0.5 or 1.0 on each request as it is made, and that stamp outranks the
    #: rule — a fallback that happened to run off-peak paid half, and a hold
    #: that drained into a peak block paid list, whatever the path was called.
    paid_fraction: float | None = None

    @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.paid_fraction is not None:
            listed = self.list_usd
            return None if listed is None else listed * self.paid_fraction
        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)"
        if self.paid_fraction is not None:
            where += f" (paid {self.paid_fraction:g}x list)"
        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)"
    if self.paid_fraction is not None:
        where += f" (paid {self.paid_fraction:g}x list)"
    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 {_prices.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 {_prices.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
    #: The provider's own status word, unmapped ("expired" survives here even
    #: though it maps to "failed" for run()'s purposes). None when the driver
    #: has nothing beyond the mapped status.
    raw_status: str | None = None
    #: When the provider says the batch finished (ISO 8601, UTC), if it says.
    #: A poller that checks once a day still learns the true completion time
    #: from this field; without it, resolution is bounded by the check times.
    completed_at_utc: str | None = None
    #: When the provider says it accepted the batch (ISO 8601, UTC), if it says.
    #: Turnaround measured between two provider stamps is the venue's own
    #: elapsed time; measured against our submit() clock it also carries
    #: whatever gap sits between us and them. Both stamps, or neither.
    created_at_utc: str | None = None

    @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.

Anthropic and OpenAI only. Every other venue in the tree — Groq, Mistral, Gemini, DeepSeek, Qwen — is opt-in: it wants its own key and its own extra, and a model name should not start costing money at a venue nobody asked for. Pass them explicitly::

from offpeak.venues import DeepSeekClock, QwenBatch
offpeak.run(jobs, "06:00", venues=[DeepSeekClock(), QwenBatch()])

Source code in src/offpeak/client.py

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

    Anthropic and OpenAI only. Every other venue in the tree — Groq, Mistral,
    Gemini, DeepSeek, Qwen — is opt-in: it wants its own key and its own
    extra, and a model name should not start costing money at a venue nobody
    asked for. Pass them explicitly::

        from offpeak.venues import DeepSeekClock, QwenBatch
        offpeak.run(jobs, "06:00", venues=[DeepSeekClock(), QwenBatch()])
    """
    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, Google, Groq, Mistral and Alibaba 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. Anthropic publishes a fast tier too, on Claude Opus 5 and Opus 4.8, also at twice list. 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.

Two venues price the same 2.0x spread on a different axis. DeepSeek publishes no batch tier; it publishes a clock, with peak hours on weekdays and half price everywhere else. The sheet stores its peak rate as the standard row, so BATCH_DISCOUNT reproduces the off-peak rate exactly — but the lane is a clock and not a queue, and :func:lane_for says which. See :mod:offpeak.venues.deepseek_clock.

Corrections

2026-08-30 — two venues added; nothing already on the sheet moved.

DeepSeek, from the rendered page at api-docs.deepseek.com/quick_start/pricing (read 2026-08-28, re-read 2026-08-30 for the per-model columns). The standard row is the peak rate — $0.44 / $1.32 on deepseek-v4-flash, $1.32 / $3.96 on deepseek-v4-pro — and the page's own words are "off-peak rates are half of the peak rates", so the batch rule gives $0.22 / $0.66 and $0.66 / $1.98, which is what the page prints. Cache-hit input ($0.007 / $0.022 off-peak) is on the page and not on the sheet: there is no cache dimension here, every input token settles at the miss rate, and a cache-heavy run overstates. The lane is "clock" in :data:_LANES, since DeepSeek has no batch API and the discount is decided per request by the wall clock.

Qwen, from alibabacloud.com/help/en/model-studio/model-pricing, international (Singapore) region: qwen3.7-max at $2.50 / $7.50 (first read 2026-08-21, confirmed 2026-08-30) and qwen3.8-max at $2.00 / $6.00 (read 2026-08-30). Batch is 50% on both, per the same page and the batch-interface docs, so the rule covers it. The page marks qwen3.7-max's rate "Limited-time 50% off" without a date it runs through, and a :class:PromoNote needs one — so there is no note, and a reader of this sheet should know the number may step up unannounced. The Beijing region is priced separately and in its own currency; it is not on the sheet.

2026-08-28 — two rows that had been wrong since before the sheet watch took its first reading, so no hash diff could have found them; tools/sheet_reconcile.py did, by reading the committed page text against this table.

Fast mode is no longer an OpenAI-only tier. Anthropic publishes one on Claude Opus 5 and Opus 4.8 at $10 / $50 per 1M — research preview, first-party Claude API only, and explicitly not available with the Batch API. The rows are in :data:_FAST_PRICES; :func:urgency_spread therefore answers 4.0 for those two models where it previously answered None. Nothing that was priced before this date changes: a fast row is an addition, and no standard or batch number moved.

Claude Sonnet 5's $2 / $10 is now the standard price. It shipped as introductory pricing through 2026-08-31, with a scheduled step up to $3 / $15 on 2026-09-01; Anthropic has cancelled that increase. The numbers here were already right — Sonnet 5 never carried a :class:PromoNote, so no quote ever promised the step-up — but prose elsewhere that called the rate introductory was describing a decay that will not happen, and has been removed.

2026-08-23 — the Mistral and Google blocks were read off mistral.ai/pricing/api and ai.google.dev/pricing on this date, and the snapshot date moved with them. The Anthropic, OpenAI and Groq blocks are carried forward unchanged from the 2026-08-21 reading; the date on the sheet is when it was last touched, not a claim that every row was re-verified.

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

SheetLoad dataclass

What :func:load_sheet did — reported, never assumed.

Source code in src/offpeak/prices.py

@dataclass(frozen=True)
class SheetLoad:
    """What :func:`load_sheet` did — reported, never assumed."""

    sheet_date: str
    source: str
    models: int
    added: int
    changed: int
    unchanged: int
    fast_models: int
    promo_notes: int

    def __str__(self) -> str:
        return (
            f"loaded price sheet {self.sheet_date} from {self.source}: "
            f"{self.models} model(s) — {self.added} new, {self.changed} changed, "
            f"{self.unchanged} unchanged; {self.fast_models} fast row(s), "
            f"{self.promo_notes} promo note(s)"
        )

sheet_date()

The date of the sheet currently in force.

:data:PRICE_SHEET_DATE is the sheet this release bundles and never moves. This is what is actually pricing jobs right now, which is the figure a quote or a receipt should print.

Source code in src/offpeak/prices.py

def sheet_date() -> str:
    """The date of the sheet currently in force.

    :data:`PRICE_SHEET_DATE` is the sheet this *release* bundles and never
    moves. This is what is actually pricing jobs right now, which is the figure
    a quote or a receipt should print.
    """
    return _SHEET_DATE

export_sheet()

The sheet in force, as the published wire format.

This is the whole publishing story: the sheet is data, so it serializes. No service, no database — a dated JSON file that anyone can fetch, diff, pin, or check against the provider pages named in sources.

Source code in src/offpeak/prices.py

def export_sheet() -> dict:
    """The sheet in force, as the published wire format.

    This is the whole publishing story: the sheet is data, so it serializes.
    No service, no database — a dated JSON file that anyone can fetch, diff,
    pin, or check against the provider pages named in ``sources``.
    """
    return {
        "schema": SHEET_SCHEMA,
        "sheet_date": _SHEET_DATE,
        "generated_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "batch_discount": BATCH_DISCOUNT,
        "prices": {
            model: {"input_per_m": rates[0], "output_per_m": rates[1]}
            for model, rates in sorted(_PRICES.items())
        },
        "fast_prices": {
            model: {"input_per_m": rates[0], "output_per_m": rates[1]}
            for model, rates in sorted(_FAST_PRICES.items())
        },
        "promo_notes": {
            model: {
                "through": note.through,
                "post_promo": list(note.post_promo),
                "source": note.source,
                "note": note.note,
            }
            for model, note in sorted(PROMO_NOTES.items())
        },
        # Additive. A reader of schema /1 that predates this key ignores it —
        # every key it does know keeps its shape — so the major does not move.
        "lanes": {prefix: lane for prefix, lane in sorted(_LANES.items())},
    }

load_sheet(source, *, replace=False)

Load a published price sheet over the bundled one. Opt in, always.

source is an https:// URL, a filesystem path, or an already-parsed dict. Nothing in the library calls this for you: the default sheet is the one this release shipped with, so offpeak keeps working offline and a receipt settled today can still be checked next year against the numbers that settled it.

replace=True clears the table first, so the loaded sheet is the whole truth and a model it omits resolves to None. The default merges, which keeps any :func:register_price overrides and older models you still run.

A sheet declaring a different batch_discount than this release is refused rather than applied. The discount is a rule the venues publish identically, not a row — and client and quote bound their copy of it at import, so honouring it here would price some arithmetic at the new rate and some at the old. That is a release, not a download.

Source code in src/offpeak/prices.py

def load_sheet(source: str | Path | dict, *, replace: bool = False) -> SheetLoad:
    """Load a published price sheet over the bundled one. **Opt in, always.**

    *source* is an ``https://`` URL, a filesystem path, or an already-parsed
    dict. Nothing in the library calls this for you: the default sheet is the
    one this release shipped with, so ``offpeak`` keeps working offline and a
    receipt settled today can still be checked next year against the numbers
    that settled it.

    ``replace=True`` clears the table first, so the loaded sheet is the whole
    truth and a model it omits resolves to ``None``. The default merges, which
    keeps any :func:`register_price` overrides and older models you still run.

    A sheet declaring a different ``batch_discount`` than this release is
    **refused** rather than applied. The discount is a rule the venues publish
    identically, not a row — and ``client`` and ``quote`` bound their copy of it
    at import, so honouring it here would price some arithmetic at the new rate
    and some at the old. That is a release, not a download.
    """
    global _SHEET_DATE, _SHEET_SOURCE

    if isinstance(source, dict):
        document, origin = source, "<dict>"
    else:
        document, origin = _read_source(source)

    schema = str(document.get("schema", ""))
    family, _, major = schema.partition("/")
    if family != SHEET_SCHEMA.partition("/")[0] or major != SHEET_SCHEMA.rpartition("/")[2]:
        raise ValueError(
            f"unsupported price-sheet schema {schema!r}; this build reads {SHEET_SCHEMA}"
        )

    date = document.get("sheet_date")
    if not date:
        raise ValueError("price sheet has no sheet_date — a sheet with no date is not checkable")

    declared = document.get("batch_discount", BATCH_DISCOUNT)
    if float(declared) != BATCH_DISCOUNT:
        raise ValueError(
            f"price sheet declares batch_discount {declared}, this build applies "
            f"{BATCH_DISCOUNT}. The discount is a published rule rather than a row; "
            "upgrade offpeak rather than loading a sheet that disagrees with it"
        )

    rows = document.get("prices") or {}
    if not rows:
        raise ValueError("price sheet carries no prices")

    parsed: dict[str, tuple[float, float]] = {}
    for model, rates in rows.items():
        try:
            parsed[str(model)] = (float(rates["input_per_m"]), float(rates["output_per_m"]))
        except (KeyError, TypeError, ValueError) as exc:
            raise ValueError(f"price sheet row {model!r} is not a pair of rates: {exc}") from None

    before = dict(_PRICES)
    added = sum(1 for m in parsed if m not in before)
    changed = sum(1 for m, r in parsed.items() if m in before and before[m] != r)
    unchanged = sum(1 for m, r in parsed.items() if m in before and before[m] == r)

    if replace:
        _PRICES.clear()
    _PRICES.update(parsed)

    fast = document.get("fast_prices") or {}
    if replace:
        _FAST_PRICES.clear()
    for model, rates in fast.items():
        _FAST_PRICES[str(model)] = (float(rates["input_per_m"]), float(rates["output_per_m"]))

    notes = document.get("promo_notes") or {}
    if replace:
        PROMO_NOTES.clear()
    for model, note in notes.items():
        PROMO_NOTES[str(model)] = PromoNote(
            through=str(note["through"]),
            post_promo=(float(note["post_promo"][0]), float(note["post_promo"][1])),
            source=str(note.get("source", "")),
            note=str(note.get("note", "")),
        )

    # Lanes are a fact about how a venue sells its discount, not a rate, so a
    # sheet that says nothing about them retracts nothing: ``replace`` clears
    # the rate tables and leaves the lane table alone unless the document
    # carries one of its own.
    lanes = document.get("lanes")
    if isinstance(lanes, dict):
        if replace:
            _LANES.clear()
        for prefix, lane in lanes.items():
            if lane not in ("batch", "clock"):
                raise ValueError(f"price sheet lane {prefix!r} is {lane!r}, not batch or clock")
            _LANES[str(prefix)] = str(lane)

    _SHEET_DATE = str(date)
    _SHEET_SOURCE = origin
    return SheetLoad(
        sheet_date=_SHEET_DATE,
        source=origin,
        models=len(parsed),
        added=added,
        changed=changed,
        unchanged=unchanged,
        fast_models=len(fast),
        promo_notes=len(notes),
    )

reset_sheet()

Put the release's own bundled sheet back. Returns its date.

Source code in src/offpeak/prices.py

def reset_sheet() -> str:
    """Put the release's own bundled sheet back. Returns its date."""
    global _SHEET_DATE, _SHEET_SOURCE
    _PRICES.clear()
    _PRICES.update(_BUNDLED_PRICES)
    _FAST_PRICES.clear()
    _FAST_PRICES.update(_BUNDLED_FAST)
    PROMO_NOTES.clear()
    PROMO_NOTES.update(_BUNDLED_PROMO)
    _LANES.clear()
    _LANES.update(_BUNDLED_LANES)
    _SHEET_DATE = PRICE_SHEET_DATE
    _SHEET_SOURCE = "bundled"
    return _SHEET_DATE

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 today is everywhere except OpenAI's gpt-5.6 family and Anthropic's Opus 5 / Opus 4.8. 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 today is everywhere
    except OpenAI's gpt-5.6 family and Anthropic's Opus 5 / Opus 4.8. 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)

lane_for(model)

How model's venue sells its discount: "batch" or "clock".

"batch" for every priced model that has no lane row — the default, since it is what every venue but one publishes. "clock" for a venue whose half price is a function of when the request is made rather than how long it may wait. None for a model that is not on the sheet: a lane for a rate nobody published is not information.

Source code in src/offpeak/prices.py

def lane_for(model: str) -> str | None:
    """How *model*'s venue sells its discount: ``"batch"`` or ``"clock"``.

    ``"batch"`` for every priced model that has no lane row — the default,
    since it is what every venue but one publishes. ``"clock"`` for a venue
    whose half price is a function of when the request is made rather than
    how long it may wait. ``None`` for a model that is not on the sheet: a
    lane for a rate nobody published is not information.
    """
    if get_price(model) is None:
        return None
    return _lookup(_LANES, model) or "batch"

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; on claude-opus-5, $10/$50 against $2.50/$12.50, the same 4x at a different venue.

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; on claude-opus-5, $10/$50 against
    $2.50/$12.50, the same 4x at a different venue.

    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}"