Building a Worst-Case Cost Preflight for Batch API Jobs

Batch jobs that call a paid API have a familiar failure mode. Someone uploads a big file, the job starts, and only afterward does anyone learn that a tenth of the rows were repeats, a few had no usable address, and a chunk had been looked up last month. The bill, or the quota burn, reflects all of that waste. I work on Locate Next, a skip tracing tool for property lists, and this is the pattern we lean on.

A preflight fixes this by answering one question before any request goes out: what is the most this run could use? This post builds a small one in Python. It counts unique rows, reused rows and rows that cannot be searched, then reports a ceiling in generic units. I use "credits" here, but the same idea works for searches, tokens or API calls.

The examples use property lists, because that is the domain I work in, but nothing below depends on it.

The buckets

The preflight below puts every input row into one of four buckets. A fifth, hold, is discussed later; it sits after the searchability check and before dedupe.

1. Unsearchable. The row is missing something the API needs. Here that means no street or no ZIP code. Sending it would waste a call or return an unusable result.
2. Duplicate. The same target appears earlier in the file, maybe spelled differently ("101 Example Avenue" and "101 Example Ave."). Search it once.
3. Reused. The target was searched recently, inside a reuse window, so the earlier result can be used again.
4. New search. Everything else. These are the only rows that can cost anything.

The order matters. Check searchability first, because a broken row should never be counted as a duplicate of another broken row. Deduplicate next, so a repeated row is not counted as reused twice. Check history last.

Normalizing the key

Duplicates are rarely byte-identical. A light normalization catches the common cases: lowercase, strip punctuation, map a few street suffixes to one form, and cut a ZIP+4 down to its first 5 digits. Real address normalization is a deep topic, and you should use a proper parser in production. In a preflight, a small table goes a long way: missing a duplicate only makes the estimate higher, so err toward merging too little rather than too much.

One caution about the key itself. It should be whatever the API actually searches on. If the API searches by address, key on the address. If it searches by owner plus address, include the owner, or two different owners at one address will be merged into one search.

The code

This is the whole thing, including a made-up sample file and assertions. It uses only the standard library.

import csv, io, re
from datetime import date, timedelta

ABBR = {"avenue": "ave", "street": "st", "road": "rd", "drive": "dr", "lane": "ln"}

def norm_addr(street, zip_code):
    words = re.sub(r"[^\w\s]", "", street.lower()).split()
    words = [ABBR.get(w, w) for w in words]
    zip5 = re.sub(r"\D", "", zip_code)[:5]
    return " ".join(words) + "|" + zip5

def preflight(rows, history, today, reuse_days=30):
    seen, report = set(), {"rows": 0, "duplicate": 0, "unsearchable": 0,
                           "reused": 0, "new_searches": 0}
    for r in rows:
        report["rows"] += 1
        street, zip_code = (r.get("street") or ""), (r.get("zip") or "")
        if not street.strip() or not zip_code.strip():
            report["unsearchable"] += 1
            continue
        key = norm_addr(street, zip_code)
        if key in seen:
            report["duplicate"] += 1
            continue
        seen.add(key)
        last = history.get(key)
        if last and (today - last).days <= reuse_days:  # inclusive window
            report["reused"] += 1
        else:
            report["new_searches"] += 1
    return report

def max_units(report, units_per_search=1):
    # Ceiling: assume every new search is billable
    return report["new_searches"] * units_per_search

SAMPLE = """id,owner,street,zip
1,Rowan Ellis,101 Example Avenue,90001
2,Rowan Ellis,101 Example Ave.,90001-1234
3,Casey Okoro,22 Sample Street,90002
4,Jules Marten,,90003
5,Maya Bennett,7 Test Road,
6,Priya Shah,9 Demo Lane,90004
7,Avery Lin,15 Mock Drive,90005
8,Sam Ortiz,3 Short Row Lane
"""

if __name__ == "__main__":
    today = date(2026, 10, 29)
    history = {norm_addr("22 Sample St", "90002"): today - timedelta(days=12),
               norm_addr("15 Mock Dr", "90005"): today - timedelta(days=45)}
    rows = list(csv.DictReader(io.StringIO(SAMPLE)))
    rep = preflight(rows, history, today)
    print(rep)
    print("most it can use:", max_units(rep), "credits")
    assert rep == {"rows": 8, "duplicate": 1, "unsearchable": 3, "reused": 1, "new_searches": 3}
    assert max_units(rep) == 3
    assert norm_addr("101 Example Ave.", "90001-1234") == norm_addr("101 example avenue", "90001")
    edge = {norm_addr("9 Demo Ln", "90004"): today - timedelta(days=30)}
    assert preflight([{"street": "9 Demo Lane", "zip": "90004"}], edge, today)["reused"] == 1
    print("tests passed")

Running it prints:

{'rows': 8, 'duplicate': 1, 'unsearchable': 3, 'reused': 1, 'new_searches': 3}
most it can use: 3 credits
tests passed

Walk through the sample. Rows 1 and 2 are the same address with different suffix spelling, and row 2 carries a ZIP+4, so row 2 is a duplicate. Row 4 has no street, row 5 has no ZIP, and row 8 is a short row with no ZIP column at all, so all three are unsearchable. Row 3 was searched 12 days ago, inside the 30-day window, so it is reused. Row 7 was searched 45 days ago, outside the window, so it counts as a new search again. The window is inclusive: a row searched exactly 30 days ago is still reused, which the last assertion checks. That leaves three new searches out of eight rows.

Build your history keys with the same norm_addr, or reuse checks will silently miss.

Why report a ceiling, not a guess

max_units assumes every new search is billable. Some APIs only charge when a call returns something useful, so the real number is often lower. It is tempting to multiply by an expected hit rate and show a friendlier figure. Don't do that in the preflight. A hit rate is an assumption about data you have not seen yet. A ceiling is a fact about the file in front of you. Show the ceiling, and if you want, let the person enter their own rate as a separate, clearly labeled estimate.

This is the approach the free skip tracing cost calculator takes. Its definition of an unsearchable row is stricter than the example's: a row counts as having no eligible search address unless it has a complete property address, or is an addressless parcel with an eligible person-owner mailing hint. You enter the rows in your list, the share of duplicates, the share with no eligible search address, and the share searched in the last 30 days, and it works through unique rows, searchable rows and new searches step by step. The share of searches that meet the billing rule is your own input, and setting it to 100% gives the ceiling.

Things to add for production

• A hold state. Some rows are not broken but need a person to decide, such as a company owner rather than a person. Count them separately so they are neither searched nor silently dropped.
• The right dedupe key. Key on whatever the API searches on. If that is owner plus address, two owners at one address are two searches, not one.
• Persist the reasons. Write the bucket name next to each row in the output so people can see why a row was skipped.
• A hard limit. Compare the ceiling with a per-run or per-day limit and refuse to start, or queue the overflow, when it is exceeded.
• Nothing runs until confirmed. The preflight should be read-only. Make starting the job a separate, explicit step.

A preflight costs a few dozen lines and one extra screen. In return, no one learns the cost after the fact, and the rows that would have been wasted are visible before they cost anything.