Skip to content

Operators

Arithmetic

_arithmetic

sign 

sign(df: DataFrame) -> DataFrame

Return the sign of each element in the DataFrame (excluding the time column).

Returns -1 for negative values, 0 for zero, 1 for positive values, and null for missing (null) or NaN values.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame with numeric value columns.

 required

Returns:

Type Description
DataFrame

New DataFrame with the same shape and time index, containing only -1, 0, 1 or null.

Raises:

Type Description
TypeError

If df is not an instance of the expected DataFrame class.
Source code in ctalearn/operator/_arithmetic.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def sign(df: DataFrame) -> DataFrame:
    """
    Return the sign of each element in the DataFrame (excluding the time column).

    Returns -1 for negative values, 0 for zero, 1 for positive values, and null
    for missing (null) or NaN values.

    Parameters
    ----------
    df : ctalearn.DataFrame
        Input DataFrame with numeric value columns.

    Returns
    -------
    ctalearn.DataFrame
        New DataFrame with the same shape and time index, containing only -1, 0, 1 or null.

    Raises
    ------
    TypeError
        If `df` is not an instance of the expected DataFrame class.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    df = df.copy()
    df.name = f"sign({df.name})"

    return df.select(pl.col(df.time_col), pl.exclude(df.time_col).sign().fill_nan(None))

log 

log(df: DataFrame) -> DataFrame

Calculate the natural logarithm of input array elements. Returns null for non-positive values (<= 0) and for null/NaN input.

Source code in ctalearn/operator/_arithmetic.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def log(df: DataFrame) -> DataFrame:
    """
    Calculate the natural logarithm of input array elements.
    Returns null for non-positive values (<= 0) and for null/NaN input.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    df = df.copy()
    df.name = f"log({df.name})"

    target = pl.exclude(df.time_col)

    # Polars .log() returns -inf for 0 and NaN for negative; emit null instead.
    # NaN input satisfies `target > 0` in Polars, so fill_nan(None) catches it too.
    return df.select(
        pl.col(df.time_col),
        pl.when(target > 0).then(target.log()).otherwise(None).fill_nan(None),
    )

symmetric_log 

symmetric_log(df: DataFrame) -> DataFrame

Calculate the symmetric logarithm. Formula: sign(x) * log(|x| + 1)

Source code in ctalearn/operator/_arithmetic.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def symmetric_log(df: DataFrame) -> DataFrame:
    """
    Calculate the symmetric logarithm.
    Formula: sign(x) * log(|x| + 1)
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    df = df.copy()
    df.name = f"symmetric_log({df.name})"

    target = pl.exclude(df.time_col)

    return df.select(
        pl.col(df.time_col),
        (target.sign() * (target.abs() + 1).log()).fill_nan(None),
    )

sqrt 

sqrt(df: DataFrame) -> DataFrame

Calculate the square root. Returns null for negative values and for null/NaN input.

Source code in ctalearn/operator/_arithmetic.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def sqrt(df: DataFrame) -> DataFrame:
    """
    Calculate the square root.
    Returns null for negative values and for null/NaN input.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    df = df.copy()
    df.name = f"sqrt({df.name})"

    target = pl.exclude(df.time_col)

    # Polars .sqrt() returns NaN for negatives; NaN input also satisfies
    # `target >= 0` in Polars, so emit null in both cases.
    return df.select(
        pl.col(df.time_col),
        pl.when(target >= 0).then(target.sqrt()).otherwise(None).fill_nan(None),
    )

symmetric_sqrt 

symmetric_sqrt(df: DataFrame) -> DataFrame

Calculate the symmetric square root. Formula: sign(x) * sqrt(|x|)

Source code in ctalearn/operator/_arithmetic.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def symmetric_sqrt(df: DataFrame) -> DataFrame:
    """
    Calculate the symmetric square root.
    Formula: sign(x) * sqrt(|x|)
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    df = df.copy()
    df.name = f"symmetric_sqrt({df.name})"

    target = pl.exclude(df.time_col)

    return df.select(
        pl.col(df.time_col), (target.sign() * target.abs().sqrt()).fill_nan(None)
    )

cbrt 

cbrt(df: DataFrame) -> DataFrame

Calculate the cubic root.

Source code in ctalearn/operator/_arithmetic.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
def cbrt(df: DataFrame) -> DataFrame:
    """
    Calculate the cubic root.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    df = df.copy()
    df.name = f"cbrt({df.name})"

    target = pl.exclude(df.time_col)

    return df.select(pl.col(df.time_col), target.cbrt().fill_nan(None))

identity 

identity(df: DataFrame) -> DataFrame

Return the input DataFrame as is.

Source code in ctalearn/operator/_arithmetic.py 
131
132
133
134
135
136
137
138
def identity(df: DataFrame) -> DataFrame:
    """
    Return the input DataFrame as is.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    return df.copy()

Cross-sectional

_cross_sectional

cs_mean 

cs_mean(df: DataFrame, ignore_nan: bool = True) -> DataFrame

Calculate the cross-sectional mean for each timestamp (row-wise / axis=1).

The mean value is broadcast to all original columns, maintaining the same column structure as the input DataFrame.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
ignore_nan bool

If True, ignore NaN/Null values when computing the row mean. If False, if any NaN/Null exists in a row, the row mean becomes null.

 True

Returns:

Type Description
DataFrame

DataFrame with the same column structure as input, where each column contains the row-wise mean value (broadcast to all columns).
Source code in ctalearn/operator/_cross_sectional.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def cs_mean(df: DataFrame, ignore_nan: bool = True) -> DataFrame:
    """
    Calculate the cross-sectional mean for each timestamp (row-wise / axis=1).

    The mean value is broadcast to all original columns, maintaining the same
    column structure as the input DataFrame.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    ignore_nan : bool, default=True
        If True, ignore NaN/Null values when computing the row mean.
        If False, if any NaN/Null exists in a row, the row mean becomes null.

    Returns
    -------
    DataFrame
        DataFrame with the same column structure as input, where each column
        contains the row-wise mean value (broadcast to all columns).
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    df = df.copy()
    asset_cols = [c for c in df._df.columns if c != df.time_col]

    # Convert to Float64 and map NaN -> Null so Polars horizontal
    # aggregations can treat them consistently.
    base = (
        df._df.with_columns(pl.col(asset_cols).cast(pl.Float64).fill_nan(None))
        if asset_cols
        else df._df
    )

    if not asset_cols:
        return DataFrame(
            base.select(pl.col(df.time_col)),
            df.time_col,
            df.freq,
            f"cs_mean({df.name}, ignore_nan={ignore_nan})",
            _skip_validate=True,
        )

    if ignore_nan:
        mean_val = pl.mean_horizontal(asset_cols)
    else:
        any_null = pl.any_horizontal([pl.col(c).is_null() for c in asset_cols])
        # NaN has already been converted to null above.
        mean_val = (
            pl.when(any_null).then(None).otherwise(pl.mean_horizontal(asset_cols))
        )

    # Broadcast mean to all original columns
    out_pl = base.select(pl.col(df.time_col), *[mean_val.alias(c) for c in asset_cols])
    return DataFrame(
        out_pl,
        df.time_col,
        df.freq,
        f"cs_mean({df.name}, ignore_nan={ignore_nan})",
        _skip_validate=True,
    )

cs_rank 

cs_rank(df: DataFrame, ignore_nan: bool = True) -> DataFrame

Rank the values for each timestamp (row-wise).

Rank is scaled to [0, 1] range where 0 is the lowest and 1 is the highest. Break ties by average.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
ignore_nan bool

If True, ignore NaN values and rank only valid values in each row. If False, if any Nan exists in a row, the entire row becomes NaN.

 True

Returns:

Type Description
DataFrame

DataFrame with percentile ranked values (0 to 1).

Raises:

Type Description
TypeError

If df is not a DataFrame.
Source code in ctalearn/operator/_cross_sectional.py 
 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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def cs_rank(df: DataFrame, ignore_nan: bool = True) -> DataFrame:
    """
    Rank the values for each timestamp (row-wise).

    Rank is scaled to [0, 1] range where 0 is the lowest and 1 is the highest.
    Break ties by average.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    ignore_nan : bool, default=True
        If True, ignore NaN values and rank only valid values in each row.
        If False, if any Nan exists in a row, the entire row becomes NaN.

    Returns
    -------
    DataFrame
        DataFrame with percentile ranked values (0 to 1).

    Raises
    ------
    TypeError
        If `df` is not a DataFrame.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    df = df.copy()
    target_cols = [c for c in df.columns if c != df.time_col]
    x = df.select(target_cols).cast(pl.Float64).to_numpy()

    _, num_factors = x.shape

    if num_factors == 1:
        # Lone value ranks at the midpoint, but NaN must stay NaN (-> null below).
        result = np.where(np.isnan(x), np.nan, 0.5)
    else:
        nan_policy = "omit" if ignore_nan else "propagate"
        ranks = rankdata(x, method="average", axis=1, nan_policy=nan_policy)

        if ignore_nan:
            valid_count = (~np.isnan(x)).sum(axis=1, keepdims=True).astype(float)

            with np.errstate(divide="ignore", invalid="ignore"):
                result = (ranks - 1) / (valid_count - 1)

            # Handle rows with single valid value
            result = np.where((valid_count == 1) & (~np.isnan(x)), 0.5, result)
            result = np.where(np.isnan(x), np.nan, result)
        else:
            result = (ranks - 1) / (num_factors - 1)
            nan_mask = np.isnan(x).any(axis=1, keepdims=True)
            result = np.where(nan_mask, np.nan, result)

    res_pl = pl.DataFrame(result, schema=target_cols)
    final_pl = pl.concat(
        [df._df.select(df.time_col), res_pl], how="horizontal"
    ).fill_nan(None)

    return DataFrame(
        final_pl, df.time_col, df.freq, f"cs_rank({df.name}, ignore_nan={ignore_nan})"
    )

cs_zscore 

cs_zscore(df: DataFrame, ignore_nan: bool = True) -> DataFrame

Calculate cross-sectional z-score for each timestamp.

Z-score is calculated as (value - mean) / std for each row. When std is 0 (all values are the same), z-score is set to 0.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
ignore_nan bool

If True, ignore NaN values. If False, if any NaN exists in a row, the entire row becomes NaN.

 True

Returns:

Type Description
DataFrame

DataFrame with z-score normalized values.
Source code in ctalearn/operator/_cross_sectional.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
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
def cs_zscore(df: DataFrame, ignore_nan: bool = True) -> DataFrame:
    """
    Calculate cross-sectional z-score for each timestamp.

    Z-score is calculated as (value - mean) / std for each row.
    When std is 0 (all values are the same), z-score is set to 0.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    ignore_nan : bool, default=True
        If True, ignore NaN values.
        If False, if any NaN exists in a row, the entire row becomes NaN.

    Returns
    -------
    DataFrame
        DataFrame with z-score normalized values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    df = df.copy()
    target_cols = [c for c in df.columns if c != df.time_col]
    x = df.select(target_cols).cast(pl.Float64).to_numpy()

    if ignore_nan:
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", category=RuntimeWarning)
            mean = np.nanmean(x, axis=1, keepdims=True)
            std = np.nanstd(x, axis=1, keepdims=True, ddof=0)
    else:
        mean = np.mean(x, axis=1, keepdims=True)
        std = np.std(x, axis=1, keepdims=True, ddof=0)

    with np.errstate(divide="ignore", invalid="ignore"):
        result = (x - mean) / std

    result = np.where(std == 0, 0, result)

    if ignore_nan:
        result = np.where(np.isnan(x), np.nan, result)
    else:
        nan_mask = np.isnan(x).any(axis=1, keepdims=True)
        result = np.where(nan_mask, np.nan, result)

    res_pl = pl.DataFrame(result, schema=target_cols)
    final_pl = pl.concat(
        [df._df.select(df.time_col), res_pl], how="horizontal"
    ).fill_nan(None)

    return DataFrame(
        final_pl, df.time_col, df.freq, f"cs_zscore({df.name}, ignore_nan={ignore_nan})"
    )

cs_winsorize 

cs_winsorize(df: DataFrame, std: float, ignore_nan: bool = True) -> DataFrame

Winsorize values for each timestamp based on standard deviation.

Winsorizes x to make sure that all values in x are clipped between the lower and upper limits, which are specified as multiple of std.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
std float

Number of standard deviations from mean to use as limits.

 required
ignore_nan bool

If True, ignore NaN values. If False, if any NaN exists in a row, the entire row becomes NaN.

 True

Returns:

Type Description
DataFrame

DataFrame with winsorized values.
Source code in ctalearn/operator/_cross_sectional.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def cs_winsorize(df: DataFrame, std: float, ignore_nan: bool = True) -> DataFrame:
    """
    Winsorize values for each timestamp based on standard deviation.

    Winsorizes x to make sure that all values in x are clipped between the lower and upper limits,
    which are specified as multiple of std.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    std : float
        Number of standard deviations from mean to use as limits.
    ignore_nan : bool, default=True
        If True, ignore NaN values.
        If False, if any NaN exists in a row, the entire row becomes NaN.

    Returns
    -------
    DataFrame
        DataFrame with winsorized values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    df = df.copy()
    target_cols = [c for c in df.columns if c != df.time_col]
    x = df.select(target_cols).cast(pl.Float64).to_numpy()

    if ignore_nan:
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", category=RuntimeWarning)
            mean = np.nanmean(x, axis=1, keepdims=True)
            std_dev = np.nanstd(x, axis=1, keepdims=True, ddof=0)
    else:
        mean = np.mean(x, axis=1, keepdims=True)
        std_dev = np.std(x, axis=1, keepdims=True, ddof=0)

    lower_limit = mean - std * std_dev
    upper_limit = mean + std * std_dev
    result = np.clip(x, lower_limit, upper_limit)

    if ignore_nan:
        result = np.where(np.isnan(x), np.nan, result)
    else:
        nan_mask = np.isnan(x).any(axis=1, keepdims=True)
        result = np.where(nan_mask, np.nan, result)

    res_pl = pl.DataFrame(result, schema=target_cols)
    final_pl = pl.concat(
        [df._df.select(df.time_col), res_pl], how="horizontal"
    ).fill_nan(None)

    return DataFrame(
        final_pl,
        df.time_col,
        df.freq,
        f"cs_winsorize({df.name}, std={std}, ignore_nan={ignore_nan})",
    )

vector_neut 

vector_neut(x: DataFrame, y: DataFrame) -> DataFrame

Perform cross-sectional vector neutralization. Formula: x - (dot(x, y) / dot(y, y)) * y

For given vectors x and y, it finds a new vector x' (output) such that x' is orthogonal to y.

Parameters:

Name Type Description Default
x DataFrame

Input DataFrame X

 required
y DataFrame

Input DataFrame Y (Target)

 required

Returns:

Type Description
DataFrame

DataFrame with neutralized values.
Source code in ctalearn/operator/_cross_sectional.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def vector_neut(x: DataFrame, y: DataFrame) -> DataFrame:
    """
    Perform cross-sectional vector neutralization.
    Formula: x - (dot(x, y) / dot(y, y)) * y

    For given vectors x and y, it finds a new vector x' (output) such that x' is orthogonal to y.

    Parameters
    ----------
    x : DataFrame
        Input DataFrame X
    y : DataFrame
        Input DataFrame Y (Target)

    Returns
    -------
    DataFrame
        DataFrame with neutralized values.
    """
    if not isinstance(x, DataFrame):
        raise TypeError(f"Type of `x` should be DataFrame, got {type(x).__name__}")

    if not isinstance(y, DataFrame):
        raise TypeError(f"Type of `y` should be DataFrame, got {type(y).__name__}")

    x = x.copy()
    y = y.copy()

    x, y = x.align(y, how="inner")

    # align() only aligns timestamps, not columns. The projection reads y[c] for
    # every column c of x, so a column present in one but not the other would
    # otherwise surface as an opaque KeyError.
    x_cols = set(x._df.columns) - {x.time_col}
    y_cols = set(y._df.columns) - {y.time_col}
    if x_cols != y_cols:
        raise ValueError("`x` and `y` must have the same asset columns")

    value_cols = sorted(list(x_cols))

    xy_dot = (x * y).select(
        pl.col(x.time_col), pl.sum_horizontal(pl.exclude(x.time_col)).alias("dot")
    )
    yy_dot_exprs = [
        pl.when(x.get_column(c).is_not_null())
        .then(pl.col(c) * pl.col(c))
        .otherwise(None)
        for c in value_cols
    ]
    yy_dot = y.select(pl.col(y.time_col), pl.sum_horizontal(yy_dot_exprs).alias("dot"))
    proj_coeff = xy_dot / (yy_dot + 1e-12)

    df = x.select(
        pl.col(x.time_col),
        *[
            (pl.col(c) - proj_coeff.get_column("dot") * y.get_column(c)).alias(c)
            for c in value_cols
        ],
    )
    df.name = f"vector_neut({x.name}, {y.name})"
    return df

regression_neut 

regression_neut(target: DataFrame, neut_factors: DataFrame | list[DataFrame], ridge_alpha: float = 1e-08) -> DataFrame

Fully vectorized cross-sectional regression using Batch Linear Algebra. No explicit Python loops over timestamps.

Source code in ctalearn/operator/_cross_sectional.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
def regression_neut(
    target: DataFrame,
    neut_factors: DataFrame | list[DataFrame],
    ridge_alpha: float = 1e-8,
) -> DataFrame:
    """
    Fully vectorized cross-sectional regression using Batch Linear Algebra.
    No explicit Python loops over timestamps.
    """
    if isinstance(neut_factors, DataFrame):
        neut_factors = [neut_factors]

    neut_names = ",".join([f.name for f in neut_factors])
    new_name = f"regression_neut({target.name}, [{neut_names}])"

    # 1. Align DataFrames
    aligned_target = target
    for f in neut_factors:
        aligned_target, _ = aligned_target.align(f, how="inner")

    if len(aligned_target._df) == 0:
        cols = sorted(list(set(target._df.columns) - {target.time_col}))
        return _return_empty_aligned(target, target, lambda x, y, z: x, cols, 1)

    aligned_neuts: list[DataFrame] = []
    for f in neut_factors:
        _, f_aligned = aligned_target.align(f, how="inner")
        aligned_neuts.append(f_aligned)

    time_col = target.time_col
    common_cols = set(aligned_target._df.columns) - {time_col}
    for f in aligned_neuts:
        common_cols &= set(f._df.columns) - {f.time_col}

    if not common_cols:
        raise ValueError("No common columns (assets) found.")

    sorted_cols = sorted(list(common_cols))

    # Y: (T, N)
    Y_mat = aligned_target._df.select(sorted_cols).to_numpy().astype(np.float64)
    # X: (T, N, K)
    X_list = [
        f._df.select(sorted_cols).to_numpy().astype(np.float64) for f in aligned_neuts
    ]
    X_mat = np.stack(X_list, axis=-1)

    # 2. Batch Linear Regression
    T, N = Y_mat.shape
    K = len(aligned_neuts)

    nan_mask = np.isnan(Y_mat) | np.isnan(X_mat).any(axis=2)
    valid_mask = ~nan_mask

    Y_filled = np.nan_to_num(Y_mat, nan=0.0)
    X_filled = np.nan_to_num(X_mat, nan=0.0)

    intercept = valid_mask.astype(float)  # (T, N)
    X_design = np.concatenate([intercept[:, :, None], X_filled], axis=2)

    X_design = X_design * valid_mask[:, :, None]

    Xt = X_design.transpose(0, 2, 1)
    XtX = Xt @ X_design

    Y_target = (Y_filled * valid_mask)[:, :, None]
    XtY = Xt @ Y_target  # (T, K+1, 1)

    if ridge_alpha > 0:
        I_ridge = np.eye(K + 1)
        I_ridge[0, 0] = 0
        XtX += ridge_alpha * I_ridge

    n_valid_counts = valid_mask.sum(axis=1)
    bad_dof_mask = n_valid_counts < (K + 2)

    if bad_dof_mask.any():
        XtX[bad_dof_mask] = np.eye(K + 1)
        XtY[bad_dof_mask] = 0

    try:
        betas = np.linalg.solve(XtX, XtY)
    except np.linalg.LinAlgError:
        return DataFrame(
            pl.concat(
                [
                    aligned_target._df.select(time_col),
                    pl.DataFrame(np.full((T, N), np.nan), schema=sorted_cols),
                ],
                how="horizontal",
            ),
            time_col,
            aligned_target.freq,
            name=new_name,
            _skip_validate=True,
        )

    preds = X_design @ betas
    residuals = Y_target - preds
    residuals = residuals.squeeze(-1)  # (T, N)

    residuals[~valid_mask] = np.nan
    residuals[bad_dof_mask, :] = np.nan

    # 3. Reconstruct DataFrame
    res_df_pl = pl.DataFrame(residuals, schema=sorted_cols, orient="row")
    final_pl = pl.concat(
        [aligned_target._df.select(time_col), res_df_pl], how="horizontal"
    )

    return DataFrame(
        final_pl, time_col, freq=aligned_target.freq, name=new_name, _skip_validate=True
    )

process_alpha_weights 

process_alpha_weights(alpha_signals: DataFrame, neutralize: bool = True) -> DataFrame

Processes raw alpha signal matrix by applying optional neutralization and subsequent normalization to each row independently.

This function transforms a matrix of raw alpha scores into a matrix of portfolio weight vectors, ensuring NaN values are converted to zero weights.

Parameters:

Name Type Description Default
alpha_signals DataFrame

A DataFrame of raw alpha values, where each row represents a different period or asset universe.

 required
neutralize bool

If True, each row (signal vector) is market-neutralized by subtracting its mean (calculated from non-NaN values), resulting in row sums of zero (Long/Short strategy). If False, no mean subtraction is performed.

 True

Returns:

Type Description
DataFrame

For each row, the absolute sum of non-zero weights is scaled to 1.0. All input NaN values are set to 0.0, indicating no capital allocation.
Source code in ctalearn/operator/_cross_sectional.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
def process_alpha_weights(
    alpha_signals: DataFrame, neutralize: bool = True
) -> DataFrame:
    """
    Processes raw alpha signal matrix by applying optional neutralization
    and subsequent normalization to each row independently.

    This function transforms a matrix of raw alpha scores into a matrix of
    portfolio weight vectors, ensuring NaN values are converted to zero weights.

    Parameters
    ----------
    alpha_signals : DataFrame
        A DataFrame of raw alpha values, where each row represents a
        different period or asset universe.
    neutralize : bool, default=True
        If True, each row (signal vector) is market-neutralized by
        subtracting its mean (calculated from non-NaN values), resulting
        in row sums of zero (Long/Short strategy).
        If False, no mean subtraction is performed.

    Returns
    -------
    DataFrame
        For each row, the absolute sum of non-zero weights is scaled to 1.0.
        All input NaN values are set to 0.0, indicating no capital allocation.
    """
    if not isinstance(alpha_signals, DataFrame):
        raise TypeError(
            f"Type of `df` should be DataFrame, got {type(alpha_signals).__name__}"
        )

    df = alpha_signals._df
    asset_cols = [col for col in df.columns if col != alpha_signals.time_col]
    asset_exprs = pl.col(asset_cols)

    # 1. Neutralization
    if neutralize:
        row_means = pl.mean_horizontal(asset_cols)
        df = df.with_columns(
            (pl.col(name) - row_means).alias(name) for name in asset_cols
        )

    # 2. Normalization
    row_abs_sums = pl.sum_horizontal(asset_exprs.abs())
    df = df.with_columns(
        (pl.col(name) / row_abs_sums).fill_nan(0.0).fill_null(0.0).alias(name)
        for name in asset_cols
    )

    return DataFrame(
        df, alpha_signals.time_col, freq=alpha_signals.freq, _skip_validate=True
    )

cs_pca 

cs_pca(df: DataFrame, window: int, n_components: int = 1, output: Literal['scores', 'loadings', 'explained_variance'] = 'scores') -> DataFrame | tuple[DataFrame, ...]

Perform high-performance vectorized rolling cross-sectional PCA with Sign Correction.

Includes Max Loading Sign Correction to ensure consistent signal direction.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame with time column and feature columns (symbols).

 required
window int

Rolling window size. Must be > 1.

 required
n_components int

Number of principal components to compute.

 1
output (scores, loadings, explained_variance)

Type of output to return: - "scores": Principal component scores - "loadings": Principal component loadings (eigenvectors) - "explained_variance": Explained variance ratio for each component

 "scores"

Returns:

Type Description
DataFrame or tuple[DataFrame, ...]
  • If n_components == 1: Returns a single DataFrame
  • If n_components > 1: Returns a tuple of DataFrames (pc1_df, pc2_df, ...)

For all output types, each DataFrame has the same column structure as the input: - "loadings": Each column contains the loading value for that feature/symbol - "scores": The scalar score is broadcast to all original columns - "explained_variance": The scalar variance ratio is broadcast to all original columns

Raises:

Type Description
TypeError

If df is not a DataFrame.
ValueError

If window <= 1, n_components <= 0, n_components > number of features, or output is not a valid option.
Source code in ctalearn/operator/_cross_sectional.py 
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
def cs_pca(
    df: DataFrame,
    window: int,
    n_components: int = 1,
    output: Literal["scores", "loadings", "explained_variance"] = "scores",
) -> "DataFrame | tuple[DataFrame, ...]":
    """
    Perform high-performance vectorized rolling cross-sectional PCA with Sign Correction.

    Includes Max Loading Sign Correction to ensure consistent signal direction.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame with time column and feature columns (symbols).
    window : int
        Rolling window size. Must be > 1.
    n_components : int, default=1
        Number of principal components to compute.
    output : {"scores", "loadings", "explained_variance"}, default="scores"
        Type of output to return:
        - "scores": Principal component scores
        - "loadings": Principal component loadings (eigenvectors)
        - "explained_variance": Explained variance ratio for each component

    Returns
    -------
    DataFrame or tuple[DataFrame, ...]
        - If n_components == 1: Returns a single DataFrame
        - If n_components > 1: Returns a tuple of DataFrames (pc1_df, pc2_df, ...)

        For all output types, each DataFrame has the same column structure as the input:
        - "loadings": Each column contains the loading value for that feature/symbol
        - "scores": The scalar score is broadcast to all original columns
        - "explained_variance": The scalar variance ratio is broadcast to all original columns

    Raises
    ------
    TypeError
        If `df` is not a DataFrame.
    ValueError
        If `window` <= 1, `n_components` <= 0, `n_components` > number of features,
        or `output` is not a valid option.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 1:
        raise ValueError(f"`window` must be an integer > 1, got {window}")

    if n_components <= 0:
        raise ValueError(f"`n_components` must be a positive integer")

    valid_outputs = ("scores", "loadings", "explained_variance")
    if output not in valid_outputs:
        raise ValueError(f"`output` must be one of {valid_outputs}, got '{output}'")

    # 1. Prepare data
    df_copy = df.copy()
    target_cols = [c for c in df_copy.columns if c != df_copy.time_col]
    num_features = len(target_cols)

    if n_components > num_features:
        raise ValueError(
            f"`n_components` ({n_components}) exceeds number of features ({num_features})"
        )

    # To Numpy matrix
    X = df_copy.select(target_cols).cast(pl.Float64).to_numpy()
    n_rows, n_cols = X.shape

    def _build_result_dataframes(
        result_data_list: list[np.ndarray],
        target_cols: list[str],
        time_df: pl.DataFrame,
        time_col: str,
        freq: int | None,
        df_name: str,
        output_type: str,
    ) -> "DataFrame | tuple[DataFrame, ...]":
        """Build result DataFrames from computed data arrays."""
        result_dfs = []
        for pc_idx, data in enumerate(result_data_list):
            res_pl = pl.DataFrame(data, schema=target_cols)
            final_pl = pl.concat([time_df, res_pl], how="horizontal").fill_nan(None)
            pc_name = f"cs_pca({df_name}, {window}, {n_components}, output='{output_type}')_PC{pc_idx + 1}"
            result_dfs.append(
                DataFrame(final_pl, time_col, freq, pc_name, _skip_validate=True)
            )
        if len(result_dfs) == 1:
            return result_dfs[0]
        return tuple(result_dfs)

    # Not enough rows, only pad result with nan
    if n_rows < window:
        nan_data = np.full((n_rows, num_features), np.nan)
        result_data_list = [nan_data for _ in range(n_components)]
        return _build_result_dataframes(
            result_data_list,
            target_cols,
            df._df.select(df.time_col),
            df.time_col,
            df.freq,
            df.name,
            output,
        )

    # 2. Sliding window
    X_windows = sliding_window_view(X, window_shape=(window, n_cols)).reshape(
        -1, window, n_cols
    )

    # 3. Vectorized Z-Score normalization
    means = np.mean(X_windows, axis=1, keepdims=True)
    stds = np.std(X_windows, axis=1, ddof=0, keepdims=True)
    stds = np.where(stds < 1e-12, 1.0, stds)
    X_norm = (X_windows - means) / stds

    # 4. Vectorized covariance
    cov_matrices = (X_norm.transpose(0, 2, 1) @ X_norm) / (window - 1)

    # 5. Eigenvalues & vectors
    try:
        eig_vals, eig_vecs = np.linalg.eigh(cov_matrices)
    except np.linalg.LinAlgError:
        eig_vals = np.full((len(cov_matrices), n_cols), np.nan)
        eig_vecs = np.full((len(cov_matrices), n_cols, n_cols), np.nan)

    eig_vals = eig_vals[:, ::-1]
    eig_vecs = eig_vecs[:, :, ::-1]

    top_eig_vals = eig_vals[:, :n_components]  # (N_win, k)
    top_eig_vecs = eig_vecs[:, :, :n_components]  # (N_win, F, k)

    # -- Sign correction
    if output in ["scores", "loadings"]:
        max_abs_idx = np.argmax(np.abs(top_eig_vecs), axis=1)  # (N_win, k)
        n_win_range = np.arange(top_eig_vecs.shape[0])[:, None]  # (N_win, 1)
        k_range = np.arange(n_components)[None, :]  # (1, k)
        max_vals = top_eig_vecs[n_win_range, max_abs_idx, k_range]
        signs = np.sign(max_vals)
        signs[signs == 0] = 1.0
        top_eig_vecs = top_eig_vecs * signs[:, np.newaxis, :]

    # Pad
    pad_len = window - 1

    # -- Output: Build result data list for each PC
    result_data_list = []

    if output == "scores":
        current_obs = X_norm[:, -1, :][:, np.newaxis, :]
        scores = (current_obs @ top_eig_vecs).squeeze(axis=1)  # (N_win, k)
        # Broadcast each scalar score to all columns
        for pc_idx in range(n_components):
            score_col = scores[:, pc_idx : pc_idx + 1]  # (N_win, 1)
            broadcasted = np.broadcast_to(score_col, (scores.shape[0], num_features))
            padded = np.vstack([np.full((pad_len, num_features), np.nan), broadcasted])
            result_data_list.append(padded)

    elif output == "loadings":
        # top_eig_vecs: (N_win, F, k) - each column is already the loading for that feature
        for pc_idx in range(n_components):
            loadings = top_eig_vecs[:, :, pc_idx]  # (N_win, F)
            padded = np.vstack([np.full((pad_len, num_features), np.nan), loadings])
            result_data_list.append(padded)

    else:  # explained_variance
        total_var = eig_vals.sum(axis=1, keepdims=True)
        total_var[total_var == 0] = 1.0
        explained_ratio = top_eig_vals / total_var  # (N_win, k)
        # Broadcast each scalar variance ratio to all columns
        for pc_idx in range(n_components):
            var_col = explained_ratio[:, pc_idx : pc_idx + 1]  # (N_win, 1)
            broadcasted = np.broadcast_to(
                var_col, (explained_ratio.shape[0], num_features)
            )
            padded = np.vstack([np.full((pad_len, num_features), np.nan), broadcasted])
            result_data_list.append(padded)

    return _build_result_dataframes(
        result_data_list,
        target_cols,
        df._df.select(df.time_col),
        df.time_col,
        df.freq,
        df.name,
        output,
    )

Time-series

_time_series

ts_rank 

ts_rank(df: DataFrame, window: int, constant: float = 0.0) -> DataFrame

Rank the values for each column over the past window size.

Returns the rank of the current value plus constant. Rank is scaled to [0, 1] range.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
window int

Size of the rolling window.

 required
constant float

Constant to add to the rank.

 0.0

Returns:

Type Description
DataFrame

DataFrame with ranked values.

Raises:

Type Description
TypeError

If df is not of type DataFrame.
ValueError

If window is not a positive integer.
Source code in ctalearn/operator/_time_series.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 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
 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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def ts_rank(df: DataFrame, window: int, constant: float = 0.0) -> DataFrame:
    """
    Rank the values for each column over the past window size.

    Returns the rank of the current value plus constant.
    Rank is scaled to [0, 1] range.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    window : int
        Size of the rolling window.
    constant : float, default=0.0
        Constant to add to the rank.

    Returns
    -------
    DataFrame
        DataFrame with ranked values.

    Raises
    ------
    TypeError
        If `df` is not of type DataFrame.
    ValueError
        If `window` is not a positive integer.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df = df.copy()
    df.name = f"ts_rank({df.name}, {window}, constant={constant})"

    # Edge case: window == 1 -> rank of a lone value is the midpoint 0.5.
    # Still honor `constant` and null out NaN/inf, like the window>=2 path.
    if window == 1:
        return df.select(
            pl.col(df.time_col),
            (pl.exclude(df.time_col) * 0 + 0.5 + constant).fill_nan(None),
        )

    @njit(  # type: ignore[untyped-decorator]
        cache=True,
        nogil=True,
        parallel=False,
        fastmath={"nsz", "ninf", "reassoc", "arcp", "contract", "afn"},
    )
    def _numba_rolling_rank(
        arr: NDArray[np.float64], window: int, constant: float
    ) -> NDArray[np.float64]:
        n = len(arr)
        result = np.full(n, np.nan, dtype=np.float64)

        if n < window:
            return result

        denom = 1.0 / (window - 1)
        raw_buffer = np.full(window, np.nan, dtype=np.float64)
        nan_count = window

        for i in range(n):
            current_val = arr[i]
            idx = i % window

            if np.isnan(raw_buffer[idx]):
                nan_count -= 1

            raw_buffer[idx] = current_val

            if np.isnan(raw_buffer[idx]):
                nan_count += 1

            if nan_count > 0:
                continue

            less_count = 0
            equal_count = 0

            for x in raw_buffer:
                if x < current_val:
                    less_count += 1
                elif x == current_val:
                    equal_count += 1

            rank = less_count + (equal_count - 1) * 0.5
            result[i] = (rank * denom) + constant

        return result

    target_cols = [c for c in df._df.columns if c != df.time_col]

    return df.select(
        pl.col(df.time_col),
        *[
            pl.col(col)
            .map_batches(
                lambda s: _numba_rolling_rank(
                    s.to_numpy().astype(np.float64), window, constant
                ),
                return_dtype=pl.Float64,
            )
            .fill_nan(None)
            .alias(col)
            for col in target_cols
        ],
    )

ts_mean 

ts_mean(df: DataFrame, window: int) -> DataFrame

Returns average value over the past window periods.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
window int

Size of the rolling window.

 required

Returns:

Type Description
DataFrame

DataFrame with rolling mean values.
Source code in ctalearn/operator/_time_series.py 
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
148
149
def ts_mean(df: DataFrame, window: int) -> DataFrame:
    """
    Returns average value over the past `window` periods.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    window : int
        Size of the rolling window.

    Returns
    -------
    DataFrame
        DataFrame with rolling mean values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df = df.copy()
    df.name = f"ts_mean({df.name}, {window})"

    return df.select(
        pl.col(df.time_col), pl.exclude(df.time_col).rolling_mean(window_size=window)
    )

ts_median 

ts_median(df: DataFrame, window: int) -> DataFrame

Returns median value over the past window periods.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
window int

Size of the rolling window.

 required

Returns:

Type Description
DataFrame

DataFrame with rolling median values. If window is even, return the average of the 2 center values.
Source code in ctalearn/operator/_time_series.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def ts_median(df: DataFrame, window: int) -> DataFrame:
    """
    Returns median value over the past `window` periods.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    window : int
        Size of the rolling window.

    Returns
    -------
    DataFrame
        DataFrame with rolling median values. If window is even, return the
        average of the 2 center values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df = df.copy()
    df.name = f"ts_median({df.name}, {window})"

    return df.select(
        pl.col(df.time_col), pl.exclude(df.time_col).rolling_median(window_size=window)
    )

ts_std_dev 

ts_std_dev(df: DataFrame, window: int, ddof: int = 0) -> DataFrame

Return standard deviation over the past window periods.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
window int

Size of the rolling window.

 required
ddof int

Delta degree of freedom.

 0

Returns:

Type Description
DataFrame

DataFrame with rolling standard deviation values.
Source code in ctalearn/operator/_time_series.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def ts_std_dev(df: DataFrame, window: int, ddof: int = 0) -> DataFrame:
    """
    Return standard deviation over the past `window` periods.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    window : int
        Size of the rolling window.
    ddof : int
        Delta degree of freedom.

    Returns
    -------
    DataFrame
        DataFrame with rolling standard deviation values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df = df.copy()
    df.name = f"ts_std_dev({df.name}, {window}, ddof={ddof})"

    return df.select(
        pl.col(df.time_col),
        pl.exclude(df.time_col).rolling_std(window_size=window, ddof=ddof),
    )

ts_zscore 

ts_zscore(df: DataFrame, window: int) -> DataFrame

Return z-score over the past window periods.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
window int

Size of the rolling window.

 required

Returns:

Type Description
DataFrame

DataFrame with rolling z-score values.
Source code in ctalearn/operator/_time_series.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def ts_zscore(df: DataFrame, window: int) -> DataFrame:
    """
    Return z-score over the past `window` periods.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    window : int
        Size of the rolling window.

    Returns
    -------
    DataFrame
        DataFrame with rolling z-score values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df = df.copy()
    df.name = f"ts_zscore({df.name}, {window})"

    target = pl.exclude(df.time_col)

    zscore_expr = (
        (target - target.rolling_mean(window))
        / target.rolling_std(window, ddof=0).fill_null(0)
    ).fill_nan(0)

    return df.select(pl.col(df.time_col), zscore_expr)

ts_robust_zscore 

ts_robust_zscore(df: DataFrame, window: int) -> DataFrame

Calculate robust Z-score using median and median absolute deviation (MAD).

Formula: (x - median) / (c * MAD) Constant c = 1.4826. Returns 0 if MAD is 0 (constant window).

Source code in ctalearn/operator/_time_series.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
def ts_robust_zscore(df: DataFrame, window: int) -> DataFrame:
    """
    Calculate robust Z-score using median and median absolute deviation (MAD).

    Formula: (x - median) / (c * MAD)
    Constant c = 1.4826.
    Returns 0 if MAD is 0 (constant window).
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df = df.copy()
    final_name = f"ts_robust_zscore({df.name}, {window})"

    @njit(  # type: ignore[untyped-decorator]
        cache=True,
        nogil=True,
        parallel=False,
        fastmath={"nsz", "ninf", "reassoc", "arcp", "contract", "afn"},
    )
    def _numba_robust_zscore(
        arr: NDArray[np.float64], medians: NDArray[np.float64], window: int
    ) -> NDArray[np.float64]:
        n = len(arr)
        result = np.full(n, np.nan, dtype=np.float64)

        if n < window:
            return result

        mad_scale = 1.4826
        raw_buffer = np.full(window, np.nan, dtype=np.float64)
        nan_count = window

        for i in range(n):
            idx = i % window

            if np.isnan(raw_buffer[idx]):
                nan_count -= 1

            raw_buffer[idx] = arr[i]

            if np.isnan(raw_buffer[idx]):
                nan_count += 1

            if nan_count > 0:
                continue

            median = medians[i]
            if np.isnan(median):
                continue

            mad = np.median(np.abs(raw_buffer - median))

            adjusted_mad = mad_scale * mad
            if adjusted_mad > 1e-12:
                result[i] = (arr[i] - median) / adjusted_mad
            else:
                result[i] = 0.0

        return result

    target_cols = [c for c in df._df.columns if c != df.time_col]

    df_median = ts_median(df, window=window).rename(
        {col: f"{col}_temp" for col in target_cols}
    )
    df = df.concat(df_median)

    df = df.select(
        pl.col(df.time_col),
        *[
            pl.struct(
                [
                    pl.col(col),
                    pl.col(f"{col}_temp"),
                ]
            )
            .map_batches(
                lambda s, col=col: _numba_robust_zscore(  # type: ignore[misc]
                    s.struct.field(col).to_numpy().astype(np.float64),
                    s.struct.field(f"{col}_temp").to_numpy().astype(np.float64),
                    window,
                ),
                return_dtype=pl.Float64,
            )
            .fill_nan(None)
            .alias(col)
            for col in target_cols
        ],
    )

    df.name = final_name
    return df

ts_sum 

ts_sum(df: DataFrame, window: int) -> DataFrame

Returns sum of values over the past window periods.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
window int

Size of the rolling window.

 required

Returns:

Type Description
DataFrame

DataFrame with rolling sum values.
Source code in ctalearn/operator/_time_series.py 
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
def ts_sum(df: DataFrame, window: int) -> DataFrame:
    """
    Returns sum of values over the past `window` periods.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    window : int
        Size of the rolling window.

    Returns
    -------
    DataFrame
        DataFrame with rolling sum values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df = df.copy()
    df.name = f"ts_sum({df.name}, {window})"

    return df.select(
        pl.col(df.time_col), pl.exclude(df.time_col).rolling_sum(window_size=window)
    )

ts_min 

ts_min(df: DataFrame, window: int) -> DataFrame

Returns min of values over the past window periods.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
window int

Size of the rolling window.

 required

Returns:

Type Description
DataFrame

DataFrame with rolling min values.
Source code in ctalearn/operator/_time_series.py 
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
def ts_min(df: DataFrame, window: int) -> DataFrame:
    """
    Returns min of values over the past `window` periods.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    window : int
        Size of the rolling window.

    Returns
    -------
    DataFrame
        DataFrame with rolling min values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df = df.copy()
    df.name = f"ts_min({df.name}, {window})"

    return df.select(
        pl.col(df.time_col), pl.exclude(df.time_col).rolling_min(window_size=window)
    )

ts_max 

ts_max(df: DataFrame, window: int) -> DataFrame

Returns max of values over the past window periods.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
window int

Size of the rolling window.

 required

Returns:

Type Description
DataFrame

DataFrame with rolling max values.
Source code in ctalearn/operator/_time_series.py 
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
def ts_max(df: DataFrame, window: int) -> DataFrame:
    """
    Returns max of values over the past `window` periods.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    window : int
        Size of the rolling window.

    Returns
    -------
    DataFrame
        DataFrame with rolling max values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df = df.copy()
    df.name = f"ts_max({df.name}, {window})"

    return df.select(
        pl.col(df.time_col), pl.exclude(df.time_col).rolling_max(window_size=window)
    )

ts_scale 

ts_scale(df: DataFrame, window: int, constant: float = 0.0) -> DataFrame

Returns min-max scaling over the past window periods.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
window int

Size of the rolling window.

 required

Returns:

Type Description
DataFrame

DataFrame with rolling max values.
Source code in ctalearn/operator/_time_series.py 
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
def ts_scale(df: DataFrame, window: int, constant: float = 0.0) -> DataFrame:
    """
    Returns min-max scaling over the past `window` periods.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    window : int
        Size of the rolling window.

    Returns
    -------
    DataFrame
        DataFrame with rolling max values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df = df.copy()
    final_name = f"ts_scale({df.name}, {window}, constant={constant})"

    mins = ts_min(df, window=window)
    maxs = ts_max(df, window=window)
    result_df = (df - mins) / (maxs - mins + 1e-12) + constant
    result_df.name = final_name
    return result_df

ts_decay_linear 

ts_decay_linear(df: DataFrame, window: int, dense: bool = True) -> DataFrame

Returns the linear decay on values over the past window periods.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
window int

Size of the rolling window.

 required
dense bool

If True, returns NaN if any value in window is NaN. If False, treat NaN as 0.

 True

Returns:

Type Description
DataFrame

DataFrame with linearly decayed values.
Source code in ctalearn/operator/_time_series.py 
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
def ts_decay_linear(df: DataFrame, window: int, dense: bool = True) -> DataFrame:
    """
    Returns the linear decay on values over the past `window` periods.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    window : int
        Size of the rolling window.
    dense : bool, default=True
        If True, returns NaN if any value in window is NaN.
        If False, treat NaN as 0.

    Returns
    -------
    DataFrame
        DataFrame with linearly decayed values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df = df.copy()
    df.name = f"ts_decay_linear({df.name}, {window}, dense={dense})"

    weights = [float(w) for w in range(1, window + 1)]

    def _apply_decay(col_name: str) -> pl.Expr:
        c = pl.col(col_name)
        values = (
            c.fill_null(0).fill_nan(0).rolling_mean(weights=weights, window_size=window)
        )

        if dense:
            has_invalid = (c.is_null() | c.is_nan()).rolling_max(window_size=window)
            mask = pl.when(has_invalid).then(None).otherwise(1.0)
            return (values * mask).alias(col_name)
        else:
            return values.alias(col_name)

    target_cols = [c for c in df._df.columns if c != df.time_col]

    return df.select(pl.col(df.time_col), *[_apply_decay(c) for c in target_cols])

ts_delay 

ts_delay(df: DataFrame, d: int) -> DataFrame

Returns values delayed by d periods.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
d int

Lag periods.

 required

Returns:

Type Description
DataFrame

DataFrame with lagged values.
Source code in ctalearn/operator/_time_series.py 
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
def ts_delay(df: DataFrame, d: int) -> DataFrame:
    """
    Returns values delayed by `d` periods.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    d : int
        Lag periods.

    Returns
    -------
    DataFrame
        DataFrame with lagged values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if d <= 0:
        raise ValueError(f"`d` must be a positive integer, got {d} instead")

    df = df.copy()
    df.name = f"ts_delay({df.name}, {d})"

    return df.select(pl.col(df.time_col), pl.exclude(df.time_col).shift(d))

ts_delta 

ts_delta(df: DataFrame, d: int) -> DataFrame

Returns x - ts_delay(x, d).

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
d int

Lag periods.

 required

Returns:

Type Description
DataFrame

DataFrame with delta values.
Source code in ctalearn/operator/_time_series.py 
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
def ts_delta(df: DataFrame, d: int) -> DataFrame:
    """
    Returns x - ts_delay(x, d).

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    d : int
        Lag periods.

    Returns
    -------
    DataFrame
        DataFrame with delta values.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if d <= 0:
        raise ValueError(f"`d` must be a positive integer, got {d} instead")

    df = df.copy()
    df.name = f"ts_delta({df.name}, {d})"

    return df.select(pl.col(df.time_col), pl.exclude(df.time_col).diff(d))

ts_ffill 

ts_ffill(df: DataFrame, limit: int | None = None) -> DataFrame

Perform forward fill.

Parameters:

Name Type Description Default
df DataFrame

Input DataFrame.

 required
limit int | None

Max consecutive values to fill.

 None

Returns:

Type Description
DataFrame

Forward-filled DataFrame.
Source code in ctalearn/operator/_time_series.py 
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
def ts_ffill(df: DataFrame, limit: int | None = None) -> DataFrame:
    """
    Perform forward fill.

    Parameters
    ----------
    df : DataFrame
        Input DataFrame.
    limit : int | None
        Max consecutive values to fill.

    Returns
    -------
    DataFrame
        Forward-filled DataFrame.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if limit is not None and limit <= 0:
        raise ValueError(f"`limit` must be a positive integer, got {limit} instead")

    df = df.copy()
    df.name = f"ts_ffill({df.name}, limit={limit})"

    return df.select(
        pl.col(df.time_col),
        pl.exclude(df.time_col).fill_null(strategy="forward", limit=limit),
    )

ts_corr 

ts_corr(df_a: DataFrame, df_b: DataFrame, window: int) -> DataFrame

Compute rolling correlation between common columns of two DataFrames.

Performs an inner join on their respective time_col to align timestamps, then calculates the rolling correlation for columns that exist in both DataFrames.

Parameters:

Name Type Description Default
df_a DataFrame

First input DataFrame.

 required
df_b DataFrame

Second input DataFrame.

 required
window int

Size of the rolling window.

 required

Returns:

Type Description
DataFrame

DataFrame with rolling correlation values for common columns. The time column name will follow df_a.time_col.
Source code in ctalearn/operator/_time_series.py 
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
def ts_corr(df_a: DataFrame, df_b: DataFrame, window: int) -> DataFrame:
    """
    Compute rolling correlation between common columns of two DataFrames.

    Performs an inner join on their respective `time_col` to align timestamps,
    then calculates the rolling correlation for columns that exist in both DataFrames.

    Parameters
    ----------
    df_a : DataFrame
        First input DataFrame.
    df_b : DataFrame
        Second input DataFrame.
    window : int
        Size of the rolling window.

    Returns
    -------
    DataFrame
        DataFrame with rolling correlation values for common columns.
        The time column name will follow `df_a.time_col`.
    """
    if not isinstance(df_a, DataFrame):
        raise TypeError(
            f"Type of `df_a` should be DataFrame, got {type(df_a).__name__}"
        )
    if not isinstance(df_b, DataFrame):
        raise TypeError(
            f"Type of `df_b` should be DataFrame, got {type(df_b).__name__}"
        )

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    df_a = df_a.copy()
    df_b = df_b.copy()

    time_a = df_a.time_col
    time_b = df_b.time_col

    cols_a = set(df_a._df.columns) - {time_a}
    cols_b = set(df_b._df.columns) - {time_b}
    common_cols = sorted(list(cols_a & cols_b))

    if not common_cols:
        return DataFrame(
            df_a._df.select(pl.col(time_a)).clear(),
            time_a,
            df_a.freq,
            "None",
            _skip_validate=True,
        )

    if time_a == time_b:
        joined = df_a._df.join(df_b._df, on=time_a, how="inner", suffix="_right")
    else:
        joined = df_a._df.join(
            df_b._df, left_on=time_a, right_on=time_b, how="inner", suffix="_right"
        )

    select_exprs = [pl.col(time_a)]

    for col in common_cols:
        expr = pl.rolling_corr(
            pl.col(col), pl.col(f"{col}_right"), window_size=window
        ).alias(col)
        select_exprs.append(expr)

    result_pl = joined.select(select_exprs)

    return DataFrame(
        result_pl, time_a, df_a.freq, f"ts_corr({df_a.name}, {df_b.name}, {window})"
    )

ts_hurst_exponent 

ts_hurst_exponent(df: DataFrame, window: int, lag_start: int = 2, lag_end: int = 20) -> DataFrame

Rolling Hurst exponent estimate (slope of log-log plot of lag vs std of differences).

For each rolling window, compute: tau(lag) = std(x[t] - x[t-lag]) hurst = slope of log10(tau) regressed on log10(lag)

Notes
  • The degree-1 fit is solved in closed form (weights . y) inside a single Numba kernel; mathematically identical to a least-squares polyfit, agreeing to ~1e-11 relative.
  • If a window contains NaN/null or has insufficient valid lags, the output is null for that row.
  • lag_start/lag_end follow Python range(start, end) semantics (end is exclusive), matching the common reference implementation.
Source code in ctalearn/operator/_time_series.py 
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
def ts_hurst_exponent(
    df: DataFrame, window: int, lag_start: int = 2, lag_end: int = 20
) -> DataFrame:
    """
    Rolling Hurst exponent estimate (slope of log-log plot of lag vs std of differences).

    For each rolling window, compute:
      tau(lag) = std(x[t] - x[t-lag])
      hurst = slope of log10(tau) regressed on log10(lag)

    Notes
    -----
    - The degree-1 fit is solved in closed form (`weights . y`) inside a single
      Numba kernel; mathematically identical to a least-squares polyfit, agreeing
      to ~1e-11 relative.
    - If a window contains NaN/null or has insufficient valid lags, the output is null for that row.
    - `lag_start`/`lag_end` follow Python `range(start, end)` semantics (end is exclusive),
      matching the common reference implementation.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if window <= 0:
        raise ValueError(f"`window` must be a positive integer, got {window} instead")

    lag_start, lag_end = int(lag_start), int(lag_end)
    lag_end = min(lag_end, window)

    if lag_start <= 0 or lag_end <= 0 or lag_end < lag_start:
        raise ValueError(
            f"`lag_start`/`lag_end` must satisfy 0 < start <= end, "
            f"got ({lag_start}, {lag_end})"
        )
    if lag_end - lag_start < 2:
        raise ValueError(f"lag_end - lag_start < 2, all values will be null.")

    df = df.copy()
    df.name = f"ts_hurst_exponent({df.name}, {window}, {lag_start}, {lag_end})"

    # OLS slope weights for the fixed regressor log10(lag): slope = weights . y.
    x = np.log10(np.arange(lag_start, lag_end, dtype=np.float64))
    xc = x - x.mean()
    weights = xc / (xc @ xc)

    target_cols = [col for col in df._df.columns if col != df.time_col]

    return df.select(
        pl.col(df.time_col),
        *[
            pl.col(col)
            .map_batches(
                lambda s: _hurst_slopes(
                    s.to_numpy().astype(np.float64),
                    window,
                    lag_start,
                    lag_end,
                    weights,
                ),
                return_dtype=pl.Float64,
            )
            .fill_nan(None)
            .alias(col)
            for col in target_cols
        ],
    )

Strategy

_strategy

trend 

trend(df: DataFrame, threshold: float) -> DataFrame

Calculate trend positions based on threshold values.

Logic: - 1 if x >= threshold - -1 if x <= -threshold - 0 if x is NaN (Input NaN handling) - Hold (Forward Fill) otherwise

Source code in ctalearn/operator/_strategy.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def trend(df: DataFrame, threshold: float) -> DataFrame:
    """
    Calculate trend positions based on threshold values.

    Logic:
    - 1  if x >= threshold
    - -1 if x <= -threshold
    - 0  if x is NaN (Input NaN handling)
    - Hold (Forward Fill) otherwise
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if threshold <= 0:
        raise ValueError(f"threshold must be greater than 0, got {threshold} instead")

    df = df.copy()
    df.name = f"trend({df.name}, {threshold})"

    target = pl.exclude(df.time_col)

    # Polars expression equivalent to the numpy logic:
    # Priority 1: Input is Null -> Signal 0
    # Priority 2: x >= threshold -> Signal 1
    # Priority 3: x <= -threshold -> Signal -1
    # Priority 4: Otherwise -> Null (which represents "Hold")
    # Finally: Forward Fill to propagate signals during "Hold" periods

    return df.select(
        pl.col(df.time_col),
        pl.when(target.is_null() | target.is_nan())
        .then(0.0)
        .when(target >= threshold)
        .then(1.0)
        .when(target <= -threshold)
        .then(-1.0)
        .otherwise(None)
        .fill_null(strategy="forward")
        .name.keep(),
    )

trend_reverse 

trend_reverse(df: DataFrame, threshold: float) -> DataFrame

Calculate trend reverse positions. Simple negation of trend strategy.

Source code in ctalearn/operator/_strategy.py 
49
50
51
52
53
54
def trend_reverse(df: DataFrame, threshold: float) -> DataFrame:
    """
    Calculate trend reverse positions.
    Simple negation of trend strategy.
    """
    return -trend(df, threshold)

trend_mean_reversion 

trend_mean_reversion(df: DataFrame, threshold: float) -> DataFrame

Calculate mean reversion positions.

Logic: - 1 (Long) if x >= threshold - -1 (Short) if x <= -threshold - 0 (Close) if x crosses 0 (positive or negative crossover) AND not opening a position - 0 if x is NaN - Hold (Forward Fill) otherwise

Source code in ctalearn/operator/_strategy.py 
 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
101
102
103
104
105
106
107
108
109
110
def trend_mean_reversion(df: DataFrame, threshold: float) -> DataFrame:
    """
    Calculate mean reversion positions.

    Logic:
    - 1 (Long) if x >= threshold
    - -1 (Short) if x <= -threshold
    - 0 (Close) if x crosses 0 (positive or negative crossover) AND not opening a position
    - 0 if x is NaN
    - Hold (Forward Fill) otherwise
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if threshold <= 0:
        raise ValueError(f"threshold must be greater than 0, got {threshold} instead")

    df = df.copy()
    df.name = f"trend_mean_reversion({df.name}, {threshold})"

    # Define expressions for clarity
    curr = pl.exclude(df.time_col)

    # Prev value, fill initial null/NaN with 0 to match numpy logic (x_shifted[0, :] = 0)
    prev = curr.shift(1).fill_null(0.0).fill_nan(0.0)

    # 1. Crossing Logic
    # Cross positive: Prev < 0 AND Curr > 0
    cross_pos = (prev < 0) & (curr > 0)
    # Cross negative: Prev > 0 AND Curr < 0
    cross_neg = (prev > 0) & (curr < 0)

    cross_zero = cross_pos | cross_neg

    # 2. Construct Signal Logic
    # Note: The order of `when` clauses determines priority.
    # In original numpy code: open triggers override close triggers.
    # signals[close_triggers] = 0 where close_triggers = (cross) & (~open)
    # This implies Open Logic has precedence over Close Logic.

    return df.select(
        pl.col(df.time_col),
        pl.when(curr.is_null() | curr.is_nan())
        .then(0.0)  # Input NaN -> 0
        .when(curr >= threshold)
        .then(1.0)  # Open Long
        .when(curr <= -threshold)
        .then(-1.0)  # Open Short
        .when(cross_zero)
        .then(0.0)  # Close Position (Cross 0)
        .otherwise(None)  # Hold
        .fill_null(strategy="forward")
        .name.keep(),
    )

trend_reverse_mean_reversion 

trend_reverse_mean_reversion(df: DataFrame, threshold: float) -> DataFrame

Calculate reverse mean reversion positions. Negation of trend_mean_reversion.

Source code in ctalearn/operator/_strategy.py 
113
114
115
116
117
118
def trend_reverse_mean_reversion(df: DataFrame, threshold: float) -> DataFrame:
    """
    Calculate reverse mean reversion positions.
    Negation of trend_mean_reversion.
    """
    return -trend_mean_reversion(df, threshold)

trend_fast 

trend_fast(df: DataFrame, threshold: float) -> DataFrame

Calculate fast trend positions. Logic: - 1 (Long) if x >= threshold - -1 (Short) if x <= -threshold - 0 (Close) if x crosses threshold (x < threshold or x > -threshold) AND not opening a position - 0 if x is NaN - Hold (Forward Fill) otherwise

Source code in ctalearn/operator/_strategy.py 
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
148
149
150
151
152
153
154
155
156
157
158
159
def trend_fast(df: DataFrame, threshold: float) -> DataFrame:
    """
    Calculate fast trend positions.
    Logic:
    - 1 (Long) if x >= threshold
    - -1 (Short) if x <= -threshold
    - 0 (Close) if x crosses threshold (x < threshold or x > -threshold) AND not opening a position
    - 0 if x is NaN
    - Hold (Forward Fill) otherwise
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if threshold <= 0:
        raise ValueError(f"threshold must be greater than 0, got {threshold} instead")

    df = df.copy()
    df.name = f"trend_fast({df.name}, {threshold})"

    curr = pl.exclude(df.time_col)
    prev = curr.shift(1).fill_null(0.0).fill_nan(0.0)
    cross_threshold = (prev >= threshold) & (curr < threshold) | (
        prev <= -threshold
    ) & (curr > -threshold)

    return df.select(
        pl.col(df.time_col),
        pl.when(curr.is_null() | curr.is_nan())
        .then(0.0)
        .when(curr >= threshold)
        .then(1.0)
        .when(curr <= -threshold)
        .then(-1.0)
        .when(cross_threshold)
        .then(0.0)
        .otherwise(None)
        .fill_null(strategy="forward")
        .name.keep(),
    )

trend_reverse_fast 

trend_reverse_fast(df: DataFrame, threshold: float) -> DataFrame

Calculate reverse fast trend positions. Negation of trend_fast.

Source code in ctalearn/operator/_strategy.py 
162
163
164
165
166
167
def trend_reverse_fast(df: DataFrame, threshold: float) -> DataFrame:
    """
    Calculate reverse fast trend positions.
    Negation of trend_fast.
    """
    return -trend_fast(df, threshold)

trend_time 

trend_time(df: DataFrame, threshold: float, step: int) -> DataFrame

Fixed-duration trend signal.

Logic (per column): - Start long (1) when x >= threshold - Start short (-1) when x <= -threshold - Once started, keep the position for step rows (including the trigger row) - After step rows, revert back to 0 - Null values do not create triggers; output is determined by the most recent trigger within the last step rows (so the window can continue through nulls)

Notes

This is a "pulse"-style signal (duration-limited), not a forward-filled hold-until-close.

Source code in ctalearn/operator/_strategy.py 
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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def trend_time(df: DataFrame, threshold: float, step: int) -> DataFrame:
    """Fixed-duration trend signal.

    Logic (per column):
    - Start long (1) when x >= threshold
    - Start short (-1) when x <= -threshold
    - Once started, keep the position for `step` rows (including the trigger row)
    - After `step` rows, revert back to 0
    - Null values do not create triggers; output is determined by the most recent trigger
      within the last `step` rows (so the window can continue through nulls)

    Notes
    -----
    This is a "pulse"-style signal (duration-limited), not a forward-filled hold-until-close.
    """
    if not isinstance(df, DataFrame):
        raise TypeError(f"Type of `df` should be DataFrame, got {type(df).__name__}")

    if threshold <= 0:
        raise ValueError(f"threshold must be greater than 0, got {threshold} instead")

    if step <= 0:
        raise ValueError(f"step must be greater than 0, got {step} instead")

    df = df.copy()
    df.name = f"trend_time({df.name}, {threshold}, step={step})"

    time_col = df.time_col
    value_cols = [c for c in df.columns if c != time_col]
    idx = pl.int_range(0, pl.len())

    exprs: list[pl.Expr] = []
    for c in value_cols:
        x = pl.col(c)
        # NaN/null must not trigger (note: NaN >= threshold is True in Polars)
        valid = x.is_not_null() & x.is_not_nan()
        trigger = (
            pl.when(valid & (x >= threshold))
            .then(1.0)
            .when(valid & (x <= -threshold))
            .then(-1.0)
            .otherwise(None)
        )

        # O(n) per column:
        # 1) Forward-fill last trigger value and its row index
        # 2) Keep it only if it's within `step` rows; otherwise output 0
        last_val = trigger.fill_null(strategy="forward")
        last_idx = (
            pl.when(trigger.is_not_null())
            .then(idx)
            .otherwise(None)
            .fill_null(strategy="forward")
        )

        exprs.append(
            pl.when(last_idx.is_null())
            .then(0.0)
            .when((idx - last_idx) < step)
            .then(last_val)
            .otherwise(0.0)
            .alias(c)
        )

    return df.select(pl.col(time_col), *exprs)