Skip to content

API Reference

Reference documentation for the public SignalFlow API. Every symbol below is re-exported from the top-level signalflow package, so it can be imported directly:

import signalflow as sf

ds = sf.Dataset(...)
flow = sf.Flow(...)

Modules

  • Data - Dataset, the data loader, and market data sources
  • Transform & Features - feature transforms, pipelines, and selection
  • Target - labeling specs and sample selection
  • Models - forecast models and validator combinators
  • Detector - signal detectors
  • Engine - execution engine, brokers, orders, and fills
  • Strategy - rules-based strategies and decision rules
  • Flow & Live - the Flow object, runs, and live feeds
  • Experiment - experiments, scorecards, and statistics
  • CLI - command-line interface
  • Technical Analysis (ta) - indicators from signalflow-ta

Enums and registry

The package also exposes shared enums and the component registry:

signalflow.registry

Central component registry - the backbone of serialization.

Every core class registers under a name; that name is what flow.yaml serializes, sf list enumerates, and plugin packages inject into. Construction from config is registry.create(ComponentType.TRANSFORM, "revert_confluence", **params)-shaped. Seven types instead of the old 21.

The design (lazy autodiscovery, dataclass-field schema introspection) is the proven one from the previous framework, trimmed to the current type set.

registry module-attribute 

registry = Registry()

Global default registry singleton.

ComponentInfo dataclass 

ComponentInfo(cls: type[Any], role: str = '', docstring: str = '', summary: str = '', module: str = '')

Metadata about a registered component (class + extracted docs + role).

Registry dataclass 

Registry(_items: dict[ComponentType, dict[str, ComponentInfo]] = dict(), _discovered: bool = False)

Maps ComponentType -> name -> ComponentInfo with lazy autodiscovery.

create 

create(component_type: ComponentType, name: str, **kwargs: Any) -> Any

Instantiate the registered class for name with kwargs.

Source code in src/signalflow/registry.py 
105
106
107
def create(self, component_type: ComponentType, name: str, **kwargs: Any) -> Any:
    """Instantiate the registered class for ``name`` with ``kwargs``."""
    return self.get(component_type, name)(**kwargs)

get 

get(component_type: ComponentType, name: str) -> type[Any]

Return the registered class for name; raise UnknownComponentError if absent.

Source code in src/signalflow/registry.py 
83
84
85
86
87
88
89
90
91
92
def get(self, component_type: ComponentType, name: str) -> type[Any]:
    """Return the registered class for ``name``; raise ``UnknownComponentError`` if absent."""
    self._discover_if_needed()
    bucket = self._items.get(component_type, {})
    key = name.lower()
    try:
        return bucket[key].cls
    except KeyError as e:
        available = ", ".join(sorted(bucket))
        raise UnknownComponentError(f"{component_type.value}:{key} not found. Available: [{available}]") from e

get_info 

get_info(component_type: ComponentType, name: str) -> ComponentInfo

Return the ComponentInfo (class, role, docs) for name.

Source code in src/signalflow/registry.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
def get_info(self, component_type: ComponentType, name: str) -> ComponentInfo:
    """Return the ``ComponentInfo`` (class, role, docs) for ``name``."""
    self._discover_if_needed()
    bucket = self._items.get(component_type, {})
    key = name.lower()
    try:
        return bucket[key]
    except KeyError as e:
        available = ", ".join(sorted(bucket))
        raise UnknownComponentError(f"{component_type.value}:{key} not found. Available: [{available}]") from e

get_schema 

get_schema(component_type: ComponentType, name: str) -> dict[str, Any]

Introspected config schema for name: class, role, and dataclass parameters.

Source code in src/signalflow/registry.py 
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
148
149
def get_schema(self, component_type: ComponentType, name: str) -> dict[str, Any]:
    """Introspected config schema for ``name``: class, role, and dataclass parameters."""
    info = self.get_info(component_type, name)
    cls = info.cls
    params: list[dict[str, Any]] = []
    if dataclasses.is_dataclass(cls):
        for f in dataclasses.fields(cls):
            if f.name.startswith("_"):
                continue
            type_str = self._type_to_str(f.type)
            if "ClassVar" in type_str:
                continue
            has_default = f.default is not dataclasses.MISSING
            has_factory = f.default_factory is not dataclasses.MISSING
            params.append(
                {
                    "name": f.name,
                    "type": type_str,
                    "default": f.default if has_default else None,
                    "required": not has_default and not has_factory,
                }
            )
    return {
        "name": name,
        "class_name": cls.__name__,
        "component_type": component_type.value,
        "role": info.role,
        "description": info.summary,
        "module": info.module,
        "parameters": params,
    }

list 

list(component_type: ComponentType) -> list[str]

Sorted registered names for one component type.

Source code in src/signalflow/registry.py 
109
110
111
112
def list(self, component_type: ComponentType) -> list[str]:
    """Sorted registered names for one component type."""
    self._discover_if_needed()
    return sorted(self._items.get(component_type, {}))

register 

register(component_type: ComponentType, name: str, cls: type[Any], *, role: str = '', override: bool = False) -> None

Register cls under name for component_type; raise unless override when the name is taken.

Source code in src/signalflow/registry.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
def register(
    self,
    component_type: ComponentType,
    name: str,
    cls: type[Any],
    *,
    role: str = "",
    override: bool = False,
) -> None:
    """Register ``cls`` under ``name`` for ``component_type``; raise unless ``override``
    when the name is taken."""
    if not isinstance(name, str) or not name.strip():
        raise ValueError("name must be a non-empty string")
    key = name.strip().lower()
    bucket = self._items.setdefault(component_type, {})
    if key in bucket and not override:
        existing = bucket[key].cls
        raise ValueError(
            f"{component_type.value}:{key} already registered by "
            f"{existing.__module__}.{existing.__qualname__}; refusing to replace it with "
            f"{cls.__module__}.{cls.__qualname__} (pass override=True to shadow deliberately)"
        )
    if key in bucket and override:
        logger.warning(f"Overriding {component_type.value}:{key} with {cls.__name__}")
    bucket[key] = ComponentInfo.from_class(cls, role=role)

snapshot 

snapshot() -> dict[str, list[str]]

Every registered name grouped by component type: {type: [names]}.

Source code in src/signalflow/registry.py 
114
115
116
117
def snapshot(self) -> "dict[str, list[str]]":
    """Every registered name grouped by component type: ``{type: [names]}``."""
    self._discover_if_needed()
    return {t.value: sorted(v) for t, v in self._items.items()}

signalflow.ComponentType

Bases: StrEnum

The registry component types.

signalflow.Signal

Bases: StrEnum

Discrete detector output.

signalflow.RunMode

Bases: StrEnum

Execution mode of a Run (backtest, paper, live, or quicktest triage).

signalflow.Provenance

Bases: StrEnum

How forecast columns in a frame were produced (full vs out-of-fold).

signalflow.Side

Bases: StrEnum

Order side.

signalflow.IntentKind

Bases: StrEnum

What a strategy proposes for a pair (a proposal, not an order).