Skip to main content
Horizon has full test coverage across both Rust and Python layers.

​
Running Tests

cargo test

​
Test Coverage

LayerTestsDescription
Rust98 testsCore engine, risk pipeline, orders, positions, exchanges, feeds, persistence, contingent orders
Python395 testsFull SDK: engine, pipeline, exchanges, feeds, risk, backtest, algos, metrics, advanced orders

​
Test Structure

tests/
├── test_engine.py           # Engine creation, orders, fills, P&L
├── test_exchange_routing.py # Exchange backend selection, cancel logic
├── test_feeds.py            # Feed snapshot management
├── test_live_exchange.py    # Polymarket/Kalshi construction, EIP-712, env config
├── test_multi_exchange.py   # Multi-exchange engine, netting, routing
├── test_paper.py            # Paper exchange matching logic
├── test_pipeline.py         # Pipeline composition, context building
├── test_production_fixes.py # Production fixes regression tests
├── test_risk.py             # Risk pipeline checks (kill switch, limits, dedup)
├── test_advanced_orders.py  # Amendment, stop-loss, take-profit, bracket, OCO
├── test_algos.py            # TWAP, VWAP, Iceberg execution algorithms
├── test_backtest.py         # Backtesting engine, data normalization, analytics
└── test_metrics.py          # Prometheus metrics, counters, gauges, histograms, server

​
Rust Tests

Rust tests (98 tests) cover the core engine, risk pipeline, order management, position tracking, exchange backends, feed system, persistence layer, contingent orders, and order amendment. Tests are located in #[cfg(test)] mod tests blocks within each source file.

​
Python Tests

Python tests (395 tests) cover the full SDK: engine creation, pipeline composition, exchange configuration, feed management, multi-exchange routing, risk behavior, backtesting, execution algorithms, advanced orders, and Prometheus metrics.

​
Writing Strategy Tests

Use the Engine directly for unit testing your strategy logic: 
from horizon import Engine, RiskConfig, OrderRequest, Side, OrderSide, Quote, Fill

def test_my_strategy_logic():
    config = RiskConfig(max_position_per_market=1000)
    engine = Engine(risk_config=config)

    # Submit an order
    order_id = engine.submit_order(OrderRequest(
        market_id="test",
        side=Side.Yes,
        order_side=OrderSide.Buy,
        size=10.0,
        price=0.55,
    ))

    # Tick to fill
    fills = engine.tick("test", 0.55)
    assert fills == 1

    # Check position
    positions = engine.positions()
    assert len(positions) == 1
    assert positions[0].size == 10.0

​
Testing with fills

Since Position has no Python constructor, build positions by processing fills: 
def test_position_tracking():
    engine = Engine()

    # Inject a fill directly
    engine.process_fill(Fill(
        fill_id="f1",
        order_id="o1",
        market_id="test",
        side=Side.Yes,
        order_side=OrderSide.Buy,
        price=0.50,
        size=10.0,
    ))

    positions = engine.positions()
    assert len(positions) == 1
    assert positions[0].avg_entry_price == 0.50

​
Testing risk limits

def test_position_limit():
    config = RiskConfig(max_position_per_market=10.0)
    engine = Engine(risk_config=config)

    # Fill up to the limit
    engine.process_fill(Fill(
        fill_id="f1", order_id="o1", market_id="test",
        side=Side.Yes, order_side=OrderSide.Buy,
        price=0.50, size=10.0,
    ))

    # This should be rejected by the position limit
    import pytest
    with pytest.raises(Exception):
        engine.submit_order(OrderRequest(
            market_id="test",
            side=Side.Yes,
            order_side=OrderSide.Buy,
            size=5.0,
            price=0.55,
        ))

​
Testing pipeline functions

Test pipeline functions in isolation with a mock context: 
from horizon.context import Context, FeedData, InventorySnapshot

def test_fair_value():
    ctx = Context(
        feeds={"default": FeedData(price=0.55, bid=0.54, ask=0.56)},
    )
    result = fair_value(ctx)
    assert 0 < result < 1  # Valid probability

def test_quoter():
    ctx = Context(
        feeds={"default": FeedData(price=0.55)},
        inventory=InventorySnapshot(positions=[]),
    )
    quotes = quoter(ctx, 0.60)
    assert len(quotes) > 0
    assert quotes[0].bid < quotes[0].ask

​
Testing with persistence

def test_crash_recovery():
    import tempfile, os

    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
        db_path = f.name

    try:
        # First run: build some state
        engine = Engine(db_path=db_path)
        engine.process_fill(Fill(
            fill_id="f1", order_id="o1", market_id="test",
            side=Side.Yes, order_side=OrderSide.Buy,
            price=0.50, size=10.0,
        ))
        engine.snapshot_positions()
        del engine

        # Second run: recover state
        engine2 = Engine(db_path=db_path)
        recovered = engine2.recover_state()
        assert recovered == 1
        positions = engine2.positions()
        assert len(positions) == 1
    finally:
        os.unlink(db_path)

​
Test Tips

Use Engine(risk_config=RiskConfig(max_position_per_market=10000)) in tests to avoid hitting risk limits unintentionally.
  • Paper exchange is the default. No credentials needed for tests
  • engine.tick(market_id, mid_price) triggers paper matching
  • engine.process_fill(fill) injects fills without going through the exchange
  • Position snapshots require engine.positions() which returns from the live tracker, not the DB
  • The fills vector is capped at 1000 entries in the Engine