Custom metrics

This guide explains how to define, configure, and register custom metrics in factrix so that they can be resolved by the DAG executor and evaluated using fx.evaluate.

There are two primary ways to create custom metrics:

1. The @metric decorator (recommended): Converts a Python function into a MetricBase subclass and registers it automatically.
2. The @metric_spec decorator + register(): Stamps a MetricSpec object onto a function and manually registers it.

---

1. Using the @metric decorator

The @metric decorator is the most common and robust way to define custom metrics. It dynamically wraps your implementation function into a subclass of MetricBase and registers it with the global metric registry.

How it works

When you decorate a function with @metric, the decorator: - Inspects the function signature (ignoring the first argument, which represents the input data frame or series). - Creates a frozen dataclass whose fields correspond to your function's configuration parameters. - Registers the class under its function name. - Returns the class, which behaves like a callable metric instance when called (e.g. my_metric()).

Example

Here is how to implement a custom Information Coefficient (IC) metric that accepts a custom trimming parameter:

import polars as pl
import factrix as fx
from factrix import Aggregation, InputShape, MetricResult, SampleThreshold, cell
from factrix.metrics import compute_ic, metric

# 1. Define the metric cell and thresholds
_CUSTOM_CELL = cell(
    scope=fx.FactorScope.INDIVIDUAL,
    density=fx.FactorDensity.DENSE,
    structure=fx.DataStructure.PANEL,
)

# 2. Decorate the function
@metric(
    cell=_CUSTOM_CELL,
    aggregation=Aggregation.CS_THEN_TS,
    input_shape=InputShape.SERIES,
    requires={"ic_df": compute_ic},
    sample_threshold=SampleThreshold(min_periods=10),
)
def custom_trimmed_ic(
    ic_df: pl.DataFrame,
    trim_ratio: float = 0.05,
) -> MetricResult:
    """Compute a trimmed mean IC significance test."""
    ic_vals = ic_df["ic"].drop_nulls()

    # Custom trimming logic
    n = len(ic_vals)
    k = int(n * trim_ratio)
    trimmed = ic_vals.sort()[k : n - k]

    mean_val = float(trimmed.mean()) if len(trimmed) > 0 else 0.0

    return MetricResult(
        value=mean_val,
        n_obs=len(trimmed),
        n_obs_axis="periods",
        metadata={
            "n_periods": n,
            "trim_ratio": trim_ratio,
        }
    )

n_obs and n_obs_axis form one contract. A custom metric that reports an effective sample count must also identify its unit with one of the supported sample-axis tokens; omit both fields when one integer count is not meaningful.

Now you can evaluate your custom metric:

raw = fx.datasets.make_cs_panel(n_assets=15, n_dates=80)
data = fx.preprocess.compute_forward_return(raw, forward_periods=5)

results = fx.evaluate(
    data,
    metrics={"trimmed_ic": custom_trimmed_ic(trim_ratio=0.1)},
    factor_cols=["factor"],
)

Validating knobs

Knob validation is not written into the body. @metric builds the metric's constructor, and every knob contract is enforced there — once, for both the config form (custom_trimmed_ic(trim_ratio=1.5)) and the direct-call form (custom_trimmed_ic(ic_df, trim_ratio=1.5)), which builds the same object before it runs.

Two rules need no declaration:

• a parameter annotated with a Literal alias is checked against that alias, so the annotation is the runtime contract (tie_policy: TiePolicy rejects anything outside "ordinal" / "average");
• a parameter named inference is checked against the applicable_inference frozenset your module declares at top level.

Anything a closed set cannot express — a numeric bound, a pair of knobs that are mutually exclusive — goes in a validate= hook: a module-level callable taking the constructed instance and raising UserInputError.

Illustrative

from factrix import UserInputError


def _validate_custom_trimmed_ic(m) -> None:
    if not 0.0 <= m.trim_ratio < 0.5:
        raise UserInputError(
            func_name="custom_trimmed_ic",
            field="trim_ratio",
            value=m.trim_ratio,
            expected="a fraction in [0, 0.5): each tail is trimmed once",
            docs_path="api/metrics",
        )


@metric(
    cell=_CUSTOM_CELL,
    aggregation=Aggregation.CS_THEN_TS,
    input_shape=InputShape.SERIES,
    requires={"ic_df": compute_ic},
    sample_threshold=SampleThreshold(min_periods=10),
    validate=_validate_custom_trimmed_ic,
)
def custom_trimmed_ic(ic_df: pl.DataFrame, trim_ratio: float = 0.05):
    ...

overlap_periods and expected_warnings are injected at dispatch rather than configured, so neither is validated here. They differ at the constructor: overlap_periods is a property of the data and is rejected outright, while expected_warnings is accepted keyword-only on every metric — declare it on a standalone call and evaluate overrides it with the study-level declaration.

---

2. Using @metric_spec and register()

If you prefer a manual, lower-level approach where you retain full control over the function object (rather than wrapping it in a dataclass subclass), you can use @metric_spec paired with factrix.metrics.register.

How it works

• @metric_spec stamps metadata (a MetricSpec object) onto the callable's __metric_spec__ attribute.
• factrix.metrics.register() registers the callable with the toolkit, making it visible to list_metrics() / spec_by_name(), resolvable as a requires= producer, and accepted directly as an fx.evaluate(metrics=...) value.

A registered callable carries no configuration object, so it always runs on its signature defaults — pass the function itself, not a call of it:

Illustrative

results = fx.evaluate(
    data,
    metrics={"my_custom_stat": my_custom_stat},   # the function, uncalled
    factor_cols=["factor"],
    forward_periods=5,
)

Use the @metric decorator instead when the metric needs per-run configuration (my_metric(trim_ratio=0.1)).

Both paths work from any module: a metric defined in your own package (or in __main__) is resolved through the metric registry, not by import path.

Example

import polars as pl
import factrix as fx
from factrix import Aggregation, MetricSpec, MetricResult, cell, metric_spec
from factrix.metrics import register

# 1. Stamp the spec onto the callable
@metric_spec(
    MetricSpec(
        name="my_custom_stat",
        cell=cell(fx.FactorScope.INDIVIDUAL, fx.FactorDensity.DENSE),
        aggregation=Aggregation.CS_THEN_TS,
    )
)
def my_custom_stat(data: pl.DataFrame) -> MetricResult:
    val = float(data["forward_return"].mean())
    return MetricResult(value=val)

# 2. Register it manually
register(my_custom_stat)

---

Defining the execution shape

When setting up your MetricSpec (or passing arguments to @metric), you should configure the following parameters to ensure proper DAG routing:

• requires: A dictionary maps your function's parameter names to the upstream producer callables they depend on (e.g. requires={"ic_df": compute_ic}).
• input_shape / output_shape: Specifies the shape of the data passed into and out of the metric. Options include InputShape.PANEL, InputShape.SERIES, or InputShape.SCALAR.
• batchable: If set to True, the DAG executor calls the function once for the entire batch of factors (receiving a dictionary of data) instead of looping over factors individually.