quantlab.dataset.raw.RawDataset#

class quantlab.dataset.raw.RawDataset(config)[source]#

Bases: BaseDataset

A dataset converted from raw vendor files into a store it owns.

A subclass implements _raw_data_to_xr, which reads the raw files and returns a dense panel (a full timestamp by symbol grid, with NaN where a symbol has no value). Everything else is inherited: the read interface of BaseDataset, Zarr storage through XrBackend, the cleaning hook, and chunked ingestion with its resume ledger and its handling of new listings (symbols that appear in the raw tier but not yet in the store). Only the build path (from_raw_data, from_raw_data_chunked, update) holds a panel; reading stays a query that holds nothing. Datasets that are not price bars, such as a rate or a boolean membership panel, derive from this class directly; price bars derive from RawMarketDataset.

Parameters:

configBaseDatasetConfig: The dataset config. It is not modified: the dataset holds a normalised copy; see the config property.

Attributes:

data_backendXrBackend: Holds the panel the build path converted, and writes the store. Requests read through a fresh instance of its type, so a dataset reads the medium it writes.
last_chunk_resultConversionResult or None: Summary of the last from_raw_data_chunked or update run, or None if neither has completed on this object.

Examples

A minimal subclass and the whole storage lifecycle:

class MembershipDataset(RawDataset):
    def _raw_data_to_xr(self) -> xr.Dataset:
        return load_membership_panel(self.config.kwargs["source"])

ds = MembershipDataset(config)
ds.from_raw_data().save()
panel = MembershipDataset(config).panel("2020-01-01", "2020-12-31")

The method examples below use the DemoDataset described on BaseDataset.

NEW_LISTING_REPORT_LIMIT = 20#: Maximum number of symbols named in a new-listing log message.

NEW_LISTING_STRATEGIES = ('refuse', 'rebuild', 'widen')#: Accepted values of on_new_listing: stop ("refuse"), rewrite the whole store ("rebuild"), or add NaN columns ("widen").

SUPERSEDED_SUFFIX = '.superseded.tmp'#: Suffix added to the store and ledger paths while a rebuild is running and the originals are set aside. It reuses XrBackend.SUPERSEDED_SUFFIX so the backend, which refuses to widen while a set-aside copy exists, always recognises one left behind by a rebuild that was killed.

copy()[source]#

Return a copy with its own config and an empty storage backend.

The config is deep-copied, so the copy shares no mutable state with this object, not even a kwargs dict. The copy holds no panel until it is built.

Examples

>>> other = ds.copy()
>>> other.config is ds.config, other.data_backend is ds.data_backend
(False, False)

from_raw_data()[source]#

Convert the raw files for the whole configured range, in memory.

Runs _raw_data_to_xr, then _clean, and gives the result to the backend. Nothing is written to disk until save is called. Every call converts again; nothing is cached.

Returns:

Self: self, for chaining.

Examples

>>> ds = DemoDataset(config).from_raw_data()
>>> list(ds.get_xarray_dataset().data_vars)  # cleaning adds the flag
['open', 'high', 'low', 'close', 'volume', 'anomaly_flag']

from_raw_data_chunked(granularity='year', ledger_path=None, append_dim='timestamp', on_new_listing='refuse', *, reporter=None, cancel=None)[source]#

Convert the raw files one time window at a time, appending to Zarr.

The pinned symbol axis, the list of symbols for the whole date range, is worked out once before any window, and every window is built on exactly that axis so that all windows line up column by column. Completed windows are recorded in a ledger file, so a crashed or cancelled run resumes by skipping them. Everything up to the ledger’s last end counts as written, so a window that straddles it is cut to its new bars: a daily run extends a store whose last window is still growing, at any granularity. Every window appended to a store that existed before the run passes through _continue_store first. Cleaning runs per window, so a price jump that spans two windows is not flagged.

When the store already exists with a different symbol axis, on_new_listing decides what happens. "refuse" stops. "widen" adds the new symbols with NaN over the store’s existing history. "rebuild" moves the store aside and rebuilds every window from the raw files; if the rebuild fails or is cancelled, the original store is put back.

Parameters:

granularitystr, default “year”: Window size, passed to TimeChunkPlanner ("year", "quarter", "month", "day" or "hour").
ledger_pathstr or None, default None: Ledger file path; None uses <store>.chunks.json.
append_dimstr, default “timestamp”: Dimension windows are appended along.
on_new_listingstr, default “refuse”: One of NEW_LISTING_STRATEGIES. update() passes a private marker here instead, to have the strategy chosen automatically.
reporterProgressReporter or None, default None: Receives an event for each window.
cancelCancelToken or None, default None: Checked before each window; when set, the run stops there.

Returns:

Self: self, with last_chunk_result describing the run.

Raises:

ValueError: If on_new_listing is unknown, if a window comes back on a symbol axis other than the pinned one, or if the ledger and the store disagree (see ChunkLedger.assert_consistent).

Examples

>>> ds = DemoDataset(config).from_raw_data_chunked(granularity="month")
>>> ds.last_chunk_result.windows_written
1

get_lazyframe()[source]#

Return the built panel as a long-format polars LazyFrame.

Long format means one row per (timestamp, symbol) pair, with one column per variable.

Examples

>>> ds.get_lazyframe().collect().shape  # 4 days x 3 symbols, 8 columns
(12, 8)

get_xarray_dataset()[source]#

Return the built panel indexed by (timestamp, symbol).

Examples

>>> tuple(ds.get_xarray_dataset().dims)
('timestamp', 'symbol')

property num_symbols#

Return the number of symbols in the built panel.

Examples

>>> ds.num_symbols
3

resample(freq, how=None)[source]#

Return a copy resampled onto freq, holding the built panel resampled.

As BaseDataset.resample; when this dataset holds a panel it built, the copy holds that panel resampled, in memory of its own, so save() writes it. A resampled dataset cannot be built from raw files: from_raw_data, from_raw_data_chunked and update refuse. save() writes to store_path, the resampled store beside the source.

Raises:

ValueError: As BaseDataset.resample, or, when this dataset holds a panel, if freq is not coarser than its bars or how does not name every variable.

Examples

>>> built = DemoDataset(minute_config).from_raw_data()
>>> built.resample("1d", "last").get_xarray_dataset().sizes["timestamp"]
2

save(**kwargs)[source]#

Narrow the built panel to the config’s range and write it to Zarr.

The write replaces the whole store directory at store_path: the config’s zarr_file_path, or the resampled store beside it when the dataset is resampled. A resampled dataset that holds no panel writes the resample of its whole source store, narrowed to the config’s range.

Parameters:

**kwargs: Passed to XrBackend.create, which lays the store out as widen_and_append does (sorted symbol axis, pinned chunk grid and fill values) so it can grow at its end.

Examples

>>> DemoDataset(config).from_raw_data().save()
>>> Path(config.zarr_file_path).is_dir()
True
>>> DemoDataset(minute_config).resample("1d", "last").save()
>>> Path(minute_config.zarr_file_path.replace(".zarr", "_resample_1d.zarr")).is_dir()
True

property symbols#

Return the symbol labels of the built panel, in axis order.

Examples

>>> ds.symbols
['AAA', 'BBB', 'CCC']

property time_interval#

Return the most common spacing between consecutive timestamps.

The most common spacing (the mode) is used rather than the minimum so that gaps such as weekends and holidays do not distort the answer.

Examples

>>> ds.time_interval  # daily bars built from raw files
np.timedelta64(86400000000000,'ns')

update(granularity='year', ledger_path=None, append_dim='timestamp')[source]#

Bring the store up to date, choosing the new-listing strategy itself.

This is from_raw_data_chunked with on_new_listing chosen from the data. When the raw tier has symbols the store lacks, the raw tier is checked for rows of those symbols inside the store’s existing date range. If there are none, the symbols are genuinely new listings and the store is widened with NaN history. If there are some, the store is rebuilt so that their real history is included. If a symbol in the store has disappeared from the raw tier, the run refuses. The choice is logged before it runs.

Parameters:

granularitystr, default “year”: Window size, passed to TimeChunkPlanner.
ledger_pathstr or None, default None: Ledger file path; None uses <store>.chunks.json.
append_dimstr, default “timestamp”: Dimension windows are appended along.

Returns:

Self: self, with last_chunk_result populated.

Examples

>>> ds = DemoDataset(config).update(granularity="month")
>>> ds.last_chunk_result.windows_written
1
>>> again = DemoDataset(config).update(granularity="month")
>>> again.last_chunk_result.windows_skipped, again.last_chunk_result.resumed
(1, True)