Skip to content

Data

Market data is held in a Dataset and produced by data sources. The data helper is a convenience loader that returns a ready-to-use Dataset.

Dataset

signalflow.Dataset dataclass 

Dataset(frame: DataFrame, source_name: str = '', source_params: dict = dict(), quote: str = 'USDT', provenance: Provenance = Provenance.FULL)

Immutable view over canonical OHLCV plus any computed columns.

cross_rate 

cross_rate(base: str, quote: str, prices: dict[str, float]) -> float

Price of base denominated in quote given a pair->close map.

Source code in src/signalflow/data/dataset.py 
91
92
93
94
95
def cross_rate(self, base: str, quote: str, prices: dict[str, float]) -> float:
    """Price of ``base`` denominated in ``quote`` given a pair->close map."""
    from signalflow.engine.types import cross_rate

    return cross_rate(base, quote, prices)

index 

index() -> pl.DataFrame

(pair, ts) of every row - the universe a sampler selects from.

Source code in src/signalflow/data/dataset.py 
53
54
55
def index(self) -> pl.DataFrame:
    """(pair, ts) of every row - the universe a sampler selects from."""
    return self.frame.select(["pair", "ts"])

iter_bars 

iter_bars(columns: list[str] | None = None) -> Iterator[Bar]

Yield one :class:Bar per timestamp in order (the replay backbone).

Source code in src/signalflow/data/dataset.py 
 97
 98
 99
100
101
102
103
def iter_bars(self, columns: list[str] | None = None) -> Iterator[Bar]:
    """Yield one :class:`Bar` per timestamp in order (the replay backbone)."""
    frame = self.frame if columns is None else self.frame.select(["pair", "ts", "close", *columns])
    for ts, sub in frame.sort("ts").group_by("ts", maintain_order=True):
        ts_val = ts[0] if isinstance(ts, tuple) else ts
        prices = dict(zip(sub.get_column("pair"), sub.get_column("close"), strict=True))
        yield Bar(ts=ts_val, frame=sub, prices=prices)

slice_time 

slice_time(start=None, end=None) -> Dataset

Rows with start <= ts < end (either bound optional). For walk-forward windows.

Source code in src/signalflow/data/dataset.py 
75
76
77
78
79
80
81
82
def slice_time(self, start=None, end=None) -> "Dataset":
    """Rows with start <= ts < end (either bound optional). For walk-forward windows."""
    expr = pl.lit(True)
    if start is not None:
        expr = expr & (pl.col("ts") >= start)
    if end is not None:
        expr = expr & (pl.col("ts") < end)
    return replace(self, frame=self.frame.filter(expr))

with_forecasts 

with_forecasts(cols: DataFrame, *, provenance: Provenance = Provenance.FULL) -> Dataset

Join forecast columns (keyed by pair, ts) and record their provenance.

Source code in src/signalflow/data/dataset.py 
70
71
72
73
def with_forecasts(self, cols: pl.DataFrame, *, provenance: Provenance = Provenance.FULL) -> "Dataset":
    """Join forecast columns (keyed by pair, ts) and record their provenance."""
    merged = self.frame.join(cols, on=["pair", "ts"], how="left")
    return replace(self, frame=merged, provenance=provenance)

Loader

signalflow.data

Data layer: Dataset container and source plugins.

Bar

Bases: NamedTuple

One timestamp's cross-section, fed to the decision loop.

BinanceSource dataclass 

BinanceSource(name: str = 'binance', base_url: str = _BASE, timeout: float = 20.0, max_retries: int = 3)

Bases: Source

Fetch spot OHLCV klines from Binance public REST.

CachedSource 

CachedSource(inner: Source, root: str | Path)

Bases: Source

Wrap inner, serving repeated requests from one growing parquet per (pair, interval).

Source code in src/signalflow/data/source/cached.py 
32
33
34
35
def __init__(self, inner: Source, root: "str | Path") -> None:
    self.inner = inner
    self.root = Path(root)
    self.name = getattr(inner, "name", "cached")

Dataset dataclass 

Dataset(frame: DataFrame, source_name: str = '', source_params: dict = dict(), quote: str = 'USDT', provenance: Provenance = Provenance.FULL)

Immutable view over canonical OHLCV plus any computed columns.

cross_rate 

cross_rate(base: str, quote: str, prices: dict[str, float]) -> float

Price of base denominated in quote given a pair->close map.

Source code in src/signalflow/data/dataset.py 
91
92
93
94
95
def cross_rate(self, base: str, quote: str, prices: dict[str, float]) -> float:
    """Price of ``base`` denominated in ``quote`` given a pair->close map."""
    from signalflow.engine.types import cross_rate

    return cross_rate(base, quote, prices)

index 

index() -> pl.DataFrame

(pair, ts) of every row - the universe a sampler selects from.

Source code in src/signalflow/data/dataset.py 
53
54
55
def index(self) -> pl.DataFrame:
    """(pair, ts) of every row - the universe a sampler selects from."""
    return self.frame.select(["pair", "ts"])

iter_bars 

iter_bars(columns: list[str] | None = None) -> Iterator[Bar]

Yield one :class:Bar per timestamp in order (the replay backbone).

Source code in src/signalflow/data/dataset.py 
 97
 98
 99
100
101
102
103
def iter_bars(self, columns: list[str] | None = None) -> Iterator[Bar]:
    """Yield one :class:`Bar` per timestamp in order (the replay backbone)."""
    frame = self.frame if columns is None else self.frame.select(["pair", "ts", "close", *columns])
    for ts, sub in frame.sort("ts").group_by("ts", maintain_order=True):
        ts_val = ts[0] if isinstance(ts, tuple) else ts
        prices = dict(zip(sub.get_column("pair"), sub.get_column("close"), strict=True))
        yield Bar(ts=ts_val, frame=sub, prices=prices)

slice_time 

slice_time(start=None, end=None) -> Dataset

Rows with start <= ts < end (either bound optional). For walk-forward windows.

Source code in src/signalflow/data/dataset.py 
75
76
77
78
79
80
81
82
def slice_time(self, start=None, end=None) -> "Dataset":
    """Rows with start <= ts < end (either bound optional). For walk-forward windows."""
    expr = pl.lit(True)
    if start is not None:
        expr = expr & (pl.col("ts") >= start)
    if end is not None:
        expr = expr & (pl.col("ts") < end)
    return replace(self, frame=self.frame.filter(expr))

with_forecasts 

with_forecasts(cols: DataFrame, *, provenance: Provenance = Provenance.FULL) -> Dataset

Join forecast columns (keyed by pair, ts) and record their provenance.

Source code in src/signalflow/data/dataset.py 
70
71
72
73
def with_forecasts(self, cols: pl.DataFrame, *, provenance: Provenance = Provenance.FULL) -> "Dataset":
    """Join forecast columns (keyed by pair, ts) and record their provenance."""
    merged = self.frame.join(cols, on=["pair", "ts"], how="left")
    return replace(self, frame=merged, provenance=provenance)

MemorySource dataclass 

MemorySource(name: str = 'memory', seed: int = 7, drift: float = 1e-05, vol: float = 0.002, start_price: float = 100.0)

Bases: Source

Synthetic OHLCV generator (deterministic).

Source

Bases: Protocol

Fetches market data as canonical OHLCV rows.

fetch 

fetch(pairs: list[str], start: str, end: str | None = None, interval: str = '1m') -> pl.DataFrame

Return a frame with at least :data:CANONICAL_COLUMNS, sorted by (pair, ts).

Source code in src/signalflow/data/source/base.py 
17
18
19
20
21
22
23
24
25
def fetch(
    self,
    pairs: list[str],
    start: str,
    end: str | None = None,
    interval: str = "1m",
) -> pl.DataFrame:
    """Return a frame with at least :data:`CANONICAL_COLUMNS`, sorted by (pair, ts)."""
    ...

data 

data(source: str | Source, pairs: list[str], start: str, end: str | None = None, interval: str = '1m', quote: str = 'USDT', cache_dir: str | Path | None = None, **source_kwargs) -> Dataset

Build a Dataset from a registered source name or a Source instance.

When cache_dir is set the resolved source is wrapped in a disk cache that fetches only spans absent from <cache_dir>/<interval>/<PAIR>.parquet.

Source code in src/signalflow/data/dataset.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def data(
    source: str | Source,
    pairs: list[str],
    start: str,
    end: str | None = None,
    interval: str = "1m",
    quote: str = "USDT",
    cache_dir: "str | Path | None" = None,
    **source_kwargs,
) -> Dataset:
    """Build a Dataset from a registered source name or a Source instance.

    When ``cache_dir`` is set the resolved source is wrapped in a disk cache that
    fetches only spans absent from ``<cache_dir>/<interval>/<PAIR>.parquet``.
    """
    from signalflow.enums import ComponentType
    from signalflow.registry import registry

    src = registry.create(ComponentType.SOURCE, source, **source_kwargs) if isinstance(source, str) else source
    if cache_dir is not None:
        from signalflow.data.source.cached import CachedSource

        src = CachedSource(src, cache_dir)
    return Dataset.from_source(src, pairs=pairs, start=start, end=end, interval=interval, quote=quote)

Sources

signalflow.BinanceSource dataclass 

BinanceSource(name: str = 'binance', base_url: str = _BASE, timeout: float = 20.0, max_retries: int = 3)

Bases: Source

Fetch spot OHLCV klines from Binance public REST.

signalflow.MemorySource dataclass 

MemorySource(name: str = 'memory', seed: int = 7, drift: float = 1e-05, vol: float = 0.002, start_price: float = 100.0)

Bases: Source

Synthetic OHLCV generator (deterministic).