diff --git a/README.md b/README.md index 2f1e086..8d3523e 100644 --- a/README.md +++ b/README.md @@ -237,6 +237,14 @@ never served as a cache hit. Eviction is LRU by fetch time against a 10 GiB budget (`RWPYTOOLS_CACHE_MAX_BYTES`). The cache is thread-safe within a process; it is not coordinated across processes. +Reading a cached export is memory-bounded: `start`, `end` and `symbols` are +pushed into a batch-by-batch scan of feather and parquet exports, so a narrow +window over a multi-GB export holds only the matching rows, never the whole +file. Iterating a large export one window at a time (e.g. a year-by-year +backfill of `liquid_universe_prices`) therefore stays flat, and freed Arrow +memory is returned to the OS after every read so long-running workers do not +ratchet up. The per-pair FX exports (Feather V1) are small and read whole. + ```python client.cache.total_bytes() client.cache.entries() diff --git a/src/rwpytools/formats.py b/src/rwpytools/formats.py index 0e50228..5bc9bc1 100644 --- a/src/rwpytools/formats.py +++ b/src/rwpytools/formats.py @@ -42,14 +42,18 @@ -------------- :func:`read_frame` narrows by date range and symbol while reading. For the -Arrow formats the predicate runs on the Arrow table *before* pandas -conversion, so ``symbols=["BTCUSDT"]`` against a 20M-row export only ever -materialises BTCUSDT's rows. :func:`max_timestamp` finds an export's -watermark by reading just its date column. +Arrow formats the predicate is pushed into a batch-by-batch dataset scan, +so ``start="2024-01-01", symbols=["BTCUSDT"]`` against a 20M-row export +never holds more than the matching rows plus one record batch — the whole +export is never materialised. :func:`max_timestamp` finds an export's +watermark by reading just its date column. After each read, Arrow's +allocator is asked to return freed memory to the OS +(:func:`release_arrow_memory`), so long-running processes do not ratchet up. """ from __future__ import annotations +import contextlib import datetime as _dt import enum import logging @@ -61,6 +65,7 @@ import pandas as pd import pyarrow as pa import pyarrow.compute as pc +import pyarrow.dataset as ds import pyarrow.feather import pyarrow.parquet @@ -344,8 +349,8 @@ def _arrow_bound(column_type: Any, day: _dt.date) -> Any: return None -def _filter_table( - table: Any, +def _filter_expression( + schema: Any, *, date_column: str | None, start: _dt.date | None, @@ -353,50 +358,109 @@ def _filter_table( symbol_column: str | None, symbols: Sequence[str] | None, ) -> tuple[Any, bool]: - """Apply what filters Arrow can evaluate natively. + """Translate the date / symbol filters into an Arrow dataset expression. - Returns ``(table, fully_filtered)``; when ``fully_filtered`` is False - the caller finishes the job in pandas (e.g. dictionary-encoded symbols - or a date column stored in an unusual type). + Returns ``(expression, complete)``. ``expression`` is ``None`` when no + filter applies to this schema. ``complete`` is False when some filter + could not be expressed (e.g. a dictionary-encoded symbol column or a + date column in an unusual type); the caller then finishes the job in + pandas, so the result is always fully filtered. """ complete = True - mask: Any = None + expression: Any = None def _and(clause: Any) -> None: - nonlocal mask - mask = clause if mask is None else pc.and_(mask, clause) + nonlocal expression + expression = clause if expression is None else (expression & clause) - if symbols and symbol_column and symbol_column in table.column_names: - column = table[symbol_column] - try: - value_set = pa.array(list(symbols)).cast(column.type) - _and(pc.is_in(column, value_set=value_set)) - except (pa.ArrowInvalid, pa.ArrowNotImplementedError, pa.ArrowTypeError): + names = set(schema.names) + if symbols and symbol_column and symbol_column in names: + column_type = schema.field(symbol_column).type + if pa.types.is_dictionary(column_type): complete = False - - if (start is not None or end is not None) and date_column and date_column in table.column_names: - column = table[date_column] - try: - if start is not None: - low = _arrow_bound(column.type, start) - if low is None: - raise pa.ArrowNotImplementedError(str(column.type)) - _and(pc.greater_equal(column, low)) - if end is not None: - high = _arrow_bound(column.type, end + _dt.timedelta(days=1)) - if high is None: - raise pa.ArrowNotImplementedError(str(column.type)) - _and(pc.less(column, high)) - except (pa.ArrowInvalid, pa.ArrowNotImplementedError, pa.ArrowTypeError): + else: + try: + value_set = pa.array(list(symbols)).cast(column_type) + _and(pc.field(symbol_column).isin(value_set)) + except (pa.ArrowInvalid, pa.ArrowNotImplementedError, pa.ArrowTypeError): + complete = False + + if (start is not None or end is not None) and date_column and date_column in names: + column_type = schema.field(date_column).type + low = _arrow_bound(column_type, start) if start is not None else None + high = _arrow_bound(column_type, end + _dt.timedelta(days=1)) if end is not None else None + if (start is not None and low is None) or (end is not None and high is None): complete = False + else: + if low is not None: + _and(pc.field(date_column) >= low) + if high is not None: + _and(pc.field(date_column) < high) + + return expression, complete + + +def _scan_filtered( + path: Path, + fmt: DataFormat, + *, + date_column: str | None, + start: _dt.date | None, + end: _dt.date | None, + symbol_column: str | None, + symbols: Sequence[str] | None, + memory_pool: Any = None, +) -> tuple[Any, bool] | None: + """Scan an Arrow-format export with the filter pushed into the scan. + + The file is read record batch by record batch and each batch is + filtered as it is decoded, so peak memory is the *matching* rows plus + one batch — never the whole export. That is what keeps a narrow + ``start`` / ``end`` cheap on a multi-GB file, on the first read and on + every different window after it. + + Returns ``(table, complete)``, or ``None`` when the file cannot be + scanned as a dataset (Feather V1, which the per-pair FX exports use, is + not an Arrow IPC file) and the caller must fall back to a full read. + """ - if mask is not None: - # Null comparisons yield null; treat them as "not selected". - table = table.filter(pc.fill_null(mask, False)) + try: + dataset = ds.dataset(str(path), format="ipc" if fmt is DataFormat.FEATHER else "parquet") + except (pa.ArrowInvalid, pa.ArrowNotImplementedError, OSError): + return None + expression, complete = _filter_expression( + dataset.schema, + date_column=date_column, + start=start, + end=end, + symbol_column=symbol_column, + symbols=symbols, + ) + try: + table = dataset.to_table(filter=expression, memory_pool=memory_pool) + except (pa.ArrowInvalid, pa.ArrowNotImplementedError, pa.ArrowTypeError): + # An expression the scanner rejects for this file: read it all and + # let pandas filter instead. + table = dataset.to_table(memory_pool=memory_pool) + complete = False return table, complete +def release_arrow_memory() -> None: + """Return Arrow's freed-but-cached memory to the operating system. + + Arrow's allocator (jemalloc / mimalloc) keeps freed memory for reuse, so + after reading a large export a process's RSS stays near its peak even + once every table is gone — and a long-running worker that reads export + after export ratchets upward until it hits its memory limit. Called + after every read of a cached export. + """ + + with contextlib.suppress(AttributeError, NotImplementedError): + pa.default_memory_pool().release_unused() + + def _read_arrow_table(path: Path, fmt: DataFormat, columns: Sequence[str] | None = None) -> Any: cols = list(columns) if columns is not None else None if fmt is DataFormat.FEATHER: @@ -419,25 +483,38 @@ def read_frame( With no filters this is :func:`read_auto`. ``end`` is inclusive of the whole day, so an intraday column keeps every bar on ``end``. + + Feather (Arrow IPC) and parquet exports are scanned with the filter + pushed down (see :func:`_scan_filtered`), so memory scales with the + rows returned rather than with the size of the export. """ fmt = detect_format(object_name) filtering = bool(symbols) or start is not None or end is not None - if not filtering: - return read(path, fmt, max_bytes=max_bytes) - _check_memory_guard(path, max_bytes=max_bytes) - kwargs: dict[str, Any] = { - "date_column": date_column, - "start": start, - "end": end, - "symbol_column": symbol_column, - "symbols": symbols, - } - if fmt in (DataFormat.FEATHER, DataFormat.PARQUET): - table, complete = _filter_table(_read_arrow_table(path, fmt), **kwargs) - frame = _arrow_to_df(table) - return frame if complete else filter_frame(frame, **kwargs) - return filter_frame(_BY_FORMAT[fmt].reader(path), **kwargs) + try: + if not filtering: + return read(path, fmt, max_bytes=max_bytes) + _check_memory_guard(path, max_bytes=max_bytes) + kwargs: dict[str, Any] = { + "date_column": date_column, + "start": start, + "end": end, + "symbol_column": symbol_column, + "symbols": symbols, + } + if fmt in (DataFormat.FEATHER, DataFormat.PARQUET): + scanned = _scan_filtered(path, fmt, **kwargs) + if scanned is None: # Feather V1: no batch index, read it whole + table = _read_arrow_table(path, fmt) + complete = False + else: + table, complete = scanned + frame = _arrow_to_df(table) + del table + return frame if complete else filter_frame(frame, **kwargs) + return filter_frame(_BY_FORMAT[fmt].reader(path), **kwargs) + finally: + release_arrow_memory() def max_timestamp(path: Path, *, object_name: str, column: str) -> pd.Timestamp | None: @@ -455,6 +532,8 @@ def max_timestamp(path: Path, *, object_name: str, column: str) -> pd.Timestamp except (KeyError, pa.ArrowInvalid): return None series = _arrow_to_df(table)[column] + del table + release_arrow_memory() else: compression: CsvCompression = "gzip" if fmt is DataFormat.CSV_GZ else None try: @@ -535,5 +614,6 @@ def read_csv_chunks( "read_frame", "read_head", "register_dtype_hints", + "release_arrow_memory", "to_timestamps", ] diff --git a/tests/test_session.py b/tests/test_session.py index 809bb96..80755e0 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -690,3 +690,115 @@ def test_snapshot_export_is_served_from_cache_without_replanning_a_download( assert not live.called assert again.route.bulk_action is not None and again.route.bulk_action.value == "cache" assert "no date watermark" in again.route.reason + + +# ---- results are never shared between calls ------------------------------------------ +# +# Each get() must return data read fresh for that call. Any in-memory reuse of +# a parsed export (e.g. an LRU of DataFrames) has to keep these passing: the +# seam refresh writes live values into the bulk half in place, and callers +# are free to mutate the frame they are handed. + + +def _mock_provisional_prices(respx_mock: respx.MockRouter) -> None: + table = pa.table( + { + "ticker": ["HBM", "HBM"], + "date": pa.array([dt.date(2026, 9, 21), dt.date(2026, 9, 22)], pa.date32()), + "close": [27.0, 27.5], + "volume": [4_000_000.0, 3_176_000.0], + } + ) + buf = io.BytesIO() + pyarrow.feather.write_feather(table, buf) + _mock_file( + respx_mock, + namespace="equities", + dataset="statarb", + schema="prices", + bucket="equity_factors_research_pod", + obj="statarb/prices.feather", + payload=buf.getvalue(), + fmt="feather", + ) + respx_mock.get("/v1/equities/statarb/prices").respond( + json=_page([{"ticker": "HBM", "date": "2026-09-22", "close": 27.74, "volume": 5_836_000.0}]) + ) + + +@respx.mock(base_url=FAKE_API, assert_all_called=False) +def test_a_bulk_only_read_is_never_changed_by_an_earlier_live_top_up( + respx_mock: respx.MockRouter, client: Client +) -> None: + _mock_provisional_prices(respx_mock) + topped_up = client.get("statarb_prices") + assert topped_up.seam_updates == 1 # live values were applied to that result... + + # ...but the export itself, read again for the same window, is untouched. + bulk = client.get("statarb_prices", end="2026-09-22", source="bulk").df + assert bulk["volume"].tolist() == [4_000_000.0, 3_176_000.0] + assert bulk["close"].tolist() == [27.0, 27.5] + + +@respx.mock(base_url=FAKE_API, assert_all_called=False) +def test_mutating_a_result_never_changes_the_next_one( + respx_mock: respx.MockRouter, client: Client +) -> None: + _mock_provisional_prices(respx_mock) + first = client.get("statarb_prices", source="bulk") + first.df.loc[0, "close"] = -1.0 + first.df["extra"] = 1 + first.df.sort_values("close", ascending=False, inplace=True) + + second = client.get("statarb_prices", source="bulk") + assert second.df["close"].tolist() == [27.0, 27.5] + assert "extra" not in second.df.columns + assert second.df is not first.df + + +@respx.mock(base_url=FAKE_API, assert_all_called=False) +def test_windowed_gets_scan_the_export_instead_of_loading_it( + respx_mock: respx.MockRouter, client: Client, monkeypatch: pytest.MonkeyPatch +) -> None: + """End to end: a year-by-year backfill through Client.get() never reads a + whole non-templated export into memory (only the one-column watermark + scan touches every row).""" + + import rwpytools.formats as formats_module + + days = pd.bdate_range("2015-01-01", "2018-12-31") + table = pa.table( + { + "ticker": ["AAPL"] * len(days) + ["MSFT"] * len(days), + "date": pa.array(list(days.date) * 2, pa.date32()), + "close": list(range(len(days) * 2)), + } + ) + buf = io.BytesIO() + pyarrow.feather.write_feather(table, buf, chunksize=100) + _mock_file( + respx_mock, + namespace="equities", + dataset="equity.factors", + schema="prices", + bucket="equity_factors_research_pod", + obj="equity_factors/prices.feather", + payload=buf.getvalue(), + fmt="feather", + ) + real = formats_module._read_arrow_table + + def _only_single_columns(path: Any, fmt: Any, columns: Any = None) -> Any: + assert columns is not None and len(columns) == 1, "a whole export was loaded" + return real(path, fmt, columns) + + monkeypatch.setattr(formats_module, "_read_arrow_table", _only_single_columns) + + rows = 0 + for year in range(2015, 2019): + result = client.get( + "liquid_universe_prices", start=f"{year}-01-01", end=f"{year}-12-31", source="bulk" + ) + assert set(pd.to_datetime(result.df["date"]).dt.year) == {year} + rows += len(result) + assert rows == len(days) * 2 diff --git a/tests/test_streaming_reads.py b/tests/test_streaming_reads.py new file mode 100644 index 0000000..272f566 --- /dev/null +++ b/tests/test_streaming_reads.py @@ -0,0 +1,288 @@ +"""Filtered reads of cached exports must not load the whole export. + +A long-running worker backfilling a large export one window at a time +(e.g. ``liquid_universe_prices``, ~5M rows, year by year) used to read the +*entire* export into memory on every ``get()`` and filter afterwards, and +Arrow's allocator kept the freed memory — RSS ratcheted up until the +worker was OOM-killed. These tests pin the fix: the filter is pushed into +a batch-by-batch scan, results match the old full-read-then-filter path +exactly, and freed Arrow memory is handed back to the OS after every read. +""" + +from __future__ import annotations + +import datetime as dt +from pathlib import Path +from typing import Any + +import numpy as np +import pandas as pd +import pyarrow as pa +import pyarrow.feather +import pyarrow.parquet +import pytest + +from rwpytools import formats +from rwpytools.errors import RwPyToolsError +from rwpytools.formats import DataFormat, filter_frame, read_auto, read_frame + +#: Metered pools must outlive every buffer allocated from them; see the +#: memory-bound test. (The library itself always uses Arrow's process-wide +#: default pool, so it has no such lifetime hazard.) +_METERED_POOLS: list[Any] = [] + +N_TICKERS = 200 +N_DAYS = 2_000 # 400k rows + + +def _wide_table(*, tz: str | None = None) -> pa.Table: + """A long export: every ticker on every day, like the equity price exports.""" + + rng = np.random.default_rng(0) + days = pd.date_range("2015-01-01", periods=N_DAYS, freq="D") + tickers = np.repeat([f"T{i:03d}" for i in range(N_TICKERS)], N_DAYS) + stamps = pd.DatetimeIndex(np.tile(days.values, N_TICKERS)).as_unit("us") + if tz: + stamps = stamps.tz_localize(tz) + n = N_TICKERS * N_DAYS + return pa.table( + { + "ticker": tickers, + "date": pa.array(stamps.date, pa.date32()), + "Datetime": pa.array(stamps.to_pydatetime(), pa.timestamp("us", tz=tz)), + "iso": pa.array(np.tile(days.strftime("%Y-%m-%d").to_numpy(), N_TICKERS)), + "close": rng.random(n), + "volume": rng.random(n), + } + ) + + +@pytest.fixture(scope="module") +def export_dir(tmp_path_factory: pytest.TempPathFactory) -> Path: + """The same data in every format the exports use, written once.""" + + root = tmp_path_factory.mktemp("exports") + table = _wide_table() + # pandas' to_feather default: Arrow IPC (Feather V2), lz4, many batches. + pyarrow.feather.write_feather(table, root / "v2.feather", chunksize=2_000) + pyarrow.feather.write_feather( + table, root / "v2_raw.feather", chunksize=2_000, compression="uncompressed" + ) + pyarrow.feather.write_feather(table.drop_columns(["Datetime"]), root / "v1.feather", version=1) + pyarrow.parquet.write_table(table, root / "x.parquet", row_group_size=20_000) + pyarrow.feather.write_feather(_wide_table(tz="UTC"), root / "tz.feather", chunksize=2_000) + dictionary = table.set_column(0, "ticker", pa.compute.dictionary_encode(table.column("ticker"))) + pyarrow.feather.write_feather(dictionary, root / "dict.feather", chunksize=2_000) + table.to_pandas().to_csv(root / "x.csv", index=False) + return root + + +# ---- the memory property -------------------------------------------------------------- + + +def test_a_narrow_window_never_loads_the_whole_export(export_dir: Path) -> None: + """The fix itself: peak Arrow memory for a one-year window over a multi- + batch export is a small fraction of the export, not all of it.""" + + path = export_dir / "v2.feather" + full_bytes = pyarrow.feather.read_table(path).nbytes + + def peak(**filters: Any) -> tuple[int, int]: + pool = pa.proxy_memory_pool(pa.system_memory_pool()) + # The scanner's background read-ahead can still release buffers to + # this pool after the scan returns; a pool destroyed first crashes the + # interpreter later, in some unrelated test. Keep it for the process. + _METERED_POOLS.append(pool) + scanned = formats._scan_filtered( + path, + DataFormat.FEATHER, + date_column=filters.get("date_column"), + start=filters.get("start"), + end=filters.get("end"), + symbol_column=None, + symbols=None, + memory_pool=pool, + ) + assert scanned is not None + table, complete = scanned + assert complete + measured = (pool.max_memory(), table.num_rows) + # The table's buffers belong to ``pool``: free them while the pool + # is still alive, or local teardown order can crash the interpreter. + del table, scanned + return measured + + # Control: an unfiltered scan really does allocate the whole export, so + # the meter is measuring what we think it is. + whole_peak, whole_rows = peak() + assert whole_rows == N_TICKERS * N_DAYS + assert whole_peak >= 0.9 * full_bytes + + year_peak, year_rows = peak( + date_column="date", start=dt.date(2016, 1, 1), end=dt.date(2016, 12, 31) + ) + assert year_rows == N_TICKERS * 366 + assert year_peak < 0.35 * full_bytes, ( + f"one-year window peaked at {year_peak:,} bytes of a {full_bytes:,}-byte export" + ) + + +def test_filtered_reads_do_not_fall_back_to_a_full_table_read( + export_dir: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + def _full_read_forbidden(*_: Any, **__: Any) -> Any: + raise AssertionError("filtered read loaded the whole export") + + monkeypatch.setattr(formats, "_read_arrow_table", _full_read_forbidden) + monkeypatch.setattr(pyarrow.feather, "read_table", _full_read_forbidden) + monkeypatch.setattr(pyarrow.parquet, "read_table", _full_read_forbidden) + + for name in ("v2.feather", "x.parquet"): + out = read_frame( + export_dir / name, + object_name=name, + date_column="date", + start=dt.date(2016, 1, 1), + end=dt.date(2016, 1, 31), + symbol_column="ticker", + symbols=["T007"], + ) + assert len(out) == 31 + assert set(out["ticker"]) == {"T007"} + + +def test_arrow_memory_is_released_after_every_read( + export_dir: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Without this, freed Arrow memory stays in the allocator and a worker's + RSS ratchets up read after read.""" + + calls: list[str] = [] + monkeypatch.setattr(formats, "release_arrow_memory", lambda: calls.append("released")) + + read_frame( + export_dir / "v2.feather", + object_name="v2.feather", + start=dt.date(2016, 1, 1), + date_column="date", + ) + read_frame(export_dir / "v2.feather", object_name="v2.feather") # unfiltered + formats.max_timestamp(export_dir / "v2.feather", object_name="v2.feather", column="date") + assert calls == ["released"] * 3 + + calls.clear() + with pytest.raises(RwPyToolsError, match="max_bytes"): # released on the error path too + read_frame( + export_dir / "v2.feather", + object_name="v2.feather", + date_column="date", + start=dt.date(2016, 1, 1), + max_bytes=1, + ) + assert calls == ["released"] + + +def test_release_arrow_memory_is_safe_to_call() -> None: + formats.release_arrow_memory() + + +# ---- results identical to the old full-read-then-filter path --------------------------- + +CASES = [ + pytest.param( + {"date_column": "date", "start": dt.date(2016, 3, 1), "end": dt.date(2016, 3, 31)}, + id="date32-range", + ), + pytest.param({"date_column": "Datetime", "start": dt.date(2017, 1, 1)}, id="timestamp-start"), + pytest.param( + {"date_column": "Datetime", "end": dt.date(2015, 2, 1)}, id="timestamp-end-inclusive" + ), + pytest.param( + {"date_column": "iso", "start": dt.date(2016, 1, 1), "end": dt.date(2016, 1, 2)}, + id="iso-string", + ), + pytest.param({"symbol_column": "ticker", "symbols": ["T001", "T150"]}, id="symbols"), + pytest.param( + { + "date_column": "date", + "start": dt.date(2016, 6, 1), + "end": dt.date(2016, 6, 30), + "symbol_column": "ticker", + "symbols": ["T042"], + }, + id="window-and-symbol", + ), + pytest.param({"date_column": "date", "start": dt.date(2030, 1, 1)}, id="empty-result"), +] + + +def _reference(path: Path, object_name: str, **filters: Any) -> pd.DataFrame: + return filter_frame(read_auto(path, object_name=object_name), **filters) + + +@pytest.mark.parametrize("filters", CASES) +@pytest.mark.parametrize( + "name", ["v2.feather", "v2_raw.feather", "x.parquet", "dict.feather", "x.csv"] +) +def test_filtered_read_matches_full_read_then_filter( + export_dir: Path, name: str, filters: dict[str, Any] +) -> None: + if name == "x.csv" and filters.get("date_column") == "Datetime": + pytest.skip("CSV round-trips timestamps as text; covered by the iso case") + out = read_frame(export_dir / name, object_name=name, **filters) + expected = _reference(export_dir / name, name, **filters) + pd.testing.assert_frame_equal( + out.reset_index(drop=True), expected.reset_index(drop=True), check_categorical=False + ) + + +@pytest.mark.parametrize( + "filters", [c for c in CASES if c.values[0].get("date_column") != "Datetime"] +) +def test_feather_v1_falls_back_and_still_filters(export_dir: Path, filters: Any) -> None: + """Feather V1 (the per-pair FX exports) is not an IPC file, so it cannot + be scanned; the fallback must still return exactly the filtered rows.""" + + assert ( + formats._scan_filtered( + export_dir / "v1.feather", + DataFormat.FEATHER, + date_column=None, + start=None, + end=None, + symbol_column=None, + symbols=None, + ) + is None + ) + out = read_frame(export_dir / "v1.feather", object_name="v1.feather", **filters) + expected = _reference(export_dir / "v1.feather", "v1.feather", **filters) + pd.testing.assert_frame_equal(out.reset_index(drop=True), expected.reset_index(drop=True)) + + +def test_timezone_aware_timestamps_filter_by_utc_day(export_dir: Path) -> None: + filters = {"date_column": "Datetime", "start": dt.date(2016, 1, 1), "end": dt.date(2016, 1, 3)} + out = read_frame(export_dir / "tz.feather", object_name="tz.feather", **filters) + expected = _reference(export_dir / "tz.feather", "tz.feather", **filters) + assert len(out) == N_TICKERS * 3 + pd.testing.assert_frame_equal(out.reset_index(drop=True), expected.reset_index(drop=True)) + + +def test_a_year_by_year_backfill_returns_every_row_exactly_once(export_dir: Path) -> None: + """The workload that OOM'd: iterate the export one calendar year at a time.""" + + path = export_dir / "v2.feather" + years = range(2015, 2021) + parts = [ + read_frame( + path, + object_name="v2.feather", + date_column="date", + start=dt.date(y, 1, 1), + end=dt.date(y, 12, 31), + ) + for y in years + ] + combined = pd.concat(parts, ignore_index=True) + assert len(combined) == N_TICKERS * N_DAYS + assert not combined.duplicated(["ticker", "date"]).any()