Skip to content

API

Tick Expressions

Activity

calc_burstiness(self, per='s')

Calculates burstiness as std(inter-trade time) / mean(inter-trade time).

Parameters:

Name Type Description Default
self ExprOrStr

timestamp column

 required
per str

time unit ("s", "ms", "us", "ns")

 's'

Returns:

Type Description
Expr

Float representing burstiness of trading activity.
Source code in ffn_polars/expr/tick/activity.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Datetime)
@auto_alias("burstiness")
def calc_burstiness(self: ExprOrStr, per: str = "s") -> pl.Expr:
    """
    Calculates burstiness as std(inter-trade time) / mean(inter-trade time).

    Args:
        self: timestamp column
        per: time unit ("s", "ms", "us", "ns")

    Returns:
        Float representing burstiness of trading activity.
    """
    scale = SCALE.get(per)
    if scale is None:
        raise ValueError(f"Unsupported unit: {per}")

    itt_ns = self.diff().dt.total_nanoseconds()
    return itt_ns.std().cast(pl.Float64) / itt_ns.mean().cast(pl.Float64)

calc_inter_trade_time(self, per='s')

Calculates the average time between consecutive trades.

Parameters:

Name Type Description Default
self ExprOrStr

Timestamp column

 required
per str

Time unit — "s", "ms", "us", or "ns"

 's'

Returns:

Type Description
Expr

Float expression representing mean inter-trade time in desired unit
Example

df.group_by("ticker").agg( pl.col("timestamp").calc_inter_trade_time(per="ms") )

Source code in ffn_polars/expr/tick/activity.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Datetime)
@auto_alias("inter_trade_time")
def calc_inter_trade_time(self: ExprOrStr, per: str = "s") -> pl.Expr:
    """
    Calculates the average time between consecutive trades.

    Args:
        self: Timestamp column
        per: Time unit — "s", "ms", "us", or "ns"

    Returns:
        Float expression representing mean inter-trade time in desired unit

    Example:
        df.group_by("ticker").agg(
            pl.col("timestamp").calc_inter_trade_time(per="ms")
        )
    """
    scale = SCALE.get(per)
    if scale is None:
        raise ValueError(f"Unsupported time unit: {per}")

    return self.diff().dt.total_nanoseconds().mean() / scale

calc_trade_rate(self, per='ms')

Calculates trade rate as number of trades per second.

Assumes self is a timestamp column from tick data.

Returns:

Type Description
Expr

An expression representing trades per second:
Expr

(count of rows) / (max(timestamp) - min(timestamp)).seconds
Example

df.group_by("ticker").agg( pl.col("timestamp").calc_trade_rate() )

Source code in ffn_polars/expr/tick/activity.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Datetime)
@auto_alias("trade_rate")
def calc_trade_rate(self: ExprOrStr, per: str = "ms") -> pl.Expr:
    """
    Calculates trade rate as number of trades per second.

    Assumes `self` is a timestamp column from tick data.

    Returns:
        An expression representing trades per second:
        (count of rows) / (max(timestamp) - min(timestamp)).seconds

    Example:
        df.group_by("ticker").agg(
            pl.col("timestamp").calc_trade_rate()
        )
    """
    scale = SCALE.get(per)
    return pl.count().cast(pl.Float64) / (
        (self.last() - self.first()).dt.total_nanoseconds() / scale
    )

Bars

Direction

apply_tick_rule_to_volume(self, price)

Applies the tick rule to volume data. Args: self: Volume column price: Price column Returns: Volume with sign based on tick rule

Source code in ffn_polars/expr/tick/direction.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Datetime)
@guard_expr("price", expected_dtype=pl.Float64)
def apply_tick_rule_to_volume(self: ExprOrStr, price: ExprOrStr) -> pl.Expr:
    """
    Applies the tick rule to volume data.
    Args:
        self: Volume column
        price: Price column
    Returns:
        Volume with sign based on tick rule
    """
    return self.cast(pl.Float64) * price.ffn.tick_rule()

calc_tick_imbalance(self)

Calculates tick imbalance using signed volume.

Parameters:

Name Type Description Default
self ExprOrStr

signed tick rule / direction column

 required

Returns:

Type Description
Expr

Float between -1 and 1
Source code in ffn_polars/expr/tick/direction.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("tick_imbalance")
def calc_tick_imbalance(self: ExprOrStr) -> pl.Expr:
    """
    Calculates tick imbalance using signed volume.

    Args:
        self: signed tick rule / direction column

    Returns:
        Float between -1 and 1
    """
    return self.sum().cast(pl.Float64) / self.len().cast(pl.Float64)

tick_rule(self)

Infers trade direction using the tick rule

+1 if price > prev_price -1 if price < prev_price 0 otherwise

Source code in ffn_polars/expr/tick/direction.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("direction")
def tick_rule(self: ExprOrStr) -> pl.Expr:
    """
    Infers trade direction using the tick rule:
        +1 if price > prev_price
        -1 if price < prev_price
         0 otherwise
    """
    return (
        pl.when(self > self.shift(1))
        .then(1)
        .when(self < self.shift(1))
        .then(-1)
        .otherwise(0)
    )

Flow

calc_order_flow_imbalance(self)

Calculates Order Flow Imbalance (OFI) as the sum of signed volume.

Assumes volume is signed

+V = buyer-initiated -V = seller-initiated

Returns:

Type Description
Expr

Float (positive = net buying, negative = net selling)
Example

df.group_by("ticker").agg( pl.col("signed_volume").calc_order_flow_imbalance() )

Source code in ffn_polars/expr/tick/flow.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("order_flow_imbalance")
def calc_order_flow_imbalance(self: ExprOrStr) -> pl.Expr:
    """
    Calculates Order Flow Imbalance (OFI) as the sum of signed volume.

    Assumes volume is signed:
        +V = buyer-initiated
        -V = seller-initiated

    Returns:
        Float (positive = net buying, negative = net selling)

    Example:
        df.group_by("ticker").agg(
            pl.col("signed_volume").calc_order_flow_imbalance()
        )
    """
    return self.sum().cast(pl.Float64)

calc_traded_value(self, volume)

Calculates traded value (price × volume sum).

Parameters:

Name Type Description Default
self ExprOrStr

Float column of trade prices

 required
volume ExprOrStr

Numeric column of trade volumes

 required

Returns:

Name Type Description
Float Expr

total traded value (dollar volume)
Example

df.group_by("ticker").agg( calc_traded_value("price", "volume") )

Source code in ffn_polars/expr/tick/flow.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Float64)
@guard_expr("volume", expected_dtype=pl.Float64)
@auto_alias("traded_value")
def calc_traded_value(self: ExprOrStr, volume: ExprOrStr) -> pl.Expr:
    """
    Calculates traded value (price × volume sum).

    Args:
        self: Float column of trade prices
        volume: Numeric column of trade volumes

    Returns:
        Float: total traded value (dollar volume)

    Example:
        df.group_by("ticker").agg(
            calc_traded_value("price", "volume")
        )
    """
    return (self * volume).sum().cast(pl.Float64)

calc_volume_rate(self, ts, per='s')

Calculates volume traded per unit time.

Parameters:

Name Type Description Default
self ExprOrStr

Numeric column of trade volumes

 required
ts ExprOrStr

Datetime column

 required
per str

"s", "ms", "us", or "ns" (default "s")

 's'

Returns:

Type Description
Expr

Float expression representing volume per unit time
Example

df.group_by("ticker").agg( calc_volume_rate("volume", "timestamp", per="ms") )

Source code in ffn_polars/expr/tick/flow.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
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Int64)
@guard_expr("ts", expected_dtype=pl.Datetime)
@auto_alias("volume_rate")
def calc_volume_rate(self: ExprOrStr, ts: ExprOrStr, per: str = "s") -> pl.Expr:
    """
    Calculates volume traded per unit time.

    Args:
        self: Numeric column of trade volumes
        ts: Datetime column
        per: "s", "ms", "us", or "ns" (default "s")

    Returns:
        Float expression representing volume per unit time

    Example:
        df.group_by("ticker").agg(
            calc_volume_rate("volume", "timestamp", per="ms")
        )
    """
    scale = SCALE.get(per)
    if scale is None:
        raise ValueError(f"Unsupported time unit: {per}")

    return self.sum().cast(pl.Float64) / (
        (ts.max() - ts.min()).dt.total_nanoseconds() / scale
    )

calc_vwap(self, volume)

Calculates volume-weighted average price (VWAP).

Formula

VWAP = sum(price * volume) / sum(volume)

Returns:

Type Description
Expr

A float expression representing VWAP
Example

df.group_by("ticker").agg( calc_vwap("price", "volume") )

Source code in ffn_polars/expr/tick/flow.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@register(namespace="tick")
@guard_expr("price", expected_dtype=pl.Float64)
@guard_expr("volume", expected_dtype=pl.Float64)
@auto_alias("vwap")
def calc_vwap(self: ExprOrStr, volume: ExprOrStr) -> pl.Expr:
    """
    Calculates volume-weighted average price (VWAP).

    Formula:
        VWAP = sum(price * volume) / sum(volume)

    Returns:
        A float expression representing VWAP

    Example:
        df.group_by("ticker").agg(
            calc_vwap("price", "volume")
        )
    """
    return (self * volume).sum() / volume.sum().cast(pl.Float64)

Latency

Price

calc_micro_returns(self)

Calculates log returns at tick level

log(p_t) - log(p_{t-1})

Returns:

Type Description
Expr

Tick-level log return series
Source code in ffn_polars/expr/tick/price.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("micro_returns")
def calc_micro_returns(self: ExprOrStr) -> pl.Expr:
    """
    Calculates log returns at tick level:
        log(p_t) - log(p_{t-1})

    Returns:
        Tick-level log return series
    """
    return self.log() - self.log().shift(1)

calc_price_impact(self, volume)

Calculates absolute price impact

(last price - first price) / sum(volume)

Assumes volume is unsigned and price is a float column.

Returns:

Type Description
Expr

Float representing absolute price impact per unit volume
Example

df.group_by("ticker").agg( calc_price_impact("price", "volume") )

Source code in ffn_polars/expr/tick/price.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@register(namespace="tick")
@guard_expr("price", expected_dtype=pl.Float64)
@guard_expr("volume", expected_dtype=pl.Float64)
@auto_alias("price_impact")
def calc_price_impact(self: ExprOrStr, volume: ExprOrStr) -> pl.Expr:
    """
    Calculates absolute price impact:
        (last price - first price) / sum(volume)

    Assumes volume is unsigned and price is a float column.

    Returns:
        Float representing absolute price impact per unit volume

    Example:
        df.group_by("ticker").agg(
            calc_price_impact("price", "volume")
        )
    """
    return (self.last() - self.first()) / volume.sum().cast(pl.Float64)

calc_price_volatility_ratio(self)

Computes the coefficient of variation

std(price) / mean(price)

Returns:

Name Type Description
Float Expr

unitless relative volatility
Source code in ffn_polars/expr/tick/price.py 
21
22
23
24
25
26
27
28
29
30
31
32
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("price_volatility_ratio")
def calc_price_volatility_ratio(self: ExprOrStr) -> pl.Expr:
    """
    Computes the coefficient of variation:
        std(price) / mean(price)

    Returns:
        Float: unitless relative volatility
    """
    return self.std().cast(pl.Float64) / self.mean().cast(pl.Float64)

Volatility

calc_realized_volatility(self)

Calculates realized volatility (non-annualized) from a price series.

Formula

sqrt(Σ (log(p_t) - log(p_{t-1}))^2)

Assumes self is a price series. Use inside .select() or .group_by().agg().

Returns:

Type Description
Expr

Realized volatility over the window
Example

df.group_by("ticker").agg( pl.col("price").calc_realized_volatility() )

Source code in ffn_polars/expr/tick/volatility.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@register(namespace="tick")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("realized_volatility")
def calc_realized_volatility(self: ExprOrStr) -> pl.Expr:
    """
    Calculates realized volatility (non-annualized) from a price series.

    Formula:
        sqrt(Σ (log(p_t) - log(p_{t-1}))^2)

    Assumes `self` is a price series. Use inside `.select()` or `.group_by().agg()`.

    Returns:
        Realized volatility over the window

    Example:
        df.group_by("ticker").agg(
            pl.col("price").calc_realized_volatility()
        )
    """
    log_returns = self.log() - self.log().shift(1)
    return (log_returns**2).sum().sqrt()

EOD Expressions

Ratios

calc_calmar_ratio(self, date_col)

Returns a Polars expression to compute the Calmar ratio: CAGR / |Max Drawdown|

Parameters:

Name Type Description Default
self ExprOrStr

Column name of price series

 required
date_col str

Column name of date series

 required

Returns:

Type Description
Expr

pl.Expr: Calmar ratio expression
Source code in ffn_polars/expr/eod/ratios.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("calmar_ratio")
def calc_calmar_ratio(self: ExprOrStr, date_col: str) -> pl.Expr:
    """
    Returns a Polars expression to compute the Calmar ratio: CAGR / |Max Drawdown|

    Args:
        self: Column name of price series
        date_col: Column name of date series

    Returns:
        pl.Expr: Calmar ratio expression
    """
    cagr_expr = self.ffn.calc_cagr(date_col=date_col)
    max_dd_expr = self.ffn.calc_max_drawdown().abs()

    return cagr_expr / max_dd_expr

calc_information_ratio(self, benchmark)

Returns a Polars expression that computes the Information Ratio.

Parameters:

Name Type Description Default
self ExprOrStr

name of the column with asset returns

 required
benchmark ExprOrStr

name of the column with benchmark returns

 required
Source code in ffn_polars/expr/eod/ratios.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@guard_expr("benchmark", expected_dtype=pl.Float64)
@auto_alias("ir")
def calc_information_ratio(self: ExprOrStr, benchmark: ExprOrStr) -> pl.Expr:
    """
    Returns a Polars expression that computes the Information Ratio.

    Args:
        self: name of the column with asset returns
        benchmark: name of the column with benchmark returns
    """
    diff = self - benchmark

    return (diff.mean() / diff.std(ddof=1)).fill_nan(0.0).fill_null(0.0)

calc_prob_mom(self, b)

Polars expression that computes probabilistic momentum between two return columns. If Rust plugin is available, uses it. Otherwise, falls back to a Polars map_batches version.

Source code in ffn_polars/expr/eod/ratios.py 
117
118
119
120
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
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@guard_expr("b", expected_dtype=pl.Float64)
@auto_alias("prob_mom")
def calc_prob_mom(self: ExprOrStr, b: ExprOrStr) -> pl.Expr:
    """
    Polars expression that computes probabilistic momentum between two return columns.
    If Rust plugin is available, uses it. Otherwise, falls back to a Polars map_batches version.
    """

    name1 = self.meta.output_name()
    name2 = b.meta.output_name()
    if _HAS_RUST:
        return pl.struct([name1, name2]).map_batches(
            lambda s: _rust.prob_mom(s.struct.field(name1), s.struct.field(name2)),
            return_dtype=pl.Float64,
        )

    # fallback: pure Polars map
    diff = self - b
    ir = (diff.mean() / diff.std()).alias("information_ratio")
    n = pl.count().alias("n_obs")

    return (
        pl.struct([ir, n])
        .map_batches(
            lambda s: pl.Series([_prob_mom_cdf(s[0])]),
            return_dtype=pl.Float64,
        )
        .alias("prob_momentum")
    )

calc_risk_return_ratio(self)

Calculates the return / risk ratio. Basically the Sharpe ratio <https://www.investopedia.com/terms/s/sharperatio.asp> without factoring in the risk-free rate <https://www.investopedia.com/terms/r/risk-freerate.asp>.

Source code in ffn_polars/expr/eod/ratios.py 
89
90
91
92
93
94
95
96
97
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("risk_return_ratio")
def calc_risk_return_ratio(self) -> pl.Expr:
    """
    Calculates the return / risk ratio. Basically the
    `Sharpe ratio <https://www.investopedia.com/terms/s/sharperatio.asp>`_ without factoring in the `risk-free rate <https://www.investopedia.com/terms/r/risk-freerate.asp>`_.
    """
    return calc_sharpe(self)

calc_sharpe(self, rf=0.0, n=252, annualize=True)

Polars expression that computes the Sharpe ratio in-place without aliasing.

Source code in ffn_polars/expr/eod/ratios.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("sharpe")
def calc_sharpe(
    self: str,
    rf: Union[float, str] = 0.0,
    n: int = 252,
    annualize: bool = True,
) -> pl.Expr:
    """
    Polars expression that computes the Sharpe ratio in-place without aliasing.
    """
    excess_expr = self.ffn.to_excess_returns(rf, n)

    sharpe_expr = (
        (excess_expr.mean() / excess_expr.std(ddof=1)) * math.sqrt(n)
        if annualize
        else (excess_expr.mean() / excess_expr.std(ddof=1))
    )

    return sharpe_expr

Returns

calc_cagr(self, date_col)

Calculates the CAGR (compound annual growth rate) <https://www.investopedia.com/terms/c/cagr.asp>_ for a given price series.

Returns:

Type Description
Expr
  • float -- cagr.
Source code in ffn_polars/expr/eod/returns.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Datetime)
@guard_expr("date_col", expected_dtype=pl.Datetime)
@auto_alias("cagr")
def calc_cagr(self: ExprOrStr, date_col: ExprOrStr) -> pl.Expr:
    """
    Calculates the `CAGR (compound annual growth rate) <https://www.investopedia.com/terms/c/cagr.asp>`_ for a given price series.

    Returns:
        * float -- cagr.

    """
    return (self.last() / self.first()) ** (1 / date_col.ffn.year_frac()) - 1

calc_mtd(self, date_col='Date')

Calculate Month-To-Date return using daily prices only.

Logic: - Latest price = last row - Reference price = last price from previous month - MTD = (latest / reference) - 1

Source code in ffn_polars/expr/eod/returns.py 
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
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@guard_expr("date_col", expected_dtype=pl.Datetime)
@auto_alias("mtd")
def calc_mtd(self: pl.Expr, date_col: ExprOrStr = "Date") -> pl.Expr:
    """
    Calculate Month-To-Date return using daily prices only.

    Logic:
    - Latest price = last row
    - Reference price = last price from previous month
    - MTD = (latest / reference) - 1
    """
    prices = self
    latest_date = date_col.max()

    # Extract month & year
    latest_month = latest_date.dt.month()
    latest_year = latest_date.dt.year()

    return pl.when(True).then(
        prices.filter(
            (date_col.dt.month() != latest_month) | (date_col.dt.year() != latest_year)
        )
        .last()
        .pipe(lambda ref: prices.last() / ref - 1)
    )

calc_total_return(self)

Calculates the total return of a series.

last / first - 1

Source code in ffn_polars/expr/eod/returns.py 
152
153
154
155
156
157
158
159
160
161
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("total_return")
def calc_total_return(self: ExprOrStr) -> pl.Expr:
    """
    Calculates the total return of a series.

    last / first - 1
    """
    return (self.last() / self.first()) - 1

calc_ytd(self, date_col='Date')

Calculate Year-To-Date (YTD) return using daily prices.

Logic: - Identify current year from latest date - First price = first row of current year - Latest price = most recent row - YTD = (latest / first_of_year) - 1

Assumes date_col is sorted ascending.

Source code in ffn_polars/expr/eod/returns.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@register(namespace="eod")
@auto_alias("ytd")
@guard_expr("self", expected_dtype=pl.Float64)
@guard_expr("date_col", expected_dtype=pl.Datetime)
def calc_ytd(self: pl.Expr, date_col: ExprOrStr = "Date") -> pl.Expr:
    """
    Calculate Year-To-Date (YTD) return using daily prices.

    Logic:
    - Identify current year from latest date
    - First price = first row of current year
    - Latest price = most recent row
    - YTD = (latest / first_of_year) - 1

    Assumes `date_col` is sorted ascending.
    """

    latest_date_expr = date_col.max()
    current_year = latest_date_expr.dt.year()

    # Filter to current year only
    current_year_prices = self.filter(date_col.dt.year() == current_year)

    return (current_year_prices.last() / current_year_prices.first()) - 1

rebase(self, value=100)

Rebase a price series to a given value.

Formula is: (p / p0) * value

Source code in ffn_polars/expr/eod/returns.py 
140
141
142
143
144
145
146
147
148
149
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("rebased")
def rebase(self, value=100):
    """
    Rebase a price series to a given value.

    Formula is: (p / p0) * value
    """
    return self / self.first() * value

to_excess_returns(self, rf, n)

Returns a Polars expression that computes excess returns.

Source code in ffn_polars/expr/eod/returns.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("excess")
def to_excess_returns(self: ExprOrStr, rf: Union[float, str], n: int) -> pl.Expr:
    """
    Returns a Polars expression that computes excess returns.

    """
    if isinstance(rf, float):
        if rf == 0:
            return self
        else:
            return self - ((1 + rf) ** (1 / n) - 1)
    elif isinstance(rf, str):
        return self - pl.col(rf)
    else:
        raise TypeError("rf must be either a float or a column name string")

to_log_returns(self)

Calculates the log returns of a price series.

Formula is: ln(p1/p0)

Source code in ffn_polars/expr/eod/returns.py 
24
25
26
27
28
29
30
31
32
33
34
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("log_returns")
def to_log_returns(self: ExprOrStr) -> pl.Expr:
    """
    Calculates the log returns of a price series.

    Formula is: ln(p1/p0)

    """
    return (self / self.shift(1)).log()

to_price_index(self, start=100)

Returns a price index given a series of returns.

Assumes arithmetic returns.

Formula is: cumprod (1+r)

Source code in ffn_polars/expr/eod/returns.py 
126
127
128
129
130
131
132
133
134
135
136
137
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64, required_substring="returns")
@auto_alias("price_index")
def to_price_index(self, start=100):
    """
    Returns a price index given a series of returns.

    Assumes arithmetic returns.

    Formula is: cumprod (1+r)
    """
    return (self.fill_null(0.0) + 1).cum_prod() * start

to_returns(self)

Calculates the simple arithmetic returns of a price series.

Formula is: (t1 / t0) - 1

Source code in ffn_polars/expr/eod/returns.py 
11
12
13
14
15
16
17
18
19
20
21
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("returns")
def to_returns(self: ExprOrStr) -> pl.Expr:
    """
    Calculates the simple arithmetic returns of a price series.

    Formula is: (t1 / t0) - 1

    """
    return self / self.shift(1) - 1

Risk

calc_max_drawdown(self)

Calculates the max drawdown of a price series. If you want the actual drawdown series, please use to_drawdown_series.

Source code in ffn_polars/expr/eod/risk.py 
89
90
91
92
93
94
95
96
97
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("max_drawdown")
def calc_max_drawdown(self: ExprOrStr) -> pl.Expr:
    """
    Calculates the max drawdown of a price series. If you want the
    actual drawdown series, please use to_drawdown_series.
    """
    return self.ffn.to_drawdown_series().min()

to_drawdown_series(self)

Calculates the drawdown <https://www.investopedia.com/terms/d/drawdown.asp>_ series.

This returns a series representing a drawdown. When the price is at all time highs, the drawdown is 0. However, when prices are below high water marks, the drawdown series = current / hwm - 1

The max drawdown can be obtained by simply calling .min() on the result (since the drawdown series is negative)

Method ignores all gaps of NaN's in the price series.

Parameters:

Name Type Description Default
self ExprOrStr

prices

 required
Source code in ffn_polars/expr/eod/risk.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("drawdowns")
def to_drawdown_series(self: ExprOrStr) -> pl.Expr:
    """
    Calculates the `drawdown <https://www.investopedia.com/terms/d/drawdown.asp>`_ series.

    This returns a series representing a drawdown.
    When the price is at all time highs, the drawdown
    is 0. However, when prices are below high water marks,
    the drawdown series = current / hwm - 1

    The max drawdown can be obtained by simply calling .min()
    on the result (since the drawdown series is negative)

    Method ignores all gaps of NaN's in the price series.

    Args:
        self: prices

    """
    prices_clean = self.forward_fill()
    hwm = prices_clean.cum_max()
    return prices_clean / hwm - 1

ulcer_index(self)

Returns a Polars expression to compute the Ulcer Index from a price series.

Formula
  1. Compute cumulative max of prices.
  2. Calculate drawdowns: ((price - cummax) / cummax) * 100.
  3. Square drawdowns, take mean, then square root.
Source code in ffn_polars/expr/eod/risk.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("ulcer_index")
def ulcer_index(self: ExprOrStr) -> pl.Expr:
    """
    Returns a Polars expression to compute the Ulcer Index from a price series.

    Formula:
        1. Compute cumulative max of prices.
        2. Calculate drawdowns: ((price - cummax) / cummax) * 100.
        3. Square drawdowns, take mean, then square root.
    """
    cummax = self.cum_max()
    drawdown_pct = ((self - cummax) / cummax) * 100
    squared_drawdowns = drawdown_pct.pow(2)

    return squared_drawdowns.mean().sqrt()

ulcer_performance_index(self, rf=0.0, n=None)

Returns a Polars expression to compute Ulcer Performance Index (UPI).

UPI = mean(excess returns) / ulcer index Must be used inside .select() or .with_columns()

Parameters:

Name Type Description Default
self ExprOrStr

column with prices

 required
rf Union[float, str]

either a float (annualized) or a column name containing RF series

 0.0
n int

required if rf is float and nonzero

 None
Source code in ffn_polars/expr/eod/risk.py 
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
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("ulcer_performance_index")
def ulcer_performance_index(
    self: ExprOrStr, rf: Union[float, str] = 0.0, n: int = None
) -> pl.Expr:
    """
    Returns a Polars expression to compute Ulcer Performance Index (UPI).

    UPI = mean(excess returns) / ulcer index
    Must be used inside `.select()` or `.with_columns()`

    Args:
        self: column with prices
        rf: either a float (annualized) or a column name containing RF series
        n: required if rf is float and nonzero
    """
    if isinstance(rf, float):
        if rf != 0 and n is None:
            raise ValueError("nperiods must be set when rf is a non-zero float")

        excess_returns = self.ffn.to_returns() - (rf / n if rf != 0 else 0)

    elif isinstance(rf, str):
        # Subtract column rf from returns
        excess_returns = self.ffn.to_returns() - pl.col(rf)
    else:
        raise TypeError("rf must be a float or a string (column name)")

    return excess_returns.mean() / self.ffn.ulcer_index()

Temporal

annualize(self, durations, one_year=365.0)

Returns a Polars expression to annualize returns given durations.

Parameters:

Name Type Description Default
self ExprOrStr

Name of the column with returns (e.g., 0.05 = 5%).

 required
durations ExprOrStr

Name of the column with durations (e.g., days held).

 required
one_year float

Number of periods in a year (default 365.0 for days).

 365.0

Returns:

Type Description
Expr

pl.Expr: Expression computing annualized return.
Source code in ffn_polars/expr/eod/temporal.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@guard_expr("durations", expected_dtype=pl.Float64)
@auto_alias("annualized")
def annualize(
    self: ExprOrStr, durations: ExprOrStr, one_year: float = 365.0
) -> pl.Expr:
    """
    Returns a Polars expression to annualize returns given durations.

    Args:
        self: Name of the column with returns (e.g., 0.05 = 5%).
        durations: Name of the column with durations (e.g., days held).
        one_year: Number of periods in a year (default 365.0 for days).

    Returns:
        pl.Expr: Expression computing annualized return.
    """
    return (1.0 + self) ** (one_year / durations) - 1.0

deannualize(self, n)

Returns a Polars expression that converts annualized returns to periodic returns.

Parameters:

Name Type Description Default
self ExprOrStr

column containing annualized returns

 required
n int

Number of periods per year (e.g., 252 for daily)

 required
Source code in ffn_polars/expr/eod/temporal.py 
13
14
15
16
17
18
19
20
21
22
23
24
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Float64)
@auto_alias("deannualized")
def deannualize(self: ExprOrStr, n: int) -> pl.Expr:
    """
    Returns a Polars expression that converts annualized returns to periodic returns.

    Args:
        self: column containing annualized returns
        n: Number of periods per year (e.g., 252 for daily)
    """
    return (self + 1.0) ** (1.0 / n) - 1.0

infer_freq(self)

Infers human-readable calendar frequency label from a datetime column. Works best for: yearly, quarterly, monthly, weekly, daily.

Returns: "yearly", "quarterly", "monthly", "weekly", "daily", or None

Source code in ffn_polars/expr/eod/temporal.py 
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
@register(namespace="eod")
@guard_expr("date_col", expected_dtype=pl.Datetime)
@auto_alias("inferred_freq")
def infer_freq(self: ExprOrStr) -> pl.Expr:
    """
    Infers human-readable calendar frequency label from a datetime column.
    Works best for: yearly, quarterly, monthly, weekly, daily.

    Returns: "yearly", "quarterly", "monthly", "weekly", "daily", or None
    """
    deltas = (
        (
            self.cast(pl.Datetime).sort().diff().dt.total_nanoseconds().cast(pl.Float64)
            / 86400
            / 1_000_000_000
        )  # convert to float days
        .drop_nulls()
        .alias("delta_days")
    )

    std_expr = deltas.std().alias("delta_std")
    mode_expr = deltas.mode().first().alias("mode_days")

    return (
        pl.struct(
            [
                std_expr,
                mode_expr,
            ]
        )
        .map_batches(_map_mode_days_with_tolerance, return_dtype=pl.Utf8)
        .alias("freq")
    )

year_frac(self)

Returns a Polars expression that computes the year fraction between the first and last date in a column, assuming average year length (365.25 days = 31557600 seconds).

Source code in ffn_polars/expr/eod/temporal.py 
158
159
160
161
162
163
164
165
166
167
168
@register(namespace="eod")
@guard_expr("self", expected_dtype=pl.Datetime)
@auto_alias("year_frac")
def year_frac(self) -> pl.Expr:
    """
    Returns a Polars expression that computes the year fraction between the first and last date
    in a column, assuming average year length (365.25 days = 31557600 seconds).
    """
    return (
        self.cast(pl.Datetime).last() - self.cast(pl.Datetime).first()
    ).dt.total_seconds() / 31_557_600

Main Namespace Registry