Skip to content

Experiment

Experiments run flows under controlled conditions and summarize their results in a Scorecard. The artifact cache reuses expensive intermediate results, and the statistics helpers quantify uncertainty in performance metrics.

Experiment

signalflow.Experiment 

Experiment(name: str, baseline=None, *, clock: Callable[[], object] | None = None)

Lifecycle wrapper: run a Flow, store the Run, score it against a baseline.

Source code in src/signalflow/experiment/experiment.py 
23
24
25
26
27
28
29
30
31
def __init__(self, name: str, baseline=None, *, clock: Callable[[], object] | None = None) -> None:
    self.name = name
    self.baseline = baseline
    self._clock = clock
    self.last_run = None
    self._baseline_run = None
    self._last_run_args: dict | None = None

    self.log: dict = {"created": None, "first_result": None, "stages": []}

run 

run(flow, data, capital, target=None, *, ts=None)

Backtest flow on data, store the Run, and stamp first_result.

Source code in src/signalflow/experiment/experiment.py 
43
44
45
46
47
48
49
50
51
def run(self, flow, data, capital, target=None, *, ts=None):
    """Backtest ``flow`` on ``data``, store the Run, and stamp ``first_result``."""
    if self.log["created"] is None:
        self.stamp("created", ts=ts)
    run = flow.backtest(data, capital, target=target)
    self.last_run = run
    self._last_run_args = {"data": data, "capital": capital, "target": target}
    self.stamp("first_result", ts=ts)
    return run

stamp 

stamp(stage: str, ts=None)

Record stage with timestamp ts (or the clock's value).

Source code in src/signalflow/experiment/experiment.py 
34
35
36
37
38
39
40
41
def stamp(self, stage: str, ts=None):
    """Record ``stage`` with timestamp ``ts`` (or the clock's value)."""
    if ts is None and self._clock is not None:
        ts = self._clock()
    self.log["stages"].append({"stage": stage, "ts": ts})
    if stage in self.log and self.log[stage] is None:
        self.log[stage] = ts
    return ts

signalflow.Scorecard

Namespace for scorecard construction; the product is a plain dict.

from_run staticmethod 

from_run(run, baseline=None, *, n: int = 1000, alpha: float = 0.05, seed: int = 0) -> dict

Summarize run into a stable-shape dict, optionally vs baseline.

Source code in src/signalflow/experiment/scorecard.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@staticmethod
def from_run(run, baseline=None, *, n: int = 1000, alpha: float = 0.05, seed: int = 0) -> dict:
    """Summarize ``run`` into a stable-shape dict, optionally vs ``baseline``."""
    returns = run.returns
    lo, hi = bootstrap_ci(returns, n=n, alpha=alpha, seed=seed)
    mc = monte_carlo_bounds(returns, n=n, seed=seed)

    card: dict = {
        "name": run.name,
        "mode": run.mode,
        "total_return": float(run.total_return),
        "sharpe": float(run.sharpe()),
        "max_drawdown": float(run.max_drawdown),
        "n_fills": len(run.fills),
        "bootstrap_ci": {"low": lo, "high": hi, "alpha": alpha},
        "monte_carlo": mc,
        "delta": None,
    }

    if baseline is not None:
        card["delta"] = {
            "total_return": float(run.total_return - baseline.total_return),
            "sharpe": float(run.sharpe() - baseline.sharpe()),
            "max_drawdown": float(run.max_drawdown - baseline.max_drawdown),
            "n_fills": len(run.fills) - len(baseline.fills),
        }
    return card

Artifact cache

signalflow.ArtifactCache 

ArtifactCache(root: str)

Disk cache of polars frames addressed by a content + code-fingerprint key.

Source code in src/signalflow/experiment/cache.py 
30
31
32
def __init__(self, root: str) -> None:
    self.root = Path(root)
    self.root.mkdir(parents=True, exist_ok=True)

compute_cached 

compute_cached(key: str, fn: Callable[[], DataFrame]) -> pl.DataFrame

Return the cached frame for key or compute it via fn and store it.

Source code in src/signalflow/experiment/cache.py 
66
67
68
69
70
71
72
73
def compute_cached(self, key: str, fn: Callable[[], pl.DataFrame]) -> pl.DataFrame:
    """Return the cached frame for ``key`` or compute it via ``fn`` and store it."""
    cached = self.get(key)
    if cached is not None:
        return cached
    df = fn()
    self.put(key, df)
    return df

key 

key(parts: dict, *, producer=None, kind: str = 'inference') -> str

Compute the content-addressed key for parts.

Source code in src/signalflow/experiment/cache.py 
34
35
36
37
38
39
40
41
42
43
def key(self, parts: dict, *, producer=None, kind: str = "inference") -> str:
    """Compute the content-addressed key for ``parts``."""
    if kind not in VALID_KINDS:
        raise ArtifactError(f"unknown artifact kind {kind!r}; expected one of {VALID_KINDS}")
    material = {
        "kind": kind,
        "parts": parts,
        "code": code_fingerprint(producer) if producer is not None else None,
    }
    return stable_hash(material)

Statistics

signalflow.bootstrap_ci 

bootstrap_ci(returns, n: int = 1000, alpha: float = 0.05, seed: int = 0) -> tuple[float, float]

Percentile bootstrap CI for the mean return.

Source code in src/signalflow/experiment/stats.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
def bootstrap_ci(returns, n: int = 1000, alpha: float = 0.05, seed: int = 0) -> tuple[float, float]:
    """Percentile bootstrap CI for the *mean* return."""
    arr = _as_array(returns)
    if arr.size == 0:
        return (0.0, 0.0)
    if arr.size == 1:
        return (float(arr[0]), float(arr[0]))
    rng = np.random.default_rng(seed)
    idx = rng.integers(0, arr.size, size=(n, arr.size))
    means = arr[idx].mean(axis=1)
    lo = float(np.percentile(means, 100.0 * (alpha / 2.0)))
    hi = float(np.percentile(means, 100.0 * (1.0 - alpha / 2.0)))
    return (lo, hi)

signalflow.monte_carlo_bounds 

monte_carlo_bounds(returns, n: int = 1000, horizon: int | None = None, seed: int = 0) -> dict

Resample per-period returns into n synthetic equity paths and report the terminal-equity distribution.

Each path is a sequence of horizon returns drawn with replacement from the observed returns (horizon defaults to the number of observed returns); terminal equity is the cumulative product starting from 1.0. Returns 5th/50th/95th percentiles plus mean/min/max. Deterministic for a fixed seed.

Source code in src/signalflow/experiment/stats.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def monte_carlo_bounds(returns, n: int = 1000, horizon: int | None = None, seed: int = 0) -> dict:
    """
    Resample per-period returns into ``n`` synthetic equity paths and report
    the terminal-equity distribution.

    Each path is a sequence of ``horizon`` returns drawn with replacement from
    the observed returns (``horizon`` defaults to the number of observed
    returns); terminal equity is the cumulative product starting from 1.0.
    Returns 5th/50th/95th percentiles plus mean/min/max. Deterministic for a
    fixed ``seed``.
    """
    arr = _as_array(returns)
    h = int(horizon) if horizon is not None else arr.size
    if arr.size == 0 or h <= 0:
        return {
            "p5": 1.0, "p50": 1.0, "p95": 1.0,
            "mean": 1.0, "min": 1.0, "max": 1.0,
            "horizon": h, "n_paths": int(n),
        }
    rng = np.random.default_rng(seed)
    idx = rng.integers(0, arr.size, size=(n, h))
    paths = arr[idx]
    terminal = np.prod(1.0 + paths, axis=1)
    return {
        "p5": float(np.percentile(terminal, 5)),
        "p50": float(np.percentile(terminal, 50)),
        "p95": float(np.percentile(terminal, 95)),
        "mean": float(terminal.mean()),
        "min": float(terminal.min()),
        "max": float(terminal.max()),
        "horizon": h,
        "n_paths": int(n),
    }