Overview

This page documents 8 operators (role: BALANCE).

Quick Reference

Operator	Role	Key Parameters	Ephemeral
CashBalance	`BALANCE`	—	No
EquityCalculator	`BALANCE`	`cash`, `positions`, `prices`	No
TotalEquityCalculator	`BALANCE`	`equity_inputs`	No
AvailableMargin	`BALANCE`	—	No
PositionQuantity	`BALANCE`	—	No
PositionValue	`BALANCE`	`prices_key='prices'`	No
BookSize	`BALANCE`	`multiplier=1.0`, `min_book_size=0.0`, `max_book_size=None`	No
MarginCalculator	`BALANCE`	`cash`, `positions`, `prices`	No

CashBalance

Query cash balance from STATE. Role: BALANCE | Ephemeral: No

Parameters

Parameter	Type	Default	Description
`input`	`'Input'`	Required

Usage

cash = CashBalance(
    input=Input("STATE:gateio:spot:cash", timeframe="1m", lookback=0),
)

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    # STATE:cash returns scalar value
    if hasattr(data, "tolist"):
        value = float(data.tolist()) if data.tolist() else 0.0
    elif isinstance(data, (int, float)):
        value = float(data)
    else:
        value = 0.0

    return TaggedArray(
        value=np.array([value]),
        exists=np.array([True]),
        valid=np.array([value >= 0]),
        updated=np.array([True]),
    )

_{Source: apps/trading/operators/balance/query.py}

EquityCalculator

Calculate equity based on market type (auto-detected from Input source). Market type is automatically detected from the cash Input source:

“STATE:gateio:spot:cash” → market_type=“spot”
“STATE:gateio:futures:cash” → market_type=“futures”

SPOT: equity = cash + sum(current_price * qty)

Cash is reduced when positions are opened
Position value is added to remaining cash

FUTURES: equity = margin + sum((current_price - entry_price) * qty)

Margin doesn’t change when positions are opened
Unrealized PnL is added to margin

Inputs (4 required):

cash: Cash/margin balance from STATE:venue:cash
positions: Position quantities from STATE:venue:pos_quantity
prices: Current prices from FIELD:close
entry_prices: Entry prices from STATE:venue:pos_entry_price

Role: BALANCE | Ephemeral: No

Parameters

Parameter	Type	Default
`cash`	`'Input'`	Required
`positions`	`'Input'`	Required
`prices`	`'Input'`	Required
`entry_prices`	`'Input'`	Required
`axis_keys`	`List[str]`	Required
`market_type`	`Optional[str]`	`None`

Usage

# With SymbolSourceMap
symbol_source_map = SymbolSourceMap({"gateio:spot": ["BTC/USDT", "ETH/USDT"]})

equity = EquityCalculator(
    cash=Input("STATE:gateio:spot:cash", timeframe="1m", lookback=0),
    positions=Input("STATE:gateio:spot:pos_quantity", timeframe="1m", lookback=0),
    prices=Input("FIELD:gateio:spot:ohlcv:close", timeframe="1m", lookback=0),
    entry_prices=Input("STATE:gateio:spot:pos_entry_price", timeframe="1m", lookback=0),
    axis_keys=symbol_source_map.axis_keys_for("gateio:spot"),
)

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if not isinstance(data, list) or len(data) < 4:
        return self._empty_result()

    cash_data, positions_data, prices_data, entry_prices_data = data[0], data[1], data[2], data[3]

    # Extract cash (scalar)
    cash = self._extract_scalar(cash_data)

    # Get axis alignment info
    all_axis_keys = self._get_axis_keys_from_context(context)
    axis_indices = []
    if self._axis_keys and all_axis_keys:
        axis_indices = self._get_axis_indices(all_axis_keys, self._axis_keys)

    # Extract arrays aligned to primary axis, then filter to our indices
    positions = self._extract_aligned_values(positions_data, axis_indices, all_axis_keys)
    prices = self._extract_aligned_values(prices_data, axis_indices, all_axis_keys)
    entry_prices = self._extract_aligned_values(entry_prices_data, axis_indices, all_axis_keys)

    # Calculate equity based on market type
    # Track if any position has missing price data (invalid tick)
    n_symbols = len(self._symbols)
    has_missing_price = False

    if self._market_type == "futures":
        # FUTURES: equity = margin + unrealized_pnl
        unrealized_pnl = 0.0
        for i in range(n_symbols):
            qty = positions[i] if i < len(positions) else 0.0
            current_price = prices[i] if i < len(prices) else 0.0
            entry_price = entry_prices[i] if i < len(entry_prices) else current_price

            # Check for missing price on held position
            if abs(qty) > 1e-10 and (np.isnan(current_price) or current_price <= 0):
                has_missing_price = True
                # Use entry_price as fallback to avoid equity crash
                current_price = entry_price

            if abs(qty) > 1e-10 and current_price > 0:
                unrealized_pnl += (current_price - entry_price) * qty

        equity = cash + unrealized_pnl
    else:
        # SPOT: equity = cash + position_value
        position_value = 0.0
        for i in range(n_symbols):
            qty = positions[i] if i < len(positions) else 0.0
            current_price = prices[i] if i < len(prices) else 0.0
            entry_price = entry_prices[i] if i < len(entry_prices) else 0.0

            # Check for missing price on held position
            if abs(qty) > 1e-10 and (np.isnan(current_price) or current_price <= 0):
                has_missing_price = True
                # Use entry_price as fallback to avoid equity crash
                if entry_price > 0:
                    current_price = entry_price

            if abs(qty) > 1e-10 and current_price > 0:
                position_value += current_price * qty

        equity = cash + position_value

    # Mark as invalid if any position has missing price data
    # Downstream (metrics) should handle invalid ticks appropriately
    is_valid = (equity >= 0) and not has_missing_price

    return TaggedArray(
        value=[equity],
        exists=[True],
        valid=[is_valid],
        updated=np.array([True]),
    )

_{Source: apps/trading/operators/balance/query.py}

TotalEquityCalculator

Calculate total equity across multiple venues. Aggregates equity from multiple EquityCalculator inputs to provide a unified portfolio-level equity value. Useful for:

Multi-venue strategies (spot + futures)
Cross-exchange arbitrage
Portfolio-level risk management

Inputs: Multiple EquityCalculator outputs, one per venue. Example:

Calculate equity for each venue separately

spot_equity = EquityCalculator( cash=Input(“STATE:binance:spot:cash”, …), positions=Input(“STATE:binance:spot:pos_quantity”, …), prices=Input(“FIELD:binance:spot:close”, …), entry_prices=Input(“STATE:binance:spot:pos_entry_price”, …), symbols=spot_symbols, ) futures_equity = EquityCalculator( cash=Input(“STATE:binance:futures:cash”, …), positions=Input(“STATE:binance:futures:pos_quantity”, …), prices=Input(“FIELD:binance:futures:close”, …), entry_prices=Input(“STATE:binance:futures:pos_entry_price”, …), symbols=futures_symbols, )

Aggregate into total equity

total_equity = TotalEquityCalculator( equity_inputs=[ Input(“spot_equity”, timeframe=“1m”, lookback=1), Input(“futures_equity”, timeframe=“1m”, lookback=1), ] ) Role: BALANCE | Ephemeral: No

Parameters

Parameter	Type	Default	Description
`equity_inputs`	`List['Input']`	Required

Usage

# Calculate equity for each venue separately
spot_equity = EquityCalculator(
    cash=Input("STATE:binance:spot:cash", ...),
    positions=Input("STATE:binance:spot:pos_quantity", ...),
    prices=Input("FIELD:binance:spot:close", ...),
    entry_prices=Input("STATE:binance:spot:pos_entry_price", ...),
    symbols=spot_symbols,
)
futures_equity = EquityCalculator(
    cash=Input("STATE:binance:futures:cash", ...),
    positions=Input("STATE:binance:futures:pos_quantity", ...),
    prices=Input("FIELD:binance:futures:close", ...),
    entry_prices=Input("STATE:binance:futures:pos_entry_price", ...),
    symbols=futures_symbols,
)

# Aggregate into total equity
total_equity = TotalEquityCalculator(
    equity_inputs=[
        Input("spot_equity", timeframe="1m", lookback=1),
        Input("futures_equity", timeframe="1m", lookback=1),
    ]
)

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if not isinstance(data, list):
        data = [data]

    total_equity = 0.0
    venue_equities = {}

    for i, equity_data in enumerate(data):
        equity = self._extract_equity(equity_data)
        total_equity += equity
        venue_equities[f"venue_{i}"] = equity

    return TaggedArray(
        value=[total_equity],
        exists=[True],
        valid=[total_equity >= 0],
        updated=np.array([True]),
    )

_{Source: apps/trading/operators/balance/query.py}

AvailableMargin

Query available margin from STATE. Role: BALANCE | Ephemeral: No

Parameters

Parameter	Type	Default	Description
`input`	`'Input'`	Required

Usage

margin = AvailableMargin(
    input=Input("STATE:binance:futures:margin", timeframe="1m", lookback=0),
)

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    if hasattr(data, "tolist"):
        value = float(data.tolist()) if data.tolist() else 0.0
    elif isinstance(data, (int, float)):
        value = float(data)
    else:
        value = 0.0

    return TaggedArray(
        value=np.array([value]),
        exists=np.array([True]),
        valid=np.array([value >= 0]),
        updated=np.array([True]),
    )

_{Source: apps/trading/operators/balance/query.py}

PositionQuantity

Query position quantities from STATE. Role: BALANCE | Ephemeral: No

Parameters

Parameter	Type	Default	Description
`input`	`'Input'`	Required

Usage

positions = PositionQuantity(
    input=Input("STATE:gateio:spot:pos_quantity", timeframe="1m", lookback=0),
)

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    # STATE:pos_quantity returns Dict[str, float]
    if hasattr(data, "tolist"):
        positions = data.tolist()
    elif isinstance(data, dict):
        positions = data
    else:
        positions = {}

    if not isinstance(positions, dict):
        positions = {}

    # Convert to arrays for Graph compatibility
    symbols = list(positions.keys())
    values = [float(positions[s]) for s in symbols]

    if not symbols:
        return TaggedArray(
            value=[],
            exists=[],
            valid=[],
            updated=np.array([]),
            symbols=np.array([]),
        )

    return TaggedArray(
        value=values,
        exists=np.ones(len(values), dtype=bool),
        valid=np.ones(len(values), dtype=bool),
        updated=np.ones(len(values), dtype=bool),
        symbols=np.array(symbols),
    )

_{Source: apps/trading/operators/balance/query.py}

PositionValue

Query position notional values from STATE. Role: BALANCE | Ephemeral: No

Parameters

Parameter	Type	Default	Description
`input`	`'Input'`	Required
`prices_key`	`str`	`'prices'`

Usage

pos_value = PositionValue(
    input=Input("STATE:gateio:spot:pos_quantity", timeframe="1m", lookback=0),
    prices_key="prices",
)

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    if hasattr(data, "tolist"):
        positions = data.tolist()
    elif isinstance(data, dict):
        positions = data
    else:
        positions = {}

    if not isinstance(positions, dict):
        positions = {}

    prices = context.get(self._prices_key, {}) if context else {}

    symbols = list(positions.keys())
    values = []
    for s in symbols:
        qty = float(positions[s])
        # Extract symbol from venue:symbol key (e.g., "gateio:futures:BTC" -> "BTC")
        symbol_only = s.split(":")[-1] if ":" in s else s
        price = prices.get(symbol_only, prices.get(s, 0.0))
        values.append(qty * price)

    if not symbols:
        return TaggedArray(
            value=[],
            exists=[],
            valid=[],
            updated=np.array([]),
            symbols=np.array([]),
        )

    return TaggedArray(
        value=values,
        exists=np.ones(len(values), dtype=bool),
        valid=np.ones(len(values), dtype=bool),
        updated=np.ones(len(values), dtype=bool),
        symbols=np.array(symbols),
    )

_{Source: apps/trading/operators/balance/query.py}

BookSize

Compute book size from balance inputs. Role: BALANCE | Ephemeral: No

Parameters

Parameter	Type	Default
`input`	`'Input'`	Required
`multiplier`	`float`	`1.0`
`min_book_size`	`float`	`0.0`
`max_book_size`	`Optional[float]`	`None`

Usage

book = BookSize(
    input=Input("STATE:gateio:spot:equity", timeframe="1m", lookback=0),
    multiplier=0.95,
    min_book_size=100.0,
    max_book_size=1_000_000.0,
)

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    # Extract value from TaggedTensor or scalar
    value = self._extract_value(data)

    # Apply multiplier
    book_size = value * self._multiplier

    # Apply limits
    book_size = max(book_size, self._min_book_size)
    if self._max_book_size is not None:
        book_size = min(book_size, self._max_book_size)

    return TaggedArray(
        value=[book_size],
        exists=[True],
        valid=[book_size > 0],
        updated=np.array([True]),
    )

_{Source: apps/trading/operators/balance/query.py}

MarginCalculator

Calculate margin status for futures positions. Uses exchange-specific margin configuration from venues.py for accurate margin ratio calculation and liquidation detection. Supports two MMR modes:

Dynamic (default): Per-symbol MMR from CCXT fetch_leverage_tiers()
Static (use_dynamic_tiers=False): Flat MMR from venues.py MarginConfig

Dynamic mode (default) uses LeverageTierResolver which:

Fetches real-time tier data for supported exchanges (Gateio, Bybit, Kraken, Binance, Hyperliquid)
Falls back to static MarginConfig for unsupported exchanges (Coinbase, Gemini)
Caches tier data for 1 hour to minimize API calls

Exchange-specific formulas:

Binance/OKX (inverted): margin_ratio = MMR / equity * 100 (liquidation >= 100%)
Bybit/Gateio/Kraken (normal): margin_ratio = equity / MMR (liquidation < threshold)

The calculator also accounts for:

Exchange-specific default MMR (maintenance margin rate)
Liquidation fees (deducted from equity before ratio calculation)
Exchange-specific liquidation thresholds (1.0 or 1.5)

Inputs (4 required):

cash: Cash balance from STATE:venue:cash
positions: Position quantities from STATE:venue:pos_quantity
prices: Current prices from FIELD:close
entry_prices: Entry prices from STATE:venue:pos_entry_price

Output TaggedArray value contains margin_ratio:

For inverted exchanges: 0-100+ (100 = liquidation threshold)
For normal exchanges: 0-inf (threshold varies by exchange)

Role: BALANCE | Ephemeral: No

Parameters

Parameter	Type	Default
`cash`	`'Input'`	Required
`positions`	`'Input'`	Required
`prices`	`'Input'`	Required
`entry_prices`	`'Input'`	Required
`symbols`	`List[str]`	Required
`exchange`	`str`	`'binance'`
`maintenance_margin_rate`	`Optional[float]`	`None`
`use_dynamic_tiers`	`bool`	`True`
`api_key`	`Optional[str]`	`None`
`api_secret`	`Optional[str]`	`None`

Usage

# Default: dynamic per-symbol MMR (recommended)
margin = MarginCalculator(
    cash=Input("STATE:gateio:futures:cash", timeframe="1m", lookback=0),
    positions=Input("STATE:gateio:futures:pos_quantity", timeframe="1m", lookback=0),
    prices=Input("FIELD:gateio:futures:close", timeframe="1m", lookback=0),
    entry_prices=Input("STATE:gateio:futures:pos_entry_price", timeframe="1m", lookback=0),
    symbols=["BTC/USDT", "ETH/USDT"],
    exchange="gateio",
)

# Static flat MMR (opt-in override)
margin = MarginCalculator(
    ...,
    exchange="gateio",
    use_dynamic_tiers=False,  # Use flat rate from venues.py
)

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if not isinstance(data, list) or len(data) < 4:
        return self._empty_result()

    cash_data, positions_data, prices_data, entry_prices_data = data[0], data[1], data[2], data[3]

    # Extract cash (scalar)
    cash = self._extract_scalar(cash_data)

    # Extract positions (dict: symbol -> qty)
    positions = self._extract_dict(positions_data)

    # Extract current prices (dict: symbol -> price)
    prices = self._extract_dict(prices_data)

    # Extract entry prices (dict: symbol -> entry_price)
    entry_prices = self._extract_dict(entry_prices_data)

    # Calculate equity and maintenance margin
    unrealized_pnl = 0.0
    total_notional = 0.0
    total_maintenance_margin = 0.0

    for symbol in self._symbols:
        qty = positions.get(symbol, 0.0)
        current_price = prices.get(symbol, 0.0)
        entry_price = entry_prices.get(symbol, current_price)

        if abs(qty) > 1e-10 and current_price > 0:
            # Unrealized PnL: (current - entry) * qty
            unrealized_pnl += (current_price - entry_price) * qty
            # Position notional value
            position_notional = abs(qty * current_price)
            total_notional += position_notional

            # Get MMR for this position (dynamic or static)
            if self._use_dynamic_tiers and self._tier_resolver:
                # Convert spot symbol to perpetual format for tier lookup
                ccxt_symbol = self._to_ccxt_perpetual_symbol(symbol)
                mmr = self._tier_resolver.get_mmr(
                    self._exchange,
                    ccxt_symbol,
                    position_notional,
                    self._api_key,
                    self._api_secret,
                )
            else:
                mmr = self._maintenance_margin_rate

            total_maintenance_margin += position_notional * mmr

    equity = cash + unrealized_pnl
    maintenance_margin = total_maintenance_margin

    # Apply liquidation fee to get effective equity (conservative estimate)
    liquidation_fee = total_notional * self._config.liquidation_fee
    effective_equity = equity - liquidation_fee

    # Calculate margin ratio using exchange-specific formula
    if maintenance_margin > 1e-10 and effective_equity > 1e-10:
        if self._config.margin_ratio_inverted:
            # Binance/OKX style: MMR / Equity * 100 (100% = liquidation)
            margin_ratio = (maintenance_margin / effective_equity) * 100.0
            is_liquidatable = 1.0 if margin_ratio >= 100.0 else 0.0
        else:
            # Bybit/Gateio/Kraken style: Equity / MMR (< threshold = liquidation)
            margin_ratio = effective_equity / maintenance_margin
            is_liquidatable = (
                1.0 if margin_ratio < self._config.liquidation_threshold else 0.0
            )
    elif maintenance_margin <= 1e-10:
        # No positions - safe
        margin_ratio = float('inf') if not self._config.margin_ratio_inverted else 0.0
        is_liquidatable = 0.0
    else:
        # No effective equity - liquidatable
        margin_ratio = 0.0 if not self._config.margin_ratio_inverted else float('inf')
        is_liquidatable = 1.0

    return TaggedArray(
        value=np.array([margin_ratio]),
        exists=np.array([True]),
        valid=np.array([equity >= 0]),
        updated=np.array([True]),
    )

_{Source: apps/trading/operators/futures.py}

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG

​Overview

​Quick Reference

​CashBalance

​Parameters

​Usage

​Source Code

​EquityCalculator

​Parameters

​Usage

​Source Code

​TotalEquityCalculator

​Calculate equity for each venue separately

​Aggregate into total equity

​Parameters

​Usage

​Source Code

​AvailableMargin

​Parameters

​Usage

​Source Code

​PositionQuantity

​Parameters

​Usage

​Source Code

​PositionValue

​Parameters

​Usage

​Source Code

​BookSize

​Parameters

​Usage

​Source Code

​MarginCalculator

​Parameters

​Usage

​Source Code

​Related Pages

Operator Protocol

StatefulGraph

Overview

Quick Reference

CashBalance

Parameters

Usage

Source Code

EquityCalculator

Parameters

Usage

Source Code

TotalEquityCalculator

Calculate equity for each venue separately

Aggregate into total equity

Parameters

Usage

Source Code

AvailableMargin

Parameters

Usage

Source Code

PositionQuantity

Parameters

Usage

Source Code

PositionValue

Parameters

Usage

Source Code

BookSize

Parameters

Usage

Source Code

MarginCalculator

Parameters

Usage

Source Code

Related Pages