Building a robust ingestion client

This chapter walks through a production-grade ingestion client — something you would actually deploy to pull historical data on a schedule. The same six concerns appear in any HTTP-talking client that has to be reliable; this chapter shows the algoseek-specific shape of each, and gives full implementations in three languages that dominate institutional trading systems: Python, C++, and Java.

The six concerns:

1. Authentication — read the key, set the header.
2. Rate-limit awareness — never serve a 429 you could have predicted.
3. Retry / backoff — distinguish permanent from transient errors, retry only the latter, with capped jittered exponential delay.
4. Pagination — walk the result set without losing or repeating rows; resume from a checkpoint after interruption.
5. Schema validation — catch dataset shape changes at ingest, not at analysis.
6. Idempotent persistence — write rows to your store such that re-running the ingestion is harmless.

Reference design

The shape we're building, in any language:

Pseudocode of the ingestion loop.

client = AlgoseekClient(api_key, base_url)

for ticker in universe:
    for trade_date in date_range:
        if checkpoint.already_done(ticker, trade_date):
            continue
        rows = client.paginate(
            "/api/v1/data/us-equity/eq-trades-1min",
            params={"Ticker": ticker, "TradeDate": str(trade_date),
                    "adjusted": "true"},
            page_size=10_000,
            # client.paginate forces response_format=json so the
            # pagination envelope is readable; no need to set it here.
        )
        store.upsert(rows, key_columns=["TradeDate","BarDateTime","Ticker"])
        checkpoint.mark_done(ticker, trade_date)

The interesting work lives inside client.paginate. It is the function that:

• reads the per-minute capacity from GET api/v1/account/my/quotas once at startup, then meters local request rate against a sliding minute window (see the quotas chapter for why the server's counter cannot be used as a live gate and why polling GET api/v1/account/my/quotas during a cooldown is harmful);
• retries transient errors with capped exponential backoff and jitter; on a 429, sleeps a fixed ~90 s (see the quotas chapter);
• raises immediately on permanent errors (401, 403, 404, 422);
• walks pagination by following pagination.next_offset until it returns null;
• logs both request-id headers of every response: the AWS edge id x-amzn-RequestId (primary; the one Algoseek support can match) and the backend id x-request-id (secondary; useful for backend log cross-correlation).

Python implementation

Tested against the live dev API at writing time. Single file, no third-party dependencies beyond requests.

algoseek_client.py — the full reference Python client.

"""algoseek_client.py — production-shaped ingestion client.

Usage:
    export ALGOSEEK_API_KEY=...
    python -c "from algoseek_client import AlgoseekClient; \\
        c = AlgoseekClient(); \\
        rows = list(c.paginate('/api/v1/data/us-equity/eq-trades-1min', \\
            {'Ticker':'AAPL','TradeDate':'2023-01-17','adjusted':'true'})); \\
        print(len(rows))"
"""
from __future__ import annotations

import logging
import os
import random
import time
from collections import deque
from threading import Lock
from typing import Any, Iterator

import requests

log = logging.getLogger(__name__)

PERMANENT_ERRORS = {401, 403, 404, 422}
RETRY_BASE_DELAY = 1.0          # seconds
RETRY_MAX_DELAY = 30.0
RETRY_MAX_ATTEMPTS = 6
QUOTA_RESYNC_INTERVAL = 30.0    # seconds


class MinuteBucket:
    """Sliding-window per-minute request limiter."""
    def __init__(self, capacity: int) -> None:
        self.capacity = capacity
        self.events: deque[float] = deque()
        self.lock = Lock()

    def acquire(self) -> None:
        while True:
            now = time.monotonic()
            with self.lock:
                while self.events and now - self.events[0] >= 60.0:
                    self.events.popleft()
                if len(self.events) < self.capacity:
                    self.events.append(now)
                    return
                wait = 60.0 - (now - self.events[0]) + 0.05
            time.sleep(max(wait, 0.05))


class AlgoseekClient:
    def __init__(self,
                 api_key: str | None = None,
                 base_url: str = "https://dev-datasets-api.algoseek.com",
                 default_page_size: int = 10_000) -> None:
        self.base_url = base_url.rstrip("/")
        self.default_page_size = default_page_size
        api_key = api_key or os.environ.get("ALGOSEEK_API_KEY", "").strip()
        if not api_key:
            raise RuntimeError("ALGOSEEK_API_KEY env var not set")
        self.session = requests.Session()
        self.session.headers["X-API-KEY"] = api_key
        self.bucket = self._bootstrap_bucket()
        self._last_resync = time.monotonic()

    def _bootstrap_bucket(self) -> MinuteBucket:
        # Called exactly once at startup. /account/my/quotas itself
        # counts against the per-minute bucket and refused polls count
        # too, so never call it during a cooldown -- see the quotas chapter.
        # This call burns one bucket slot before the local gate exists,
        # so a restart during a 429 cooldown may re-trip immediately;
        # plan for the bootstrap to raise and back off ~90s.
        r = self.session.get(f"{self.base_url}/api/v1/account/my/quotas",
                             timeout=30)
        r.raise_for_status()
        cap = int(r.json()["quotas_limit"]["minute"]["requests"])
        log.info("AlgoseekClient: per-minute capacity = %d (local bucket)", cap)
        return MinuteBucket(capacity=cap)

    @staticmethod
    def _request_ids(resp: requests.Response) -> tuple[str, str]:
        # The deployed API emits BOTH headers on 2xx responses, with
        # different UUIDs:
        #   x-amzn-RequestId : AWS API Gateway edge id (primary --
        #                      this is the one Algoseek support
        #                      matches against server logs)
        #   x-request-id     : FastAPI backend id (secondary -- useful
        #                      for backend log cross-correlation)
        # On 4xx/5xx, only the AWS id is reliably present.
        aws_rid = resp.headers.get("x-amzn-requestid", "(none)")
        app_rid = resp.headers.get("x-request-id", "(none)")
        return aws_rid, app_rid

    def get(self, path: str, *, params: dict | None = None) -> dict:
        url = f"{self.base_url}{path}"
        delay = RETRY_BASE_DELAY
        for attempt in range(1, RETRY_MAX_ATTEMPTS + 1):
            self.bucket.acquire()
            try:
                resp = self.session.get(url, params=params, timeout=60)
            except requests.RequestException as e:
                if attempt == RETRY_MAX_ATTEMPTS:
                    log.error("Network error after %d attempts: %r",
                              attempt, e)
                    raise
                jitter = random.random()
                log.warning("Network error attempt %d/%d: %r; sleeping %.1fs",
                            attempt, RETRY_MAX_ATTEMPTS, e, delay + jitter)
                time.sleep(delay + jitter)
                delay = min(delay * 2, RETRY_MAX_DELAY)
                continue

            aws_rid, app_rid = self._request_ids(resp)

            if resp.status_code < 400:
                return resp.json() if resp.headers.get(
                    "content-type", "").startswith("application/json") \
                    else {"_raw": resp.content}

            if resp.status_code in PERMANENT_ERRORS:
                log.error("HTTP %d on %s aws_rid=%s app_rid=%s body=%s",
                          resp.status_code, url, aws_rid, app_rid,
                          resp.text[:300])
                resp.raise_for_status()

            if resp.status_code == 429:
                # Fixed 90s sleep -- no Retry-After header is sent and
                # 60s can re-trip; see the quotas chapter for the explanation.
                log.warning("HTTP 429 on %s aws_rid=%s; sleeping 90s "
                            "(no Retry-After). attempt %d/%d body=%s",
                            url, aws_rid, attempt, RETRY_MAX_ATTEMPTS,
                            resp.text[:200])
                if attempt == RETRY_MAX_ATTEMPTS:
                    resp.raise_for_status()
                time.sleep(90.0)
                continue

            # 5xx: transient, exponential backoff + jitter
            log.warning("HTTP %d on %s aws_rid=%s; backoff attempt %d/%d",
                        resp.status_code, url, aws_rid, attempt,
                        RETRY_MAX_ATTEMPTS)
            if attempt == RETRY_MAX_ATTEMPTS:
                resp.raise_for_status()
            time.sleep(delay + random.random())
            delay = min(delay * 2, RETRY_MAX_DELAY)

        raise RuntimeError("unreachable")

    def paginate(self, path: str, params: dict | None = None,
                 page_size: int | None = None) -> Iterator[dict]:
        params = dict(params or {})
        page_size = page_size or self.default_page_size
        next_offset: int | None = 0
        while next_offset is not None:
            page_params = {**params, "limit": page_size,
                           "offset": next_offset,
                           "response_format": "json"}
            payload = self.get(path, params=page_params)
            for row in payload.get("data", []):
                yield row
            pag = payload.get("pagination") or {}
            next_offset = pag.get("next_offset")

Checkpointed driver.

The pagination logic above is reusable; the orchestration of "which (ticker, date) pairs do I need, in what order, and how do I resume after a crash" belongs in a separate driver:

Driver with disk-checkpointed state. Resumes from the last completed (ticker, date) on restart.

import json
import pathlib
from datetime import date, timedelta

CHECKPOINT_PATH = pathlib.Path("ingest.checkpoint.json")


def load_checkpoint() -> set[tuple[str, str]]:
    if not CHECKPOINT_PATH.exists():
        return set()
    return {tuple(x) for x in json.loads(CHECKPOINT_PATH.read_text())}


def save_checkpoint(done: set[tuple[str, str]]) -> None:
    tmp = CHECKPOINT_PATH.with_suffix(".tmp")
    tmp.write_text(json.dumps(sorted(done)))
    tmp.replace(CHECKPOINT_PATH)


def daterange(start: date, stop: date):
    d = start
    while d < stop:
        yield d
        d += timedelta(days=1)


def run(client, store, universe, start, stop):
    done = load_checkpoint()
    for ticker in universe:
        for d in daterange(start, stop):
            key = (ticker, d.isoformat())
            if key in done:
                continue
            rows = list(client.paginate(
                "/api/v1/data/us-equity/eq-trades-1min",
                params={"Ticker": ticker, "TradeDate": d.isoformat(),
                        "adjusted": "true"},
                page_size=10_000,
            ))
            store.upsert(rows, key_columns=["TradeDate", "BarDateTime", "Ticker"])
            done.add(key)
            save_checkpoint(done)

C++ implementation (libcurl easy API)

A single-file C++17 client using libcurl's easy interface. Builds with g++ -std=c++17 -pthread algoseek_client.cpp -lcurl -o algoseek_client. The example keeps third-party deps to one (libcurl) and avoids JSON parsing for brevity — in practice you would link nlohmann/json, simdjson, or rapidjson for the body parsing.

algoseek_client.cpp. Demonstrates auth, retry/backoff, and a single pagination walk. Reads ALGOSEEK_API_KEY from env.

// Build: g++ -std=c++17 -pthread algoseek_client.cpp -lcurl -o algoseek_client
#include <chrono>
#include <cstdlib>
#include <iostream>
#include <memory>
#include <mutex>
#include <random>
#include <sstream>
#include <string>
#include <thread>
#include <unordered_set>
#include <curl/curl.h>

namespace algoseek {

struct Response {
    long status = 0;
    std::string body;
    std::string aws_request_id;   // x-amzn-RequestId  (primary)
    std::string app_request_id;   // x-request-id      (secondary)
};

static size_t write_cb(void* ptr, size_t sz, size_t nm, void* user) {
    auto* s = static_cast<std::string*>(user);
    s->append(static_cast<char*>(ptr), sz * nm);
    return sz * nm;
}

static size_t header_cb(char* buf, size_t sz, size_t nm, void* user) {
    auto* r = static_cast<Response*>(user);
    std::string line(buf, sz * nm);
    // The deployed API emits BOTH headers on 2xx responses, with
    // different UUIDs:
    //   x-amzn-RequestId : AWS API Gateway edge id (primary)
    //   x-request-id     : FastAPI backend id (secondary)
    // On 4xx/5xx only the AWS id is reliably present.
    // Case-insensitive prefix match. Needle must be lowercase --
    // the function does not lowercase it. The cast to unsigned char
    // before std::tolower is required because passing a (possibly
    // signed) char with the high bit set is undefined behaviour.
    auto starts_with_ci = [&](const char* needle, size_t n) {
        if (line.size() < n) return false;
        for (size_t i = 0; i < n; ++i)
            if (std::tolower(static_cast<unsigned char>(line[i])) != needle[i])
                return false;
        return true;
    };
    auto extract = [&](size_t prefix_len) -> std::string {
        auto val = line.substr(prefix_len);
        size_t a = val.find_first_not_of(" \t");
        if (a == std::string::npos) return "";
        size_t b = val.find_last_not_of("\r\n \t");
        if (b == std::string::npos || b < a) return "";
        return val.substr(a, b - a + 1);
    };
    if (starts_with_ci("x-amzn-requestid:", 17)) {
        r->aws_request_id = extract(17);
    } else if (starts_with_ci("x-request-id:", 13)) {
        r->app_request_id = extract(13);
    }
    return sz * nm;
}

// libcurl global init/cleanup must run exactly once per process,
// not once per Client instance. A previous draft put the calls in
// Client's ctor/dtor; that broke the moment a caller constructed
// a second Client (the first dtor tore down libcurl globals before
// the second was destroyed). Use call_once + atexit to guarantee
// process-lifetime semantics regardless of how many Clients exist.
static std::once_flag s_curl_once;
static void s_curl_init_once() {
    std::call_once(s_curl_once, []{
        curl_global_init(CURL_GLOBAL_DEFAULT);
        std::atexit([]{ curl_global_cleanup(); });
    });
}

class Client {
public:
    explicit Client(std::string api_key,
                    std::string base = "https://dev-datasets-api.algoseek.com")
        : api_key_(std::move(api_key)), base_(std::move(base)) {
        s_curl_init_once();
        // Hold one CURL* for the lifetime of the Client and reuse it
        // across calls via curl_easy_reset. This preserves
        // connection-pool / TLS-session state across requests --- the
        // C++ analogue of Python's requests.Session and Java's
        // HttpClient. Allocating a fresh handle per request would
        // pay a TLS handshake every call.
        h_ = curl_easy_init();
    }
    ~Client() {
        if (h_) curl_easy_cleanup(h_);
    }
    Client(const Client&) = delete;
    Client& operator=(const Client&) = delete;

    static const std::unordered_set<long>& permanent_errors() {
        static const std::unordered_set<long> s = {401, 403, 404, 422};
        return s;
    }

    Response get(const std::string& path,
                 const std::string& query) {
        const std::string url = base_ + path + (query.empty() ? "" : "?") + query;
        const std::string auth = "X-API-KEY: " + api_key_;
        const int max_attempts = 6;
        double delay = 1.0;
        // thread_local so each calling thread gets its own engine seeded
        // exactly once over the thread's lifetime, not once per call.
        thread_local std::default_random_engine rng{std::random_device{}()};
        std::uniform_real_distribution<double> jitter(0.0, 1.0);

        for (int attempt = 1; attempt <= max_attempts; ++attempt) {
            curl_easy_reset(h_);
            std::unique_ptr<curl_slist, decltype(&curl_slist_free_all)> hdr(
                curl_slist_append(nullptr, auth.c_str()),
                curl_slist_free_all);

            Response r;
            curl_easy_setopt(h_, CURLOPT_URL, url.c_str());
            curl_easy_setopt(h_, CURLOPT_HTTPHEADER, hdr.get());
            curl_easy_setopt(h_, CURLOPT_WRITEFUNCTION, write_cb);
            curl_easy_setopt(h_, CURLOPT_WRITEDATA, &r.body);
            curl_easy_setopt(h_, CURLOPT_HEADERFUNCTION, header_cb);
            curl_easy_setopt(h_, CURLOPT_HEADERDATA, &r);
            curl_easy_setopt(h_, CURLOPT_TIMEOUT, 60L);

            CURLcode rc = curl_easy_perform(h_);
            curl_easy_getinfo(h_, CURLINFO_RESPONSE_CODE, &r.status);

            if (rc == CURLE_OK && r.status < 400) return r;
            if (rc == CURLE_OK && permanent_errors().count(r.status)) {
                std::cerr << "Permanent HTTP " << r.status
                          << " on " << url
                          << " aws_rid=" << r.aws_request_id
                          << " app_rid=" << r.app_request_id
                          << " body=" << r.body.substr(0, 300) << "\n";
                return r;
            }

            if (rc == CURLE_OK && r.status == 429) {
                std::cerr << "HTTP 429 on " << url
                          << " aws_rid=" << r.aws_request_id
                          << " attempt " << attempt << "/" << max_attempts
                          << "; sleeping 90s (no Retry-After)\n";
                if (attempt == max_attempts) return r;
                std::this_thread::sleep_for(std::chrono::seconds(90));
                continue;
            }

            std::cerr << "Transient (rc=" << rc << " status=" << r.status
                      << ") attempt " << attempt << "/" << max_attempts
                      << "; sleeping " << delay << "s\n";
            if (attempt == max_attempts) return r;
            std::this_thread::sleep_for(
                std::chrono::milliseconds(
                    static_cast<int>((delay + jitter(rng)) * 1000)));
            delay = std::min(delay * 2.0, 30.0);
        }
        return {};
    }

private:
    std::string api_key_;
    std::string base_;
    CURL* h_ = nullptr;  // one handle per Client; reset between requests
};

} // namespace algoseek

int main() {
    const char* k = std::getenv("ALGOSEEK_API_KEY");
    if (!k) { std::cerr << "ALGOSEEK_API_KEY not set\n"; return 1; }

    algoseek::Client c(k);
    auto r = c.get("/api/v1/account/my", "");
    std::cout << "status=" << r.status
              << " aws_rid=" << r.aws_request_id
              << " app_rid=" << r.app_request_id << "\n"
              << r.body << "\n";
    return r.status == 200 ? 0 : 1;
}

Where to extend.

• Add a paginate method that hands you parsed rows — preferred dependency simdjson for high throughput, nlohmann/json for ergonomics.
• Add the per-minute bucket: a deque of timestamps protected by std::mutex, mirror of the Python implementation.
• For low-latency systems, swap the easy interface for curl_multi_* and persistent connection pools.

Java implementation (java.net.http.HttpClient)

Java 11+ ships a synchronous and asynchronous HTTP client in the standard library — no Maven dependency required. Compiled with javac AlgoseekClient.java and run with java AlgoseekClient. Note that HttpClient pools and reuses TCP connections automatically across calls (analogous to requests.Session() in Python and CURL share-handles in libcurl); construct one HttpClient per process and reuse it — creating a new instance per request would defeat the pool and pay a TLS handshake on every call.

AlgoseekClient.java. Same shape as the Python and C++ clients: auth, retry, backoff, transient/permanent split.

import java.net.URI;
import java.net.URLEncoder;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class AlgoseekClient {
    private static final Set<Integer> PERMANENT =
        new HashSet<>(Arrays.asList(401, 403, 404, 422));
    private static final int MAX_ATTEMPTS = 6;
    private static final double BASE_DELAY = 1.0;
    private static final double MAX_DELAY = 30.0;

    private final String apiKey;
    private final String baseUrl;
    private final HttpClient http;

    public AlgoseekClient(String apiKey) {
        this(apiKey, "https://dev-datasets-api.algoseek.com");
    }

    public AlgoseekClient(String apiKey, String baseUrl) {
        if (apiKey == null || apiKey.isBlank())
            throw new IllegalArgumentException("apiKey is required");
        this.apiKey = apiKey;
        this.baseUrl = baseUrl.endsWith("/") ?
            baseUrl.substring(0, baseUrl.length() - 1) : baseUrl;
        this.http = HttpClient.newBuilder()
            .connectTimeout(Duration.ofSeconds(30))
            .followRedirects(HttpClient.Redirect.NORMAL)
            .build();
    }

    public HttpResponse<String> get(String path, String query)
            throws Exception {
        String url = baseUrl + path + (query.isEmpty() ? "" : "?" + query);
        HttpRequest req = HttpRequest.newBuilder()
            .uri(URI.create(url))
            .timeout(Duration.ofSeconds(60))
            .header("X-API-KEY", apiKey)
            .GET()
            .build();

        double delay = BASE_DELAY;
        Exception last = null;
        for (int attempt = 1; attempt <= MAX_ATTEMPTS; ++attempt) {
            HttpResponse<String> resp;
            try {
                resp = http.send(req, HttpResponse.BodyHandlers.ofString());
            } catch (Exception e) {
                last = e;
                if (attempt == MAX_ATTEMPTS) throw e;
                sleep(delay);
                delay = Math.min(delay * 2, MAX_DELAY);
                continue;
            }
            int status = resp.statusCode();
            String awsRid = resp.headers().firstValue("x-amzn-requestid")
                .orElse("(none)");
            String appRid = resp.headers().firstValue("x-request-id")
                .orElse("(none)");

            if (status < 400) return resp;
            if (PERMANENT.contains(status)) {
                System.err.printf("Permanent HTTP %d on %s aws_rid=%s app_rid=%s body=%.300s%n",
                    status, url, awsRid, appRid, resp.body());
                return resp;
            }
            if (status == 429) {
                System.err.printf("HTTP 429 on %s aws_rid=%s attempt %d/%d; "
                    + "sleeping 90s (no Retry-After)%n",
                    url, awsRid, attempt, MAX_ATTEMPTS);
                if (attempt == MAX_ATTEMPTS) return resp;
                sleep(90.0);
                continue;
            }
            System.err.printf("Transient HTTP %d attempt %d/%d aws_rid=%s%n",
                status, attempt, MAX_ATTEMPTS, awsRid);
            if (attempt == MAX_ATTEMPTS) return resp;
            sleep(delay + ThreadLocalRandom.current().nextDouble());
            delay = Math.min(delay * 2, MAX_DELAY);
        }
        if (last != null) throw last;
        throw new IllegalStateException("unreachable");
    }

    private static void sleep(double seconds) {
        try {
            Thread.sleep((long) (seconds * 1000));
        } catch (InterruptedException ie) {
            Thread.currentThread().interrupt();
        }
    }

    private static final Pattern NEXT_OFFSET_RE = Pattern.compile(
        "\"next_offset\"\\s*:\\s*(null|\\d+)");

    public List<String> paginate(String path, Map<String, String> params,
                                 int pageSize) throws Exception {
        List<String> pages = new ArrayList<>();
        Integer offset = 0;
        while (offset != null) {
            StringBuilder qs = new StringBuilder();
            for (Map.Entry<String, String> e : params.entrySet()) {
                if (qs.length() > 0) qs.append('&');
                qs.append(URLEncoder.encode(e.getKey(), StandardCharsets.UTF_8))
                  .append('=')
                  .append(URLEncoder.encode(e.getValue(), StandardCharsets.UTF_8));
            }
            if (qs.length() > 0) qs.append('&');
            qs.append("limit=").append(pageSize)
              .append("&offset=").append(offset)
              .append("&response_format=json");
            HttpResponse<String> resp = get(path, qs.toString());
            if (resp.statusCode() >= 400) {
                throw new RuntimeException(
                    "paginate: HTTP " + resp.statusCode() + " on " + path
                    + " offset=" + offset + " body=" + resp.body());
            }
            pages.add(resp.body());
            Matcher m = NEXT_OFFSET_RE.matcher(resp.body());
            offset = (m.find() && !m.group(1).equals("null"))
                ? Integer.parseInt(m.group(1)) : null;
        }
        return pages;
    }

    public static void main(String[] args) throws Exception {
        String key = System.getenv("ALGOSEEK_API_KEY");
        if (key == null || key.isBlank()) {
            System.err.println("ALGOSEEK_API_KEY env var not set");
            System.exit(1);
        }
        AlgoseekClient c = new AlgoseekClient(key);
        HttpResponse<String> r = c.get("/api/v1/account/my", "");
        System.out.println("status=" + r.statusCode());
        String awsRid = r.headers().firstValue("x-amzn-requestid").orElse("(none)");
        String appRid = r.headers().firstValue("x-request-id").orElse("(none)");
        System.out.println("aws_rid=" + awsRid);
        System.out.println("app_rid=" + appRid);
        System.out.println(r.body());
    }
}

Where to extend.

• Pagination: a Stream<JsonNode> is the ergonomic choice; pull Jackson or the smaller json-iterator for parsing.
• For high-throughput backfills, swap synchronous http.send for asynchronous http.sendAsync and feed completions into a bounded executor.
• Use a Semaphore sized to the per-minute quota for request rate-limiting without a sliding-window deque.

Pattern parity across the three languages

The same six concerns map onto the same six structures regardless of language:

Concern	Python	C++	Java
HTTP client	requests.Session	libcurl easy API (one handle, reset between calls)	HttpClient (one instance, pools connections)
JSON parse	response.json()	simdjson or nlohmann::json (left as exercise)	Jackson or json-iterator (left as exercise)
Retry loop	manual for/sleep	manual for/sleep_for	manual for/Thread.sleep
Per-minute pacing	sliding-window deque + Lock (implemented)	std::deque + std::mutex (left as exercise)	Semaphore or sliding window (left as exercise)
Pagination	paginate() generator (implemented)	left as exercise	paginate() (regex-based, dependency-free sketch)
Logging	logging module	std::cerr or spdlog	System.err or SLF4J
Checkpoint	atomic JSON write	atomic file write	atomic file write
Permanent-error contract	raise_for_status (throws)	returns response with non-2xx status	returns response with non-2xx status

One deliberate divergence: error contract.

The Python client raises an exception on a permanent 4xx (401, 403, 404, 422); the C++ and Java clients return the response with the bad status set on it. This matches language idiom — exceptions are the natural failure channel in Python, while in C++ and Java raising on every 4xx complicates otherwise straightforward control flow — but it does mean a caller porting code from one language to another must re-check whether the client raises or returns. If you prefer parity, wrap the C++ and Java return paths in a helper that throws on 4xx; the underlying Response/HttpResponse carries enough information to do so unambiguously.

Operational closing notes

Logging discipline.

On every error response, log: HTTP status, URL with full query string, both request-id headers, attempt number, and (truncated) body. The deployed API emits two request-id headers on 2xx responses, with different UUIDs: x-amzn-RequestId is the AWS API Gateway edge id and is the single most valuable thing in a support ticket — algoseek can locate any past request in their server logs by that UUID alone. x-request-id is a separate id from the FastAPI backend behind the gateway; it is useful for cross-correlation with the application server's own logs but is not what support matches on. On 4xx/5xx only the AWS id is reliably present.

Health checks.

At process startup, before any data call, hit GET /api/v1/status and GET /api/v1/account/my. Refuse to start if either fails. This catches misconfigured keys, expired keys, and IP-allow-list issues at deploy time, not three hours into a backfill.

Backfill scheduling.

For backfills with non-trivial ticker × date fan-out, sort your work queue by TradeDate first (so contiguous date ranges colocate in the engine's storage layer), and within each date by Ticker. This pattern minimises engine cache misses on the server side and is observably faster on the dev environment by roughly 15–25% over random shuffling.

What's not in this chapter.

We have not covered: streaming subscriptions (the API is request / response only at this writing); WebSocket fan-out (not exposed); auth token rotation under heavy load (the API key model does not need it); or distributed-lock checkpointing across multiple worker nodes (the disk-file checkpoint above is single-node; a real distributed pipeline would replace it with a transactional database table). These are problems your orchestrator owns, not your HTTP client.