Flow & Live

Flow wires data, detectors, strategy, and engine into a single runnable pipeline. A Run captures the result of executing a flow. Feeds drive the flow over historical, replayed, or live data.

Flow

signalflow.Flow dataclass

Flow(name: str, forecasts: dict = dict(), detectors: list = list(), strategy: object = RulesStrategy(), risk: Risk = Risk(), validator: object | None = None, quote: str = 'USDT')

The framework's central noun: what you research, promote, and trade.

required_warmup property

required_warmup: int

Bars of history the flow needs before its outputs are valid.

Max over each detector's warmup, each forecast model's feature-pipe warmup, and the validator's feature warmup. Zero when the flow has none of these.

backtest

backtest(data, capital, target: str | None = None, broker=None, oos: bool = False)

Backtest the flow. oos=True scores leak-free out-of-fold predictions and stamps the Run promotable; the default in-sample run of a model flow is not promotable.

Source code in src/signalflow/flow/flow.py

def backtest(self, data, capital, target: str | None = None, broker=None, oos: bool = False):
    """Backtest the flow. ``oos=True`` scores leak-free out-of-fold predictions and
    stamps the Run promotable; the default in-sample run of a model flow is not promotable.
    """
    broker = broker or self._sim_broker()
    return run_event_loop(self, data, capital, target, broker, RunMode.BACKTEST, oos=oos)

live

live(feed, capital, target: str | None = None, broker=None, armed: bool = False, max_bars: int | None = None, state_path: str | None = None)

Trade a live (or replayed) feed via the real-time loop.

feed may be a LiveFeed or a Dataset (wrapped in a ReplayFeed). Armed trading requires an explicit ExchangeBroker; SimBroker is paper-only.

Source code in src/signalflow/flow/flow.py

def live(
    self,
    feed,
    capital,
    target: str | None = None,
    broker=None,
    armed: bool = False,
    max_bars: int | None = None,
    state_path: str | None = None,
):
    """Trade a live (or replayed) feed via the real-time loop.

    ``feed`` may be a LiveFeed or a Dataset (wrapped in a ReplayFeed). Armed
    trading requires an explicit ExchangeBroker; SimBroker is paper-only.
    """
    if armed and broker is None:
        raise FlowConfigError(
            "armed live requires an explicit ExchangeBroker; refusing to send real orders via SimBroker"
        )
    broker = broker or self._sim_broker()
    if not hasattr(feed, "stream"):
        feed = ReplayFeed(feed)
    return run_live_loop(self, feed, capital, broker, target=target, max_bars=max_bars, state_path=state_path)

paper

paper(data, capital, target: str | None = None, broker=None)

Replay a Dataset with simulated fills - the same loop as backtest, paper mode.

Source code in src/signalflow/flow/flow.py

def paper(self, data, capital, target: str | None = None, broker=None):
    """Replay a Dataset with simulated fills - the same loop as backtest, paper mode."""
    broker = broker or self._sim_broker()
    return run_event_loop(self, data, capital, target, broker, RunMode.PAPER)

save

save(path: str, model_dir: str | None = None) -> str

Serialize the flow to YAML at path and return it.

Each forecast/validator must already have a pinned URI, or pass model_dir to save the trained artifacts there. load restores a byte-identical backtest.

Source code in src/signalflow/flow/flow.py

def save(self, path: str, model_dir: str | None = None) -> str:
    """Serialize the flow to YAML at ``path`` and return it.

    Each forecast/validator must already have a pinned URI, or pass ``model_dir`` to
    save the trained artifacts there. ``load`` restores a byte-identical backtest.
    """
    from signalflow.flow.yaml import save_flow

    return save_flow(self, path, model_dir=model_dir)

simulate

simulate(data, capital, target: str | None = None, broker=None, warmup: int | None = None, maxlen: int = 5000, state_path: str | None = None)

Full-speed incremental live simulation (walk-forward).

Replays a Dataset through the live decision loop with no real-time wait: the flow sees only data up to each bar, recomputed step by step, exactly as in live. warmup reserves a leading lookback window that fills the buffer without trading; None resolves to :attr:required_warmup while an explicit 0 is honored. Use it to confirm the live path before arming.

Source code in src/signalflow/flow/flow.py

def simulate(
    self,
    data,
    capital,
    target: str | None = None,
    broker=None,
    warmup: int | None = None,
    maxlen: int = 5000,
    state_path: str | None = None,
):
    """Full-speed incremental live simulation (walk-forward).

    Replays a Dataset through the live decision loop with no real-time wait:
    the flow sees only data up to each bar, recomputed step by step, exactly
    as in live. ``warmup`` reserves a leading lookback window that fills the
    buffer without trading; ``None`` resolves to :attr:`required_warmup` while
    an explicit ``0`` is honored. Use it to confirm the live path before arming.
    """
    broker = broker or self._sim_broker()
    warmup = self.required_warmup if warmup is None else warmup
    feed = ReplayFeed(data, warmup_bars=warmup)
    return run_live_loop(self, feed, capital, broker, target=target, maxlen=maxlen, state_path=state_path)

signalflow.Run dataclass

Run(name: str, mode: str, equity_curve: DataFrame, fills: list[Fill] = list(), target: str = 'USDT', promotable: bool = True, oos: bool = False)

Result of executing a Flow: equity curve, fills, and derived metrics.

promotable is false for an in-sample model backtest; oos flags a leak-free out-of-sample run.

periods_per_year

periods_per_year() -> float

Annualization factor derived from the equity curve's median bar spacing.

Source code in src/signalflow/flow/run.py

def periods_per_year(self) -> float:
    """Annualization factor derived from the equity curve's median bar spacing."""
    seconds_per_year = 365.0 * 24.0 * 3600.0
    ec = self.equity_curve
    if ec.height >= 3 and "ts" in ec.columns:
        secs = ec.get_column("ts").sort().diff().drop_nulls().dt.total_seconds()
        positive = secs.filter(secs > 0)
        if positive.len() > 0:
            median_dt = float(positive.median())
            if median_dt > 0:
                return seconds_per_year / median_dt
    return 8760.0

scorecard

scorecard() -> dict

Standard metric dict: name, mode, target, promotable, oos, n_fills, initial_equity, final_equity, total_return, max_drawdown, and sharpe.

Source code in src/signalflow/flow/run.py

def scorecard(self) -> dict:
    """Standard metric dict: ``name``, ``mode``, ``target``, ``promotable``, ``oos``,
    ``n_fills``, ``initial_equity``, ``final_equity``, ``total_return``,
    ``max_drawdown``, and ``sharpe``.
    """
    return {
        "name": self.name,
        "mode": self.mode,
        "target": self.target,
        "promotable": self.promotable,
        "oos": self.oos,
        "n_fills": len(self.fills),
        "initial_equity": round(self.initial_equity, 2),
        "final_equity": round(self.final_equity, 2),
        "total_return": round(self.total_return, 4),
        "max_drawdown": round(self.max_drawdown, 4),
        "sharpe": round(self.sharpe(), 3),
    }

sharpe

sharpe(periods_per_year: float | None = None) -> float

Annualized Sharpe of per-bar equity returns (zero risk-free rate).

periods_per_year defaults to the factor implied by the equity curve's median bar spacing (:meth:periods_per_year), so hourly bars annualize by ~8760.

Source code in src/signalflow/flow/run.py

def sharpe(self, periods_per_year: float | None = None) -> float:
    """Annualized Sharpe of per-bar equity returns (zero risk-free rate).

    ``periods_per_year`` defaults to the factor implied by the equity curve's median
    bar spacing (:meth:`periods_per_year`), so hourly bars annualize by ~8760.
    """
    r = self.returns
    if r.size < 2 or r.std() == 0:
        return 0.0
    ppy = self.periods_per_year() if periods_per_year is None else periods_per_year
    return float(r.mean() / r.std() * np.sqrt(ppy))

Feeds

signalflow.LiveFeed

Bases: Protocol

Warmup history (preloaded, no trades) + a stream of newly closed bars.

signalflow.ReplayFeed dataclass

ReplayFeed(data: Dataset, interval: str | None = None, warmup_bars: int = 0)

Bases: LiveFeed

Replays a finished Dataset bar-by-bar - drives the live loop at full speed.

warmup_bars splits off a leading lookback window that fills the buffer without trading; decisions then run incrementally over the remaining bars - a faithful walk-forward of the live cycle.

signalflow.PollingFeed dataclass

PollingFeed(source: object, pairs: list[str], interval: str = '1m', quote: str = 'USDT', warmup_bars: int | None = None, max_bars: int | None = None, lag_seconds: float = 3.0, clock: Clock = (lambda: Clock(RunMode.LIVE))())

Bases: LiveFeed

Polls a Source for newly closed klines, one yield per new bar boundary.

warmup() fetches a closed-bar history prefix to fill the buffer. stream() sleeps to each interval boundary, fetches the latest CLOSED bar(s), and never yields the still-forming current candle.

Live loop

signalflow.run_live_loop

run_live_loop(flow, feed: LiveFeed, capital, broker, target: str | None = None, maxlen: int = 5000, max_bars: int | None = None, mandate: dict | None = None, state_path: str | None = None, on_bar=None, max_latency_s: float = 10.0)

Drive a Flow over a live (or replayed) feed.

Warmup history fills the buffer without trading; decisions begin on the first streamed bar. Per live bar, decision latency (execution wall-clock minus the bar's close time) is measured and a breach of max_latency_s is logged. Runs are not promotable.

Source code in src/signalflow/flow/live.py

def run_live_loop(
    flow,
    feed: LiveFeed,
    capital,
    broker,
    target: "str | None" = None,
    maxlen: int = 5000,
    max_bars: "int | None" = None,
    mandate: "dict | None" = None,
    state_path: "str | None" = None,
    on_bar=None,
    max_latency_s: float = 10.0,
):
    """Drive a Flow over a live (or replayed) feed.

    Warmup history fills the buffer without trading; decisions begin on the first
    streamed bar. Per live bar, decision latency (execution wall-clock minus the
    bar's close time) is measured and a breach of ``max_latency_s`` is logged.
    Runs are not promotable.
    """
    from signalflow.engine.broker import ExchangeBroker
    from signalflow.flow.run import Run

    armed = isinstance(broker, ExchangeBroker)
    required_warmup = resolve_warmup_bars(flow, feed)
    logger.info(f"live: resolved warmup = {required_warmup} bars")
    if maxlen < required_warmup:
        raise ValueError(
            f"maxlen={maxlen} is below the flow's required warmup of {required_warmup} bars; "
            f"raise maxlen to at least {required_warmup}"
        )
    coverage = getattr(feed, "warmup_bars", None)
    if coverage is not None and coverage < required_warmup:
        logger.warning(
            f"live: feed warmup coverage {coverage} < required {required_warmup}; "
            f"outputs may be NaN-driven until the buffer grows"
        )

    target = target or feed.quote
    engine = Engine(capital, target=target, quote=feed.quote)
    if state_path and load_state(engine, state_path):
        logger.info(f"live: resumed book from {state_path}")

    buf: list = []
    for wbar in feed.warmup().iter_bars():
        buf.append(wbar.frame)
    step_s = _INTERVAL_S.get(getattr(feed, "interval", None) or "")

    eq_ts, eq_val = [], []
    peak = float("-inf")
    started = False
    for n, bar in enumerate(feed.stream()):
        buf.append(bar.frame)
        if len(buf) > maxlen:
            buf = buf[-maxlen:]
        hist = Dataset(frame=pl.concat(buf), quote=feed.quote)
        signals = enriched_signals(flow, hist)
        sig_frame = (
            signals.filter(pl.col("ts") == bar.ts) if signals.height else pl.DataFrame(schema=_EMPTY_SIGNALS_SCHEMA)
        )
        snap = engine.snapshot(bar.ts, bar.prices)
        if not started:
            eq_ts.append(bar.ts)
            eq_val.append(snap.equity)
            started = True
        peak = max(peak, snap.equity)
        obs = Observation(bar.ts, sig_frame, snap, mandate or {})
        intents = flow.risk.clip(flow.strategy.decide(obs), snap, peak, raise_on_trip=armed)
        fills = broker.execute(_orders(intents, bar.prices, bar.ts), bar)
        engine.apply(fills)
        eq_ts.append(bar.ts)
        eq_val.append(engine.equity(bar.prices))

        latency = (time.time() - _close_epoch(bar.ts, step_s)) if step_s else None
        if fills and latency is not None and latency > max_latency_s:
            logger.warning(f"live: order latency {latency:.1f}s exceeds {max_latency_s}s budget at close {bar.ts}")

        if state_path:
            save_state(engine, state_path)
        if on_bar is not None:
            on_bar(engine, bar, fills, latency)
        if max_bars is not None and n + 1 >= max_bars:
            break

    curve = pl.DataFrame({"ts": eq_ts, "equity": eq_val})
    return Run(flow.name, RunMode.LIVE.value, curve, engine.event_log, target, promotable=False)