quantlab.acquisition.massive.client.MassiveClient#

class quantlab.acquisition.massive.client.MassiveClient(*, credentials=None, store=None, http_get=None, streams=16, files=4, part_bytes=33554432, read_timeout=30.0, retries=5, backoff=2.0, sleep=<built-in function sleep>)[source]#

Bases: object

Download Massive’s daily files and condition table into the raw tier.

Parameters:

credentialsMassiveCredentials, optional: Read from the environment when omitted, so a missing variable fails here, before any transfer.
storeObjectStore, optional: The S3 seam; an S3ObjectStore when omitted.
http_getcallable, optional: The REST seam: http_get(url, params=..., headers=...) returns an HttpResponse; rest_transport when omitted.
streamsint, default 16: Byte ranges in flight at once, all of one file; also the S3 connection pool’s size.
filesint, default 4: Files download_days has in flight at once: one being fetched, the others being verified (one core each).
part_bytesint, default 32 MiB: Size of one byte range.
read_timeoutfloat, default 30: Seconds a stream may send nothing before it is given up and its range resumed; passed to the S3ObjectStore made when store is omitted.
retriesint, default 5: Attempts after the first for a transport error; a range whose stream breaks after delivering bytes resumes at once and does not use one up.
backofffloat, default 2.0: Seconds before the first retry, doubled on each later one.
sleepcallable, default time.sleep: How the client waits.

Raises:

MassiveCredentialError: If a credential variable is unset.
ValueError: If streams, files or part_bytes is below 1.

Examples

Needs the credentials and the network:

client = MassiveClient(streams=32, files=4)
client.download_day("trades", date(2024, 11, 29), "/data/quantlab/downloads")

condition_table(download_root)[source]#

Pull the stock condition table over REST and store it as a snapshot.

Every page is followed; the snapshot is the vendor’s records verbatim, named by the pull’s UTC time (quantlab.dataset.massive.raw.conditions_file).

Parameters:

download_rootstr or Path: The download directory; the snapshot lands under its massive/.

Returns:

Path: The snapshot.

Raises:

MassiveEntitlementError: If Massive refused the API key.

Examples

Needs the credentials and the network (94 records on 2026-10-09):

path = client.condition_table("/data/quantlab/downloads")

download_day(data_type, day, download_root)[source]#

Make one data type’s file for one trading day present and verified in the raw tier.

A file already there with the vendor’s size is kept as is.

Parameters:

data_typestr: "trades", "minute_aggs" or "day_aggs".
daydate: The trading day.
download_rootstr or Path: The download directory; the file lands under its massive/.

Returns:

Path: The file in the raw tier.

Raises:

MassiveNotPublishedError: If Massive has no file for the day.
MassiveEntitlementError: If Massive refused the credentials or the day.
MassiveTransportError: If the transfer kept failing, or the file kept failing its checks.

Examples

Needs the credentials and the network; the file lands at /data/quantlab/downloads/massive/trades/2016/2016-11-25.csv.gz, and Thanksgiving (2016-11-24) raises MassiveNotPublishedError:

client.download_day("trades", date(2016, 11, 25), "/data/quantlab/downloads")
client.download_day("trades", date(2016, 11, 24), "/data/quantlab/downloads")

download_days(data_type, start, end, download_root)[source]#

Download one data type’s files over the XNYS sessions of a date range, oldest first.

files files are in flight at once, one being fetched with all streams byte ranges while the others are verified, and the days are yielded in date order as they finish, so a caller converting day d while d+1 downloads keeps only a few files ahead: a day is started only once the caller has moved past the day files before it, so however slow the caller, the day it holds and at most files - 1 after it are on disk. Only XNYS sessions are asked for: weekends and holidays are not. A session Massive has no file for (not published yet) is yielded unpublished and skipped. Each published day moves the data type’s watermark to it (raw.read_watermark) while no earlier day of the run is missing, and a later run starts after the watermark: a day not published yet holds it, so a later run asks for that day again. The watermark says what was downloaded, not what was converted; a caller converting the files keeps its own record.

Parameters:

data_typestr: "trades", "minute_aggs" or "day_aggs".
start, enddate: The first and last day, inclusive; days on or before the watermark are left out.
download_rootstr or Path: The download directory; the files land under its massive/.

Yields:

DayDownload: One per XNYS session after the watermark, in date order.

Raises:

MassiveEntitlementError: If Massive refused the credentials or a day; the run stops, and the watermark stays on the last day before it.
MassiveTransportError: If a transfer kept failing; likewise.

Examples

Needs the credentials and the network; 2016-11-24 (Thanksgiving) is not asked for:

client = MassiveClient()
for done in client.download_days("trades", date(2016, 11, 21), date(2016, 11, 25),
                                 "/data/quantlab/downloads"):
    print(done.day, done.published, done.fetched_bytes)

published_size(data_type, day)[source]#

Return the size of one data type’s file for one trading day, downloading nothing.

One HEAD request, not retried: a caller probing which days the plan serves (the oldest day of its rolling window) tells the answers apart by the exception.

Parameters:

data_typestr: "trades", "minute_aggs" or "day_aggs".
daydate: The trading day.

Returns:

int: The file’s size in bytes.

Raises:

MassiveNotPublishedError: If Massive has no file for the day.
MassiveEntitlementError: If Massive refused the credentials or the day (outside the plan’s window).
MassiveTransportError: If the request failed.

Examples

Needs the credentials and the network:

client.published_size("trades", date(2016, 11, 25))  # 217911421

pull(endpoint, download_root, *, end=None, today=None)[source]#

Pull one REST endpoint’s new rows into the raw tier.

The first page asks for <date_field>.gte the day after the endpoint’s watermark less its overlap_days (earliest without one), and .lte end when given, sorted by the date field ascending; every next_url is followed with no other parameter (its cursor carries the filters), after any apiKey in it is dropped. Each page is written as the vendor returned it (raw.rest_page_file), and the watermark moves after each one: to the day before its last row’s day while more pages follow (that day may continue on the next page), and to the last day asked for once the last page is in. The watermark never passes yesterday in New York, so today’s and future rows (a split announced ahead) are asked for again by the next pull, and it never moves back. A snapshot endpoint (date_field None) is pulled whole and keeps no watermark.

Parameters:

endpointRestEndpoint or str: The endpoint, or its name in rest.ENDPOINTS.
download_rootstr or Path: The download directory; pages land under its massive/rest/.
enddate, optional: The last day asked for; no upper bound when omitted.
todaydate, optional: Today in New York; the system clock’s when omitted.

Returns:

RestPull: The pages and rows written and the watermark after the pull.

Raises:

ValueError: If endpoint is a name not in rest.ENDPOINTS.
MassiveEntitlementError: If Massive refused the key (401) or the endpoint (403), naming it; the pages written before stay, and so does the watermark.
MassiveTransportError: If a page kept failing after retries back-offs; likewise, so the next pull resumes from the watermark.
MassiveResponseError: If a page is not JSON, or a record lacks the field the watermark or the id it was asked by needs; likewise.

Examples

Needs the credentials and the network:

client = MassiveClient()
done = client.pull("dividends", "/data/quantlab/downloads")
print(done.rows, done.through)