Advanced Multi-Detector Strategies¶

This tutorial demonstrates how to build advanced backtesting strategies using SignalFlow's named component system and multi-detector aggregation. We will combine multiple detectors, explore different aggregation modes, and show how the same configuration can be expressed in both Python and YAML.

What you'll learn:

Configure multiple detectors with the name= parameter
Use aggregation modes (merge, any, majority, weighted, unanimous)
Validate configurations before running
Express the same strategy as a YAML configuration file

1. Setup¶

In [1]:

Copied!





from datetime import datetime
from pathlib import Path

import signalflow as sf
from signalflow.data import RawDataFactory
from signalflow.data.raw_store import DuckDbSpotStore
from signalflow.data.source import VirtualDataProvider

# Generate synthetic OHLCV data: 10,000 bars for 3 pairs
db_path = Path("/tmp/advanced_strategies.duckdb")
store = DuckDbSpotStore(db_path=db_path)
VirtualDataProvider(store=store, seed=42).download(
    pairs=["BTCUSDT", "ETHUSDT", "SOLUSDT"],
    n_bars=10_000,
)

raw_data = RawDataFactory.from_duckdb_spot_store(
    spot_store_path=db_path,
    pairs=["BTCUSDT", "ETHUSDT", "SOLUSDT"],
    start=datetime(2020, 1, 1),
    end=datetime(2030, 1, 1),
)
print(f"Loaded {len(raw_data.pairs)} pairs")
print(f"Spot data shape: {raw_data.get('spot').shape}")
from datetime import datetime
from pathlib import Path

import signalflow as sf
from signalflow.data import RawDataFactory
from signalflow.data.raw_store import DuckDbSpotStore
from signalflow.data.source import VirtualDataProvider

# Generate synthetic OHLCV data: 10,000 bars for 3 pairs
db_path = Path("/tmp/advanced_strategies.duckdb")
store = DuckDbSpotStore(db_path=db_path)
VirtualDataProvider(store=store, seed=42).download(
    pairs=["BTCUSDT", "ETHUSDT", "SOLUSDT"],
    n_bars=10_000,
)

raw_data = RawDataFactory.from_duckdb_spot_store(
    spot_store_path=db_path,
    pairs=["BTCUSDT", "ETHUSDT", "SOLUSDT"],
    start=datetime(2020, 1, 1),
    end=datetime(2030, 1, 1),
)
print(f"Loaded {len(raw_data.pairs)} pairs")
print(f"Spot data shape: {raw_data.get('spot').shape}")

2026-02-15 00:51:08.525 | INFO     | signalflow.data.raw_store.duckdb_stores:_ensure_tables:153 - Database initialized: /tmp/advanced_strategies.duckdb (data_type=spot, timeframe=1m)
2026-02-15 00:51:08.618 | DEBUG    | signalflow.data.raw_store.duckdb_stores:insert_klines:220 - Inserted 10,000 rows for BTCUSDT
2026-02-15 00:51:08.619 | INFO     | signalflow.data.source.virtual:download:255 - VirtualDataProvider: generated 10000 bars for BTCUSDT
2026-02-15 00:51:08.691 | DEBUG    | signalflow.data.raw_store.duckdb_stores:insert_klines:220 - Inserted 10,000 rows for ETHUSDT
2026-02-15 00:51:08.692 | INFO     | signalflow.data.source.virtual:download:255 - VirtualDataProvider: generated 10000 bars for ETHUSDT
2026-02-15 00:51:08.755 | DEBUG    | signalflow.data.raw_store.duckdb_stores:insert_klines:220 - Inserted 10,000 rows for SOLUSDT
2026-02-15 00:51:08.756 | INFO     | signalflow.data.source.virtual:download:255 - VirtualDataProvider: generated 10000 bars for SOLUSDT
2026-02-15 00:51:08.764 | INFO     | signalflow.data.raw_store.duckdb_stores:_ensure_tables:153 - Database initialized: /tmp/advanced_strategies.duckdb (data_type=spot, timeframe=1m)

Loaded 3 pairs
Spot data shape: (30000, 8)

2. Single Detector Baseline¶

Before combining detectors, let's establish a baseline with a single SMA crossover detector.

In [2]:

Copied!





baseline = (
    sf.Backtest("baseline")
    .data(raw=raw_data)
    .detector("example/sma_cross", fast_period=20, slow_period=50)
    .exit(tp=0.03, sl=0.015)
    .capital(50_000)
    .run()
)
print(baseline.summary())
baseline = (
    sf.Backtest("baseline")
    .data(raw=raw_data)
    .detector("example/sma_cross", fast_period=20, slow_period=50)
    .exit(tp=0.03, sl=0.015)
    .capital(50_000)
    .run()
)
print(baseline.summary())

2026-02-15 00:51:08.805 | DEBUG    | signalflow.core.registry:_discover_internal_packages:152 - autodiscover: failed to import signalflow.detector.adapter
Backtesting: 100%|██████████| 10000/10000 [00:00<00:00, 22379.10it/s]

==================================================
           BACKTEST SUMMARY
==================================================
  Trades:                 636
  Win Rate:              0.0%
  Profit Factor:         0.00
--------------------------------------------------
  Initial Capital: $   50,000.00
  Final Capital:   $        0.00
  Total Return:       -100.0%
--------------------------------------------------
==================================================

/home/alastor/sf-project/sf/src/signalflow/api/result.py:398: UserWarning: Using default source 'default' for 'spot'. Specify explicitly: raw.spot.default
  spot = accessor.to_polars()

3. Named Multi-Detector Configuration¶

The name= parameter lets you attach a label to each detector instance. This is essential when you add several detectors of the same type with different parameters -- each name must be unique within the strategy.

Below we create two SMA crossover detectors:

fast_sma -- a short-term cross (5/15) that reacts quickly
slow_sma -- a longer-term cross (20/50) that filters noise

In [3]:

Copied!





result = (
    sf.Backtest("multi_detector")
    .data(raw=raw_data)
    .detector("example/sma_cross", fast_period=5, slow_period=15, name="fast_sma")
    .detector("example/sma_cross", fast_period=20, slow_period=50, name="slow_sma")
    .aggregation(mode="any")
    .exit(tp=0.03, sl=0.015)
    .capital(50_000)
    .run()
)
print(result.summary())
result = (
    sf.Backtest("multi_detector")
    .data(raw=raw_data)
    .detector("example/sma_cross", fast_period=5, slow_period=15, name="fast_sma")
    .detector("example/sma_cross", fast_period=20, slow_period=50, name="slow_sma")
    .aggregation(mode="any")
    .exit(tp=0.03, sl=0.015)
    .capital(50_000)
    .run()
)
print(result.summary())

Backtesting: 100%|██████████| 10000/10000 [00:03<00:00, 3135.48it/s]

==================================================
           BACKTEST SUMMARY
==================================================
  Trades:                2812
  Win Rate:              0.0%
  Profit Factor:         0.00
--------------------------------------------------
  Initial Capital: $   50,000.00
  Final Capital:   $        0.00
  Total Return:       -100.0%
--------------------------------------------------
==================================================

4. Aggregation Modes¶

When multiple detectors are present, the aggregation mode controls how their signals are combined into a single trading decision:

Mode	Description
`merge`	Combine all signals as a union -- every signal from every detector is kept
`any`	Fire a signal if any detector produces one (logical OR)
`majority`	Fire a signal if the majority of detectors agree
`weighted`	Weighted combination of detector signals with explicit `weights`
`unanimous`	Fire a signal only if all detectors agree (logical AND)

Let's compare them side by side.

In [4]:

Copied!





for mode in ["merge", "any", "majority", "unanimous"]:
    try:
        r = (
            sf.Backtest(f"agg_{mode}")
            .data(raw=raw_data)
            .detector("example/sma_cross", fast_period=5, slow_period=15, name="fast")
            .detector("example/sma_cross", fast_period=20, slow_period=50, name="slow")
            .aggregation(mode=mode)
            .exit(tp=0.03, sl=0.015)
            .capital(50_000)
            .run()
        )
        print(f"{mode:>12s}: trades={r.n_trades:>4}, return={r.total_return:>+7.2%}, win_rate={r.win_rate:.1%}")
    except Exception as e:
        print(f"{mode:>12s}: {e}")
for mode in ["merge", "any", "majority", "unanimous"]:
    try:
        r = (
            sf.Backtest(f"agg_{mode}")
            .data(raw=raw_data)
            .detector("example/sma_cross", fast_period=5, slow_period=15, name="fast")
            .detector("example/sma_cross", fast_period=20, slow_period=50, name="slow")
            .aggregation(mode=mode)
            .exit(tp=0.03, sl=0.015)
            .capital(50_000)
            .run()
        )
        print(f"{mode:>12s}: trades={r.n_trades:>4}, return={r.total_return:>+7.2%}, win_rate={r.win_rate:.1%}")
    except Exception as e:
        print(f"{mode:>12s}: {e}")

       merge: unable to find column "probability"; valid columns: ["pair", "timestamp", "signal_type", "signal", "_source_detector", "_src"]

Backtesting: 100%|██████████| 10000/10000 [00:03<00:00, 2854.86it/s]

         any: trades=2812, return=-100.00%, win_rate=0.0%

Backtesting: 100%|██████████| 10000/10000 [00:03<00:00, 2752.39it/s]

    majority: trades=2814, return=-100.00%, win_rate=0.0%

Backtesting: 100%|██████████| 10000/10000 [00:00<00:00, 121654.78it/s]

   unanimous: trades=  18, return=-100.00%, win_rate=0.0%

5. Weighted Aggregation¶

The weighted mode lets you assign explicit importance to each detector. Weights are applied in the order the detectors were added. Here we give 30% weight to the fast detector and 70% to the slow one.

In [5]:

Copied!





weighted_result = (
    sf.Backtest("weighted")
    .data(raw=raw_data)
    .detector("example/sma_cross", fast_period=5, slow_period=15, name="fast")
    .detector("example/sma_cross", fast_period=20, slow_period=50, name="slow")
    .aggregation(mode="weighted", weights=[0.3, 0.7])
    .exit(tp=0.03, sl=0.015)
    .capital(50_000)
    .run()
)
print(weighted_result.summary())
weighted_result = (
    sf.Backtest("weighted")
    .data(raw=raw_data)
    .detector("example/sma_cross", fast_period=5, slow_period=15, name="fast")
    .detector("example/sma_cross", fast_period=20, slow_period=50, name="slow")
    .aggregation(mode="weighted", weights=[0.3, 0.7])
    .exit(tp=0.03, sl=0.015)
    .capital(50_000)
    .run()
)
print(weighted_result.summary())

Backtesting: 100%|██████████| 10000/10000 [00:03<00:00, 2798.02it/s]

==================================================
           BACKTEST SUMMARY
==================================================
  Trades:                2814
  Win Rate:              0.0%
  Profit Factor:         0.00
--------------------------------------------------
  Initial Capital: $   50,000.00
  Final Capital:   $        0.00
  Total Return:       -100.0%
--------------------------------------------------
==================================================

6. Configuration Validation¶

The builder exposes a .validate() method that checks the configuration for common mistakes (missing data, incompatible parameters, duplicate names) without actually running the backtest.

In [6]:

Copied!





builder = (
    sf.Backtest("validate_test")
    .data(raw=raw_data)
    .detector("example/sma_cross", fast_period=20, slow_period=50)
    .exit(tp=0.03, sl=0.015)
    .capital(50_000)
)

# Validate without running
errors = builder.validate()
if errors:
    print("Validation errors:")
    for e in errors:
        print(f"  - {e}")
else:
    print("Configuration is valid!")
builder = (
    sf.Backtest("validate_test")
    .data(raw=raw_data)
    .detector("example/sma_cross", fast_period=20, slow_period=50)
    .exit(tp=0.03, sl=0.015)
    .capital(50_000)
)

# Validate without running
errors = builder.validate()
if errors:
    print("Validation errors:")
    for e in errors:
        print(f"  - {e}")
else:
    print("Configuration is valid!")

Configuration is valid!

7. YAML Configuration Equivalent¶

The same multi-detector strategy can be defined declaratively in a YAML file and executed via the SignalFlow CLI. This is useful for reproducibility, version control, and parameter sweeps.

strategy:
  name: multi_detector

data:
  source: data/binance.duckdb
  pairs: [BTCUSDT, ETHUSDT, SOLUSDT]
  start: "2024-01-01"
  timeframe: 1m

detectors:
  fast_sma:
    name: example/sma_cross
    params:
      fast_period: 5
      slow_period: 15
  slow_sma:
    name: example/sma_cross
    params:
      fast_period: 20
      slow_period: 50

aggregation:
  mode: weighted
  weights: [0.3, 0.7]

exit:
  tp: 0.03
  sl: 0.015

capital: 50000

Run with the CLI:

sf run strategy.yaml --plot

8. Comparing Strategies¶

Finally, let's put everything together and compare a single-detector baseline against several multi-detector configurations.

In [7]:

Copied!





# Compare single vs multi-detector
strategies = {
    "Single (SMA 20/50)": {
        "detectors": [("example/sma_cross", {"fast_period": 20, "slow_period": 50})],
        "aggregation": None,
    },
    "Fast + Slow (any)": {
        "detectors": [
            ("example/sma_cross", {"fast_period": 5, "slow_period": 15}),
            ("example/sma_cross", {"fast_period": 20, "slow_period": 50}),
        ],
        "aggregation": "any",
    },
    "Fast + Slow (majority)": {
        "detectors": [
            ("example/sma_cross", {"fast_period": 5, "slow_period": 15}),
            ("example/sma_cross", {"fast_period": 20, "slow_period": 50}),
        ],
        "aggregation": "majority",
    },
}

print(f"{'Strategy':>30s} | {'Trades':>6s} | {'Return':>8s} | {'Win Rate':>8s}")
print("-" * 65)

for name, config in strategies.items():
    b = sf.Backtest(name).data(raw=raw_data)
    for i, (det_name, params) in enumerate(config["detectors"]):
        b.detector(det_name, name=f"d{i}", **params)
    if config["aggregation"]:
        b.aggregation(mode=config["aggregation"])
    b.exit(tp=0.03, sl=0.015).capital(50_000)

    try:
        r = b.run()
        print(f"{name:>30s} | {r.n_trades:>6} | {r.total_return:>+7.2%} | {r.win_rate:>7.1%}")
    except Exception as e:
        print(f"{name:>30s} | Error: {e}")
# Compare single vs multi-detector
strategies = {
    "Single (SMA 20/50)": {
        "detectors": [("example/sma_cross", {"fast_period": 20, "slow_period": 50})],
        "aggregation": None,
    },
    "Fast + Slow (any)": {
        "detectors": [
            ("example/sma_cross", {"fast_period": 5, "slow_period": 15}),
            ("example/sma_cross", {"fast_period": 20, "slow_period": 50}),
        ],
        "aggregation": "any",
    },
    "Fast + Slow (majority)": {
        "detectors": [
            ("example/sma_cross", {"fast_period": 5, "slow_period": 15}),
            ("example/sma_cross", {"fast_period": 20, "slow_period": 50}),
        ],
        "aggregation": "majority",
    },
}

print(f"{'Strategy':>30s} | {'Trades':>6s} | {'Return':>8s} | {'Win Rate':>8s}")
print("-" * 65)

for name, config in strategies.items():
    b = sf.Backtest(name).data(raw=raw_data)
    for i, (det_name, params) in enumerate(config["detectors"]):
        b.detector(det_name, name=f"d{i}", **params)
    if config["aggregation"]:
        b.aggregation(mode=config["aggregation"])
    b.exit(tp=0.03, sl=0.015).capital(50_000)

    try:
        r = b.run()
        print(f"{name:>30s} | {r.n_trades:>6} | {r.total_return:>+7.2%} | {r.win_rate:>7.1%}")
    except Exception as e:
        print(f"{name:>30s} | Error: {e}")

                      Strategy | Trades |   Return | Win Rate
-----------------------------------------------------------------

Backtesting: 100%|██████████| 10000/10000 [00:00<00:00, 22352.67it/s]

            Single (SMA 20/50) |    636 | -100.00% |    0.0%

Backtesting: 100%|██████████| 10000/10000 [00:03<00:00, 2815.27it/s]

             Fast + Slow (any) |   2812 | -100.00% |    0.0%

Backtesting: 100%|██████████| 10000/10000 [00:04<00:00, 2176.14it/s]

        Fast + Slow (majority) |   2816 | -100.00% |    0.0%

9. Clean Up¶

In [8]:

Copied!

store.close()
db_path.unlink(missing_ok=True)
print("Done!")
store.close()
db_path.unlink(missing_ok=True)
print("Done!")

Done!

Key Takeaways¶

Use the name= parameter to create named component instances when adding multiple detectors of the same type.
Aggregation modes control how multiple detector signals are combined: merge, any, majority, weighted, and unanimous.
Call .validate() to catch configuration errors before running a backtest.
The same configuration works in Python code and YAML files, so you can develop interactively and deploy declaratively.

Next Steps¶

CLI Guide: Full CLI reference
Signal Architecture: Deep dive into signal processing
Advanced Strategies Guide: Position sizing and entry filters