GEX
Public feedInitializing terminal
The official Python SDK for GEX Engine. Provides a typed, Pythonic interface to all API endpoints with built-in retry logic, rate limit handling, and WebSocket streaming support.
pip install gex-engine
Requires Python 3.9 or later. The SDK has minimal dependencies: httpx for HTTP, websockets for streaming, and pydantic for response models.
from gex_engine import GEXClient client = GEXClient(api_key="YOUR_API_KEY") # Or use an environment variable (recommended) # export GEX_API_KEY=your_key client = GEXClient() # reads from GEX_API_KEY
gex = client.gex("SPX")
print(f"Total GEX: {gex.total_gex:,.0f}")
print(f"Net GEX: {gex.net_gex:,.0f}")
print(f"Call GEX: {gex.call_gex:,.0f}")
print(f"Put GEX: {gex.put_gex:,.0f}")
print(f"Spot: {gex.meta.spot:.2f}")
# Per-strike data
for strike, exposure in zip(gex.strikes, gex.gex_per_strike):
if abs(exposure) > 1_000_000_000:
print(f" {strike}: {exposure:>14,.0f}")dsi = client.dsi("SPX")
print(f"DSI: {dsi.dsi:+.1f}")
print(f"Gamma component: {dsi.gamma_component:+.2f}")
print(f"Vol component: {dsi.vol_component:+.2f}")regime = client.regime("SPX")
print(f"Regime: {regime.regime}")
print(f"Strength: {regime.strength:.2f}")
print(f"Amplification: {regime.amplification_factor:.2f}")levels = client.levels("SPX")
print(f"Gamma flip: {levels.gamma_flip:.2f}")
print(f"Put wall: {levels.put_wall:.0f}")
print(f"Call wall: {levels.call_wall:.0f}")
print(f"Max pain: {levels.max_pain:.0f}")import asyncio
async def stream_gex():
async for update in client.stream("SPX", channels=["gex", "signals", "regime"]):
print(f"[{update.channel}] {update.data}")
asyncio.run(stream_gex())from gex_engine import GEXClient, GEXError, RateLimitError
client = GEXClient()
try:
gex = client.gex("SPX")
except RateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after}s")
except GEXError as e:
print(f"API error {e.status_code}: {e.detail}")from gex_engine import GEXClient
import asyncio
client = GEXClient(api_key="YOUR_API_KEY")
# Fetch all key data points
gex = client.gex("SPX")
regime = client.regime("SPX")
levels = client.levels("SPX")
dsi = client.dsi("SPX")
print(f"=== SPX Dashboard ===")
print(f"Spot: {gex.meta.spot:>10,.2f}")
print(f"Total GEX: {gex.total_gex:>10,.0f}")
print(f"Regime: {regime.regime:>10}")
print(f"Strength: {regime.strength:>10.2f}")
print(f"DSI: {dsi.dsi:>+10.1f}")
print(f"Gamma Flip: {levels.gamma_flip:>10,.2f}")
print(f"Put Wall: {levels.put_wall:>10,.0f}")
print(f"Call Wall: {levels.call_wall:>10,.0f}")
print()
# Stream real-time updates
async def monitor():
async for update in client.stream("SPX", channels=["gex", "regime"]):
if update.channel == "regime":
print(f"Regime: {update.data['regime']} "
f"(strength: {update.data['strength']:.2f})")
elif update.channel == "gex":
print(f"GEX: {update.data['total_gex']:,.0f}")
asyncio.run(monitor())