Features

FeatureStore + built-in feature classes

FeatureStore

python
from horizon.features import FeatureStore

store = FeatureStore(max_history=1000)

# Each tick
store.update_feeds(feeds, now=now)

# Compute features for a universe
namespace = store.compute(
    feature_spec={"z": Zscore(20), "vol": RealizedVol(60)},
    universe_market_ids=["AAPL", "MSFT"],
    feeds=feeds,
)

# Access
z_aapl = namespace.z["AAPL"]
vol_msft = namespace.vol["MSFT"]

Methods

  • update_feeds(feeds: dict[str, FeedData], now: datetime | None = None): append new prices to histories, clear per-tick cache. Silently rejects NaN/Inf/≤0 prices.
  • history(market_id: str) -> PriceHistory: get the rolling history for a market.
  • has_market(market_id: str) -> bool: check if a market has any history.
  • compute(feature_spec, universe_market_ids, feeds) -> FeatureNamespace: compute every feature in the spec for every market in the universe.

PriceHistory

python
@dataclass
class PriceHistory:
    max_len: int = 1000
    prices: list[float]
    log_returns: list[float]
    timestamps: list[datetime]

Methods

  • update(price: float, ts: datetime | None = None): append a new price. Silently rejects NaN/Inf/≤0.
  • last_price() -> float | None
  • last_n_prices(n: int) -> list[float]
  • last_n_returns(n: int) -> list[float]
  • __len__() -> int

Feature base class

python
from horizon.features.base import Feature, PriceHistory
from horizon.context import FeedData

class Feature(ABC):
    def __init__(self, *, market: str | None = None):
        """``market`` pins this feature to a specific market id."""
        self.market = market

    @abstractmethod
    def compute(
        self,
        market_id: str,
        history: PriceHistory,
        feeds: dict[str, FeedData],
    ) -> float:
        ...

    @property
    def key(self) -> str:
        """Unique cache key. class name + all attribute params."""

Built-in features

Price

python
Price(market: str | None = None)

Most recent price for a market. Useful as cross-asset reference (pin with market=...).

Return

python
Return(window: int = 1, market: str | None = None)

Log return over the last window periods. Returns NaN if insufficient history.

Zscore

python
Zscore(window: int = 20, market: str | None = None)

Rolling z-score of the most recent log return against the last window returns. Returns 0.0 if the std is 0.

RealizedVol

python
RealizedVol(
    window: int = 60,
    annualize: bool = True,
    periods_per_year: float = 252.0,
    market: str | None = None,
)

Annualized (by default) standard deviation of log returns over the last window periods.

SMA

python
SMA(window: int = 20, market: str | None = None)

Simple moving average of the last window prices.

EMA

python
EMA(span: int = 20, alpha: float | None = None, market: str | None = None)

Exponential moving average. If alpha is None, derived from span as 2 / (span + 1).

RSI

python
RSI(window: int = 14, market: str | None = None)

Wilder’s Relative Strength Index. Returns a value in [0, 100]. Neutral is 50.

BollingerZ

python
BollingerZ(window: int = 20, market: str | None = None)

Distance from rolling mean price in stddev units. Like Zscore but over price levels rather than returns.

MovingAverageCross

python
MovingAverageCross(fast: int = 10, slow: int = 30, market: str | None = None)

Normalized distance between a fast and slow SMA: (fast_ma - slow_ma) / slow_ma. Positive in an uptrend, negative in a downtrend.

FeatureNamespace

The object returned from store.compute(). Access features by name, then by market id:

python
ns = store.compute({"z": Zscore(20), "vol": RealizedVol(60)}, ["AAPL"], feeds)
ns.z["AAPL"]         # scalar
ns.vol["AAPL"]       # scalar
"z" in ns            # True
ns.nonexistent       # AttributeError

Writing custom features

Subclass Feature and implement compute:

python
import math
from horizon.features.base import Feature, PriceHistory
from horizon.context import FeedData

class MaxDrawdownOver(Feature):
    def __init__(self, window: int = 20, market: str | None = None):
        super().__init__(market=market)
        self.window = window

    def compute(self, market_id, history, feeds):
        prices = history.last_n_prices(self.window)
        if len(prices) < self.window:
            return float("nan")
        peak = prices[0]
        max_dd = 0.0
        for p in prices:
            peak = max(peak, p)
            if peak > 0:
                max_dd = max(max_dd, (peak - p) / peak)
        return max_dd

Use immediately in a strategy: no registration needed.

Tests

21 unit tests in tests/test_features.py cover every feature against hand-calculated expected values.

Next

Writing strategies Use features in a strategy. Quant library Prebuilts that use the feature library.