Custom Feature

# custom_feature.py
import math

import horizon as hz
from horizon import Signal, Strategy
from horizon.asset_classes import AssetClass, Equity
from horizon.context import FeedData
from horizon.data import SyntheticGBM
from horizon.discovery import StaticUniverse
from horizon.discovery.base import Market
from horizon.features import Zscore
from horizon.features.base import Feature, PriceHistory
from horizon.portfolio import KellyOptimizer
from horizon.risk import RiskProfile


# -----------------------------------------------------------------------------
# Custom feature
# -----------------------------------------------------------------------------
class MaxDrawdown(Feature):
    """Largest peak-to-trough drop over the last `window` bars.

    Returns a value in [0, 1]. `0` means no drawdown, `0.5` means 50% drawdown.
    """

    def __init__(self, window: int = 20, market: str | None = None):
        super().__init__(market=market)
        if window < 2:
            raise ValueError(f"window must be >= 2, got {window}")
        self.window = window

    def compute(
        self,
        market_id: str,
        history: PriceHistory,
        feeds: dict[str, FeedData],
    ) -> float:
        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:
                dd = (peak - p) / peak
                max_dd = max(max_dd, dd)
        return max_dd


# -----------------------------------------------------------------------------
# Strategy using the custom feature
# -----------------------------------------------------------------------------
class DrawdownAwareMeanReversion(Strategy):
    """Mean reversion, but only on names with meaningful recent drawdown.

    Rationale: a name that just fell 10% is more likely to mean-revert than
    a name that's been flat. We filter the universe by recent drawdown and
    then apply a standard z-score-based mean reversion.
    """

    name = "dd_aware_mr"
    asset_classes = [Equity]
    features = {
        "dd": MaxDrawdown(window=20),
        "z": Zscore(window=10),
    }

    def evaluate(self, f, universe):
        return [
            Signal.from_score(
                m,
                score=-f.z[m.id],       # negative z → long
                edge_per_stdev=25,
                horizon="3d",
                reason=f"dd={f.dd[m.id]:.2%} z={f.z[m.id]:.2f}",
            )
            for m in universe
            if not math.isnan(f.dd[m.id])
            and f.dd[m.id] > 0.05            # recent drawdown > 5%
            and abs(f.z[m.id]) > 1.5          # and an oversold z-score
        ]


# -----------------------------------------------------------------------------
# Run
# -----------------------------------------------------------------------------
def main():
    universe = StaticUniverse([
        Market(id=t, asset_class=AssetClass.Equity)
        for t in ["AAPL", "MSFT", "NVDA", "GOOGL", "AMZN"]
    ])

    data = SyntheticGBM(
        market_ids=["AAPL", "MSFT", "NVDA", "GOOGL", "AMZN"],
        n_bars=252,
        mu=0.08,
        sigma=0.25,           # higher vol → more drawdowns
        seed=42,
    )

    result = hz.run(
        mode="backtest",
        strategies=[DrawdownAwareMeanReversion],
        asset_classes=[Equity],
        universe=universe,
        portfolio=KellyOptimizer(kelly_fraction=0.25, max_gross_leverage=1.0),
        risk=RiskProfile.moderate(),
        data_source=data,
        backtest=hz.BacktestConfig(initial_cash_usd=100_000),
    )

    print(f"Trades:       {result.n_trades}")
    print(f"Total return: {result.total_return:+.2%}")
    print(f"Sharpe:       {result.sharpe:+.3f}")
    print(f"Max drawdown: {result.max_drawdown:.2%}")


if __name__ == "__main__":
    main()

PYTHONPATH=. python3 custom_feature.py