Skip to content

markowitz.data

markowitz.data

Data layer for markowitz-optimizer.

Public surface area (re-exported here for ergonomic imports):

  • Loaders: :func:load_prices, :func:compute_returns, :func:align_universe, :func:summary_stats
  • Risk-free: :func:risk_free_rate, :func:de_annualize
  • Factors: :func:load_french_factors
  • Calendars: :func:trading_sessions, :func:reindex_to_sessions
  • Cache: :func:cache_path, :func:read_cache, :func:write_cache
  • Providers: :class:PriceProvider, :class:YFinanceProvider, :class:CachedProvider, :class:FallbackProvider
  • Exceptions: :class:DataLayerError and all subclasses

CacheCorruptError

Bases: DataLayerError

Raised when a cache file exists but cannot be deserialized.

Note: :func:markowitz.data.cache.read_cache deliberately does not raise this — it warns and returns None so callers can transparently fall back to a fresh fetch. The exception exists for explicit callers that want hard failures.

CacheCorruptionWarning

Bases: UserWarning

Emitted when a cache file exists but cannot be deserialized.

CachedProvider(inner: PriceProvider, root: str | os.PathLike[str])

Wrap a :class:PriceProvider with on-disk Parquet caching.

Source code in src/markowitz/data/providers.py 
134
135
136
137
def __init__(self, inner: PriceProvider, root: str | os.PathLike[str]) -> None:
    self._inner = inner
    self._root = Path(root)
    self.name = f"cached({inner.name})"

CalendarMismatchError

Bases: DataLayerError

Raised when an index does not align with the expected trading calendar.

DataIntegrityError

Bases: DataLayerError

Raised when fetched data violates structural invariants.

Examples include duplicate timestamps, non-monotonic indices, unexpected NaNs in critical columns, or timezone inconsistencies.

DataLayerError

Bases: Exception

Base class for all errors raised by the markowitz.data package.

EmptyDataError

Bases: DataLayerError

Raised when a provider returns no rows for the requested window.

FallbackProvider(providers: list[PriceProvider])

Try each provider in order, advancing past rate-limit / unavailable.

Source code in src/markowitz/data/providers.py 
167
168
169
170
def __init__(self, providers: list[PriceProvider]) -> None:
    if not providers:
        raise ValueError("FallbackProvider requires at least one inner provider")
    self._providers = providers

InsufficientDataError

Bases: DataLayerError

Raised when fewer observations are available than required.

PriceProvider

Bases: Protocol

Protocol every price source must satisfy.

fetch(ticker: str, start: DateLike, end: DateLike, *, frequency: str = '1d') -> pd.DataFrame

Return a frame indexed by tz-naive midnight, single close column.

Source code in src/markowitz/data/providers.py 
36
37
38
39
40
41
42
43
44
45
def fetch(
    self,
    ticker: str,
    start: DateLike,
    end: DateLike,
    *,
    frequency: str = "1d",
) -> pd.DataFrame:
    """Return a frame indexed by tz-naive midnight, single ``close`` column."""
    ...

ProviderUnavailableError

Bases: DataLayerError

Raised when an upstream data provider is unreachable or misconfigured.

This includes optional dependencies that are not installed, missing credentials, transient network failures, and HTTP 5xx responses.

RateLimitError

Bases: DataLayerError

Raised when an upstream provider signals a rate-limit condition.

YFinanceProvider()

Provider backed by yfinance.download.

Source code in src/markowitz/data/providers.py 
83
84
85
86
87
88
89
def __init__(self) -> None:
    try:
        import yfinance as _yf  # noqa: F401, PLC0415
    except ImportError as exc:
        raise ProviderUnavailableError(
            "yfinance is not installed; install with `pip install yfinance`"
        ) from exc

align_universe(returns: pd.DataFrame, *, min_obs_per_ticker: int = 252, common_window: bool = True) -> pd.DataFrame

Drop tickers with sparse history; optionally restrict to common window.

Source code in src/markowitz/data/loaders.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def align_universe(
    returns: pd.DataFrame,
    *,
    min_obs_per_ticker: int = 252,
    common_window: bool = True,
) -> pd.DataFrame:
    """Drop tickers with sparse history; optionally restrict to common window."""
    if returns.empty:
        raise EmptyDataError("returns is empty; nothing to align")

    obs_per_ticker = returns.notna().sum()
    survivors = obs_per_ticker[obs_per_ticker >= min_obs_per_ticker].index.tolist()
    if not survivors:
        raise InsufficientDataError(
            f"no ticker has >= {min_obs_per_ticker} observations; "
            f"max available is {int(obs_per_ticker.max())}"
        )

    aligned = returns[survivors]
    if common_window:
        aligned = aligned.dropna(how="any")
        if aligned.shape[0] < min_obs_per_ticker:
            raise InsufficientDataError(
                f"common window has {aligned.shape[0]} rows; need >= {min_obs_per_ticker}"
            )
    return cast(pd.DataFrame, aligned)

annualize_rate(per_period: pd.Series, frequency: str) -> pd.Series

Inverse of :func:de_annualize. Provided for round-trip testing.

Source code in src/markowitz/data/risk_free.py 
103
104
105
106
def annualize_rate(per_period: pd.Series, frequency: str) -> pd.Series:
    """Inverse of :func:`de_annualize`. Provided for round-trip testing."""
    n = _periods_per_year(frequency)
    return (1.0 + per_period) ** n - 1.0

cache_path(root: str | os.PathLike[str], ticker: str, source: str, frequency: str) -> Path

Return the canonical cache file path for a (ticker, source, frequency) triple.

The path is <root>/<source>/<frequency>/<ticker>.parquet with each component sanitized. The parent directories are not created by this function.

Source code in src/markowitz/data/cache.py 
41
42
43
44
45
46
47
48
49
def cache_path(root: str | os.PathLike[str], ticker: str, source: str, frequency: str) -> Path:
    """Return the canonical cache file path for a (ticker, source, frequency) triple.

    The path is ``<root>/<source>/<frequency>/<ticker>.parquet`` with
    each component sanitized. The parent directories are *not* created
    by this function.
    """
    root_path = Path(root)
    return root_path / _sanitize(source) / _sanitize(frequency) / f"{_sanitize(ticker)}.parquet"

compute_returns(prices: pd.DataFrame, *, method: str = 'simple', dropna: bool = True) -> pd.DataFrame

Convert a price panel into period-over-period returns.

method="simple" is the Markowitz default. method="log" is provided for portfolio diagnostics and for users who prefer continuously compounded returns.

Source code in src/markowitz/data/loaders.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def compute_returns(
    prices: pd.DataFrame,
    *,
    method: str = "simple",
    dropna: bool = True,
) -> pd.DataFrame:
    """Convert a price panel into period-over-period returns.

    ``method="simple"`` is the Markowitz default. ``method="log"`` is
    provided for portfolio diagnostics and for users who prefer
    continuously compounded returns.
    """
    if prices.empty:
        raise EmptyDataError("prices is empty; cannot compute returns")
    if not isinstance(prices.index, pd.DatetimeIndex):
        raise DataIntegrityError("prices.index must be a DatetimeIndex")

    if method == "simple":
        rets = prices.pct_change(fill_method=None)
    elif method == "log":
        rets = np.log(prices / prices.shift(1))
    else:
        raise ValueError(f"unknown method {method!r}; expected 'simple' or 'log'")

    if dropna:
        rets = rets.dropna(how="any")
    if rets.empty:
        raise InsufficientDataError(
            "computed return frame is empty after dropna; check input price coverage"
        )
    return cast(pd.DataFrame, rets)

constant_rate(value: float, index: pd.DatetimeIndex) -> pd.Series

Construct a constant-rate series — useful for tests and quick demos.

Source code in src/markowitz/data/risk_free.py 
109
110
111
112
113
def constant_rate(value: float, index: pd.DatetimeIndex) -> pd.Series:
    """Construct a constant-rate series — useful for tests and quick demos."""
    if not math.isfinite(value):
        raise ValueError(f"value must be finite, got {value!r}")
    return pd.Series(value, index=index, name="rf", dtype="float64")

de_annualize(annual_pct: pd.Series, frequency: str) -> pd.Series

Convert an annual rate (already in decimal, e.g. 0.05) to per-period.

Uses geometric de-annualization::

r_period = (1 + r_annual) ** (1 / periods_per_year) - 1

so a round-trip via the matching annualizer is exact.

Source code in src/markowitz/data/risk_free.py 
33
34
35
36
37
38
39
40
41
42
43
def de_annualize(annual_pct: pd.Series, frequency: str) -> pd.Series:
    """Convert an annual rate (already in decimal, e.g. 0.05) to per-period.

    Uses geometric de-annualization::

        r_period = (1 + r_annual) ** (1 / periods_per_year) - 1

    so a round-trip via the matching annualizer is exact.
    """
    n = _periods_per_year(frequency)
    return (1.0 + annual_pct) ** (1.0 / n) - 1.0

load_french_factors(dataset: str, *, frequency: str = 'monthly', cache_root: str | os.PathLike[str] | None = None) -> pd.DataFrame

Download (or read from cache) a Ken French factor table.

Parameters:

Name Type Description Default
dataset str

Dataset identifier, e.g. "F-F_Research_Data_Factors".

 required
frequency str

"monthly" (default) or "daily".

 'monthly'
cache_root str | PathLike[str] | None

Optional directory used to cache the raw ZIP payload.

 None
Source code in src/markowitz/data/french.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def load_french_factors(
    dataset: str,
    *,
    frequency: str = "monthly",
    cache_root: str | os.PathLike[str] | None = None,
) -> pd.DataFrame:
    """Download (or read from cache) a Ken French factor table.

    Parameters
    ----------
    dataset:
        Dataset identifier, e.g. ``"F-F_Research_Data_Factors"``.
    frequency:
        ``"monthly"`` (default) or ``"daily"``.
    cache_root:
        Optional directory used to cache the raw ZIP payload.
    """
    url = _build_url(dataset, frequency)
    cache_dir = Path(cache_root) if cache_root is not None else None
    cache_file: Path | None = None
    if cache_dir is not None:
        cache_dir.mkdir(parents=True, exist_ok=True)
        cache_file = cache_dir / f"{dataset}_{frequency}.zip"

    payload: bytes
    if cache_file is not None and cache_file.exists():
        payload = cache_file.read_bytes()
    else:
        try:
            with urllib.request.urlopen(url, timeout=30) as resp:
                payload = resp.read()
        except urllib.error.URLError as exc:
            raise ProviderUnavailableError(
                f"Kenneth French data library unreachable ({url}): {exc.reason}"
            ) from exc
        except Exception as exc:
            raise ProviderUnavailableError(
                f"Failed to download French factors from {url}: {exc}"
            ) from exc
        if cache_file is not None:
            cache_file.write_bytes(payload)

    try:
        with zipfile.ZipFile(io.BytesIO(payload)) as zf:
            csv_names = [n for n in zf.namelist() if n.lower().endswith(".csv")]
            if not csv_names:
                raise DataIntegrityError(f"no CSV in French archive {dataset}")
            text = zf.read(csv_names[0]).decode("latin-1")
    except zipfile.BadZipFile as exc:
        raise DataIntegrityError(f"French archive {dataset} is corrupt") from exc

    return _parse_french_csv(text, frequency=frequency)

load_prices(tickers: str | Iterable[str], start: DateLike, end: DateLike, *, provider: PriceProvider | None = None, frequency: str = '1d', min_history_sessions: int = 252) -> pd.DataFrame

Fetch adjusted close prices for tickers as a wide DataFrame.

Each column is a ticker; the index is a tz-naive, monotonic, unique :class:~pandas.DatetimeIndex aligned across tickers via an outer join. Tickers without at least min_history_sessions non-NaN observations raise :class:InsufficientDataError.

Source code in src/markowitz/data/loaders.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def load_prices(
    tickers: str | Iterable[str],
    start: DateLike,
    end: DateLike,
    *,
    provider: PriceProvider | None = None,
    frequency: str = "1d",
    min_history_sessions: int = 252,
) -> pd.DataFrame:
    """Fetch adjusted close prices for ``tickers`` as a wide DataFrame.

    Each column is a ticker; the index is a tz-naive, monotonic,
    unique :class:`~pandas.DatetimeIndex` aligned across tickers via an
    outer join. Tickers without at least ``min_history_sessions``
    non-NaN observations raise :class:`InsufficientDataError`.
    """
    if provider is None:
        # Late import so missing optional deps don't break test collection.
        from .providers import YFinanceProvider  # noqa: PLC0415

        provider = YFinanceProvider()

    ticker_list = _coerce_tickers(tickers)
    frames: dict[str, pd.Series] = {}
    for tkr in ticker_list:
        frame = provider.fetch(tkr, start, end, frequency=frequency)
        if "close" not in frame.columns:
            raise DataIntegrityError(
                f"provider {provider.name!r} returned unexpected columns "
                f"{list(frame.columns)} for {tkr}"
            )
        series = frame["close"].astype("float64")
        non_na = int(series.notna().sum())
        if non_na < min_history_sessions:
            raise InsufficientDataError(
                f"{tkr} has {non_na} observations; need >= {min_history_sessions}"
            )
        frames[tkr] = series

    combined = pd.concat(frames, axis=1)
    combined.columns = pd.Index(list(frames.keys()))
    combined = combined.sort_index()
    if combined.index.has_duplicates:
        combined = combined.loc[~combined.index.duplicated(keep="last")]
    assert isinstance(combined, pd.DataFrame)
    return combined

read_cache(root: str | os.PathLike[str], ticker: str, source: str, frequency: str) -> pd.DataFrame | None

Read a cached dataframe, returning None on miss or corruption.

This function never raises — corruption is signalled via a :class:CacheCorruptionWarning and a None return so callers can transparently fall back to re-fetching.

Source code in src/markowitz/data/cache.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def read_cache(
    root: str | os.PathLike[str],
    ticker: str,
    source: str,
    frequency: str,
) -> pd.DataFrame | None:
    """Read a cached dataframe, returning ``None`` on miss or corruption.

    This function never raises — corruption is signalled via a
    :class:`CacheCorruptionWarning` and a ``None`` return so callers can
    transparently fall back to re-fetching.
    """
    path = cache_path(root, ticker, source, frequency)
    if not path.exists():
        return None
    try:
        return pd.read_parquet(path, engine="pyarrow")
    except Exception as exc:  # broad: any deserialization failure is corruption
        warnings.warn(
            f"Cache file {path} could not be read ({exc!r}); ignoring.",
            CacheCorruptionWarning,
            stacklevel=2,
        )
        return None

reindex_to_sessions(df: pd.DataFrame, *, exchange: str = 'XNYS', on_missing: str = 'raise', max_gap_sessions: int = 1) -> pd.DataFrame

Align df to the canonical session calendar.

Parameters:

Name Type Description Default
df DataFrame

Input frame indexed by tz-naive midnight timestamps.

 required
exchange str

Exchange MIC code understood by pandas_market_calendars.

 'XNYS'
on_missing str

"raise" (default), "ffill" or "nan".

 'raise'
max_gap_sessions int

Maximum number of consecutive missing sessions tolerated when on_missing != "raise". A larger run triggers a :class:CalendarMismatchError.

 1
Source code in src/markowitz/data/calendars.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def reindex_to_sessions(
    df: pd.DataFrame,
    *,
    exchange: str = "XNYS",
    on_missing: str = "raise",
    max_gap_sessions: int = 1,
) -> pd.DataFrame:
    """Align ``df`` to the canonical session calendar.

    Parameters
    ----------
    df:
        Input frame indexed by tz-naive midnight timestamps.
    exchange:
        Exchange MIC code understood by ``pandas_market_calendars``.
    on_missing:
        ``"raise"`` (default), ``"ffill"`` or ``"nan"``.
    max_gap_sessions:
        Maximum number of consecutive missing sessions tolerated when
        ``on_missing != "raise"``. A larger run triggers a
        :class:`CalendarMismatchError`.
    """
    if df.empty:
        return df
    if not isinstance(df.index, pd.DatetimeIndex):
        raise CalendarMismatchError("df.index must be a DatetimeIndex")
    if df.index.tz is not None:
        raise CalendarMismatchError("df.index must be tz-naive")
    if not df.index.is_monotonic_increasing:
        raise CalendarMismatchError("df.index must be monotonically increasing")

    sessions = trading_sessions(df.index[0], df.index[-1], exchange=exchange)
    missing = sessions.difference(df.index)
    if len(missing) == 0:
        return cast(pd.DataFrame, df.reindex(sessions))

    if on_missing == "raise":
        raise CalendarMismatchError(
            f"{len(missing)} expected sessions are missing from the input "
            f"(first missing: {missing[0]!s})"
        )

    # gap-length guard
    in_sessions = sessions.isin(df.index)
    longest_gap = 0
    current_gap = 0
    for present in in_sessions:
        if present:
            current_gap = 0
        else:
            current_gap += 1
            longest_gap = max(longest_gap, current_gap)
    if longest_gap > max_gap_sessions:
        raise CalendarMismatchError(
            f"Gap of {longest_gap} consecutive sessions exceeds max_gap_sessions={max_gap_sessions}"
        )

    reindexed = df.reindex(sessions)
    if on_missing == "ffill":
        return cast(pd.DataFrame, reindexed.ffill())
    if on_missing == "nan":
        return cast(pd.DataFrame, reindexed)
    raise CalendarMismatchError(f"unknown on_missing policy: {on_missing!r}")

risk_free_rate(start: DateLike, end: DateLike, *, series_id: str = 'DGS1MO', frequency: str = 'daily', api_key: str | None = None) -> pd.Series

Fetch a FRED risk-free series and return it at frequency.

The FRED yield series are quoted in annualized percent (e.g. 5.32 means 5.32% per year). This function converts to decimal, forward-fills any missing days inside the window, and resamples / de-annualizes to the requested frequency.

Source code in src/markowitz/data/risk_free.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def risk_free_rate(
    start: DateLike,
    end: DateLike,
    *,
    series_id: str = "DGS1MO",
    frequency: str = "daily",
    api_key: str | None = None,
) -> pd.Series:
    """Fetch a FRED risk-free series and return it at ``frequency``.

    The FRED yield series are quoted in annualized percent (e.g. ``5.32``
    means 5.32% per year). This function converts to decimal,
    forward-fills any missing days inside the window, and resamples /
    de-annualizes to the requested frequency.
    """
    try:
        from fredapi import Fred  # noqa: PLC0415
    except ImportError as exc:
        raise ProviderUnavailableError(
            "fredapi is not installed; install with `pip install fredapi`"
        ) from exc

    key = api_key or os.environ.get("FRED_API_KEY")
    try:
        client = Fred(api_key=key)
        raw = client.get_series(
            series_id,
            observation_start=pd.Timestamp(start).strftime("%Y-%m-%d"),
            observation_end=pd.Timestamp(end).strftime("%Y-%m-%d"),
        )
    except Exception as exc:
        raise ProviderUnavailableError(
            f"FRED fetch failed for series {series_id!r}: {exc}"
        ) from exc

    if raw is None or len(raw) == 0:
        raise EmptyDataError(f"FRED returned no data for series {series_id!r}")

    series = pd.Series(raw, dtype="float64").sort_index() / 100.0
    series.index = pd.DatetimeIndex(series.index).tz_localize(None).normalize()
    series = series[~series.index.duplicated(keep="last")]
    series.name = series_id

    # Quoted as annualized; convert to per-period.
    per_period = de_annualize(series, frequency)
    if frequency == "daily":
        return per_period
    rule = {"weekly": "W", "monthly": "ME", "quarterly": "QE", "annual": "YE"}[frequency]
    # last observation in each period; equivalent to "rate in effect at period end"
    resampled = per_period.resample(rule).last().dropna()
    if resampled.empty:
        raise EmptyDataError(
            f"FRED series {series_id!r} contained no observations after resampling"
        )
    return cast(pd.Series, resampled)

summary_stats(returns: pd.DataFrame, *, frequency: str = 'daily', annualize: bool = True) -> pd.DataFrame

Per-asset summary statistics used as covariance / mean targets.

Returned columns: mu_hat, sigma_hat, sharpe, skew, kurtosis, n_obs, min, max.

When annualize=True (default), mu_hat is scaled by the periods-per-year factor and sigma_hat by its square root.

Source code in src/markowitz/data/loaders.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def summary_stats(
    returns: pd.DataFrame,
    *,
    frequency: str = "daily",
    annualize: bool = True,
) -> pd.DataFrame:
    """Per-asset summary statistics used as covariance / mean targets.

    Returned columns: ``mu_hat``, ``sigma_hat``, ``sharpe``, ``skew``,
    ``kurtosis``, ``n_obs``, ``min``, ``max``.

    When ``annualize=True`` (default), ``mu_hat`` is scaled by the
    periods-per-year factor and ``sigma_hat`` by its square root.
    """
    if returns.empty:
        raise EmptyDataError("returns is empty; cannot compute summary stats")

    n = _periods_per_year(frequency)
    mu = returns.mean()
    sigma = returns.std(ddof=1)
    if annualize:
        mu = mu * n
        sigma = sigma * np.sqrt(n)

    sigma_arr = sigma.to_numpy()
    mu_arr = mu.to_numpy()
    safe = sigma_arr > 0
    ratio = np.full_like(mu_arr, np.nan, dtype="float64")
    np.divide(mu_arr, sigma_arr, out=ratio, where=safe)
    sharpe = pd.Series(ratio, index=mu.index, dtype="float64")

    stats = pd.DataFrame(
        {
            "mu_hat": mu.astype("float64"),
            "sigma_hat": sigma.astype("float64"),
            "sharpe": sharpe,
            "skew": returns.skew().astype("float64"),
            "kurtosis": returns.kurt().astype("float64"),
            "n_obs": returns.notna().sum().astype("int64"),
            "min": returns.min().astype("float64"),
            "max": returns.max().astype("float64"),
        }
    )
    return stats

trading_sessions(start: str | _dt.date | pd.Timestamp, end: str | _dt.date | pd.Timestamp, *, exchange: str = 'XNYS') -> pd.DatetimeIndex

Return tz-naive midnight DatetimeIndex of trading sessions.

Uses :mod:pandas_market_calendars when installed. Falls back to a NYSE weekday-minus-fixed-holidays approximation otherwise.

Source code in src/markowitz/data/calendars.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def trading_sessions(
    start: str | _dt.date | pd.Timestamp,
    end: str | _dt.date | pd.Timestamp,
    *,
    exchange: str = "XNYS",
) -> pd.DatetimeIndex:
    """Return tz-naive midnight ``DatetimeIndex`` of trading sessions.

    Uses :mod:`pandas_market_calendars` when installed. Falls back to a
    NYSE weekday-minus-fixed-holidays approximation otherwise.
    """
    start_ts = pd.Timestamp(start).normalize()
    end_ts = pd.Timestamp(end).normalize()

    try:
        import pandas_market_calendars as mcal  # noqa: PLC0415
    except ImportError:
        return _fallback_sessions(start_ts, end_ts)

    cal = mcal.get_calendar(exchange)
    sched = cal.schedule(start_date=start_ts, end_date=end_ts)
    idx = pd.DatetimeIndex(sched.index).normalize()
    if idx.tz is not None:
        idx = idx.tz_localize(None)
    return idx

write_cache(df: pd.DataFrame, root: str | os.PathLike[str], ticker: str, source: str, frequency: str) -> Path

Atomically write df to the cache and return the final path.

The dataframe is serialized via PyArrow Parquet. The write is performed to a temporary file in the same directory as the target, then renamed with :func:os.replace, which is atomic on POSIX and on Windows (Python 3.3+) when source and destination are on the same volume.

Source code in src/markowitz/data/cache.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def write_cache(
    df: pd.DataFrame,
    root: str | os.PathLike[str],
    ticker: str,
    source: str,
    frequency: str,
) -> Path:
    """Atomically write ``df`` to the cache and return the final path.

    The dataframe is serialized via PyArrow Parquet. The write is
    performed to a temporary file in the same directory as the target,
    then renamed with :func:`os.replace`, which is atomic on POSIX and
    on Windows (Python 3.3+) when source and destination are on the
    same volume.
    """
    target = cache_path(root, ticker, source, frequency)
    target.parent.mkdir(parents=True, exist_ok=True)

    # tempfile in same dir guarantees os.replace is atomic (same volume).
    fd, tmp_name = tempfile.mkstemp(
        prefix=f".{target.name}.",
        suffix=".tmp",
        dir=str(target.parent),
    )
    os.close(fd)
    tmp_path = Path(tmp_name)
    try:
        df.to_parquet(tmp_path, engine="pyarrow", index=True)
        os.replace(tmp_path, target)
    except Exception:
        # Best-effort cleanup; never let cleanup mask the original error.
        with contextlib.suppress(OSError):
            tmp_path.unlink(missing_ok=True)
        raise
    return target