adapter

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 22, 2026 License: Apache-2.0 Imports: 11 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source
var (
	ErrAccountNotFound = errors.New("account not found in execution router")
	ErrClientNotFound  = errors.New("execution client not found for account")
)

Functions

This section is empty.

Types

type AggTradeConfig

type AggTradeConfig struct{}

AggTradeConfig enables aggregate trade stream subscription (Binance @aggTrade). Presence of the key is enough; no options today. Ignored on venues without aggTrade.

type DataClient

type DataClient interface {
	HasSub() bool
	Connect(ctx context.Context) error
	Disconnect()
	SubscribeDepthUpdate(symbolID int, depthLevel int, pushRateMs int)
	SubscribeTrade(symbolID int)
	SubscribeAggTrade(symbolID int)
	SubscribeKline(symbolID int, interval string)
	ReqDepthSnapshot(symbolID int, limit int) error
	// ReqHistoricalKline fetches historical candles. Times are nanoseconds; 0 omits the bound.
	ReqHistoricalKline(symbolID int, interval string, startTimeNs, endTimeNs uint64, limit int) error
}

DataClient is the interface for exchange-specific market data stream clients. Parameters use primitives so implementations need not import this package.

type DataClientFactory

type DataClientFactory func(cat *catalog.Catalog, bus *msgbus.MsgBus) DataClient

DataClientFactory creates a DataClient for a specific exchange+product.

type DataRouter

type DataRouter struct {
	// contains filtered or unexported fields
}

func NewDataRouter

func NewDataRouter(cat *catalog.Catalog, bus *msgbus.MsgBus) *DataRouter

func (*DataRouter) Connect

func (r *DataRouter) Connect(ctx context.Context) error

func (*DataRouter) Disconnect

func (r *DataRouter) Disconnect()

func (*DataRouter) RegisterFactory

func (r *DataRouter) RegisterFactory(exchangeID, productID int, factory DataClientFactory)

RegisterFactory registers a factory for a given exchange+product pair.

func (*DataRouter) ReqDepthSnapshot

func (r *DataRouter) ReqDepthSnapshot(symbolID int) error

func (*DataRouter) ReqHistoricalKline

func (r *DataRouter) ReqHistoricalKline(symbolID int, interval string, startTimeNs, endTimeNs uint64, limit int) error

ReqHistoricalKline requests historical klines for a symbol. interval is a Binance-style string (e.g. "1m"); start/end are nanoseconds (0 = omit).

func (*DataRouter) SubscribeAggTrade

func (r *DataRouter) SubscribeAggTrade(symbolID int) error

SubscribeAggTrade subscribes to aggregate trade updates for a symbol.

func (*DataRouter) SubscribeDepthUpdate

func (r *DataRouter) SubscribeDepthUpdate(symbolID int, opts *DepthOptions) error

SubscribeDepthUpdate subscribes to depth updates for a symbol with options.

func (*DataRouter) SubscribeKline

func (r *DataRouter) SubscribeKline(symbolID int, opts *KlineOptions) error

SubscribeKline subscribes to kline updates for a symbol with options.

func (*DataRouter) SubscribeTrade

func (r *DataRouter) SubscribeTrade(symbolID int) error

SubscribeTrade subscribes to trade tick updates for a symbol.

type DataRouterEntry

type DataRouterEntry struct {
	Symbol   string          `yaml:"symbol"`             // Universal ticker (required)
	Endpoint string          `yaml:"endpoint,omitempty"` // Regional endpoint: bybit, bybit_tr, bybit_eu
	Depth    *DepthConfig    `yaml:"depth,omitempty"`    // Depth subscription options
	Trade    *TradeConfig    `yaml:"trade,omitempty"`    // Trade tick subscription
	AggTrade *AggTradeConfig `yaml:"aggTrade,omitempty"` // Aggregate trade subscription
	Kline    *KlineConfig    `yaml:"kline,omitempty"`    // Kline subscription options
}

DataRouterEntry is per-symbol data subscription config from YAML.

type DepthConfig

type DepthConfig struct {
	// Type selects the depth stream kind for Binance:
	//   delta (default) → diff depth (@depth@100ms)
	//   depth5|depth10|depth20 → partial book WS (@depthN@100ms → DepthSnapshot)
	Type     string `yaml:"type,omitempty"`
	PushRate string `yaml:"push_rate,omitempty"` // 100ms, 1000ms (binance)
	Levels   int    `yaml:"levels,omitempty"`    // 1, 50, 200, 1000 (bybit)
}

DepthConfig configures depth stream subscription.

type DepthOptions

type DepthOptions struct {
	Type     string // delta, depth5, depth10, depth20 (binance)
	PushRate string // 100ms, 1000ms (binance)
	Levels   int    // 1, 50, 200, 1000 (bybit)
}

DepthOptions contains generic depth subscription options. These are translated to primitive parameters by the router.

type ExecRouterEntry

type ExecRouterEntry struct {
	Account string `yaml:"account"`
	Wallet  string `yaml:"wallet"`
	API     string `yaml:"api"`
}

ExecRouterEntry configures a single execution client connection.

type ExecutionClient

type ExecutionClient interface {
	// Connect establishes the connection for trading
	Connect(ctx context.Context) error

	// Disconnect closes the connection
	Disconnect()

	// SubscribeOrderUpdate subscribes to order status update events
	// Events: OrderAccepted, OrderPartiallyFilled, OrderFilled, OrderCanceled, OrderRejected
	SubscribeOrderUpdate() error

	// SubscribeFill subscribes to execution/fill events
	// Events: Fill (trade execution details including price, quantity, fees)
	SubscribeFill() error

	// SubscribeBalance subscribes to wallet/balance update events
	// Events: BalanceUpdate (available, locked, total for each asset)
	SubscribeBalance() error

	// SubmitOrder submits a new order with the given client order ID
	SubmitOrder(clientOrderID int, symbolID int, side common.Side, orderType common.OrderType, timeInForce common.TimeInForce, price float64, quantity float64) error

	// CancelOrder cancels an order by orderID, with clientOrderID for response correlation
	CancelOrder(symbolID int, orderID int, clientOrderID int) error

	// CancelAllOrders cancels all open orders for a symbol
	CancelAllOrders(symbolID int) error

	// ReqBalanceSnapshot requests the current balance snapshot for the given wallet type.
	// The response will be published as a RespBalanceSnapshot event.
	ReqBalanceSnapshot(walletType common.WalletType) error
}

ExecutionClient is the interface that execution clients must implement. Each execution client handles order operations for a specific account.

Subscription Model: The interface provides granular subscription methods for different private data types. Each exchange handles these differently internally:

  • Bybit: Subscribes to individual topics (order, execution, wallet) as requested
  • Binance: Subscribes to entire user data stream on first subscribe call, then filters events internally based on which subscriptions are active

Unsubscribed events are logged as "unhandled" and not published to the event bus.

type ExecutionRouter

type ExecutionRouter struct {
	// contains filtered or unexported fields
}

ExecutionRouter routes order operations to the appropriate ExecutionClient based on account ID. It manages a collection of execution clients and provides a unified interface for order management.

func NewExecutionRouter

func NewExecutionRouter() *ExecutionRouter

NewExecutionRouter creates a new ExecutionRouter in paper mode by default.

func (*ExecutionRouter) AccountIDs

func (r *ExecutionRouter) AccountIDs() []int

AccountIDs returns the list of registered account IDs

func (*ExecutionRouter) CancelAllOrders

func (r *ExecutionRouter) CancelAllOrders(acctID int, symbolID int) error

CancelAllOrders routes the cancel all orders request to the appropriate client. Paper mode refuses the call before any venue client is touched.

func (*ExecutionRouter) CancelOrder

func (r *ExecutionRouter) CancelOrder(acctID int, symbolID int, orderID int, clientOrderID int) error

CancelOrder routes the order cancellation to the appropriate client. Paper mode refuses the call before any venue client is touched.

func (*ExecutionRouter) ClientCount

func (r *ExecutionRouter) ClientCount() int

ClientCount returns the number of registered execution clients

func (*ExecutionRouter) Connect

func (r *ExecutionRouter) Connect(ctx context.Context) error

Connect connects all registered execution clients Note: Subscriptions must be called separately after Connect

func (*ExecutionRouter) ConnectAccount

func (r *ExecutionRouter) ConnectAccount(ctx context.Context, acctID int) error

ConnectAccount connects a specific account's execution client

func (*ExecutionRouter) Disconnect

func (r *ExecutionRouter) Disconnect()

Disconnect disconnects all registered execution clients

func (*ExecutionRouter) DisconnectAccount

func (r *ExecutionRouter) DisconnectAccount(acctID int)

DisconnectAccount disconnects a specific account's execution client

func (*ExecutionRouter) GetClient

func (r *ExecutionRouter) GetClient(acctID int) (ExecutionClient, error)

GetClient returns the ExecutionClient for a specific account ID

func (*ExecutionRouter) RegisterClient

func (r *ExecutionRouter) RegisterClient(acctID int, client ExecutionClient)

RegisterClient registers an ExecutionClient for a specific account ID

func (*ExecutionRouter) ReqBalanceSnapshot

func (r *ExecutionRouter) ReqBalanceSnapshot(acctID int, walletType common.WalletType) error

ReqBalanceSnapshot requests the current balance snapshot for an account's wallet type

func (*ExecutionRouter) SetTradingMode

func (r *ExecutionRouter) SetTradingMode(mode tradingmode.Mode)

SetTradingMode sets the process trading mode gate for venue order mutations. Call once during Node.Init before Start/Connect.

func (*ExecutionRouter) SubmitOrder

func (r *ExecutionRouter) SubmitOrder(acctID int, clientOrderID int, symbolID int, side common.Side, orderType common.OrderType, timeInForce common.TimeInForce, price float64, quantity float64) error

SubmitOrder routes the order submission to the appropriate client. Paper mode refuses the call before any venue client is touched.

func (*ExecutionRouter) SubscribeBalance

func (r *ExecutionRouter) SubscribeBalance(acctID int) error

SubscribeBalance subscribes to balance update events for an account

func (*ExecutionRouter) SubscribeFill

func (r *ExecutionRouter) SubscribeFill(acctID int) error

SubscribeFill subscribes to fill/execution events for an account

func (*ExecutionRouter) SubscribeOrderUpdate

func (r *ExecutionRouter) SubscribeOrderUpdate(acctID int) error

SubscribeOrderUpdate subscribes to order update events for an account

func (*ExecutionRouter) TradingMode

func (r *ExecutionRouter) TradingMode() tradingmode.Mode

TradingMode returns the active trading mode gate.

func (*ExecutionRouter) UnregisterClient

func (r *ExecutionRouter) UnregisterClient(acctID int)

UnregisterClient removes an ExecutionClient for a specific account ID

type KlineConfig

type KlineConfig struct {
	Interval string `yaml:"interval,omitempty"` // 1m, 5m, 1h, 1d, ... (Binance form; Bybit mapped)
}

KlineConfig configures kline / candlestick stream subscription.

type KlineOptions

type KlineOptions struct {
	Interval string // 1m, 5m, 1h, 1d, ...
}

KlineOptions contains generic kline subscription options.

type RouterError

type RouterError struct {
	AccountID int
	Err       error
}

RouterError represents an error that occurred for a specific account

func (*RouterError) Error

func (e *RouterError) Error() string

func (*RouterError) Unwrap

func (e *RouterError) Unwrap() error

type TradeConfig

type TradeConfig struct{}

TradeConfig enables trade tick stream subscription. Presence of the key is enough; no options today.

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL