DataFrame

dataframe

DataFrame

DataFrame(df: DataFrame, time_col: str, freq: int | None = None, name: str = '', _skip_validate: bool = False)

Source code in ctalearn/core/dataframe.py

def __init__(
    self,
    df: pl.DataFrame,
    time_col: str,
    freq: int | None = None,
    name: str = "",
    _skip_validate: bool = False,
):
    self.time_col = time_col
    self.name = name

    if _skip_validate:
        self._df = df
        self.freq = freq
    else:
        # 1. Sort & Check (O(N) or O(N log N))
        # We assume user provides mostly clean data, but we must ensure Sort for Polars checks
        sorted_df = df.sort(time_col)

        # 2. Calculate Freq (O(N) scan)
        if freq is not None:
            self.freq = int(freq)
            # A forced freq must actually divide the observed spacing,
            # else off-grid points get silently relabeled onto grid slots.
            self._check_on_grid(sorted_df, self.freq)
        else:
            self.freq = self._calculate_freq(sorted_df)

        # 3. Upsample / Make Dense (O(N))
        # Critical: This guarantees that if two DFs have same start/end/freq, they map 1:1.
        if self.freq and len(sorted_df) > 1:
            self._df = self._ensure_dense(sorted_df, self.freq)
        else:
            self._df = sorted_df

value_col property

value_col: str

The single non-time value column.

Series-shaped consumers (e.g. performance metrics on a pnl frame) use this to locate the value column by structure instead of a hardcoded name. Raises if the frame is not single-series.

Raises:

Type	Description
ValueError	If there is not exactly one non-time column.

shift_time

shift_time(delta: timedelta) -> DataFrame

Shifts the underlying time column by a given timedelta.

Source code in ctalearn/core/dataframe.py

def shift_time(self, delta: timedelta) -> "DataFrame":
    """
    Shifts the underlying time column by a given timedelta.
    """
    new_df = self._df.with_columns(
        (pl.col(self.time_col) + delta).alias(self.time_col)
    )

    return DataFrame(
        new_df,
        self.time_col,
        self.freq,
        f"{self.name}.shift_time({delta!r})",
        _skip_validate=True,
    )

align

align(other: DataFrame, how: str = 'inner') -> tuple[DataFrame, DataFrame]

Explicitly align two DataFrames. :param how: 'inner' (Intersection) or 'outer' (Union + FFill)

Source code in ctalearn/core/dataframe.py

def align(
    self, other: "DataFrame", how: str = "inner"
) -> tuple["DataFrame", "DataFrame"]:
    """
    Explicitly align two DataFrames.
    :param how: 'inner' (Intersection) or 'outer' (Union + FFill)
    """
    if not isinstance(other, DataFrame):
        raise TypeError("Can only align with another DataFrame")

    time_col = self.time_col
    other_time_col = other.time_col

    if self.freq is None or other.freq is None:
        return self._empty(), other._empty()

    # No data on either side -> empty intersection (avoid indexing [0]/[-1])
    if len(self._df) == 0 or len(other._df) == 0:
        return self._empty(), other._empty()

    # Rename for consistency
    other_df_renamed = (
        other._df.rename({other_time_col: time_col})
        if other_time_col != time_col
        else other._df
    )

    start_self = self._df[time_col][0]
    end_self = self._df[time_col][-1]
    start_other = other_df_renamed[time_col][0]
    end_other = other_df_renamed[time_col][-1]

    # Determine Time Bounds
    if how == "outer":
        start_ts = min(start_self, start_other)
        end_ts = max(end_self, end_other)
    else:  # inner
        start_ts = max(start_self, start_other)
        end_ts = min(end_self, end_other)

    if start_ts > end_ts:
        return self._empty(), other._empty()

    # GCD Frequency
    if self.freq == other.freq:
        gcd_freq_s = self.freq
    else:
        gcd_freq_s = math.gcd(self.freq, other.freq)

    # Create Grid
    grid_lf = (
        pl.datetime_range(
            start=start_ts,
            end=end_ts,
            interval=timedelta(seconds=gcd_freq_s),
            eager=True,
        )
        .alias(time_col)
        .to_frame()
        .lazy()
    )

    # Join & Fill Logic
    def _process(base_lazy: pl.LazyFrame, original_freq: int) -> pl.DataFrame:
        # Bound the backfill to one source bar (at gcd resolution) for both
        # how modes. outer extends the grid past a frame's range, so an
        # unbounded fill would fabricate that frame's tail from its last value.
        tol = timedelta(seconds=original_freq - gcd_freq_s)

        return grid_lf.join_asof(
            base_lazy, on=time_col, strategy="backward", tolerance=tol
        ).collect()

    df1_aligned = _process(self._df.lazy(), self.freq)
    df2_aligned = _process(other_df_renamed.lazy(), other.freq)

    return (
        DataFrame(
            df1_aligned, time_col, gcd_freq_s, self.name, _skip_validate=True
        ),
        DataFrame(
            df2_aligned, time_col, gcd_freq_s, other.name, _skip_validate=True
        ),
    )

auto_align

auto_align(method: Callable[..., DataFrame]) -> Callable[..., Any]

Decorator for operators (+, -, *, /). Includes a FAST PATH to skip alignment if DFs are already matched. Otherwise, enforces 'inner' alignment.

Source code in ctalearn/core/dataframe.py

def auto_align(
    method: Callable[..., "DataFrame"],
) -> Callable[..., Any]:
    """
    Decorator for operators (+, -, *, /).
    Includes a FAST PATH to skip alignment if DFs are already matched.
    Otherwise, enforces 'inner' alignment.
    """

    @wraps(method)
    def wrapper(self: "DataFrame", other: Any) -> Any:
        if not isinstance(other, DataFrame):
            return NotImplemented

        # If metadata matches perfectly, row N of self corresponds to row N of other.
        # We assume __init__ guaranteed dense data if freq is present.
        has_data = len(self._df) > 0 and len(other._df) > 0

        if (
            has_data
            and self.freq is not None
            and other.freq is not None
            and self.freq == other.freq
            and len(self._df) == len(other._df)
        ):
            # Check Start Time and End Time
            start_match = self._df[self.time_col][0] == other._df[other.time_col][0]
            end_match = self._df[self.time_col][-1] == other._df[other.time_col][-1]

            if start_match and end_match:
                # Skip align() overhead.
                return method(self, self, other)

        # Explicit Inner Align
        df1_aligned, df2_aligned = self.align(other, how="inner")

        # Handle empty intersection gracefully
        if len(df1_aligned._df) == 0:
            time_col = self.time_col
            other_time_col = other.time_col
            cols_self = set(self._df.columns)
            cols_other = set(other._df.columns)
            common_cols = sorted(
                list((cols_self & cols_other) - {time_col, other_time_col})
            )
            return _return_empty_aligned(
                self, other, method, common_cols, df1_aligned.freq
            )

        return method(self, df1_aligned, df2_aligned)

    return wrapper