Core Module¶

signalflow.core ¶

SignalFlow Core Module.

Provides fundamental building blocks for SignalFlow trading framework: - Containers: RawData, Signals, Position, Trade, Portfolio, Order, OrderFill - Enums: SignalType, SfComponentType, PositionType, etc. - Registry: Component registration and discovery - Decorators: Semantic decorators for component registration - @sf.detector, @sf.feature, @sf.validator, @sf.labeler - @sf.entry, @sf.exit - @sf.signal_metric, @sf.strategy_metric - @sf.alert, @sf.data_source, @sf.data_store, @sf.executor, @sf.risk - Transforms: SignalsTransform protocol

RawData `dataclass` ¶

RawData(datetime_start: datetime, datetime_end: datetime, pairs: list[str] = list(), data: dict[str, DataFrame | dict[str, DataFrame]] = dict(), default_source: str | None = None)

Immutable container for raw market data.

Acts as a unified in-memory bundle for multiple raw datasets (e.g. spot prices, funding, trades, orderbook, signals).

Design principles

Canonical storage is dataset-based (dictionary by name)
Datasets accessed via string keys (e.g. raw_data["spot"])
No business logic or transformations
Immutability ensures reproducibility in pipelines

Supports two data structures

Flat: dict[str, pl.DataFrame] - single source per data type
Nested: dict[str, dict[str, pl.DataFrame]] - multi-source per data type

Attributes:

Name	Type	Description
`datetime_start`	`datetime`	Start datetime of the data snapshot.
`datetime_end`	`datetime`	End datetime of the data snapshot.
`pairs`	`list[str]`	List of trading pairs in the snapshot.
`data`	`dict`	Dictionary of datasets. Can be flat or nested.
`default_source`	`str \| None`	Default source for nested data.

Example

from signalflow.core import RawData
import polars as pl
from datetime import datetime

# Flat structure (single source)
raw_data = RawData(
    datetime_start=datetime(2024, 1, 1),
    datetime_end=datetime(2024, 12, 31),
    pairs=["BTCUSDT", "ETHUSDT"],
    data={
        "spot": spot_dataframe,
        "signals": signals_dataframe,
    }
)

# Access datasets
spot_df = raw_data["spot"]
signals_df = raw_data.get("signals")

# Nested structure (multi-source)
raw_data = RawData(
    datetime_start=datetime(2024, 1, 1),
    datetime_end=datetime(2024, 12, 31),
    pairs=["BTCUSDT", "ETHUSDT"],
    data={
        "perpetual": {
            "binance": binance_df,
            "okx": okx_df,
            "bybit": bybit_df,
        }
    },
    default_source="binance",
)

# Hierarchical access
df = raw_data.perpetual.binance      # specific source
df = raw_data.perpetual.to_polars()  # default with warning
print(raw_data.perpetual.sources)    # ["binance", "okx", "bybit"]

# Check if dataset exists
if "spot" in raw_data:
    print("Spot data available")

Note

Dataset schemas are defined by convention, not enforced. Views (pandas/polars) should be handled by RawDataView wrapper.

contains ¶

__contains__(key: str) -> bool

Check if dataset exists.

Parameters:

Name	Type	Description	Default
`key`	`str`	Dataset name to check.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if dataset exists, False otherwise.

Example

if "spot" in raw_data:
    process_spot_data(raw_data["spot"])

Source code in src/signalflow/core/containers/raw_data.py

def __contains__(self, key: str) -> bool:
    """Check if dataset exists.

    Args:
        key (str): Dataset name to check.

    Returns:
        bool: True if dataset exists, False otherwise.

    Example:
        ```python
        if "spot" in raw_data:
            process_spot_data(raw_data["spot"])
        ```
    """
    return key in self.data

getattr ¶

__getattr__(name: str) -> DataTypeAccessor

Attribute access for hierarchical pattern: raw.perpetual.binance.

Parameters:

Name	Type	Description	Default
`name`	`str`	Data type name.	required

Returns:

Name	Type	Description
`DataTypeAccessor`	`DataTypeAccessor`	Accessor for the data type.

Raises:

Type	Description
`AttributeError`	If data type doesn't exist.

Source code in src/signalflow/core/containers/raw_data.py

def __getattr__(self, name: str) -> DataTypeAccessor:
    """Attribute access for hierarchical pattern: raw.perpetual.binance.

    Args:
        name: Data type name.

    Returns:
        DataTypeAccessor: Accessor for the data type.

    Raises:
        AttributeError: If data type doesn't exist.
    """
    # Avoid recursion for special attributes
    if name.startswith("_") or name in ("data", "default_source", "datetime_start", "datetime_end", "pairs"):
        raise AttributeError(f"'{type(self).__name__}' has no attribute '{name}'")

    data = object.__getattribute__(self, "data")
    default_source = object.__getattribute__(self, "default_source")

    if name not in data:
        raise AttributeError(f"No data type '{name}'. Available: {list(data.keys())}")

    value = data[name]

    # Nested structure: return accessor
    if isinstance(value, dict) and not isinstance(value, pl.DataFrame):
        return DataTypeAccessor(value, default_source, name)

    # Flat structure: wrap single DataFrame as accessor
    source_name = default_source or "default"
    return DataTypeAccessor({source_name: value}, source_name, name)

getitem ¶

__getitem__(key: str | tuple[str, str]) -> pl.DataFrame

Dictionary-style access to datasets.

Supports both simple key and tuple (data_type, source) indexing.

Parameters:

Name	Type	Description	Default
`key`	`str \| tuple[str, str]`	Dataset name or (data_type, source) tuple.	required

Returns:

Type	Description
`DataFrame`	pl.DataFrame: Dataset as Polars DataFrame.

Example

# Flat structure
spot_df = raw_data["spot"]

# Nested structure - explicit source
df = raw_data["perpetual", "binance"]

# Nested structure - default source (warns)
df = raw_data["perpetual"]

Source code in src/signalflow/core/containers/raw_data.py

def __getitem__(self, key: str | tuple[str, str]) -> pl.DataFrame:
    """Dictionary-style access to datasets.

    Supports both simple key and tuple (data_type, source) indexing.

    Args:
        key: Dataset name or (data_type, source) tuple.

    Returns:
        pl.DataFrame: Dataset as Polars DataFrame.

    Example:
        ```python
        # Flat structure
        spot_df = raw_data["spot"]

        # Nested structure - explicit source
        df = raw_data["perpetual", "binance"]

        # Nested structure - default source (warns)
        df = raw_data["perpetual"]
        ```
    """
    if isinstance(key, tuple):
        data_type, source = key
        return self.get(data_type, source=source)
    return self.get(key)

iter ¶

__iter__() -> Iterator[str]

Iterate over data type keys.

Returns:

Type	Description
`Iterator[str]`	Iterator[str]: Iterator over dataset names.

Example

for data_type in raw_data:
    print(f"Dataset: {data_type}")

Source code in src/signalflow/core/containers/raw_data.py

def __iter__(self) -> Iterator[str]:
    """Iterate over data type keys.

    Returns:
        Iterator[str]: Iterator over dataset names.

    Example:
        ```python
        for data_type in raw_data:
            print(f"Dataset: {data_type}")
        ```
    """
    return iter(self.data)

get ¶

get(key: str, source: str | None = None) -> pl.DataFrame

Get dataset by key.

For nested (multi-source) data, returns default source with warning unless source is explicitly specified.

Parameters:

Name	Type	Description	Default
`key`	`str`	Dataset name (e.g. "spot", "perpetual").	required
`source`	`str \| None`	Source name for nested data. If None, uses default_source with warning.	`None`

Returns:

Type	Description
`DataFrame`	pl.DataFrame: Polars DataFrame if exists, empty DataFrame otherwise.

Raises:

Type	Description
`TypeError`	If dataset exists but is not a valid structure.
`KeyError`	If source specified but not found.

Example

# Flat structure
spot_df = raw_data.get("spot")

# Nested structure - explicit source
df = raw_data.get("perpetual", source="binance")

# Nested structure - default source (warns)
df = raw_data.get("perpetual")

# Returns empty DataFrame if key doesn't exist
missing_df = raw_data.get("nonexistent")
assert missing_df.is_empty()

Source code in src/signalflow/core/containers/raw_data.py

def get(self, key: str, source: str | None = None) -> pl.DataFrame:
    """Get dataset by key.

    For nested (multi-source) data, returns default source with warning
    unless source is explicitly specified.

    Args:
        key (str): Dataset name (e.g. "spot", "perpetual").
        source (str | None): Source name for nested data.
            If None, uses default_source with warning.

    Returns:
        pl.DataFrame: Polars DataFrame if exists, empty DataFrame otherwise.

    Raises:
        TypeError: If dataset exists but is not a valid structure.
        KeyError: If source specified but not found.

    Example:
        ```python
        # Flat structure
        spot_df = raw_data.get("spot")

        # Nested structure - explicit source
        df = raw_data.get("perpetual", source="binance")

        # Nested structure - default source (warns)
        df = raw_data.get("perpetual")

        # Returns empty DataFrame if key doesn't exist
        missing_df = raw_data.get("nonexistent")
        assert missing_df.is_empty()
        ```
    """
    obj = self.data.get(key)
    if obj is None:
        return pl.DataFrame()

    # Flat structure: single DataFrame
    if isinstance(obj, pl.DataFrame):
        return obj

    # Nested structure: dict of DataFrames
    if isinstance(obj, dict):
        if source is not None:
            if source not in obj:
                raise KeyError(f"Source '{source}' not found for '{key}'. Available: {list(obj.keys())}")
            return obj[source]

        # No source specified - use default with warning
        default = self.default_source
        if default is None or default not in obj:
            default = next(iter(obj.keys()), None)
        if default is None:
            return pl.DataFrame()

        warnings.warn(
            f"Using default source '{default}' for '{key}'. "
            f"Specify explicitly: raw.get('{key}', source='{default}')",
            UserWarning,
            stacklevel=2,
        )
        return obj[default]

    raise TypeError(f"Dataset '{key}' has invalid type: {type(obj)}")

items ¶

items() -> Iterator[tuple[str, pl.DataFrame | dict[str, pl.DataFrame]]]

Return (key, dataset) pairs.

Returns:

Name	Type	Description
`Iterator`	`Iterator[tuple[str, DataFrame \| dict[str, DataFrame]]]`	Iterator over (key, DataFrame) tuples.

Example

for name, df in raw_data.items():
    print(f"{name}: {df.shape}")

Source code in src/signalflow/core/containers/raw_data.py

def items(self) -> Iterator[tuple[str, pl.DataFrame | dict[str, pl.DataFrame]]]:
    """Return (key, dataset) pairs.

    Returns:
        Iterator: Iterator over (key, DataFrame) tuples.

    Example:
        ```python
        for name, df in raw_data.items():
            print(f"{name}: {df.shape}")
        ```
    """
    return iter(self.data.items())

keys ¶

keys() -> Iterator[str]

Return available dataset keys.

Returns:

Type	Description
`Iterator[str]`	Iterator[str]: Iterator over dataset names.

Example

for key in raw_data.keys():
    print(f"Dataset: {key}")

Source code in src/signalflow/core/containers/raw_data.py

def keys(self) -> Iterator[str]:
    """Return available dataset keys.

    Returns:
        Iterator[str]: Iterator over dataset names.

    Example:
        ```python
        for key in raw_data.keys():
            print(f"Dataset: {key}")
        ```
    """
    return iter(self.data.keys())

sources ¶

sources(data_type: str) -> list[str]

Return available sources for a data type.

Parameters:

Name	Type	Description	Default
`data_type`	`str`	Data type name (e.g. "perpetual", "spot").	required

Returns:

Type	Description
`list[str]`	list[str]: List of source names. Returns ["default"] for flat data.

Raises:

Type	Description
`KeyError`	If data type doesn't exist.

Example

# Nested structure
print(raw_data.sources("perpetual"))  # ["binance", "okx", "bybit"]

# Flat structure
print(raw_data.sources("spot"))  # ["default"]

Source code in src/signalflow/core/containers/raw_data.py

def sources(self, data_type: str) -> list[str]:
    """Return available sources for a data type.

    Args:
        data_type: Data type name (e.g. "perpetual", "spot").

    Returns:
        list[str]: List of source names. Returns ["default"] for flat data.

    Raises:
        KeyError: If data type doesn't exist.

    Example:
        ```python
        # Nested structure
        print(raw_data.sources("perpetual"))  # ["binance", "okx", "bybit"]

        # Flat structure
        print(raw_data.sources("spot"))  # ["default"]
        ```
    """
    if data_type not in self.data:
        raise KeyError(f"No data type '{data_type}'. Available: {list(self.data.keys())}")

    value = self.data[data_type]
    if isinstance(value, pl.DataFrame):
        return [self.default_source or "default"]
    if isinstance(value, dict):
        return list(value.keys())
    raise TypeError(f"Invalid data structure for '{data_type}'")

values ¶

values() -> Iterator[pl.DataFrame | dict[str, pl.DataFrame]]

Return dataset values.

Returns:

Name	Type	Description
`Iterator`	`Iterator[DataFrame \| dict[str, DataFrame]]`	Iterator over DataFrames.

Example

for df in raw_data.values():
    print(df.columns)

Source code in src/signalflow/core/containers/raw_data.py

def values(self) -> Iterator[pl.DataFrame | dict[str, pl.DataFrame]]:
    """Return dataset values.

    Returns:
        Iterator: Iterator over DataFrames.

    Example:
        ```python
        for df in raw_data.values():
            print(df.columns)
        ```
    """
    return iter(self.data.values())

Signals `dataclass` ¶

Signals(value: DataFrame)

Immutable container for trading signals.

Canonical in-memory format is a Polars DataFrame with long schema.

Required columns

pair (str): Trading pair identifier
timestamp (datetime): Signal timestamp
signal_type (SignalType | int): Signal type (RISE, FALL, NONE)
signal (int | float): Signal value

Optional columns

probability (float): Signal probability (required for merge logic)

Attributes:

Name	Type	Description
`value`	`DataFrame`	Polars DataFrame containing signal data.

Example

from signalflow.core import Signals, SignalType
import polars as pl
from datetime import datetime

# Create signals
signals_df = pl.DataFrame({
    "pair": ["BTCUSDT", "ETHUSDT"],
    "timestamp": [datetime.now(), datetime.now()],
    "signal_type": [SignalType.RISE.value, SignalType.FALL.value],
    "signal": [1, -1],
    "probability": [0.8, 0.7]
})

signals = Signals(signals_df)

# Apply transformation
filtered = signals.apply(filter_transform)

# Chain transformations
processed = signals.pipe(
    transform1,
    transform2,
    transform3
)

# Merge signals
combined = signals1 + signals2

Note

All transformations return new Signals instance. No in-place mutation is allowed.

add ¶

__add__(other: Signals) -> Signals

Merge two Signals objects.

Merge rules

Key: (pair, timestamp) or (pair, timestamp, signal_category) if signal_category column is present.
Signal type priority:
null signal_type has lowest priority
SignalType.NONE ("none") has lowest priority (backward compat)
Non-null/non-NONE always overrides
If both non-null, other wins
Low-priority signals normalized to probability = 0
Merge is deterministic

Parameters:

Name	Type	Description	Default
`other`	`Signals`	Another Signals object to merge.	required

Returns:

Name	Type	Description
`Signals`	`Signals`	New merged Signals instance.

Raises:

Type	Description
`TypeError`	If other is not a Signals instance.

Example

# Detector 1 signals
signals1 = detector1.run(data)

# Detector 2 signals
signals2 = detector2.run(data)

# Merge with priority to signals2
merged = signals1 + signals2

# NONE/null signals overridden by non-NONE
# Non-NONE conflicts resolved by taking signals2

Source code in src/signalflow/core/containers/signals.py

def __add__(self, other: Signals) -> Signals:
    """Merge two Signals objects.

    Merge rules:
        1. Key: (pair, timestamp) or (pair, timestamp, signal_category)
           if signal_category column is present.
        2. Signal type priority:
           - null signal_type has lowest priority
           - SignalType.NONE ("none") has lowest priority (backward compat)
           - Non-null/non-NONE always overrides
           - If both non-null, ``other`` wins
        3. Low-priority signals normalized to probability = 0
        4. Merge is deterministic

    Args:
        other (Signals): Another Signals object to merge.

    Returns:
        Signals: New merged Signals instance.

    Raises:
        TypeError: If other is not a Signals instance.

    Example:
        ```python
        # Detector 1 signals
        signals1 = detector1.run(data)

        # Detector 2 signals
        signals2 = detector2.run(data)

        # Merge with priority to signals2
        merged = signals1 + signals2

        # NONE/null signals overridden by non-NONE
        # Non-NONE conflicts resolved by taking signals2
        ```
    """
    if not isinstance(other, Signals):
        return NotImplemented

    a = self.value
    b = other.value

    all_cols = list(dict.fromkeys([*a.columns, *b.columns]))

    def align(df: pl.DataFrame) -> pl.DataFrame:
        return df.with_columns([pl.lit(None).alias(c) for c in all_cols if c not in df.columns]).select(all_cols)

    a = align(a).with_columns(pl.lit(0).alias("_src"))
    b = align(b).with_columns(pl.lit(1).alias("_src"))

    merged = pl.concat([a, b], how="vertical")

    # Determine merge key: include signal_category if present
    has_category = "signal_category" in all_cols
    key_cols = ["pair", "timestamp"]
    if has_category:
        key_cols = ["pair", "timestamp", "signal_category"]

    # Priority: null and "none" signal_type are lowest priority
    merged = merged.with_columns(
        pl.when(pl.col("signal_type").is_null() | (pl.col("signal_type") == _NONE_SIGNAL))
        .then(pl.lit(0))
        .otherwise(pl.col("probability"))
        .alias("probability")
    )

    merged = merged.with_columns(
        pl.when(pl.col("signal_type").is_null() | (pl.col("signal_type") == _NONE_SIGNAL))
        .then(pl.lit(0))
        .otherwise(pl.lit(1))
        .alias("_priority")
    )

    sort_cols = [*key_cols, "_priority", "_src"]
    sort_desc = [False] * len(key_cols) + [True, True]

    merged = (
        merged.sort(sort_cols, descending=sort_desc)
        .unique(subset=key_cols, keep="first")
        .drop(["_priority", "_src"])
        .sort(key_cols[:2])  # always sort output by pair, timestamp
    )

    return Signals(merged)

apply ¶

apply(transform: SignalsTransform) -> Signals

Apply a single transformation to signals.

Parameters:

Name	Type	Description	Default
`transform`	`SignalsTransform`	Callable transformation implementing SignalsTransform protocol.	required

Returns:

Name	Type	Description
`Signals`	`Signals`	New Signals instance with transformed data.

Example

from signalflow.core import Signals
import polars as pl

def filter_high_probability(df: pl.DataFrame) -> pl.DataFrame:
    return df.filter(pl.col("probability") > 0.7)

filtered = signals.apply(filter_high_probability)

Source code in src/signalflow/core/containers/signals.py

def apply(self, transform: SignalsTransform) -> Signals:
    """Apply a single transformation to signals.

    Args:
        transform (SignalsTransform): Callable transformation implementing
            SignalsTransform protocol.

    Returns:
        Signals: New Signals instance with transformed data.

    Example:
        ```python
        from signalflow.core import Signals
        import polars as pl

        def filter_high_probability(df: pl.DataFrame) -> pl.DataFrame:
            return df.filter(pl.col("probability") > 0.7)

        filtered = signals.apply(filter_high_probability)
        ```
    """
    out = transform(self.value)
    return Signals(out)

pipe ¶

pipe(*transforms: SignalsTransform) -> Signals

Apply multiple transformations sequentially.

Parameters:

Name	Type	Description	Default
`*transforms`	`SignalsTransform`	Sequence of transformations to apply in order.	`()`

Returns:

Name	Type	Description
`Signals`	`Signals`	New Signals instance after applying all transformations.

Example

result = signals.pipe(
    filter_none_signals,
    normalize_probabilities,
    add_metadata
)

Source code in src/signalflow/core/containers/signals.py

def pipe(self, *transforms: SignalsTransform) -> Signals:
    """Apply multiple transformations sequentially.

    Args:
        *transforms (SignalsTransform): Sequence of transformations to apply in order.

    Returns:
        Signals: New Signals instance after applying all transformations.

    Example:
        ```python
        result = signals.pipe(
            filter_none_signals,
            normalize_probabilities,
            add_metadata
        )
        ```
    """
    s = self
    for t in transforms:
        s = s.apply(t)
    return s

SignalType ¶

Bases: StrEnum

Enumeration of price direction signal types.

.. deprecated:: SignalType is deprecated and will be removed in a future version. Use plain string values instead (e.g. "rise", "fall", "flat"). Use null (Polars null) instead of SignalType.NONE for unknown/no signal.

For configuring which signal types to trade, use ``signal_type_map`` on
entry rules or ``DIRECTIONAL_SIGNAL_MAP`` from ``core.signal_registry``.

Represents the direction of a trading signal detected by signal detectors. This enum covers the PRICE_DIRECTION category. Other categories use free-form string values for signal_type (see SignalCategory).

Values

NONE: No signal detected. Deprecated -- use null for unknown or FLAT for sideways market. RISE: Bullish signal indicating potential price increase. FALL: Bearish signal indicating potential price decrease. FLAT: Sideways market / range-bound price action.

Example

# Preferred: use plain strings
signal_type = "rise"
if signal_type == "rise":
    print("Bullish signal detected")

# Legacy (deprecated):
from signalflow.core.enums import SignalType
if signal_type == SignalType.RISE.value:
    print("Bullish signal detected")

Note

Stored as string values in DataFrames for serialization. For non-PRICE_DIRECTION categories, use string values directly (e.g. "high_volatility", "local_max") instead of this enum.

SignalFlowRegistry `dataclass` ¶

SignalFlowRegistry(_items: dict[SfComponentType, dict[str, ComponentInfo]] = dict(), _raw_data_types: dict[str, set[str]] = (lambda: {k: (v.copy()) for k, v in (_BUILTIN_RAW_DATA_TYPES.items())})(), _discovered: bool = False)

Component registry for dynamic component discovery and instantiation.

Provides centralized registration and lookup for SignalFlow components. Components are organized by type (DETECTOR, EXTRACTOR, etc.) and accessed by case-insensitive names.

Also manages extensible raw data type definitions - each data type maps to a set of required columns. Built-in types (SPOT, FUTURES, PERPETUAL) are pre-registered; users can add custom types via register_raw_data_type().

Registry structure

component_type -> name -> ComponentInfo (class + metadata)

Supported component types

DETECTOR: Signal detection classes
EXTRACTOR: Feature extraction classes
LABELER: Signal labeling classes
ENTRY_RULE: Position entry rules
EXIT_RULE: Position exit rules
METRIC: Strategy metrics
EXECUTOR: Order execution engines

Attributes:

Name	Type	Description
`_items`	`dict[SfComponentType, dict[str, ComponentInfo]]`	Internal storage mapping component types to name-ComponentInfo pairs.
`_raw_data_types`	`dict[str, set[str]]`	Mapping of raw data type names to their required column sets.

Example

from signalflow.core.registry import SignalFlowRegistry, default_registry

# Register custom raw data type
default_registry.register_raw_data_type(
    name="lob",
    columns=["pair", "timestamp", "bid", "ask", "bid_size", "ask_size"],
)

# Get component info with docstring for UI
info = default_registry.get_info(SfComponentType.DETECTOR, "sma_cross")
print(info.summary)    # "Detects SMA crossover signals."
print(info.docstring)  # Full docstring with Args, Example, etc.

# Get columns for any type
cols = default_registry.get_raw_data_columns("spot")
custom_cols = default_registry.get_raw_data_columns("lob")

# List all registered raw data types
print(default_registry.list_raw_data_types())

Note

Component names are stored and looked up in lowercase. Use default_registry singleton for application-wide registration.

autodiscover ¶

autodiscover() -> None

Scan signalflow.* packages and entry-points for components.

Walks all sub-modules of the signalflow package using :func:pkgutil.walk_packages and imports them. Because semantic decorators register classes at import time, importing a module is sufficient to populate the registry.

External packages can expose components via the signalflow.components entry-point group. Each entry-point should reference a module (not a callable); importing it triggers registration through semantic decorators.

This method is idempotent - subsequent calls are no-ops once _discovered is True.

Example

from signalflow.core.registry import default_registry

# Explicit discovery (normally automatic on first get/list)
default_registry.autodiscover()

# All decorated classes are now registered
print(default_registry.snapshot())

Source code in src/signalflow/core/registry.py

def autodiscover(self) -> None:
    """Scan ``signalflow.*`` packages and entry-points for components.

    Walks all sub-modules of the ``signalflow`` package using
    :func:`pkgutil.walk_packages` and imports them.  Because
    semantic decorators register classes at import time, importing
    a module is sufficient to populate the registry.

    External packages can expose components via the
    ``signalflow.components`` entry-point group.  Each entry-point
    should reference a module (not a callable); importing it triggers
    registration through semantic decorators.

    This method is idempotent - subsequent calls are no-ops once
    ``_discovered`` is ``True``.

    Example:
        ```python
        from signalflow.core.registry import default_registry

        # Explicit discovery (normally automatic on first get/list)
        default_registry.autodiscover()

        # All decorated classes are now registered
        print(default_registry.snapshot())
        ```
    """
    if self._discovered:
        return
    self._discovered = True

    self._discover_internal_packages()
    self._discover_entry_points()

create ¶

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

Instantiate a component by registry key.

Convenient method that combines get() and instantiation.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of component to create.	required
`name`	`str`	Component name (case-insensitive).	required
`**kwargs`	`Any`	Arguments to pass to component constructor.	`{}`

Returns:

Name	Type	Description
`Any`	`Any`	Instantiated component.

Raises:

Type	Description
`KeyError`	If component not found.
`TypeError`	If kwargs don't match component constructor.

Example

# Create detector with params
detector = registry.create(
    SfComponentType.DETECTOR,
    "sma_cross",
    fast_window=10,
    slow_window=20
)

# Create extractor
extractor = registry.create(
    SfComponentType.EXTRACTOR,
    "rsi",
    window=14
)

# Create with config dict
config = {"window": 20, "threshold": 0.7}
labeler = registry.create(
    SfComponentType.LABELER,
    "fixed",
    **config
)

Source code in src/signalflow/core/registry.py

def create(self, component_type: SfComponentType, name: str, **kwargs: Any) -> Any:
    """Instantiate a component by registry key.

    Convenient method that combines get() and instantiation.

    Args:
        component_type (SfComponentType): Type of component to create.
        name (str): Component name (case-insensitive).
        **kwargs: Arguments to pass to component constructor.

    Returns:
        Any: Instantiated component.

    Raises:
        KeyError: If component not found.
        TypeError: If kwargs don't match component constructor.

    Example:
        ```python
        # Create detector with params
        detector = registry.create(
            SfComponentType.DETECTOR,
            "sma_cross",
            fast_window=10,
            slow_window=20
        )

        # Create extractor
        extractor = registry.create(
            SfComponentType.EXTRACTOR,
            "rsi",
            window=14
        )

        # Create with config dict
        config = {"window": 20, "threshold": 0.7}
        labeler = registry.create(
            SfComponentType.LABELER,
            "fixed",
            **config
        )
        ```
    """
    cls = self.get(component_type, name)
    return cls(**kwargs)

export_schemas ¶

export_schemas() -> dict[str, builtins.list[dict[str, Any]]]

Export schemas for all registered components, grouped by type.

Useful for bulk loading into UI component browsers without N+1 calls.

Returns:

Type	Description
`dict[str, list[dict[str, Any]]]`	Dict mapping component type names to lists of component schemas.

Example

all_schemas = registry.export_schemas() for det in all_schemas.get("DETECTOR", []): ... print(det["name"], len(det["parameters"]), "params")

Source code in src/signalflow/core/registry.py

def export_schemas(self) -> dict[str, builtins.list[dict[str, Any]]]:
    """Export schemas for all registered components, grouped by type.

    Useful for bulk loading into UI component browsers without N+1 calls.

    Returns:
        Dict mapping component type names to lists of component schemas.

    Example:
        >>> all_schemas = registry.export_schemas()
        >>> for det in all_schemas.get("DETECTOR", []):
        ...     print(det["name"], len(det["parameters"]), "params")
    """
    self._discover_if_needed()
    result: dict[str, list[dict[str, Any]]] = {}
    for comp_type, names_dict in self._items.items():
        schemas = []
        for name in sorted(names_dict):
            try:
                schemas.append(self.get_schema(comp_type, name))
            except Exception:
                logger.debug(f"export_schemas: failed to get schema for {comp_type.value}:{name}")
        if schemas:
            result[comp_type.value] = schemas
    return result

get ¶

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

Get a registered class by key.

Lookup is case-insensitive. Raises helpful error with available components if key not found.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of component to lookup.	required
`name`	`str`	Component name (case-insensitive).	required

Returns:

Type	Description
`type[Any]`	Type[Any]: Registered class.

Raises:

Type	Description
`KeyError`	If component not found. Error message includes available components.

Example

# Get component class
detector_cls = registry.get(SfComponentType.DETECTOR, "sma_cross")

# Case-insensitive
detector_cls = registry.get(SfComponentType.DETECTOR, "SMA_Cross")

# Instantiate manually
detector = detector_cls(fast_window=10, slow_window=20)

# Handle missing component
try:
    cls = registry.get(SfComponentType.DETECTOR, "unknown")
except KeyError as e:
    print(f"Component not found: {e}")
    # Shows: "Component not found: DETECTOR:unknown. Available: [sma_cross, ...]"

Source code in src/signalflow/core/registry.py

def get(self, component_type: SfComponentType, name: str) -> type[Any]:
    """Get a registered class by key.

    Lookup is case-insensitive. Raises helpful error with available
    components if key not found.

    Args:
        component_type (SfComponentType): Type of component to lookup.
        name (str): Component name (case-insensitive).

    Returns:
        Type[Any]: Registered class.

    Raises:
        KeyError: If component not found. Error message includes available components.

    Example:
        ```python
        # Get component class
        detector_cls = registry.get(SfComponentType.DETECTOR, "sma_cross")

        # Case-insensitive
        detector_cls = registry.get(SfComponentType.DETECTOR, "SMA_Cross")

        # Instantiate manually
        detector = detector_cls(fast_window=10, slow_window=20)

        # Handle missing component
        try:
            cls = registry.get(SfComponentType.DETECTOR, "unknown")
        except KeyError as e:
            print(f"Component not found: {e}")
            # Shows: "Component not found: DETECTOR:unknown. Available: [sma_cross, ...]"
        ```
    """
    self._discover_if_needed()
    self._ensure(component_type)
    key = name.lower()
    try:
        return self._items[component_type][key].cls
    except KeyError as e:
        available = ", ".join(sorted(self._items[component_type]))
        raise KeyError(f"Component not found: {component_type.value}:{key}. Available: [{available}]") from e

get_info ¶

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

Get full component info including docstring.

Use this method when you need metadata for UI display or documentation generation.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of component to lookup.	required
`name`	`str`	Component name (case-insensitive).	required

Returns:

Name	Type	Description
`ComponentInfo`	`ComponentInfo`	Full metadata including class, docstring, summary, module.

Raises:

Type	Description
`KeyError`	If component not found.

Example

# Get info for UI tooltip
info = registry.get_info(SfComponentType.DETECTOR, "sma_cross")
print(info.summary)     # "Detects SMA crossover signals."
print(info.docstring)   # Full docstring
print(info.module)      # "signalflow.detector.sma_cross"

# Use in sf-ui component browser
for name in registry.list(SfComponentType.FEATURE):
    info = registry.get_info(SfComponentType.FEATURE, name)
    display_component_card(name, info.summary, info.docstring)

Source code in src/signalflow/core/registry.py

def get_info(self, component_type: SfComponentType, name: str) -> ComponentInfo:
    """Get full component info including docstring.

    Use this method when you need metadata for UI display or
    documentation generation.

    Args:
        component_type (SfComponentType): Type of component to lookup.
        name (str): Component name (case-insensitive).

    Returns:
        ComponentInfo: Full metadata including class, docstring, summary, module.

    Raises:
        KeyError: If component not found.

    Example:
        ```python
        # Get info for UI tooltip
        info = registry.get_info(SfComponentType.DETECTOR, "sma_cross")
        print(info.summary)     # "Detects SMA crossover signals."
        print(info.docstring)   # Full docstring
        print(info.module)      # "signalflow.detector.sma_cross"

        # Use in sf-ui component browser
        for name in registry.list(SfComponentType.FEATURE):
            info = registry.get_info(SfComponentType.FEATURE, name)
            display_component_card(name, info.summary, info.docstring)
        ```
    """
    self._discover_if_needed()
    self._ensure(component_type)
    key = name.lower()
    try:
        return self._items[component_type][key]
    except KeyError as e:
        available = ", ".join(sorted(self._items[component_type]))
        raise KeyError(f"Component not found: {component_type.value}:{key}. Available: [{available}]") from e

get_raw_data_columns ¶

get_raw_data_columns(name: str) -> set[str]

Get required columns for a raw data type.

Parameters:

Name	Type	Description	Default
`name`	`str`	Data type identifier (case-insensitive). Accepts both `RawDataType` enum members and plain strings.	required

Returns:

Type	Description
`set[str]`	Copy of the column set for the requested type.

Raises:

Type	Description
`KeyError`	If data type is not registered.

Example

cols = default_registry.get_raw_data_columns("spot")
# {'pair', 'timestamp', 'open', 'high', 'low', 'close', 'volume'}

cols = default_registry.get_raw_data_columns("lob")
# {'pair', 'timestamp', 'bid', 'ask', ...}

Source code in src/signalflow/core/registry.py

def get_raw_data_columns(self, name: str) -> set[str]:
    """Get required columns for a raw data type.

    Args:
        name: Data type identifier (case-insensitive). Accepts both
            ``RawDataType`` enum members and plain strings.

    Returns:
        Copy of the column set for the requested type.

    Raises:
        KeyError: If data type is not registered.

    Example:
        ```python
        cols = default_registry.get_raw_data_columns("spot")
        # {'pair', 'timestamp', 'open', 'high', 'low', 'close', 'volume'}

        cols = default_registry.get_raw_data_columns("lob")
        # {'pair', 'timestamp', 'bid', 'ask', ...}
        ```
    """
    raw = getattr(name, "value", name)
    key = str(raw).strip().lower()
    try:
        return self._raw_data_types[key].copy()
    except KeyError:
        available = ", ".join(sorted(self._raw_data_types))
        raise KeyError(f"Raw data type '{key}' not registered. Available: [{available}]") from None

get_schema ¶

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

Get JSON-serializable parameter schema for a registered component.

Uses dataclasses.fields() to introspect @dataclass-decorated components and extract field names, types, and defaults.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of component.	required
`name`	`str`	Component registry name (case-insensitive).	required

Returns:

Type	Description
`dict[str, Any]`	Schema dict with keys: `name`, `class_name`, `component_type`,
`dict[str, Any]`	`description`, `docstring`, `module`, `parameters`,
`dict[str, Any]`	`requires`, `outputs`.

Raises:

Type	Description
`KeyError`	If component not found.

Example

schema = registry.get_schema(SfComponentType.DETECTOR, "example/sma_cross") print(schema["description"]) # Short summary print(schema["docstring"]) # Full docstring for UI for p in schema["parameters"]: ... print(f"{p['name']}: {p['type']} = {p['default']}")

Source code in src/signalflow/core/registry.py

def get_schema(self, component_type: SfComponentType, name: str) -> dict[str, Any]:
    """Get JSON-serializable parameter schema for a registered component.

    Uses ``dataclasses.fields()`` to introspect ``@dataclass``-decorated
    components and extract field names, types, and defaults.

    Args:
        component_type: Type of component.
        name: Component registry name (case-insensitive).

    Returns:
        Schema dict with keys: ``name``, ``class_name``, ``component_type``,
        ``description``, ``docstring``, ``module``, ``parameters``,
        ``requires``, ``outputs``.

    Raises:
        KeyError: If component not found.

    Example:
        >>> schema = registry.get_schema(SfComponentType.DETECTOR, "example/sma_cross")
        >>> print(schema["description"])  # Short summary
        >>> print(schema["docstring"])    # Full docstring for UI
        >>> for p in schema["parameters"]:
        ...     print(f"{p['name']}: {p['type']} = {p['default']}")
    """
    info = self.get_info(component_type, name)
    cls = info.cls

    parameters: list[dict[str, Any]] = []
    if dataclasses.is_dataclass(cls):
        for f in dataclasses.fields(cls):
            if f.name.startswith("_") or f.name in self._BASE_FIELDS:
                continue
            # Skip ClassVar fields (they show up as strings containing "ClassVar")
            type_str = self._type_to_str(f.type)
            if "ClassVar" in type_str:
                continue

            has_default = f.default is not dataclasses.MISSING
            has_default_factory = f.default_factory is not dataclasses.MISSING  # type: ignore[misc]

            parameters.append(
                {
                    "name": f.name,
                    "type": type_str,
                    "default": f.default if has_default else None,
                    "required": not has_default and not has_default_factory,
                }
            )

    # Extract ClassVar metadata
    requires = getattr(cls, "requires", [])
    outputs = getattr(cls, "outputs", [])

    return {
        "name": name,
        "class_name": cls.__name__,
        "component_type": component_type.value,
        "description": info.summary,
        "docstring": info.docstring,
        "module": info.module,
        "parameters": parameters,
        "requires": list(requires) if requires else [],
        "outputs": list(outputs) if outputs else [],
    }

list ¶

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

List registered components for a type.

Returns sorted list of component names for given type.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of components to list.	required

Returns:

Type	Description
`list[str]`	list[str]: Sorted list of registered component names.

Example

# List all detectors
detectors = registry.list(SfComponentType.DETECTOR)
print(f"Available detectors: {detectors}")
# Output: ['ema_cross', 'macd', 'rsi_threshold', 'sma_cross']

# Check if component exists
if "sma_cross" in registry.list(SfComponentType.DETECTOR):
    detector = registry.create(SfComponentType.DETECTOR, "sma_cross")

# List all component types
from signalflow.core.enums import SfComponentType
for component_type in SfComponentType:
    components = registry.list(component_type)
    print(f"{component_type.value}: {components}")

Source code in src/signalflow/core/registry.py

def list(self, component_type: SfComponentType) -> list[str]:
    """List registered components for a type.

    Returns sorted list of component names for given type.

    Args:
        component_type (SfComponentType): Type of components to list.

    Returns:
        list[str]: Sorted list of registered component names.

    Example:
        ```python
        # List all detectors
        detectors = registry.list(SfComponentType.DETECTOR)
        print(f"Available detectors: {detectors}")
        # Output: ['ema_cross', 'macd', 'rsi_threshold', 'sma_cross']

        # Check if component exists
        if "sma_cross" in registry.list(SfComponentType.DETECTOR):
            detector = registry.create(SfComponentType.DETECTOR, "sma_cross")

        # List all component types
        from signalflow.core.enums import SfComponentType
        for component_type in SfComponentType:
            components = registry.list(component_type)
            print(f"{component_type.value}: {components}")
        ```
    """
    self._discover_if_needed()
    self._ensure(component_type)
    return sorted(self._items[component_type])

list_raw_data_types ¶

list_raw_data_types() -> builtins.list[str]

List all registered raw data type names.

Returns:

Type	Description
`list[str]`	Sorted list of registered data type names.

Source code in src/signalflow/core/registry.py

def list_raw_data_types(self) -> builtins.list[str]:
    """List all registered raw data type names.

    Returns:
        Sorted list of registered data type names.
    """
    return sorted(self._raw_data_types)

register ¶

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

Register a class under (component_type, name).

Stores class with metadata (docstring, module) in registry for later lookup, instantiation, and UI display. Names are normalized to lowercase for case-insensitive lookup.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of component (DETECTOR, EXTRACTOR, etc.).	required
`name`	`str`	Registry name (case-insensitive, will be lowercased).	required
`cls`	`Type[Any]`	Class to register.	required
`override`	`bool`	Allow overriding existing registration. Default: False.	`False`

Raises:

Type	Description
`ValueError`	If name is empty or already registered (when override=False).

Example

# Register new component
registry.register(
    SfComponentType.DETECTOR,
    name="my_detector",
    cls=MyDetector
)

# Override existing component
registry.register(
    SfComponentType.DETECTOR,
    name="my_detector",
    cls=ImprovedDetector,
    override=True  # Logs warning
)

# Register multiple types
registry.register(SfComponentType.EXTRACTOR, "rsi", RsiExtractor)
registry.register(SfComponentType.LABELER, "fixed", FixedHorizonLabeler)

Source code in src/signalflow/core/registry.py

def register(self, component_type: SfComponentType, name: str, cls: type[Any], *, override: bool = False) -> None:
    """Register a class under (component_type, name).

    Stores class with metadata (docstring, module) in registry for later
    lookup, instantiation, and UI display.
    Names are normalized to lowercase for case-insensitive lookup.

    Args:
        component_type (SfComponentType): Type of component (DETECTOR, EXTRACTOR, etc.).
        name (str): Registry name (case-insensitive, will be lowercased).
        cls (Type[Any]): Class to register.
        override (bool): Allow overriding existing registration. Default: False.

    Raises:
        ValueError: If name is empty or already registered (when override=False).

    Example:
        ```python
        # Register new component
        registry.register(
            SfComponentType.DETECTOR,
            name="my_detector",
            cls=MyDetector
        )

        # Override existing component
        registry.register(
            SfComponentType.DETECTOR,
            name="my_detector",
            cls=ImprovedDetector,
            override=True  # Logs warning
        )

        # Register multiple types
        registry.register(SfComponentType.EXTRACTOR, "rsi", RsiExtractor)
        registry.register(SfComponentType.LABELER, "fixed", FixedHorizonLabeler)
        ```
    """
    if not isinstance(name, str) or not name.strip():
        raise ValueError("name must be a non-empty string")

    key = name.strip().lower()
    self._ensure(component_type)

    if key in self._items[component_type] and not override:
        raise ValueError(f"{component_type.value}:{key} already registered")

    if key in self._items[component_type] and override:
        logger.warning(f"Overriding {component_type.value}:{key} with {cls.__name__}")

    # Store class with extracted metadata
    self._items[component_type][key] = ComponentInfo.from_class(cls)

register_raw_data_type ¶

register_raw_data_type(name: str, columns: list[str] | set[str], *, override: bool = False) -> None

Register a custom raw data type with its required columns.

Parameters:

Name	Type	Description	Default
`name`	`str`	Data type identifier (case-insensitive, stored lowercase).	required
`columns`	`list[str] \| set[str]`	Required column names for this data type.	required
`override`	`bool`	Allow overriding an existing registration.	`False`

Raises:

Type	Description
`ValueError`	If name is empty, columns are empty, or name already registered (when override=False).

Example

default_registry.register_raw_data_type(
    name="lob",
    columns=["pair", "timestamp", "bid", "ask", "bid_size", "ask_size"],
)

Source code in src/signalflow/core/registry.py

def register_raw_data_type(
    self,
    name: str,
    columns: builtins.list[str] | set[str],
    *,
    override: bool = False,
) -> None:
    """Register a custom raw data type with its required columns.

    Args:
        name: Data type identifier (case-insensitive, stored lowercase).
        columns: Required column names for this data type.
        override: Allow overriding an existing registration.

    Raises:
        ValueError: If name is empty, columns are empty, or name already
            registered (when override=False).

    Example:
        ```python
        default_registry.register_raw_data_type(
            name="lob",
            columns=["pair", "timestamp", "bid", "ask", "bid_size", "ask_size"],
        )
        ```
    """
    if not isinstance(name, str) or not name.strip():
        raise ValueError("name must be a non-empty string")

    cols = set(columns)
    if not cols:
        raise ValueError("columns must be a non-empty collection")

    key = name.strip().lower()

    if key in self._raw_data_types and not override:
        raise ValueError(f"Raw data type '{key}' already registered")

    if key in self._raw_data_types and override:
        logger.warning(f"Overriding raw data type '{key}'")

    self._raw_data_types[key] = cols

snapshot ¶

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

Snapshot of registry for debugging.

Returns complete registry state organized by component type.

Returns:

Type	Description
`dict[str, list[str]]`	dict[str, list[str]]: Dictionary mapping component type names to sorted lists of registered component names.

Example

# Get full registry snapshot
snapshot = registry.snapshot()
print(snapshot)
# Output:
# {
#     'DETECTOR': ['ema_cross', 'sma_cross'],
#     'EXTRACTOR': ['rsi', 'sma'],
#     'LABELER': ['fixed', 'triple_barrier'],
#     'ENTRY_RULE': ['fixed_size'],
#     'EXIT_RULE': ['take_profit', 'time_based']
# }

# Use for debugging
import json
print(json.dumps(registry.snapshot(), indent=2))

# Check registration status
snapshot = registry.snapshot()
if 'DETECTOR' in snapshot and 'sma_cross' in snapshot['DETECTOR']:
    print("SMA detector is registered")

Source code in src/signalflow/core/registry.py

def snapshot(self) -> dict[str, builtins.list[str]]:
    """Snapshot of registry for debugging.

    Returns complete registry state organized by component type.

    Returns:
        dict[str, list[str]]: Dictionary mapping component type names
            to sorted lists of registered component names.

    Example:
        ```python
        # Get full registry snapshot
        snapshot = registry.snapshot()
        print(snapshot)
        # Output:
        # {
        #     'DETECTOR': ['ema_cross', 'sma_cross'],
        #     'EXTRACTOR': ['rsi', 'sma'],
        #     'LABELER': ['fixed', 'triple_barrier'],
        #     'ENTRY_RULE': ['fixed_size'],
        #     'EXIT_RULE': ['take_profit', 'time_based']
        # }

        # Use for debugging
        import json
        print(json.dumps(registry.snapshot(), indent=2))

        # Check registration status
        snapshot = registry.snapshot()
        if 'DETECTOR' in snapshot and 'sma_cross' in snapshot['DETECTOR']:
            print("SMA detector is registered")
        ```
    """
    self._discover_if_needed()
    return {t.value: sorted(v.keys()) for t, v in self._items.items()}

sf_component ¶

sf_component(*, name: str, override: bool = True) -> Callable[[type[Any]], type[Any]]

Register class as SignalFlow component.

.. deprecated:: 0.6.0 Use semantic decorators instead: - @sf.detector("name") for detectors - @sf.feature("name") for features - @sf.entry("name") for entry rules - @sf.exit("name") for exit rules - etc.

Decorator that registers a class in the global component registry, making it discoverable by name for dynamic instantiation.

The decorated class must have a component_type class attribute of type SfComponentType to indicate what kind of component it is.

Parameters:

Name	Type	Description	Default
`name`	`str`	Registry name for the component (case-insensitive).	required
`override`	`bool`	Allow overriding existing registration. Default: True.	`True`

Returns:

Type	Description
`Callable[[type[Any]], type[Any]]`	Decorator function that registers and returns the class unchanged.

Raises:

Type	Description
`ValueError`	If class doesn't define component_type attribute.

Source code in src/signalflow/core/decorators.py

def sf_component(*, name: str, override: bool = True) -> Callable[[type[Any]], type[Any]]:
    """Register class as SignalFlow component.

    .. deprecated:: 0.6.0
        Use semantic decorators instead:
        - @sf.detector("name") for detectors
        - @sf.feature("name") for features
        - @sf.entry("name") for entry rules
        - @sf.exit("name") for exit rules
        - etc.

    Decorator that registers a class in the global component registry,
    making it discoverable by name for dynamic instantiation.

    The decorated class must have a `component_type` class attribute
    of type `SfComponentType` to indicate what kind of component it is.

    Args:
        name: Registry name for the component (case-insensitive).
        override: Allow overriding existing registration. Default: True.

    Returns:
        Decorator function that registers and returns the class unchanged.

    Raises:
        ValueError: If class doesn't define component_type attribute.
    """
    warnings.warn(
        "@sf_component is deprecated. Use semantic decorators instead:\n"
        "  @sf.detector('name'), @sf.feature('name'), @sf.entry('name'), etc.\n"
        "See https://signalflow-trading.com/migration for details.",
        DeprecationWarning,
        stacklevel=2,
    )

    def decorator(cls: type[Any]) -> type[Any]:
        component_type = getattr(cls, "component_type", None)
        if not isinstance(component_type, SfComponentType):
            raise ValueError(f"{cls.__name__} must define class attribute 'component_type: SfComponentType'")

        default_registry.register(
            component_type,
            name=name,
            cls=cls,
            override=override,
        )
        return cls

    return decorator

signalflow.core.enums ¶

SfComponentType ¶

Bases: StrEnum

Enumeration of SignalFlow component types.

Defines all component types that can be registered in the component registry. Used by sf_component decorator and SignalFlowRegistry for type-safe registration.

Component categories

Data: Raw data loading and storage
Feature: Feature extraction
Signals: Signal detection, transformation, labeling, validation
Strategy: Execution, rules, metrics

Values

RAW_DATA_STORE: Raw data storage backends (e.g., DuckDB, Parquet). RAW_DATA_SOURCE: Raw data sources (e.g., Binance API). RAW_DATA_LOADER: Raw data loaders combining source + store. FEATURE: Feature extraction classes (e.g., RSI, SMA). SIGNALS_TRANSFORM: Signal transformation functions. LABELER: Signal labeling strategies (e.g., triple barrier). DETECTOR: Signal detection algorithms (e.g., SMA cross). VALIDATOR: Signal validation models. TORCH_MODULE: PyTorch neural network modules. VALIDATOR_MODEL: Pre-trained validator models. STRATEGY_STORE: Strategy state persistence backends. STRATEGY_RUNNER: Backtest/live runner implementations. STRATEGY_BROKER: Order management and position tracking. STRATEGY_EXECUTOR: Order execution engines (backtest/live). STRATEGY_EXIT_RULE: Position exit rules (e.g., take profit, stop loss). STRATEGY_ENTRY_RULE: Position entry rules (e.g., fixed size). STRATEGY_METRIC: Strategy performance metrics. STRATEGY_ALERT: Strategy monitoring alerts (e.g., max drawdown, stuck positions).

Example

import signalflow as sf
from signalflow.detector import SignalDetector

# Register detector with semantic decorator
@sf.detector("my_detector")
class MyDetector(SignalDetector):
    # ... implementation

# Register feature
@sf.feature("my_feature")
class MyFeature(Feature):
    # ... implementation

# Register exit rule
@sf.exit("my_exit")
class MyExit(ExitRule):
    # ... implementation

# Use in registry
from signalflow.core.registry import default_registry
from signalflow.core.enums import SfComponentType

detector = default_registry.create(
    SfComponentType.DETECTOR,
    "my_detector"
)

Note

All registered components must have component_type class attribute. Component types are organized hierarchically (category/subcategory).

DataFrameType ¶

Bases: StrEnum

Supported DataFrame backends.

Specifies which DataFrame library to use for data processing. Used by FeatureExtractor and other components to determine input/output format.

Values

POLARS: Polars DataFrame (faster, modern). PANDAS: Pandas DataFrame (legacy compatibility).

Example

from signalflow.core.enums import DataFrameType
from signalflow.feature import FeatureExtractor

# Polars-based extractor
class MyExtractor(FeatureExtractor):
    df_type = DataFrameType.POLARS

    def extract(self, df: pl.DataFrame) -> pl.DataFrame:
        return df.with_columns(
            pl.col("close").rolling_mean(20).alias("sma_20")
        )

# Pandas-based extractor
class LegacyExtractor(FeatureExtractor):
    df_type = DataFrameType.PANDAS

    def extract(self, df: pd.DataFrame) -> pd.DataFrame:
        df["sma_20"] = df["close"].rolling(20).mean()
        return df

# Use in RawDataView
from signalflow.core import RawDataView

view = RawDataView(raw=raw_data)

# Get data in required format
df_polars = view.get_data("spot", DataFrameType.POLARS)
df_pandas = view.get_data("spot", DataFrameType.PANDAS)

Note

New code should prefer POLARS for better performance. PANDAS supported for backward compatibility and legacy libraries.

RawDataType ¶

Bases: StrEnum

Built-in raw data types.

Defines types of market data that can be loaded and processed. Column definitions are stored in :class:SignalFlowRegistry and can be extended with custom types via default_registry.register_raw_data_type().

Values

SPOT: Spot trading data (OHLCV). FUTURES: Futures trading data (OHLCV + open_interest). PERPETUAL: Perpetual swaps data (OHLCV + funding_rate + open_interest).

Example

from signalflow.core.enums import RawDataType

# Built-in types
spot_cols = RawDataType.SPOT.columns
# {'pair', 'timestamp', 'open', 'high', 'low', 'close', 'volume'}

# Custom types - register via registry
from signalflow.core.registry import default_registry

default_registry.register_raw_data_type(
    name="lob",
    columns=["pair", "timestamp", "bid", "ask", "bid_size", "ask_size"],
)
cols = default_registry.get_raw_data_columns("lob")

Note

Use default_registry.register_raw_data_type() to add custom types. Use default_registry.get_raw_data_columns(name) to look up columns for any type (built-in or custom).

columns `property` ¶

columns: set[str]

Columns guaranteed to be present (looked up from registry).

signalflow.core.registry ¶

default_registry `module-attribute` ¶

default_registry = SignalFlowRegistry()

Global default registry instance.

Use this singleton for application-wide component registration.

Example

from signalflow.core.registry import default_registry
from signalflow.core.enums import SfComponentType

# Register to default registry
default_registry.register(
    SfComponentType.DETECTOR,
    "my_detector",
    MyDetector
)

# Access from anywhere
detector = default_registry.create(
    SfComponentType.DETECTOR,
    "my_detector"
)

ComponentInfo `dataclass` ¶

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

Metadata about a registered component.

Stores the class reference along with extracted documentation for UI display and introspection.

Attributes:

Name	Type	Description
`cls`	`type[Any]`	The registered component class.
`docstring`	`str`	Full class docstring (or empty string if none).
`summary`	`str`	First line of docstring (short description).
`module`	`str`	Module path where the class is defined.

from_class `classmethod` ¶

from_class(component_cls: type[Any]) -> ComponentInfo

Create ComponentInfo by extracting metadata from a class.

Source code in src/signalflow/core/registry.py

@classmethod
def from_class(cls, component_cls: type[Any]) -> ComponentInfo:
    """Create ComponentInfo by extracting metadata from a class."""
    docstring = (component_cls.__doc__ or "").strip()
    lines = docstring.split("\n")
    summary = lines[0] if lines else ""
    module = getattr(component_cls, "__module__", "")
    return cls(
        cls=component_cls,
        docstring=docstring,
        summary=summary,
        module=module,
    )

SignalFlowRegistry `dataclass` ¶

SignalFlowRegistry(_items: dict[SfComponentType, dict[str, ComponentInfo]] = dict(), _raw_data_types: dict[str, set[str]] = (lambda: {k: (v.copy()) for k, v in (_BUILTIN_RAW_DATA_TYPES.items())})(), _discovered: bool = False)

Component registry for dynamic component discovery and instantiation.

Provides centralized registration and lookup for SignalFlow components. Components are organized by type (DETECTOR, EXTRACTOR, etc.) and accessed by case-insensitive names.

Also manages extensible raw data type definitions - each data type maps to a set of required columns. Built-in types (SPOT, FUTURES, PERPETUAL) are pre-registered; users can add custom types via register_raw_data_type().

Registry structure

component_type -> name -> ComponentInfo (class + metadata)

Supported component types

DETECTOR: Signal detection classes
EXTRACTOR: Feature extraction classes
LABELER: Signal labeling classes
ENTRY_RULE: Position entry rules
EXIT_RULE: Position exit rules
METRIC: Strategy metrics
EXECUTOR: Order execution engines

Attributes:

Name	Type	Description
`_items`	`dict[SfComponentType, dict[str, ComponentInfo]]`	Internal storage mapping component types to name-ComponentInfo pairs.
`_raw_data_types`	`dict[str, set[str]]`	Mapping of raw data type names to their required column sets.

Example

from signalflow.core.registry import SignalFlowRegistry, default_registry

# Register custom raw data type
default_registry.register_raw_data_type(
    name="lob",
    columns=["pair", "timestamp", "bid", "ask", "bid_size", "ask_size"],
)

# Get component info with docstring for UI
info = default_registry.get_info(SfComponentType.DETECTOR, "sma_cross")
print(info.summary)    # "Detects SMA crossover signals."
print(info.docstring)  # Full docstring with Args, Example, etc.

# Get columns for any type
cols = default_registry.get_raw_data_columns("spot")
custom_cols = default_registry.get_raw_data_columns("lob")

# List all registered raw data types
print(default_registry.list_raw_data_types())

Note

Component names are stored and looked up in lowercase. Use default_registry singleton for application-wide registration.

autodiscover ¶

autodiscover() -> None

Scan signalflow.* packages and entry-points for components.

Walks all sub-modules of the signalflow package using :func:pkgutil.walk_packages and imports them. Because semantic decorators register classes at import time, importing a module is sufficient to populate the registry.

External packages can expose components via the signalflow.components entry-point group. Each entry-point should reference a module (not a callable); importing it triggers registration through semantic decorators.

This method is idempotent - subsequent calls are no-ops once _discovered is True.

Example

from signalflow.core.registry import default_registry

# Explicit discovery (normally automatic on first get/list)
default_registry.autodiscover()

# All decorated classes are now registered
print(default_registry.snapshot())

Source code in src/signalflow/core/registry.py

def autodiscover(self) -> None:
    """Scan ``signalflow.*`` packages and entry-points for components.

    Walks all sub-modules of the ``signalflow`` package using
    :func:`pkgutil.walk_packages` and imports them.  Because
    semantic decorators register classes at import time, importing
    a module is sufficient to populate the registry.

    External packages can expose components via the
    ``signalflow.components`` entry-point group.  Each entry-point
    should reference a module (not a callable); importing it triggers
    registration through semantic decorators.

    This method is idempotent - subsequent calls are no-ops once
    ``_discovered`` is ``True``.

    Example:
        ```python
        from signalflow.core.registry import default_registry

        # Explicit discovery (normally automatic on first get/list)
        default_registry.autodiscover()

        # All decorated classes are now registered
        print(default_registry.snapshot())
        ```
    """
    if self._discovered:
        return
    self._discovered = True

    self._discover_internal_packages()
    self._discover_entry_points()

create ¶

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

Instantiate a component by registry key.

Convenient method that combines get() and instantiation.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of component to create.	required
`name`	`str`	Component name (case-insensitive).	required
`**kwargs`	`Any`	Arguments to pass to component constructor.	`{}`

Returns:

Name	Type	Description
`Any`	`Any`	Instantiated component.

Raises:

Type	Description
`KeyError`	If component not found.
`TypeError`	If kwargs don't match component constructor.

Example

# Create detector with params
detector = registry.create(
    SfComponentType.DETECTOR,
    "sma_cross",
    fast_window=10,
    slow_window=20
)

# Create extractor
extractor = registry.create(
    SfComponentType.EXTRACTOR,
    "rsi",
    window=14
)

# Create with config dict
config = {"window": 20, "threshold": 0.7}
labeler = registry.create(
    SfComponentType.LABELER,
    "fixed",
    **config
)

Source code in src/signalflow/core/registry.py

def create(self, component_type: SfComponentType, name: str, **kwargs: Any) -> Any:
    """Instantiate a component by registry key.

    Convenient method that combines get() and instantiation.

    Args:
        component_type (SfComponentType): Type of component to create.
        name (str): Component name (case-insensitive).
        **kwargs: Arguments to pass to component constructor.

    Returns:
        Any: Instantiated component.

    Raises:
        KeyError: If component not found.
        TypeError: If kwargs don't match component constructor.

    Example:
        ```python
        # Create detector with params
        detector = registry.create(
            SfComponentType.DETECTOR,
            "sma_cross",
            fast_window=10,
            slow_window=20
        )

        # Create extractor
        extractor = registry.create(
            SfComponentType.EXTRACTOR,
            "rsi",
            window=14
        )

        # Create with config dict
        config = {"window": 20, "threshold": 0.7}
        labeler = registry.create(
            SfComponentType.LABELER,
            "fixed",
            **config
        )
        ```
    """
    cls = self.get(component_type, name)
    return cls(**kwargs)

export_schemas ¶

export_schemas() -> dict[str, builtins.list[dict[str, Any]]]

Export schemas for all registered components, grouped by type.

Useful for bulk loading into UI component browsers without N+1 calls.

Returns:

Type	Description
`dict[str, list[dict[str, Any]]]`	Dict mapping component type names to lists of component schemas.

Example

all_schemas = registry.export_schemas() for det in all_schemas.get("DETECTOR", []): ... print(det["name"], len(det["parameters"]), "params")

Source code in src/signalflow/core/registry.py

def export_schemas(self) -> dict[str, builtins.list[dict[str, Any]]]:
    """Export schemas for all registered components, grouped by type.

    Useful for bulk loading into UI component browsers without N+1 calls.

    Returns:
        Dict mapping component type names to lists of component schemas.

    Example:
        >>> all_schemas = registry.export_schemas()
        >>> for det in all_schemas.get("DETECTOR", []):
        ...     print(det["name"], len(det["parameters"]), "params")
    """
    self._discover_if_needed()
    result: dict[str, list[dict[str, Any]]] = {}
    for comp_type, names_dict in self._items.items():
        schemas = []
        for name in sorted(names_dict):
            try:
                schemas.append(self.get_schema(comp_type, name))
            except Exception:
                logger.debug(f"export_schemas: failed to get schema for {comp_type.value}:{name}")
        if schemas:
            result[comp_type.value] = schemas
    return result

get ¶

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

Get a registered class by key.

Lookup is case-insensitive. Raises helpful error with available components if key not found.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of component to lookup.	required
`name`	`str`	Component name (case-insensitive).	required

Returns:

Type	Description
`type[Any]`	Type[Any]: Registered class.

Raises:

Type	Description
`KeyError`	If component not found. Error message includes available components.

Example

# Get component class
detector_cls = registry.get(SfComponentType.DETECTOR, "sma_cross")

# Case-insensitive
detector_cls = registry.get(SfComponentType.DETECTOR, "SMA_Cross")

# Instantiate manually
detector = detector_cls(fast_window=10, slow_window=20)

# Handle missing component
try:
    cls = registry.get(SfComponentType.DETECTOR, "unknown")
except KeyError as e:
    print(f"Component not found: {e}")
    # Shows: "Component not found: DETECTOR:unknown. Available: [sma_cross, ...]"

Source code in src/signalflow/core/registry.py

def get(self, component_type: SfComponentType, name: str) -> type[Any]:
    """Get a registered class by key.

    Lookup is case-insensitive. Raises helpful error with available
    components if key not found.

    Args:
        component_type (SfComponentType): Type of component to lookup.
        name (str): Component name (case-insensitive).

    Returns:
        Type[Any]: Registered class.

    Raises:
        KeyError: If component not found. Error message includes available components.

    Example:
        ```python
        # Get component class
        detector_cls = registry.get(SfComponentType.DETECTOR, "sma_cross")

        # Case-insensitive
        detector_cls = registry.get(SfComponentType.DETECTOR, "SMA_Cross")

        # Instantiate manually
        detector = detector_cls(fast_window=10, slow_window=20)

        # Handle missing component
        try:
            cls = registry.get(SfComponentType.DETECTOR, "unknown")
        except KeyError as e:
            print(f"Component not found: {e}")
            # Shows: "Component not found: DETECTOR:unknown. Available: [sma_cross, ...]"
        ```
    """
    self._discover_if_needed()
    self._ensure(component_type)
    key = name.lower()
    try:
        return self._items[component_type][key].cls
    except KeyError as e:
        available = ", ".join(sorted(self._items[component_type]))
        raise KeyError(f"Component not found: {component_type.value}:{key}. Available: [{available}]") from e

get_info ¶

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

Get full component info including docstring.

Use this method when you need metadata for UI display or documentation generation.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of component to lookup.	required
`name`	`str`	Component name (case-insensitive).	required

Returns:

Name	Type	Description
`ComponentInfo`	`ComponentInfo`	Full metadata including class, docstring, summary, module.

Raises:

Type	Description
`KeyError`	If component not found.

Example

# Get info for UI tooltip
info = registry.get_info(SfComponentType.DETECTOR, "sma_cross")
print(info.summary)     # "Detects SMA crossover signals."
print(info.docstring)   # Full docstring
print(info.module)      # "signalflow.detector.sma_cross"

# Use in sf-ui component browser
for name in registry.list(SfComponentType.FEATURE):
    info = registry.get_info(SfComponentType.FEATURE, name)
    display_component_card(name, info.summary, info.docstring)

Source code in src/signalflow/core/registry.py

def get_info(self, component_type: SfComponentType, name: str) -> ComponentInfo:
    """Get full component info including docstring.

    Use this method when you need metadata for UI display or
    documentation generation.

    Args:
        component_type (SfComponentType): Type of component to lookup.
        name (str): Component name (case-insensitive).

    Returns:
        ComponentInfo: Full metadata including class, docstring, summary, module.

    Raises:
        KeyError: If component not found.

    Example:
        ```python
        # Get info for UI tooltip
        info = registry.get_info(SfComponentType.DETECTOR, "sma_cross")
        print(info.summary)     # "Detects SMA crossover signals."
        print(info.docstring)   # Full docstring
        print(info.module)      # "signalflow.detector.sma_cross"

        # Use in sf-ui component browser
        for name in registry.list(SfComponentType.FEATURE):
            info = registry.get_info(SfComponentType.FEATURE, name)
            display_component_card(name, info.summary, info.docstring)
        ```
    """
    self._discover_if_needed()
    self._ensure(component_type)
    key = name.lower()
    try:
        return self._items[component_type][key]
    except KeyError as e:
        available = ", ".join(sorted(self._items[component_type]))
        raise KeyError(f"Component not found: {component_type.value}:{key}. Available: [{available}]") from e

get_raw_data_columns ¶

get_raw_data_columns(name: str) -> set[str]

Get required columns for a raw data type.

Parameters:

Name	Type	Description	Default
`name`	`str`	Data type identifier (case-insensitive). Accepts both `RawDataType` enum members and plain strings.	required

Returns:

Type	Description
`set[str]`	Copy of the column set for the requested type.

Raises:

Type	Description
`KeyError`	If data type is not registered.

Example

cols = default_registry.get_raw_data_columns("spot")
# {'pair', 'timestamp', 'open', 'high', 'low', 'close', 'volume'}

cols = default_registry.get_raw_data_columns("lob")
# {'pair', 'timestamp', 'bid', 'ask', ...}

Source code in src/signalflow/core/registry.py

def get_raw_data_columns(self, name: str) -> set[str]:
    """Get required columns for a raw data type.

    Args:
        name: Data type identifier (case-insensitive). Accepts both
            ``RawDataType`` enum members and plain strings.

    Returns:
        Copy of the column set for the requested type.

    Raises:
        KeyError: If data type is not registered.

    Example:
        ```python
        cols = default_registry.get_raw_data_columns("spot")
        # {'pair', 'timestamp', 'open', 'high', 'low', 'close', 'volume'}

        cols = default_registry.get_raw_data_columns("lob")
        # {'pair', 'timestamp', 'bid', 'ask', ...}
        ```
    """
    raw = getattr(name, "value", name)
    key = str(raw).strip().lower()
    try:
        return self._raw_data_types[key].copy()
    except KeyError:
        available = ", ".join(sorted(self._raw_data_types))
        raise KeyError(f"Raw data type '{key}' not registered. Available: [{available}]") from None

get_schema ¶

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

Get JSON-serializable parameter schema for a registered component.

Uses dataclasses.fields() to introspect @dataclass-decorated components and extract field names, types, and defaults.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of component.	required
`name`	`str`	Component registry name (case-insensitive).	required

Returns:

Type	Description
`dict[str, Any]`	Schema dict with keys: `name`, `class_name`, `component_type`,
`dict[str, Any]`	`description`, `docstring`, `module`, `parameters`,
`dict[str, Any]`	`requires`, `outputs`.

Raises:

Type	Description
`KeyError`	If component not found.

Example

schema = registry.get_schema(SfComponentType.DETECTOR, "example/sma_cross") print(schema["description"]) # Short summary print(schema["docstring"]) # Full docstring for UI for p in schema["parameters"]: ... print(f"{p['name']}: {p['type']} = {p['default']}")

Source code in src/signalflow/core/registry.py

def get_schema(self, component_type: SfComponentType, name: str) -> dict[str, Any]:
    """Get JSON-serializable parameter schema for a registered component.

    Uses ``dataclasses.fields()`` to introspect ``@dataclass``-decorated
    components and extract field names, types, and defaults.

    Args:
        component_type: Type of component.
        name: Component registry name (case-insensitive).

    Returns:
        Schema dict with keys: ``name``, ``class_name``, ``component_type``,
        ``description``, ``docstring``, ``module``, ``parameters``,
        ``requires``, ``outputs``.

    Raises:
        KeyError: If component not found.

    Example:
        >>> schema = registry.get_schema(SfComponentType.DETECTOR, "example/sma_cross")
        >>> print(schema["description"])  # Short summary
        >>> print(schema["docstring"])    # Full docstring for UI
        >>> for p in schema["parameters"]:
        ...     print(f"{p['name']}: {p['type']} = {p['default']}")
    """
    info = self.get_info(component_type, name)
    cls = info.cls

    parameters: list[dict[str, Any]] = []
    if dataclasses.is_dataclass(cls):
        for f in dataclasses.fields(cls):
            if f.name.startswith("_") or f.name in self._BASE_FIELDS:
                continue
            # Skip ClassVar fields (they show up as strings containing "ClassVar")
            type_str = self._type_to_str(f.type)
            if "ClassVar" in type_str:
                continue

            has_default = f.default is not dataclasses.MISSING
            has_default_factory = f.default_factory is not dataclasses.MISSING  # type: ignore[misc]

            parameters.append(
                {
                    "name": f.name,
                    "type": type_str,
                    "default": f.default if has_default else None,
                    "required": not has_default and not has_default_factory,
                }
            )

    # Extract ClassVar metadata
    requires = getattr(cls, "requires", [])
    outputs = getattr(cls, "outputs", [])

    return {
        "name": name,
        "class_name": cls.__name__,
        "component_type": component_type.value,
        "description": info.summary,
        "docstring": info.docstring,
        "module": info.module,
        "parameters": parameters,
        "requires": list(requires) if requires else [],
        "outputs": list(outputs) if outputs else [],
    }

list ¶

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

List registered components for a type.

Returns sorted list of component names for given type.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of components to list.	required

Returns:

Type	Description
`list[str]`	list[str]: Sorted list of registered component names.

Example

# List all detectors
detectors = registry.list(SfComponentType.DETECTOR)
print(f"Available detectors: {detectors}")
# Output: ['ema_cross', 'macd', 'rsi_threshold', 'sma_cross']

# Check if component exists
if "sma_cross" in registry.list(SfComponentType.DETECTOR):
    detector = registry.create(SfComponentType.DETECTOR, "sma_cross")

# List all component types
from signalflow.core.enums import SfComponentType
for component_type in SfComponentType:
    components = registry.list(component_type)
    print(f"{component_type.value}: {components}")

Source code in src/signalflow/core/registry.py

def list(self, component_type: SfComponentType) -> list[str]:
    """List registered components for a type.

    Returns sorted list of component names for given type.

    Args:
        component_type (SfComponentType): Type of components to list.

    Returns:
        list[str]: Sorted list of registered component names.

    Example:
        ```python
        # List all detectors
        detectors = registry.list(SfComponentType.DETECTOR)
        print(f"Available detectors: {detectors}")
        # Output: ['ema_cross', 'macd', 'rsi_threshold', 'sma_cross']

        # Check if component exists
        if "sma_cross" in registry.list(SfComponentType.DETECTOR):
            detector = registry.create(SfComponentType.DETECTOR, "sma_cross")

        # List all component types
        from signalflow.core.enums import SfComponentType
        for component_type in SfComponentType:
            components = registry.list(component_type)
            print(f"{component_type.value}: {components}")
        ```
    """
    self._discover_if_needed()
    self._ensure(component_type)
    return sorted(self._items[component_type])

list_raw_data_types ¶

list_raw_data_types() -> builtins.list[str]

List all registered raw data type names.

Returns:

Type	Description
`list[str]`	Sorted list of registered data type names.

Source code in src/signalflow/core/registry.py

def list_raw_data_types(self) -> builtins.list[str]:
    """List all registered raw data type names.

    Returns:
        Sorted list of registered data type names.
    """
    return sorted(self._raw_data_types)

register ¶

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

Register a class under (component_type, name).

Stores class with metadata (docstring, module) in registry for later lookup, instantiation, and UI display. Names are normalized to lowercase for case-insensitive lookup.

Parameters:

Name	Type	Description	Default
`component_type`	`SfComponentType`	Type of component (DETECTOR, EXTRACTOR, etc.).	required
`name`	`str`	Registry name (case-insensitive, will be lowercased).	required
`cls`	`Type[Any]`	Class to register.	required
`override`	`bool`	Allow overriding existing registration. Default: False.	`False`

Raises:

Type	Description
`ValueError`	If name is empty or already registered (when override=False).

Example

# Register new component
registry.register(
    SfComponentType.DETECTOR,
    name="my_detector",
    cls=MyDetector
)

# Override existing component
registry.register(
    SfComponentType.DETECTOR,
    name="my_detector",
    cls=ImprovedDetector,
    override=True  # Logs warning
)

# Register multiple types
registry.register(SfComponentType.EXTRACTOR, "rsi", RsiExtractor)
registry.register(SfComponentType.LABELER, "fixed", FixedHorizonLabeler)

Source code in src/signalflow/core/registry.py

def register(self, component_type: SfComponentType, name: str, cls: type[Any], *, override: bool = False) -> None:
    """Register a class under (component_type, name).

    Stores class with metadata (docstring, module) in registry for later
    lookup, instantiation, and UI display.
    Names are normalized to lowercase for case-insensitive lookup.

    Args:
        component_type (SfComponentType): Type of component (DETECTOR, EXTRACTOR, etc.).
        name (str): Registry name (case-insensitive, will be lowercased).
        cls (Type[Any]): Class to register.
        override (bool): Allow overriding existing registration. Default: False.

    Raises:
        ValueError: If name is empty or already registered (when override=False).

    Example:
        ```python
        # Register new component
        registry.register(
            SfComponentType.DETECTOR,
            name="my_detector",
            cls=MyDetector
        )

        # Override existing component
        registry.register(
            SfComponentType.DETECTOR,
            name="my_detector",
            cls=ImprovedDetector,
            override=True  # Logs warning
        )

        # Register multiple types
        registry.register(SfComponentType.EXTRACTOR, "rsi", RsiExtractor)
        registry.register(SfComponentType.LABELER, "fixed", FixedHorizonLabeler)
        ```
    """
    if not isinstance(name, str) or not name.strip():
        raise ValueError("name must be a non-empty string")

    key = name.strip().lower()
    self._ensure(component_type)

    if key in self._items[component_type] and not override:
        raise ValueError(f"{component_type.value}:{key} already registered")

    if key in self._items[component_type] and override:
        logger.warning(f"Overriding {component_type.value}:{key} with {cls.__name__}")

    # Store class with extracted metadata
    self._items[component_type][key] = ComponentInfo.from_class(cls)

register_raw_data_type ¶

register_raw_data_type(name: str, columns: list[str] | set[str], *, override: bool = False) -> None

Register a custom raw data type with its required columns.

Parameters:

Name	Type	Description	Default
`name`	`str`	Data type identifier (case-insensitive, stored lowercase).	required
`columns`	`list[str] \| set[str]`	Required column names for this data type.	required
`override`	`bool`	Allow overriding an existing registration.	`False`

Raises:

Type	Description
`ValueError`	If name is empty, columns are empty, or name already registered (when override=False).

Example

default_registry.register_raw_data_type(
    name="lob",
    columns=["pair", "timestamp", "bid", "ask", "bid_size", "ask_size"],
)

Source code in src/signalflow/core/registry.py

def register_raw_data_type(
    self,
    name: str,
    columns: builtins.list[str] | set[str],
    *,
    override: bool = False,
) -> None:
    """Register a custom raw data type with its required columns.

    Args:
        name: Data type identifier (case-insensitive, stored lowercase).
        columns: Required column names for this data type.
        override: Allow overriding an existing registration.

    Raises:
        ValueError: If name is empty, columns are empty, or name already
            registered (when override=False).

    Example:
        ```python
        default_registry.register_raw_data_type(
            name="lob",
            columns=["pair", "timestamp", "bid", "ask", "bid_size", "ask_size"],
        )
        ```
    """
    if not isinstance(name, str) or not name.strip():
        raise ValueError("name must be a non-empty string")

    cols = set(columns)
    if not cols:
        raise ValueError("columns must be a non-empty collection")

    key = name.strip().lower()

    if key in self._raw_data_types and not override:
        raise ValueError(f"Raw data type '{key}' already registered")

    if key in self._raw_data_types and override:
        logger.warning(f"Overriding raw data type '{key}'")

    self._raw_data_types[key] = cols

snapshot ¶

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

Snapshot of registry for debugging.

Returns complete registry state organized by component type.

Returns:

Type	Description
`dict[str, list[str]]`	dict[str, list[str]]: Dictionary mapping component type names to sorted lists of registered component names.

Example

# Get full registry snapshot
snapshot = registry.snapshot()
print(snapshot)
# Output:
# {
#     'DETECTOR': ['ema_cross', 'sma_cross'],
#     'EXTRACTOR': ['rsi', 'sma'],
#     'LABELER': ['fixed', 'triple_barrier'],
#     'ENTRY_RULE': ['fixed_size'],
#     'EXIT_RULE': ['take_profit', 'time_based']
# }

# Use for debugging
import json
print(json.dumps(registry.snapshot(), indent=2))

# Check registration status
snapshot = registry.snapshot()
if 'DETECTOR' in snapshot and 'sma_cross' in snapshot['DETECTOR']:
    print("SMA detector is registered")

Source code in src/signalflow/core/registry.py

def snapshot(self) -> dict[str, builtins.list[str]]:
    """Snapshot of registry for debugging.

    Returns complete registry state organized by component type.

    Returns:
        dict[str, list[str]]: Dictionary mapping component type names
            to sorted lists of registered component names.

    Example:
        ```python
        # Get full registry snapshot
        snapshot = registry.snapshot()
        print(snapshot)
        # Output:
        # {
        #     'DETECTOR': ['ema_cross', 'sma_cross'],
        #     'EXTRACTOR': ['rsi', 'sma'],
        #     'LABELER': ['fixed', 'triple_barrier'],
        #     'ENTRY_RULE': ['fixed_size'],
        #     'EXIT_RULE': ['take_profit', 'time_based']
        # }

        # Use for debugging
        import json
        print(json.dumps(registry.snapshot(), indent=2))

        # Check registration status
        snapshot = registry.snapshot()
        if 'DETECTOR' in snapshot and 'sma_cross' in snapshot['DETECTOR']:
            print("SMA detector is registered")
        ```
    """
    self._discover_if_needed()
    return {t.value: sorted(v.keys()) for t, v in self._items.items()}

get_component ¶

get_component(type: SfComponentType, name: str) -> type[Any]

Get a registered component by type and name.

Source code in src/signalflow/core/registry.py

def get_component(type: SfComponentType, name: str) -> type[Any]:
    """Get a registered component by type and name."""
    return default_registry.get(type, name)

get_component_info ¶

get_component_info(type: SfComponentType, name: str) -> ComponentInfo

Get component info including docstring for UI display.

Source code in src/signalflow/core/registry.py

def get_component_info(type: SfComponentType, name: str) -> ComponentInfo:
    """Get component info including docstring for UI display."""
    return default_registry.get_info(type, name)

Core Module¶

signalflow.core ¶

RawData dataclass ¶

__contains__ ¶

__getattr__ ¶

__getitem__ ¶

__iter__ ¶

get ¶

items ¶

keys ¶

sources ¶

values ¶

Signals dataclass ¶

__add__ ¶

apply ¶

pipe ¶

SignalType ¶

SignalFlowRegistry dataclass ¶

autodiscover ¶

create ¶

export_schemas ¶

get ¶

get_info ¶

get_raw_data_columns ¶

get_schema ¶

list ¶

list_raw_data_types ¶

register ¶

register_raw_data_type ¶

snapshot ¶

sf_component ¶

signalflow.core.enums ¶

SfComponentType ¶

DataFrameType ¶

RawDataType ¶

columns property ¶

signalflow.core.registry ¶

default_registry module-attribute ¶

ComponentInfo dataclass ¶

from_class classmethod ¶

SignalFlowRegistry dataclass ¶

autodiscover ¶

create ¶

export_schemas ¶

get ¶

get_info ¶

get_raw_data_columns ¶

get_schema ¶

list ¶

list_raw_data_types ¶

register ¶

register_raw_data_type ¶

snapshot ¶

get_component ¶

get_component_info ¶

RawData `dataclass` ¶

contains ¶

getattr ¶

getitem ¶

iter ¶

Signals `dataclass` ¶

add ¶

SignalFlowRegistry `dataclass` ¶

columns `property` ¶

default_registry `module-attribute` ¶

ComponentInfo `dataclass` ¶

from_class `classmethod` ¶

SignalFlowRegistry `dataclass` ¶