Strategy

A strategy decides what to do with signals. RulesStrategy composes entry, exit, and risk rules; StrategyModel is the base contract for custom strategies, and Observation is the input passed to each decision.

Rules strategy

signalflow.RulesStrategy dataclass

RulesStrategy(entry: Entry = Entry(), exit: Exit = Exit())

Bases: Strategy

Turn validated signals into open/close intents via fixed rules.

Rules

signalflow.Entry dataclass

Entry(size_pct: float = 0.1, min_p_success: float = 0.0, max_positions: int = 1000, max_positions_per_pair: int | None = None)

Position entry policy.

When a validator is present (the signal frame carries a p_success column) and min_p_success is positive, a null p_success counts as not validated and the entry is skipped. Without a validator column, entries open unchanged.

max_positions caps concurrent open positions global across pairs. max_positions_per_pair (when set) caps them per pair; None disables it.

signalflow.Exit dataclass

Exit(tp: float = 0.03, sl: float = 0.015)

Take-profit / stop-loss exit policy.

signalflow.Risk dataclass

Risk(max_drawdown: float = 1.0, max_positions: int = 1000, max_notional_per_pair: float = 1.0, kill_switch_path: str | None = None)

Clip intents against drawdown, position, and notional limits.

max_positions caps concurrent open positions global across pairs.

clip

clip(intents: list[Intent], portfolio: PortfolioSnapshot, peak_equity: float, raise_on_trip: bool = False) -> list[Intent]

Clip intents; with raise_on_trip a tripped kill switch halts loudly instead of dropping.

Source code in src/signalflow/strategy/risk.py

def clip(
    self,
    intents: list[Intent],
    portfolio: PortfolioSnapshot,
    peak_equity: float,
    raise_on_trip: bool = False,
) -> list[Intent]:
    """Clip intents; with ``raise_on_trip`` a tripped kill switch halts loudly instead of dropping."""
    eq = portfolio.equity
    if peak_equity > 0 and (peak_equity - eq) / peak_equity >= self.max_drawdown:
        self.trip(f"drawdown {(peak_equity - eq) / peak_equity:.3f} >= {self.max_drawdown}")

    if self._tripped and raise_on_trip:
        raise KillSwitchTripped(f"kill switch engaged; refusing to send orders (path={self.kill_switch_path!r})")

    out: list[Intent] = []
    n_pos = len(portfolio.positions)
    for it in intents:
        if it.kind == IntentKind.CLOSE:
            out.append(it)
            n_pos = max(0, n_pos - 1)
            continue
        if self._tripped:
            continue
        if n_pos >= self.max_positions:
            continue
        if it.notional is not None:
            it.notional = min(it.notional, self.max_notional_per_pair * eq)
        out.append(it)
        n_pos += 1
    return out

Strategy contract

signalflow.StrategyModel

Bases: Protocol

signalflow.Observation dataclass

Observation(ts: object, signals: DataFrame, portfolio: PortfolioSnapshot, mandate: dict = dict(), schema_version: int = OBSERVATION_SCHEMA_VERSION)

Validated signals + portfolio + market state + constraints at one ts.

from_dict classmethod

from_dict(payload: dict) -> Observation

Rebuild an Observation, rejecting a schema version it was not trained on.

Source code in src/signalflow/strategy/observation.py

@classmethod
def from_dict(cls, payload: dict) -> "Observation":
    """Rebuild an Observation, rejecting a schema version it was not trained on."""
    version = payload.get("schema_version")
    if version != OBSERVATION_SCHEMA_VERSION:
        raise SchemaVersionError(
            f"Observation schema_version {version!r} does not match runtime {OBSERVATION_SCHEMA_VERSION!r}"
        )
    pf = payload.get("portfolio") or {}
    portfolio = PortfolioSnapshot(
        ts=pf.get("ts"),
        target=pf.get("target", ""),
        balances=dict(pf.get("balances", {})),
        positions={
            p: Position(pair=v["pair"], qty=v["qty"], avg_price=v["avg_price"])
            for p, v in (pf.get("positions") or {}).items()
        },
        equity=pf.get("equity", 0.0),
        prices=dict(pf.get("prices", {})),
    )
    return cls(
        ts=payload.get("ts"),
        signals=pl.DataFrame(payload.get("signals") or []),
        portfolio=portfolio,
        mandate=payload.get("mandate") or {},
        schema_version=version,
    )

to_dict

to_dict() -> dict

Serialize to a JSON-able dict tagged with the schema version.

Source code in src/signalflow/strategy/observation.py

def to_dict(self) -> dict:
    """Serialize to a JSON-able dict tagged with the schema version."""
    port = self.portfolio
    return {
        "schema_version": self.schema_version,
        "ts": str(self.ts),
        "mandate": self.mandate,
        "signals": self.signals.to_dicts(),
        "portfolio": {
            "ts": str(port.ts),
            "target": port.target,
            "balances": dict(port.balances),
            "equity": port.equity,
            "prices": dict(port.prices),
            "positions": {
                p: {"pair": pos.pair, "qty": pos.qty, "avg_price": pos.avg_price, "opened_ts": str(pos.opened_ts)}
                for p, pos in port.positions.items()
            },
        },
    }

to_prompt_context

to_prompt_context() -> dict

JSON-able structured context for an LLM strategy (no raw candles).

Source code in src/signalflow/strategy/observation.py

def to_prompt_context(self) -> dict:
    """JSON-able structured context for an LLM strategy (no raw candles)."""
    cols = [c for c in ("pair", "signal", "p_success") if c in self.signals.columns]
    return {
        "ts": str(self.ts),
        "mandate": self.mandate,
        "equity": self.portfolio.equity,
        "positions": {
            p: {"qty": pos.qty, "avg_price": pos.avg_price} for p, pos in self.portfolio.positions.items()
        },
        "signals": self.signals.select(cols).to_dicts() if cols else [],
    }

to_vector

to_vector() -> np.ndarray

Stable fixed-length numeric summary for RL policies.

Source code in src/signalflow/strategy/observation.py

def to_vector(self) -> np.ndarray:
    """Stable fixed-length numeric summary for RL policies."""
    sig = self.signals
    n = max(sig.height, 1)
    rise = (sig.get_column("signal") == "rise").sum() / n if "signal" in sig.columns else 0.0
    p_succ = float(sig.get_column("p_success").fill_null(0.0).mean() or 0.0) if "p_success" in sig.columns else 0.0
    eq = self.portfolio.equity
    cash = self.portfolio.balances.get(self.portfolio.target, 0.0)
    return np.array(
        [
            float(eq),
            float(cash / eq) if eq else 0.0,
            float(len(self.portfolio.positions)),
            float(rise),
            float(p_succ),
        ],
        dtype=np.float64,
    )