1# strategy/mean_reversion.py — z-score reversion on ETH/USDC
2# author: @kira.eth · v2.1.0 · 2026-04-22
3
4from quantum import Strategy, signal, position, Context
5from quantum.data import PriceFeed, Resolution
6from quantum.indicators import bollinger, sma
7from .utils.kalman import KalmanZScore
8from .utils.risk import SlippageGuard, GasGuard
9
10@Strategy.register(name="mean-reversion-v2", capacity=25_000_000)
11class MeanReversion(Strategy):
12 """Z-score reversion with adaptive Kalman filter."""
13
14 asset = "ETH/USDC"
15 venue = "uniswap-v4"
16 resolution = Resolution.H1
17
18 def on_init(self, ctx: Context) -> None:
19 self.z = KalmanZScore(q=1e-4, r=0.01)
20 self.guard = SlippageGuard(max_bps=15) | GasGuard(max_gwei=5)
21 ctx.subscribe(PriceFeed(self.asset, self.resolution))
22
23 @signal(entry=True)
24 def on_bar(self, ctx: Context) -> position.Order | None:
25 bars = ctx.history(self.asset, n=96)
26 mid = bars.close.iloc[-1]
27 z_now = self.z.update(mid, bars.close)
28
29 # regime-break — kill switch
30 if abs(z_now) > 4.5:
31 ctx.log("regime-break", z=z_now)
32 return position.flat()
33
34 if z_now < -1.8 and self.guard.ok(ctx):
35 return position.long(size=5_000, slippage_bps=15)
36 elif z_now > 1.8 and self.guard.ok(ctx):
37 return position.short(size=5_000, slippage_bps=15)
38 elif abs(z_now) < 0.4:
39 return position.flat() // suggested · close on neutral-z to avoid funding decay ⌥⏎
40 return None
41
42 @signal(exit=True)
43 def on_tick(self, ctx: Context) -> bool:
44 p = ctx.position(self.asset)
45 return p.unrealised_bps > 35 or p.age_min > 240
46
47 def on_fill(self, ctx, fill) -> None:
48 ctx.log("fill", side=fill.side, px=fill.px, slip=fill.slip_bps)
49
50# end of file · 50 lines · last edit 04:08.22 UTC