Custom Strategy

# custom_strategy.py
import math

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


class AdaptiveMeanReversion(Strategy):
    """Mean-reversion with adaptive confidence and drawdown pausing.

    Core logic:
    - Z-score-based mean reversion entry
    - Scales confidence by |z| (deeper extremes = higher conviction)
    - Pauses entirely if portfolio drawdown exceeds 5%
    - Uses rolling vol to set expected vol on signals
    - Skips markets in strong trends (|r20| > 3%)
    """

    name = "adaptive_mr"
    asset_classes = [Equity]
    features = {
        "z": Zscore(window=20),
        "vol": RealizedVol(window=60),
        "r20": Return(window=20),
    }

    def evaluate(self, f, universe, ctx):
        # Drawdown pause
        if ctx.portfolio.drawdown_pct > 0.05:
            return []

        signals = []
        for m in universe:
            z = f.z[m.id]
            vol = f.vol[m.id]
            r20 = f.r20[m.id]

            if math.isnan(z) or math.isnan(r20):
                continue

            # Skip strong-trend markets. mean reversion doesn't work
            if abs(r20) > 0.03:
                continue

            # Require meaningful z-score deviation
            if abs(z) < 2.0:
                continue

            # Scale confidence by magnitude
            confidence = min(1.0, abs(z) / 4)

            # Direction: oversold (z<0) → long, overbought (z>0) → short
            direction_score = -z

            signals.append(Signal.from_score(
                market=m,
                score=direction_score,
                edge_per_stdev=25,
                horizon="3d",
                expected_expected_vol_bps=max((vol if not math.isnan(vol) else 0.25) * 10_000, 100),
                reason=f"z={z:.2f} r20={r20:+.2%}",
                features={"z": z, "r20": r20, "vol": vol},
            ))
        return signals


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

    data = SyntheticGBM(
        market_ids=["AAPL", "MSFT", "NVDA", "GOOGL", "AMZN", "META"],
        n_bars=252,
        mu=0.10,
        sigma=0.22,
        seed=42,
    )

    result = hz.run(
        mode="backtest",
        strategies=[AdaptiveMeanReversion],
        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%}")

    # Show attribution
    print()
    print("Top 10 trades by P&L:")
    top = sorted(result.trades, key=lambda t: t.realized_pnl, reverse=True)[:10]
    for t in top:
        print(f"  {t.market_id:8s} {t.side:4s} qty={t.quantity:6.1f} P&L={t.realized_pnl:+.2f}")


if __name__ == "__main__":
    main()

PYTHONPATH=. python3 custom_strategy.py