quantlab.acquisition.symbol_batch.SymbolBatchAcquisition#

class quantlab.acquisition.symbol_batch.SymbolBatchAcquisition(config)[source]#

Bases: ABC

Abstract base of the symbol-batched downloads: a roster over a window, resumable.

A subclass declares VENDOR and RAW_COLUMNS and implements _fetch_page. The base class turns that single request into a concurrent, resumable run over the config’s symbol list, in which one failing batch does not stop the others. It writes parquet shards under config.raw_data_dir_path in a hive layout whose keys depend on the data frequency. It never touches xarray or Zarr; the matching dataset class converts these raw files into the canonical panel.

download() backfills [config.start_date, config.end_date] for every requested symbol. refresh() fetches each symbol forward from its own watermark, so repeated calls do not re-fetch history. Both skip symbols whose sidecar already covers the requested window.

Each symbol’s sidecar lives under config.watermark_path and holds last_date, optionally start_date and optionally no_data. start_date is where the downloaded range begins. When it is missing the start is unknown and is never guessed; config.kwargs["legacy_watermarks"] says whether such a symbol is skipped or fetched again. no_data means the vendor was asked about the symbol over that window and returned nothing. That is different from a failed fetch (an entry in the failure manifest and no sidecar) and from a symbol that was never fetched (neither). Both optional keys are omitted rather than written as false, so sidecars written before these keys existed still read back correctly.

Per-run tuning parameters are read from config.kwargs: batch_size, max_workers, resume, progress, wait_for_quota, quota_wait_seconds, quota_max_waits, legacy_watermarks, request_retries and retry_backoff_seconds.

Parameters:

configAcquisitionConfig: The acquisition config. start_date and end_date are filled with open-ended defaults when left None.

Attributes:

last_resultAcquisitionResult or None: The outcome of the most recent download() or refresh(), or None before either has run.

Examples

A minimal vendor client, where client stands for the vendor’s own HTTP wrapper and returns every row in one page:

class DemoAcquisition(SymbolBatchAcquisition):
    VENDOR = "tiingo"
    RAW_COLUMNS = ("timestamp", "symbol", "close", "vendor")

    def _fetch_page(self, symbols, start_date, end_date,
                    page_token=None):
        frame = client.bars(symbols, start_date, end_date)
        return frame, None

acq = DemoAcquisition(config)
acq.download()
print(acq.last_result.succeeded, acq.coverage_report())

CREDENTIAL_ENV_VARS = ()#: The environment variables whose values the request policy hides (each becomes <NAME REDACTED>). A subclass fills this from module-level constants, not from its HTTP client class, so a test double that replaces the client cannot turn the redaction off. An empty tuple means the vendor uses no credentials.

DEFAULT_BATCH_SIZE = 1#: How many symbols one _fetch_page request carries. A vendor whose API takes one symbol per call uses 1, which behaves exactly like a per-symbol loop in requests, failures and resumes. Overridable per run via config.kwargs["batch_size"].

DEFAULT_MAX_WORKERS = 8#: Concurrent in-flight batch fetches. Overridable per run via config.kwargs["max_workers"].

DEFAULT_QUOTA_MAX_WAITS = 3#: Maximum resume attempts after the quota runs out. Bounded, because retrying forever against a locked-out account is worse than stopping.

DEFAULT_QUOTA_WAIT_SECONDS = 3600#: Delay between resume attempts, in seconds. Vendors do not publish whether their quota resets on a fixed hourly bucket or a rolling window, so this is a configured interval rather than a computed reset instant; one hour covers both readings.

DEFAULT_REQUEST_RETRIES = 5#: Retries of one request before its batch counts as failed and goes to the manifest for the next run to retry. At the default back-off the waits are 2, 4, 8, 16 and 32 seconds, about a minute in all. Overridable via config.kwargs["request_retries"].

DEFAULT_RETRY_BACKOFF_SECONDS = 2.0#: Seconds before the first retry of a request; doubled on each later one, unless the server sent Retry-After. Overridable per run via config.kwargs["retry_backoff_seconds"].

DEFAULT_WAIT_FOR_QUOTA = False#: Whether a run that used up the vendor’s request quota (its “allocation”) waits for it to reset and then resumes. Off by default, so no run silently sits waiting for hours.

FAILURE_MANIFEST_NAME = '_failures.json'#: Filename of the failure manifest, written under config.watermark_path next to the per-symbol sidecars. Imported from quantlab.acquisition._support.coverage so this writer and the read-only inspector, which needs no vendor credentials, name the same file.

RAW_COLUMNS#: The columns, in order, of every raw shard this subclass writes. _write_shard selects exactly these, so every shard in a vendor’s directory has the same schema. A directory scan takes its schema from the first file and enforces it on all the others, so a single shard with an extra column, or with the columns in another order, would make the whole directory unreadable.

RETRY_STATUS_CODES = frozenset({429, 500, 502, 503, 504})#: HTTP statuses that mean “try this request again later”: the worker waits and repeats the request without stopping the run. The request policy’s RETRY_STATUSES (429 and the 5xx a server sends while it recovers). A vendor whose 429 means a used-up quota, where the right response is to stop the run, says so in _classify_error before deferring to the base (Tiingo).

SESSION_TIME_ZONE = None#: The IANA time zone (for example "America/New_York") whose calendar day defines the intraday date= hive key, so one directory holds one trading session. None means not declared, and _session_date then raises rather than falling back to a UTC date. Only intraday frequencies use it. Timestamps inside the shards stay UTC without a time zone attached; only the directory key is converted.

VENDOR#: Which vendor this subclass fetches from. Set by every subclass. It is written into every shard as a literal vendor column and hashed into every PageLedger.batch_key, so changing it changes the files.

attach(*, reporter=None, cancel=None)[source]#

Attach a progress reporter and/or a cancel token for the next run.

Both values are kept on the instance rather than on the config. The config is written to disk beside model checkpoints, a cancel token cannot be serialised, and a live reporter is not reproducible configuration. Passing None for either argument clears it, so the same object can be reused for a later, unobserved run without inheriting the first run’s reporter.

Parameters:

reporterProgressReporter or None, default None: Destination for ProgressEvent objects. When None, the progress entry of config.kwargs decides between a tqdm progress bar and silence.
cancelCancelToken or None, default None: Token the caller sets to stop the run at the next batch boundary.

Returns:

Self: self, so the call chains into download() or refresh().

Examples

token = CancelToken()
acq.attach(reporter=NullProgressReporter(), cancel=token).download()
acq.attach()  # clear both before the next run

property class_name#: Return the bare class name, used to label log and error messages.

property config#: The acquisition config this object runs with.

coverage_report(symbols=None)[source]#

Report which symbols are already downloaded for the config window.

This runs the same check _run uses to decide what is pending, so the two always agree, but it downloads nothing. Use it to see whether a wider date window would re-fetch anything before starting a multi-hour job, or to count the legacy sidecars (no recorded start) a legacy_watermarks="refetch" run would fetch again.

Parameters:

symbolslist[str] | None, default None: The symbols to classify; defaults to config.symbols.

Returns:

dict: Symbol counts under requested, pending, skipped, covered, widened (recorded start is later than the requested start), legacy (no recorded start) and no_data.

Examples

report = acq.coverage_report()
if report["pending"]:
    print(f"{report['pending']} symbol(s) would be fetched")

download(symbols=None)[source]#

Backfill [config.start_date, config.end_date] for every symbol.

Symbols whose sidecar already covers the window are skipped unless config.kwargs["resume"] is false. If start_date has been moved earlier, symbols whose recorded coverage starts later are fetched again.

Parameters:

symbolslist[str] | None, default None: The symbols to fetch; defaults to config.symbols.

Returns:

Self: self; the outcome is on last_result.

Examples

result = acq.download(["AAPL", "MSFT"]).last_result
print(result.succeeded, result.coverage["skipped"])

property import_path#

Return the dotted import path of this object’s class.

For example "quantlab.acquisition.tiingo.TiingoAcquisition".

refresh(symbols=None)[source]#

Fetch each symbol forward from its own watermark.

A refresh fetches [last_date, config.end_date] per symbol, so the recorded start stays what it was, and an unknown start stays unknown. It ignores an earlier config.start_date; extending history backwards is download()’s job.

Parameters:

symbolslist[str] | None, default None: The symbols to refresh; defaults to config.symbols.

Returns:

Self: self; the outcome is on last_result.

Examples

acq.config.end_date = "2024-01-08"
acq.refresh()