195 lines
6.9 KiB
Python
195 lines
6.9 KiB
Python
|
"""
|
||
|
|
Corporate action price adjustment utilities.
|
||
|
|
|
||
|
|
adjust_for_splits(close, split_factors, split_indices)
|
||
|
|
Apply split adjustments to a close price series (backward-adjusted).
|
||
|
|
|
||
|
|
adjust_for_dividends(close, dividends, ex_dates)
|
||
|
|
Apply dividend adjustments to a close price series (backward-adjusted).
|
||
|
|
|
||
|
|
adjust_ohlcv(open_, high, low, close, volume, split_factors=None, split_indices=None,
|
||
|
|
dividends=None, ex_date_indices=None)
|
||
|
|
Apply both split and dividend adjustments to a full OHLCV dataset.
|
||
|
|
Returns (adj_open, adj_high, adj_low, adj_close, adj_volume).
|
||
|
|
"""
|
||
|
|
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
import numpy as np
|
||
|
|
from numpy.typing import ArrayLike, NDArray
|
||
|
|
|
||
|
|
__all__ = ["adjust_for_splits", "adjust_for_dividends", "adjust_ohlcv"]
|
||
|
|
|
||
|
|
|
||
|
|
def adjust_for_splits(
|
||
|
|
close: ArrayLike,
|
||
|
|
split_factors: ArrayLike, # e.g. [2.0, 3.0] means 2-for-1 then 3-for-1
|
||
|
|
split_indices: ArrayLike, # bar indices of each split (must be sorted ascending)
|
||
|
|
) -> NDArray:
|
||
|
|
"""Backward-adjust close prices for stock splits.
|
||
|
|
|
||
|
|
All prices BEFORE a split are divided by the split factor.
|
||
|
|
e.g. a 2-for-1 split at bar 100: prices[0:100] are halved.
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
close : array-like
|
||
|
|
Raw close prices.
|
||
|
|
split_factors : array-like
|
||
|
|
Split factor for each split event (e.g. 2.0 for a 2-for-1 split).
|
||
|
|
split_indices : array-like
|
||
|
|
Bar index of each split event (0-based, must be sorted ascending).
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
NDArray of adjusted close prices.
|
||
|
|
"""
|
||
|
|
c = np.asarray(close, dtype=np.float64).copy()
|
||
|
|
factors = np.asarray(split_factors, dtype=np.float64)
|
||
|
|
indices = np.asarray(split_indices, dtype=np.intp)
|
||
|
|
|
||
|
|
# Process splits in chronological order; apply backward adjustment
|
||
|
|
# (all bars before the split are divided by the factor)
|
||
|
|
for idx, factor in zip(indices, factors):
|
||
|
|
if factor <= 0:
|
||
|
|
raise ValueError(f"split_factor must be > 0, got {factor}")
|
||
|
|
c[:idx] /= factor
|
||
|
|
|
||
|
|
return c
|
||
|
|
|
||
|
|
|
||
|
|
def adjust_for_dividends(
|
||
|
|
close: ArrayLike,
|
||
|
|
dividends: ArrayLike, # dividend amount per ex-date
|
||
|
|
ex_date_indices: ArrayLike, # bar indices of ex-dividend dates
|
||
|
|
) -> NDArray:
|
||
|
|
"""Backward-adjust close prices for cash dividends (proportional method).
|
||
|
|
|
||
|
|
Adjustment factor at ex-date i = (close[i-1] - dividend) / close[i-1].
|
||
|
|
All bars before ex-date are multiplied by the cumulative adjustment.
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
close : array-like
|
||
|
|
Raw close prices.
|
||
|
|
dividends : array-like
|
||
|
|
Dividend amount (in currency units) at each ex-dividend date.
|
||
|
|
ex_date_indices : array-like
|
||
|
|
Bar index of each ex-dividend date (0-based, sorted ascending).
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
NDArray of adjusted close prices.
|
||
|
|
"""
|
||
|
|
c = np.asarray(close, dtype=np.float64).copy()
|
||
|
|
divs = np.asarray(dividends, dtype=np.float64)
|
||
|
|
indices = np.asarray(ex_date_indices, dtype=np.intp)
|
||
|
|
|
||
|
|
# Process in chronological order
|
||
|
|
for idx, div in zip(indices, divs):
|
||
|
|
if idx == 0:
|
||
|
|
# No prior bar; skip adjustment (nothing to adjust)
|
||
|
|
continue
|
||
|
|
prev_close = c[idx - 1]
|
||
|
|
if prev_close <= 0:
|
||
|
|
continue
|
||
|
|
adj_factor = (prev_close - div) / prev_close
|
||
|
|
if adj_factor <= 0:
|
||
|
|
continue
|
||
|
|
# All prices before ex-date are multiplied by adj_factor
|
||
|
|
c[:idx] *= adj_factor
|
||
|
|
|
||
|
|
return c
|
||
|
|
|
||
|
|
|
||
|
|
def adjust_ohlcv(
|
||
|
|
open_: ArrayLike,
|
||
|
|
high: ArrayLike,
|
||
|
|
low: ArrayLike,
|
||
|
|
close: ArrayLike,
|
||
|
|
volume: ArrayLike,
|
||
|
|
split_factors: Optional[ArrayLike] = None,
|
||
|
|
split_indices: Optional[ArrayLike] = None,
|
||
|
|
dividends: Optional[ArrayLike] = None,
|
||
|
|
ex_date_indices: Optional[ArrayLike] = None,
|
||
|
|
) -> tuple[NDArray, NDArray, NDArray, NDArray, NDArray]:
|
||
|
|
"""Apply split and dividend adjustments to full OHLCV data.
|
||
|
|
|
||
|
|
Price arrays are multiplied by cumulative adjustment factor.
|
||
|
|
Volume is divided by split factors (shares outstanding adjust inversely).
|
||
|
|
Returns (adj_open, adj_high, adj_low, adj_close, adj_volume).
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
open_, high, low, close : array-like
|
||
|
|
Raw OHLCV price arrays.
|
||
|
|
volume : array-like
|
||
|
|
Raw volume array.
|
||
|
|
split_factors : array-like, optional
|
||
|
|
Split factors for each split event.
|
||
|
|
split_indices : array-like, optional
|
||
|
|
Bar indices of split events (required if split_factors provided).
|
||
|
|
dividends : array-like, optional
|
||
|
|
Dividend amounts for each ex-date.
|
||
|
|
ex_date_indices : array-like, optional
|
||
|
|
Bar indices of ex-dividend dates (required if dividends provided).
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
(adj_open, adj_high, adj_low, adj_close, adj_volume)
|
||
|
|
"""
|
||
|
|
o = np.asarray(open_, dtype=np.float64).copy()
|
||
|
|
h = np.asarray(high, dtype=np.float64).copy()
|
||
|
|
low_arr = np.asarray(low, dtype=np.float64).copy()
|
||
|
|
c = np.asarray(close, dtype=np.float64).copy()
|
||
|
|
v = np.asarray(volume, dtype=np.float64).copy()
|
||
|
|
|
||
|
|
n = len(c)
|
||
|
|
|
||
|
|
# Build a per-bar cumulative adjustment factor for prices (starts at 1.0)
|
||
|
|
price_adj = np.ones(n, dtype=np.float64)
|
||
|
|
# Separate inverse adjustment for volume (splits only)
|
||
|
|
vol_adj = np.ones(n, dtype=np.float64)
|
||
|
|
|
||
|
|
# -----------------------------------------------------------------------
|
||
|
|
# Apply split adjustments
|
||
|
|
# -----------------------------------------------------------------------
|
||
|
|
if split_factors is not None and split_indices is not None:
|
||
|
|
sf = np.asarray(split_factors, dtype=np.float64)
|
||
|
|
si = np.asarray(split_indices, dtype=np.intp)
|
||
|
|
for idx, factor in zip(si, sf):
|
||
|
|
if factor <= 0:
|
||
|
|
raise ValueError(f"split_factor must be > 0, got {factor}")
|
||
|
|
# Prices before split are divided by factor
|
||
|
|
price_adj[:idx] /= factor
|
||
|
|
# Volume before split is multiplied by factor (more shares pre-split)
|
||
|
|
vol_adj[:idx] *= factor
|
||
|
|
|
||
|
|
# -----------------------------------------------------------------------
|
||
|
|
# Apply dividend adjustments (prices only)
|
||
|
|
# -----------------------------------------------------------------------
|
||
|
|
if dividends is not None and ex_date_indices is not None:
|
||
|
|
divs = np.asarray(dividends, dtype=np.float64)
|
||
|
|
ei = np.asarray(ex_date_indices, dtype=np.intp)
|
||
|
|
# We need the split-adjusted close at (idx-1) for each dividend event.
|
||
|
|
# Instead of recomputing the full array each iteration, read the single
|
||
|
|
# element we need: c[idx-1] * price_adj[idx-1].
|
||
|
|
for idx, div in zip(ei, divs):
|
||
|
|
if idx == 0:
|
||
|
|
continue
|
||
|
|
prev_close = c[idx - 1] * price_adj[idx - 1]
|
||
|
|
if prev_close <= 0:
|
||
|
|
continue
|
||
|
|
adj_factor = (prev_close - div) / prev_close
|
||
|
|
if adj_factor <= 0:
|
||
|
|
continue
|
||
|
|
price_adj[:idx] *= adj_factor
|
||
|
|
|
||
|
|
adj_open = o * price_adj
|
||
|
|
adj_high = h * price_adj
|
||
|
|
adj_low = low_arr * price_adj
|
||
|
|
adj_close = c * price_adj
|
||
|
|
adj_volume = v * vol_adj
|
||
|
|
|
||
|
|
return adj_open, adj_high, adj_low, adj_close, adj_volume
|