Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions docs/user-guide/features.md
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,44 @@ result = compute_features(df, [
Unknown parameters, repeated output names, and output names that replace input columns
raise `ValueError` before feature execution.

### Derive precise price returns

The next minor version requires Polars 2.0 or newer. Return expressions select their
native arithmetic from the column dtype while planning a query, without collecting
lazy data. Keep integer ticks as integers until after the expression evaluates;
casting large prices to floating point first can erase a one-tick change.

<!-- ml4t-exec -->
```python
import polars as pl
from ml4t.engineer.features.utils.arithmetic import returns, log_returns

prices = pl.DataFrame({
"asset": ["A", "A", "A"],
"close": pl.Series([2**53, 2**53 + 1, 2**53 + 2], dtype=pl.Int64),
})
result = prices.lazy().with_columns(
returns("close").over("asset").alias("return"),
log_returns("close").over("asset").alias("log_return"),
).collect()
assert result["return"][0] is None
assert result["return"][1] == 1 / 2**53
assert result["log_return"][1] > 0
```

Both expressions retain the input row count and leave the first `periods` rows
undefined in each asset. Integer differences are computed before floating division;
floating input keeps its existing ratio formula and dtype. Log returns retain tiny
changes without turning a near-total loss into an artificial negative infinity.
`percentage_change` uses the same simple-return calculation multiplied by 100.

When replacing an Empyrical price-return call, note that its output omits the first
undefined observation. Trim that documented leading row only if the consuming
workflow needs that shape; do not drop every null or compress gaps. These are price
expressions, not portfolio statistics. Use explicit grouping for multiple assets and
validate the declared bar grid before deriving returns as described below. Row-only
data must already have the intended observation order.

### Calendar and bar grid

The calendar-aware default targets the next minor version (0.2), separate from the 0.1.7 patch.
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ dependencies = [
"ml4t-specs>=0.2.0,<0.3.0",

# Data processing (Polars-first)
"polars>=0.20.0",
"polars>=2.0.0",
"pandas>=2.0.0",
"numpy>=1.24.0,<2.6",

Expand Down
41 changes: 35 additions & 6 deletions src/ml4t/engineer/features/utils/arithmetic.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,14 +9,43 @@
from ml4t.engineer.core.exceptions import InvalidParameterError


def _price_return(column: str, periods: int, *, logarithmic: bool) -> pl.Expr:
def build(price: pl.Expr, dtype: pl.DataType) -> pl.Expr:
previous = price.shift(periods)
ratio = price / previous
if not dtype.is_integer():
return ratio.log() if logarithmic else ratio - 1

wide = price.cast(pl.Int128)
lagged = wide.shift(periods)
if dtype == pl.Int128:
# Opposite-sign Int128 differences can exceed even the widest native integer.
same_sign = ((wide >= 0) & (lagged >= 0)) | ((wide <= 0) & (lagged <= 0))
safe_price = pl.when(same_sign).then(wide).otherwise(0)
safe_previous = pl.when(same_sign).then(lagged).otherwise(0)
difference = (safe_price - safe_previous).cast(pl.Float64)
gain = pl.when(same_sign).then(difference / previous).otherwise(ratio - 1)
else:
gain = (wide - lagged).cast(pl.Float64) / previous
if not logarithmic:
return gain
# log1p retains tiny ticks, while the ratio retains near-total losses and domains.
return pl.when(gain.abs() < 0.5).then(gain.log1p()).otherwise(ratio.log())

return pl.col(column).pipe_with_dtype(build)


def returns(column: str, periods: int = 1) -> pl.Expr:
"""Price returns over N periods.

Calculates the percentage change in price over a specified number of periods.
This is one of the most fundamental calculations in quantitative finance.

Formula:
Return = (Price[t] / Price[t-periods]) - 1
Return = (Price[t] - Price[t-periods]) / Price[t-periods]

Integer differences are retained before floating conversion. Floating inputs
retain their existing ratio calculation and output dtype.

Args:
column: Name of the column containing price data
Expand All @@ -30,7 +59,7 @@ def returns(column: str, periods: int = 1) -> pl.Expr:

Example:
>>> import polars as pl
>>> import ml4t.engineer.features.ta as qta
>>> from ml4t.engineer.features.utils import arithmetic as qta
>>>
>>> df = pl.DataFrame({
... "close": [100, 105, 103, 107, 110]
Expand All @@ -44,7 +73,7 @@ def returns(column: str, periods: int = 1) -> pl.Expr:
if periods < 1:
raise InvalidParameterError(f"periods must be >= 1, got {periods}")

return (pl.col(column) / pl.col(column).shift(periods)) - 1
return _price_return(column, periods, logarithmic=False)


def log_returns(column: str, periods: int = 1) -> pl.Expr:
Expand All @@ -69,7 +98,7 @@ def log_returns(column: str, periods: int = 1) -> pl.Expr:

Example:
>>> import polars as pl
>>> import ml4t.engineer.features.ta as qta
>>> from ml4t.engineer.features.utils import arithmetic as qta
>>>
>>> df = pl.DataFrame({
... "close": [100, 105, 103, 107, 110]
Expand All @@ -82,7 +111,7 @@ def log_returns(column: str, periods: int = 1) -> pl.Expr:
if periods < 1:
raise InvalidParameterError(f"periods must be >= 1, got {periods}")

return (pl.col(column) / pl.col(column).shift(periods)).log()
return _price_return(column, periods, logarithmic=True)


def percentage_change(column: str, periods: int = 1) -> pl.Expr:
Expand All @@ -106,7 +135,7 @@ def percentage_change(column: str, periods: int = 1) -> pl.Expr:

Example:
>>> import polars as pl
>>> import ml4t.engineer.features.ta as qta
>>> from ml4t.engineer.features.utils import arithmetic as qta
>>>
>>> df = pl.DataFrame({
... "close": [100, 105, 103, 107, 110]
Expand Down
143 changes: 143 additions & 0 deletions tests/test_returns_precision.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,143 @@
"""Public price expressions must not discard exact integer ticks before division."""

import math

import numpy as np
import polars as pl
import pytest
from polars.testing import assert_frame_equal

from ml4t.engineer import compute_features
from ml4t.engineer.core.exceptions import InvalidParameterError
from ml4t.engineer.features.utils.arithmetic import (
log_returns,
percentage_change,
returns,
)

INTEGER_CASES = [
(pl.Int64, [2**53, 2**53 + 1, 2**53 + 2]),
(pl.Int64, [2**63 - 1, 2**63 - 2, 2**63 - 3]),
(pl.UInt64, [2**63, 2**63 + 1, 2**63 + 2]),
(pl.UInt64, [2**64 - 1, 2**64 - 2, 2**64 - 3]),
(pl.Int64, [-(2**63), 2**63 - 1, -(2**63)]),
(pl.UInt64, [2**64 - 1, 0, 1, 2**64 - 1]),
(pl.UInt64, [2**64 - 1, 1, 2]),
(pl.Int128, [2**127 - 1, 2**127 - 2, 2**127 - 3]),
(pl.Int128, [-(2**127), 2**127 - 1, -(2**127)]),
(pl.Int8, [-128, 127, -128]),
(pl.UInt8, [250, 5, 10]),
(pl.Int64, [100, 101, None, 99, 103]),
(pl.Int64, [0, 0, 10, 0, -10]),
(pl.Int64, [-100, -101, -99, -102]),
(pl.Int64, []),
(pl.Int64, [2**53 + 1]),
]
METHODS = [returns, log_returns, percentage_change]


def _expected(prices, periods, method):
result = []
for index, current in enumerate(prices):
previous = prices[index - periods] if index >= periods else None
if current is None or previous is None:
result.append(None)
continue
with np.errstate(divide="ignore", invalid="ignore"):
change = (current - previous) / previous if previous else np.divide(current, previous)
if method is log_returns:
value = (
np.log1p(change) if abs(change) < 0.5 else np.log(np.divide(current, previous))
)
else:
value = change * (100 if method is percentage_change else 1)
result.append(float(value))
return result


def _assert_values(actual, expected):
assert len(actual) == len(expected)
for observed, wanted in zip(actual, expected, strict=True):
if wanted is None:
assert observed is None
elif math.isnan(wanted):
assert math.isnan(observed)
else:
assert observed == pytest.approx(wanted, rel=2e-15, abs=0)


@pytest.mark.parametrize("dtype,prices", INTEGER_CASES)
@pytest.mark.parametrize("method", METHODS)
@pytest.mark.parametrize("periods", [1, 2])
@pytest.mark.parametrize("lazy", [False, True])
def test_exact_integer_prices_preserve_ticks_limits_and_missingness(
dtype, prices, method, periods, lazy
):
frame = pl.DataFrame({"close": pl.Series(prices, dtype=dtype)})
before = frame.clone()
data = frame.lazy() if lazy else frame
produced = data.select(method("close", periods).alias("return"))
if lazy:
assert isinstance(produced, pl.LazyFrame)
assert produced.collect_schema()["return"] == pl.Float64
produced = produced.collect()
_assert_values(produced["return"].to_list(), _expected(prices, periods, method))
assert_frame_equal(frame, before)


@pytest.mark.parametrize("method", METHODS)
@pytest.mark.parametrize("dtype", [pl.Float32, pl.Float64])
@pytest.mark.parametrize("lazy", [False, True])
def test_float_prices_keep_native_dtype_domain_and_missing_conventions(method, dtype, lazy):
frame = pl.DataFrame(
{"close": pl.Series([100.5, 101.25, None, float("nan"), 0, -1], dtype=dtype)}
)
source = pl.col("close")
ratio = source / source.shift(1)
expected = ratio.log() if method is log_returns else (ratio - 1)
if method is percentage_change:
expected = expected * 100
data = frame.lazy() if lazy else frame
produced = data.select(method("close").alias("return"))
control = data.select(expected.alias("return"))
if lazy:
produced, control = produced.collect(), control.collect()
assert_frame_equal(produced, control)


@pytest.mark.parametrize("method", METHODS)
@pytest.mark.parametrize("lazy", [False, True])
def test_public_price_validation_then_grouped_returns_does_not_mix_assets(method, lazy):
frame = pl.DataFrame(
{
"asset": ["A", "B", "A", "B", "A", "B"],
"close": [2**53, 100, 2**53 + 1, 101, 2**53 + 2, 99],
}
).with_columns(
pl.col("close").alias("open"),
pl.col("close").alias("high"),
pl.col("close").alias("low"),
pl.lit(100).alias("volume"),
)
before = frame.clone()
source = frame.lazy() if lazy else frame
validated = compute_features(source, [], group_col="asset", calendar=None)
produced = validated.with_columns(method("close").over("asset").alias("return"))
if lazy:
produced = produced.collect(engine="streaming")
for asset, prices in (("A", [2**53, 2**53 + 1, 2**53 + 2]), ("B", [100, 101, 99])):
_assert_values(
produced.filter(pl.col("asset") == asset)["return"].to_list(),
_expected(prices, 1, method),
)
assert_frame_equal(frame, before)


@pytest.mark.parametrize("method", METHODS)
@pytest.mark.parametrize("periods", [0, -1])
def test_invalid_lag_fails_before_input_or_lazy_plan_is_changed(method, periods):
frame = pl.DataFrame({"close": [100, 101]})
before = frame.clone()
with pytest.raises(InvalidParameterError, match="periods must be >= 1"):
frame.lazy().select(method("close", periods))
assert_frame_equal(frame, before)
Loading
Loading