aces.types module

This module contains types defined by this package.

These are all simple types using named tuples so can be safely ignored by downstream users provided data fields are passed in the correct order.

class aces.types.TemporalWindowBounds(left_inclusive: bool, window_size: timedelta, right_inclusive: bool, offset: timedelta | None = None)[source]

Bases: object

Named tuple to represent temporal window bounds.

left_inclusive

The start of the window, inclusive.

Type:

bool

window_size

The size of the window.

Type:

datetime.timedelta

right_inclusive

The end of the window, inclusive.

Type:

bool

offset

The offset from the start of the window to the end of the window.

Type:

datetime.timedelta | None

Example

>>> bounds = TemporalWindowBounds(
...     left_inclusive=True,
...     window_size=timedelta(days=1),
...     right_inclusive=False,
...     offset=timedelta(hours=1)
... )
>>> bounds
TemporalWindowBounds(left_inclusive=True,
                     window_size=datetime.timedelta(days=1),
                     right_inclusive=False,
                     offset=datetime.timedelta(seconds=3600))
>>> left_inclusive, window_size, right_inclusive, offset = bounds
>>> bounds.left_inclusive
True
>>> window_size
datetime.timedelta(days=1)
>>> right_inclusive
False
>>> offset
datetime.timedelta(seconds=3600)
left_inclusive : bool
offset : timedelta | None = None
property polars_gp_rolling_kwargs : dict[str, str | datetime.timedelta]

Return the parameters for a group_by rolling operation in Polars.

Examples

>>> TemporalWindowBounds(
...     left_inclusive=True,
...     window_size=timedelta(days=1),
...     right_inclusive=True,
...     offset=None
... ).polars_gp_rolling_kwargs
{'period': datetime.timedelta(days=1),
 'offset': datetime.timedelta(0),
 'closed': 'both'}
>>> TemporalWindowBounds(
...     left_inclusive=True,
...     window_size=timedelta(days=1),
...     right_inclusive=True,
...     offset=timedelta(hours=1)
... ).polars_gp_rolling_kwargs
{'period': datetime.timedelta(days=1),
 'offset': datetime.timedelta(seconds=3600),
 'closed': 'both'}
>>> TemporalWindowBounds(
...     left_inclusive=False,
...     window_size=timedelta(days=2),
...     right_inclusive=False,
...     offset=timedelta(minutes=1)
... ).polars_gp_rolling_kwargs
{'period': datetime.timedelta(days=2),
 'offset': datetime.timedelta(seconds=60),
 'closed': 'none'}
>>> TemporalWindowBounds(
...     left_inclusive=True,
...     window_size=timedelta(days=2),
...     right_inclusive=False,
...     offset=timedelta(minutes=1)
... ).polars_gp_rolling_kwargs
{'period': datetime.timedelta(days=2),
 'offset': datetime.timedelta(seconds=60),
 'closed': 'left'}
>>> TemporalWindowBounds(
...     left_inclusive=False,
...     window_size=timedelta(days=2),
...     right_inclusive=True,
...     offset=timedelta(minutes=1)
... ).polars_gp_rolling_kwargs
{'period': datetime.timedelta(days=2),
 'offset': datetime.timedelta(seconds=60),
 'closed': 'right'}
right_inclusive : bool
window_size : timedelta
class aces.types.ToEventWindowBounds(left_inclusive: bool, end_event: str, right_inclusive: bool, offset: timedelta | None = None)[source]

Bases: object

Named tuple to represent temporal window bounds.

left_inclusive

The start of the window, inclusive.

Type:

bool

end_event

The string name of the event that bounds the end of this window. Operationally, this is interpreted as the string name of the column which contains a positive value if the row corresponds to the end event of this window and a zero otherwise.

Type:

str

right_inclusive

The end of the window, inclusive.

Type:

bool

offset

The offset from the start of the window to the end of the window.

Type:

datetime.timedelta | None

Raises:

Example

>>> bounds = ToEventWindowBounds(
...     left_inclusive=True,
...     end_event="foo",
...     right_inclusive=False,
...     offset=timedelta(hours=1)
... )
>>> bounds
ToEventWindowBounds(left_inclusive=True,
                    end_event='foo',
                    right_inclusive=False,
                    offset=datetime.timedelta(seconds=3600))
>>> left_inclusive, end_event, right_inclusive, offset = bounds
>>> left_inclusive
True
>>> end_event
'foo'
>>> right_inclusive
False
>>> offset
datetime.timedelta(seconds=3600)
>>> ToEventWindowBounds(
...     left_inclusive=True,
...     end_event="",
...     right_inclusive=False,
...     offset=timedelta(hours=1)
... )
Traceback (most recent call last):
    ...
ValueError: The 'end_event' must be a non-empty string.
>>> ToEventWindowBounds(
...     left_inclusive=True,
...     end_event="_RECORD_START",
...     right_inclusive=False,
...     offset=timedelta(hours=1)
... )
Traceback (most recent call last):
    ...
ValueError: It doesn't make sense to have the start of the record _RECORD_START be an end event. Did
you mean to make that be the start event (which should result in the `end_event` parameter being
'-_RECORD_START')?
>>> ToEventWindowBounds(
...     left_inclusive=True,
...     end_event="-_RECORD_END",
...     right_inclusive=False,
...     offset=timedelta(hours=1)
... )
Traceback (most recent call last):
    ...
ValueError: It doesn't make sense to have the end of the record _RECORD_END be a start event. Did
you mean to make that be the end event (which should result in the `end_event` parameter being
'_RECORD_END')?
property boolean_expr_bound_sum_kwargs : dict[str, str | datetime.timedelta | polars.expr.expr.Expr]

Return the parameters for a group_by rolling operation in Polars.

Examples

>>> def print_kwargs(kwargs: dict):
...     for key, value in kwargs.items():
...         print(f"{key}: {value}")
>>> print_kwargs(ToEventWindowBounds(
...     left_inclusive=True, end_event="is_A", right_inclusive=False, offset=None
... ).boolean_expr_bound_sum_kwargs)
boundary_expr: [(col("is_A")) > (dyn int: 0)]
mode: row_to_bound
closed: left
offset: 0:00:00
>>> print_kwargs(ToEventWindowBounds(
...     left_inclusive=False, end_event="-is_B", right_inclusive=True, offset=None
... ).boolean_expr_bound_sum_kwargs)
boundary_expr: [(col("is_B")) > (dyn int: 0)]
mode: bound_to_row
closed: right
offset: 0:00:00
>>> print_kwargs(ToEventWindowBounds(
...     left_inclusive=False, end_event="is_B", right_inclusive=False, offset=timedelta(hours=-3)
... ).boolean_expr_bound_sum_kwargs)
boundary_expr: [(col("is_B")) > (dyn int: 0)]
mode: row_to_bound
closed: none
offset: -1 day, 21:00:00
>>> print_kwargs(ToEventWindowBounds(
...     left_inclusive=True,
...     end_event="-_RECORD_START",
...     right_inclusive=True,
...     offset=timedelta(days=2),
... ).boolean_expr_bound_sum_kwargs)
boundary_expr: [(col("timestamp")) == (col("timestamp").min().over([col("subject_id")]))]
mode: bound_to_row
closed: both
offset: 2 days, 0:00:00
>>> print_kwargs(ToEventWindowBounds(
...     left_inclusive=False,
...     end_event="_RECORD_END",
...     right_inclusive=True,
...     offset=timedelta(days=1),
... ).boolean_expr_bound_sum_kwargs)
boundary_expr: [(col("timestamp")) == (col("timestamp").max().over([col("subject_id")]))]
mode: row_to_bound
closed: right
offset: 1 day, 0:00:00
end_event : str
left_inclusive : bool
offset : timedelta | None = None
right_inclusive : bool