Horizon Pipeline Composition

Build Horizon strategies by chaining pipeline functions with automatic signature introspection.

Strategies are built by chaining pipeline functions. Each function receives the Context plus the outputs of previous functions. The final function must return list[Quote].

Basic Example

# Step 1: Just context, compute fair value
def fair_value(ctx: hz.Context) -> float:
    return ctx.feeds["btc"].price * 0.01

# Step 2: Context + previous output, compute toxicity
def toxicity(ctx: hz.Context) -> float:
    return 0.5  # VPIN, Kyle's lambda, etc.

# Step 3: Context + two previous outputs, generate quotes
def quoter(ctx: hz.Context, fair: float, tox: float) -> list[hz.Quote]:
    spread = 0.02 + tox * 0.04
    return hz.quotes(fair, spread, size=5)

hz.run(pipeline=[fair_value, toxicity, quoter], ...)

How It Works

Horizon inspects each function’s signature to determine how many arguments it expects. Previous outputs are passed in order:

Parameters	What’s passed
(ctx)	Just the context
(ctx, prev)	Context + last output
(ctx, a, b)	Context + last two outputs
(ctx, a, b, c)	Context + last three outputs

This means you can insert or remove pipeline stages without changing downstream functions.

Signature Introspection

The pipeline runner uses inspect.signature() to count parameters:

import inspect

for fn in pipeline:
    sig = inspect.signature(fn)
    n_params = len(sig.parameters)
    # n_params == 1: fn(ctx)
    # n_params == 2: fn(ctx, prev_output)
    # n_params == 3: fn(ctx, output_n-2, output_n-1)

The first parameter is always ctx: hz.Context. Additional parameters receive outputs from previous pipeline stages, most recent last.

Pipeline Rules

1. At least one function is required in the pipeline
2. The final function must return list[Quote] (or a single Quote)
3. Each function runs once per market per cycle
4. If a function raises an exception, the error is logged and that market is skipped for the cycle
5. The pipeline runs for every market in the markets list

Result Processing

After the pipeline runs, _process_result() handles the output:

1. Cancel existing orders for the market (cancel_market())
2. Tick the paper exchange with the feed mid price (or quote mid as fallback)
3. Update mark prices for unrealized P&L tracking
4. Submit new quotes, routed to the correct exchange based on market.exchange

Risk rejections during quote submission are logged at debug level (not warning), since they’re expected during normal operation (e.g., position limits hit).

Composability Examples

Adding a stage

You can add an intermediate stage without modifying existing functions:

# Before: [fair_value, quoter]
# After:  [fair_value, inventory_skew, quoter]

def inventory_skew(ctx: hz.Context, fair: float) -> float:
    """Skew the fair value based on inventory."""
    skew = ctx.inventory.net * 0.001
    return fair - skew

hz.run(pipeline=[fair_value, inventory_skew, quoter], ...)

Multiple data sources

def fair_value(ctx: hz.Context) -> float:
    poly = ctx.feeds.get("poly", hz.context.FeedData())
    kalshi = ctx.feeds.get("kalshi", hz.context.FeedData())
    return (poly.price + kalshi.price) / 2

def volatility(ctx: hz.Context) -> float:
    btc = ctx.feeds.get("btc", hz.context.FeedData())
    return abs(btc.bid - btc.ask) / max(btc.price, 1)

def quoter(ctx: hz.Context, fair: float, vol: float) -> list[hz.Quote]:
    spread = 0.02 + vol * 0.1
    return hz.quotes(fair, spread, size=5)

The hz.quotes() Helper

A convenience function to create quotes from a fair value and spread:

quotes = hz.quotes(fair=0.50, spread=0.06, size=5.0)
# → [Quote(bid=0.47, ask=0.53, size=5.0)]

Bid and ask are automatically clamped to [0.01, 0.99] for prediction markets. If the spread is too wide for the fair value (resulting in a crossed or invalid quote), an empty list is returned instead of raising an error:

hz.quotes(fair=0.02, spread=0.10, size=5.0)
# → [Quote(bid=0.01, ask=0.07, size=5.0)]  (bid clamped from -0.03 to 0.01)

hz.quotes(fair=0.99, spread=0.10, size=5.0)
# → [Quote(bid=0.94, ask=0.99, size=5.0)]  (ask clamped from 1.04 to 0.99)

Or create quotes manually:

quotes = [
    hz.Quote(bid=0.45, ask=0.55, size=10.0),
    hz.Quote(bid=0.40, ask=0.60, size=5.0),
]

Built-in Pipeline Functions

Horizon provides several ready-to-use pipeline functions that can be composed together:

Factory	Returns	Description
hz.market_maker(...)	list[Quote]	Avellaneda-Stoikov market making with multi-level quoting
hz.signal_combiner(...)	float	Combine multiple alpha signals into a composite score
hz.kelly_sizer(...)	float	Kelly criterion position sizing
hz.arb_scanner(...)	list[Quote] or None	Continuous cross-exchange arbitrage scanning
hz.hawkes_intensity(feed_name, mu, alpha, beta)	float	Hawkes self-exciting process intensity
hz.correlation_estimator(feed_names, window)	list[list[float]]	Ledoit-Wolf shrinkage correlation
hz.hot_reload(source)	dict	Runtime parameter hot-reload

Example: Signal → Kelly → Market Maker

hz.run(
    pipeline=[
        hz.signal_combiner([
            hz.price_signal("book", weight=0.5),
            hz.momentum_signal("book", lookback=20, weight=0.3),
            hz.spread_signal("book", weight=0.2),
        ], smoothing=10),
        hz.market_maker(feed_name="book", gamma=0.5, size=5.0),
    ],
    ...
)

See Market Making, Signal Combiner, Kelly Criterion, and Arbitrage Executor for detailed documentation.