Transform & Features

Transforms turn raw market data into model-ready features. Feature is the base contract, FeaturePipe chains transforms into a pipeline, and the selection helpers score and prune features by informativeness.

Transform contract

signalflow.Transform

Bases: ABC

Base contract for every computation in the framework.

outputs abstractmethod property

outputs: list[str]

Column names this transform appends.

compute abstractmethod

compute(df: DataFrame) -> pl.DataFrame

Append :pyattr:outputs to df (causal).

Source code in src/signalflow/transform/base.py

@abstractmethod
def compute(self, df: pl.DataFrame) -> pl.DataFrame:
    """Append :pyattr:`outputs` to ``df`` (causal)."""

fit

fit(df: DataFrame, target: Series | None = None) -> Transform

Stateful transforms override; stateless ones are a no-op.

Source code in src/signalflow/transform/base.py

def fit(self, df: pl.DataFrame, target: pl.Series | None = None) -> "Transform":
    """Stateful transforms override; stateless ones are a no-op."""
    return self

from_config classmethod

from_config(cfg: dict) -> Transform

Inverse of :meth:to_config, rebuilding nested transforms recursively.

Source code in src/signalflow/transform/base.py

@classmethod
def from_config(cls, cfg: dict) -> "Transform":
    """Inverse of :meth:`to_config`, rebuilding nested transforms recursively."""
    params = {k: _rebuild_value(v) for k, v in (cfg.get("params") or {}).items()}
    return cls(**params)

to_config

to_config() -> dict

Round-trippable {transform, role, params} (registry reconstructs it).

Source code in src/signalflow/transform/base.py

def to_config(self) -> dict:
    """Round-trippable ``{transform, role, params}`` (registry reconstructs it)."""
    params = {}
    if dataclasses.is_dataclass(self):
        for f in dataclasses.fields(self):
            if f.name.startswith("_"):
                continue
            val = getattr(self, f.name)
            params[f.name] = val.to_config() if isinstance(val, Transform) else val
    return {"transform": self.name, "role": self.role, "params": params}

signalflow.Feature

Bases: Transform

A causal, stateless feature expressed as Polars expressions.

exprs abstractmethod

exprs() -> list[pl.Expr]

Expressions whose .alias(...) names match :pyattr:outputs.

Source code in src/signalflow/transform/base.py

@abstractmethod
def exprs(self) -> list[pl.Expr]:
    """Expressions whose ``.alias(...)`` names match :pyattr:`outputs`."""

signalflow.FeaturePipe

FeaturePipe(*transforms: Transform)

Bases: Transform

Run child transforms in order; outputs are the union of their outputs.

Source code in src/signalflow/transform/pipe.py

def __init__(self, *transforms: Transform):
    for t in transforms:
        if SIGNAL_COL in t.outputs:
            raise PipeError(
                f"{t.name!r} outputs {SIGNAL_COL!r}; detectors go in detectors=, "
                "not in a FeaturePipe (signal-as-feature is a separate explicit step)"
            )
    self.transforms: tuple[Transform, ...] = transforms

fit

fit(df: DataFrame, target: Series | None = None) -> FeaturePipe

Fit stateful children in order, each on the frame produced so far.

Source code in src/signalflow/transform/pipe.py

def fit(self, df: pl.DataFrame, target: pl.Series | None = None) -> "FeaturePipe":
    """Fit stateful children in order, each on the frame produced so far."""
    cur = df
    for t in self.transforms:
        if t.requires_fit:
            t.fit(cur, target if t.requires_target else None)
        cur = t.compute(cur)
    return self

load classmethod

load(path: str) -> FeaturePipe

Rebuild a pipe from a YAML file written by :meth:save; reject non-pipe roots.

Source code in src/signalflow/transform/pipe.py

@classmethod
def load(cls, path: str) -> "FeaturePipe":
    """Rebuild a pipe from a YAML file written by :meth:`save`; reject non-pipe roots."""
    with open(path, encoding="utf-8") as fh:
        result = build_transform(yaml.safe_load(fh))
    if not isinstance(result, FeaturePipe):
        root = getattr(result, "name", type(result).__name__)
        raise PipeError(f"{path} does not describe a FeaturePipe; its root is {root!r}")
    return result

save

save(path: str) -> str

Serialize the pipe (config only) to a portable YAML file.

Source code in src/signalflow/transform/pipe.py

def save(self, path: str) -> str:
    """Serialize the pipe (config only) to a portable YAML file."""
    with open(path, "w", encoding="utf-8") as fh:
        yaml.safe_dump(self.to_config(), fh, sort_keys=False, allow_unicode=True)
    return path

Curated features

signalflow.SMA dataclass

SMA(length: int = 20)

Bases: Feature

Simple moving average of close.

Encoding and selection

signalflow.WoE dataclass

WoE(binning: Binning = (lambda: Binning('monotonic', 10))(), refit: str | None = '1d', window: str | None = '365d', smoothing: float = 0.5, positive_threshold: float = 0.0, positive_classes: tuple[float, ...] | None = None, columns: list[str] | None = None)

Bases: Transform

Encode every feature column via Weight of Evidence against the target.

refit is the rolling retrain cadence and window the trailing training span (duration strings such as "1d"/"365d"); either set to None (or "" for back-compat) disables rolling refit and fits once over all data. positive_threshold/positive_classes control how the target is binarized.

from_config classmethod

from_config(cfg: dict) -> WoE

Rebuild the recipe (hyperparameters); the fitted table is restored via load_state.

Source code in src/signalflow/transform/encode/woe.py

@classmethod
def from_config(cls, cfg: dict) -> "WoE":
    """Rebuild the recipe (hyperparameters); the fitted table is restored via load_state."""
    params = dict(cfg.get("params") or {})
    binning = params.get("binning")
    if isinstance(binning, dict):
        params["binning"] = Binning(**binning)
    return cls(**params)

load_state

load_state(state: dict) -> WoE

Restore fitted edges/WoE/IV produced by :meth:state_dict.

Source code in src/signalflow/transform/encode/woe.py

def load_state(self, state: dict) -> "WoE":
    """Restore fitted edges/WoE/IV produced by :meth:`state_dict`."""
    self.columns_ = list(state["columns"])
    self.edges_ = {c: np.asarray(state["edges"][c], dtype=float) for c in self.columns_}
    self.woe_ = {c: np.asarray(state["woe"][c], dtype=float) for c in self.columns_}
    self.iv_ = {c: float(state["iv"][c]) for c in self.columns_}
    return self

state_dict

state_dict() -> dict

Portable fitted state: bin edges + WoE table + IV per column (JSON-able).

Source code in src/signalflow/transform/encode/woe.py

def state_dict(self) -> dict:
    """Portable fitted state: bin edges + WoE table + IV per column (JSON-able)."""
    self._require_fitted("woe_")
    return {
        "columns": list(self.columns_),
        "binning": self.binning.to_dict(),
        "smoothing": self.smoothing,
        "edges": {c: self.edges_[c].tolist() for c in self.columns_},
        "woe": {c: self.woe_[c].tolist() for c in self.columns_},
        "iv": {c: float(self.iv_[c]) for c in self.columns_},
    }

signalflow.Binning dataclass

Binning(method: str = 'monotonic', max_bins: int = 10)

Bin-edge policy used by WoE/IV. monotonic enforces a monotone event rate.

signalflow.IVSelector dataclass

IVSelector(min_iv: float = 0.1, max_bins: int = 10, smoothing: float = 0.5, positive_threshold: float = 0.0, positive_classes: tuple[float, ...] | None = None)

Bases: Transform

Keep feature columns with IV ≥ min_iv.

positive_threshold/positive_classes control how the target is binarized.