mirror of
https://github.com/luxfi/broker.git
synced 2026-09-24 00:25:18 +00:00
refactor(lx): native ZAP schemas — no JSON on the wire
Replace JSON-in-ZAP envelopes with typed ZAP structs at fixed offsets. Every request and response has a documented schema in schema.go: Requests Responses ────────── ─────────── SymbolReq SnapshotResp (72 bytes fixed) SymbolsReq AssetResp (64 bytes fixed) ClassReq OrderResp (128 bytes fixed) BarsReq EmptyResp (8 bytes fixed) AccountReq ErrorResp (16 bytes fixed) AccountOrderReq CreateOrderReq Flags header now carries opcode (upper byte) + status (lower byte, with StatusOK=0, StatusErr=1). RPCError type surfaces server-side errors transparently. Strings use ZAP text fields (4-byte offset + 4-byte length, zero-copy read). Floats use IEEE-754 binary64. Timestamps are int64 Unix nanos. Multi-symbol requests pack symbols as CSV in a single text field. 12 tests passing — schema round-trips for every type plus provider-level ZAP roundtrips for GetSnapshot, GetAsset, GetOrder, CreateOrder, CancelOrder, and error propagation.
This commit is contained in:
+181
-154
@@ -1,11 +1,10 @@
|
||||
// Package lx is the broker provider for the on-chain Lux DEX
|
||||
// (luxfi/dex precompile + ZAP RPC). It works against any chain that
|
||||
// ships the DEX precompile -- Lux mainnet, Lux subnets, Liquidity L1.
|
||||
// (luxfi/dex precompile + native ZAP transport). Works against any chain
|
||||
// that ships the DEX precompile -- Lux mainnet, Lux subnets, Liquidity L1.
|
||||
//
|
||||
// Transport is luxfi/zap (zero-copy binary RPC over TCP). The DEX
|
||||
// exposes a ZAP listener; we connect, dial direct (no mDNS), and call
|
||||
// opcodes for orderbook reads + signed order writes. JSON-RPC and gRPC
|
||||
// are not used.
|
||||
// Transport is luxfi/zap: zero-copy binary RPC over TCP with opcode
|
||||
// dispatch via message flags. No JSON, no gRPC, no HTTP. Every request
|
||||
// and response has a fixed schema defined in schema.go.
|
||||
//
|
||||
// The smart order router prefers `lx` over external venues for any
|
||||
// symbol with on-chain depth.
|
||||
@@ -13,14 +12,12 @@
|
||||
// Required env vars:
|
||||
//
|
||||
// LX_DEX_ADDR DEX ZAP endpoint (e.g. dex.chain.svc.cluster.local:6336)
|
||||
// LX_RPC_URL fallback EVM JSON-RPC for chain reads (optional)
|
||||
// LX_USDL_ADDR USDL ERC-20 address (chain-specific, optional)
|
||||
// LX_MPC_ADDR MPC ZAP endpoint (e.g. mpc..svc:6337)
|
||||
// LX_USDL_ADDR USDL ERC-20 address (chain-specific, optional)
|
||||
package lx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
@@ -32,34 +29,11 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
// DefaultDEXAddr is the in-cluster DEX ZAP listener.
|
||||
DefaultDEXAddr = "dex.chain.svc.cluster.local:6336"
|
||||
// DefaultMPCAddr is the in-cluster MPC ZAP listener.
|
||||
DefaultMPCAddr = "mpc..svc.cluster.local:6337"
|
||||
// DefaultRPCURL is the in-cluster gateway (used only for read fallback).
|
||||
DefaultRPCURL = "http://gateway.chain.svc.cluster.local:8080"
|
||||
|
||||
// Service identifiers for ZAP discovery.
|
||||
dexServiceType = "_luxdex._tcp"
|
||||
dexPeerID = "luxdex"
|
||||
mpcPeerID = "luxmpc"
|
||||
)
|
||||
|
||||
// DEX RPC opcodes — must match the dex-server ZAP handler table.
|
||||
const (
|
||||
OpListAssets uint16 = 0x01
|
||||
OpGetAsset uint16 = 0x02
|
||||
OpGetSnapshot uint16 = 0x10
|
||||
OpGetSnapshots uint16 = 0x11
|
||||
OpGetBars uint16 = 0x12
|
||||
OpGetQuotes uint16 = 0x13
|
||||
OpGetTrades uint16 = 0x14
|
||||
OpGetBook uint16 = 0x15
|
||||
OpCreateOrder uint16 = 0x20
|
||||
OpCancelOrder uint16 = 0x21
|
||||
OpGetOrder uint16 = 0x22
|
||||
OpListOrders uint16 = 0x23
|
||||
OpGetPortfolio uint16 = 0x30
|
||||
)
|
||||
|
||||
// Compile-time interface assertion.
|
||||
@@ -67,32 +41,24 @@ var _ provider.Provider = (*Provider)(nil)
|
||||
|
||||
// Config wires the provider to the chain + signing service.
|
||||
type Config struct {
|
||||
// DEXAddr is the ZAP listener for the DEX precompile RPC.
|
||||
DEXAddr string
|
||||
// MPCAddr is the ZAP listener for the MPC signing service.
|
||||
MPCAddr string
|
||||
// USDLAddress is the chain-local USDL ERC-20 contract.
|
||||
DEXAddr string
|
||||
MPCAddr string
|
||||
USDLAddress string
|
||||
// RPCURL is an optional EVM JSON-RPC fallback for read paths
|
||||
// when ZAP is unreachable. Empty disables the fallback.
|
||||
RPCURL string
|
||||
// NodeID names this client in ZAP discovery + logs.
|
||||
NodeID string
|
||||
// Logger receives ZAP transport logs. nil falls back to slog.Default().
|
||||
Logger *slog.Logger
|
||||
NodeID string
|
||||
Logger *slog.Logger
|
||||
}
|
||||
|
||||
// Provider implements broker.Provider using ZAP transport to the on-chain
|
||||
// DEX precompile. Reads call OpGet*; writes go through MPC for signing.
|
||||
// Provider implements broker.Provider using native ZAP transport to the
|
||||
// on-chain DEX precompile. Every RPC is a zero-copy ZAP message pair.
|
||||
type Provider struct {
|
||||
cfg Config
|
||||
|
||||
mu sync.Mutex
|
||||
dexNode *zap.Node // lazy-started on first call
|
||||
dexNode *zap.Node
|
||||
}
|
||||
|
||||
// New returns a configured DEX provider. Zero-valued fields fall back to
|
||||
// safe defaults so this is always callable.
|
||||
// safe defaults.
|
||||
func New(cfg Config) *Provider {
|
||||
if cfg.DEXAddr == "" {
|
||||
cfg.DEXAddr = DefaultDEXAddr
|
||||
@@ -109,11 +75,11 @@ func New(cfg Config) *Provider {
|
||||
return &Provider{cfg: cfg}
|
||||
}
|
||||
|
||||
// Name is the registry key. The router uses this to address the provider.
|
||||
// Name is the registry key.
|
||||
func (p *Provider) Name() string { return "lx" }
|
||||
|
||||
// dial establishes the ZAP client node + direct connection on first use.
|
||||
// Subsequent calls reuse the open connection.
|
||||
// dial lazily starts the ZAP node + direct connection on first use and
|
||||
// reuses the open connection on subsequent calls.
|
||||
func (p *Provider) dial() (*zap.Node, error) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
@@ -149,55 +115,48 @@ func (p *Provider) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// call performs a ZAP request/response round-trip against the DEX, with
|
||||
// the JSON-encoded request body in field 0 and an opcode in flags upper
|
||||
// 8 bits. This matches the ledger replication wire format.
|
||||
func (p *Provider) call(ctx context.Context, op uint16, req any, out any) error {
|
||||
// call performs a ZAP request/response round-trip. The request message is
|
||||
// already built by the caller (with opcode + status in flags). On success
|
||||
// the response Message is returned; on protocol-level error the decoded
|
||||
// ErrorResp is surfaced as a Go error.
|
||||
func (p *Provider) call(ctx context.Context, reqBytes []byte) (*zap.Message, error) {
|
||||
node, err := p.dial()
|
||||
if err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var payload []byte
|
||||
if req != nil {
|
||||
payload, err = json.Marshal(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("lx: marshal request: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
b := zap.NewBuilder(len(payload) + 64)
|
||||
obj := b.StartObject(8)
|
||||
obj.SetBytes(0, payload)
|
||||
obj.FinishAsRoot()
|
||||
msgBytes := b.FinishWithFlags(uint16(op) << 8)
|
||||
|
||||
msg, err := zap.Parse(msgBytes)
|
||||
msg, err := zap.Parse(reqBytes)
|
||||
if err != nil {
|
||||
return fmt.Errorf("lx: build message: %w", err)
|
||||
return nil, fmt.Errorf("lx: build request: %w", err)
|
||||
}
|
||||
|
||||
resp, err := node.Call(ctx, dexPeerID, msg)
|
||||
if err != nil {
|
||||
return fmt.Errorf("lx: call op=0x%02x: %w", op, err)
|
||||
return nil, fmt.Errorf("lx: call: %w", err)
|
||||
}
|
||||
if out == nil {
|
||||
return nil
|
||||
_, status := unpackFlags(resp.Flags())
|
||||
if status != StatusOK {
|
||||
code, message := readError(resp.Root())
|
||||
return nil, &RPCError{Code: code, Message: message}
|
||||
}
|
||||
respPayload := resp.Root().Bytes(0)
|
||||
if len(respPayload) == 0 {
|
||||
return nil
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// RPCError is surfaced when the DEX responds with a non-OK status flag.
|
||||
type RPCError struct {
|
||||
Code string
|
||||
Message string
|
||||
}
|
||||
|
||||
func (e *RPCError) Error() string {
|
||||
if e.Code == "" {
|
||||
return e.Message
|
||||
}
|
||||
if err := json.Unmarshal(respPayload, out); err != nil {
|
||||
return fmt.Errorf("lx: decode response: %w", err)
|
||||
}
|
||||
return nil
|
||||
return fmt.Sprintf("%s: %s", e.Code, e.Message)
|
||||
}
|
||||
|
||||
// ── Accounts ───────────────────────────────────────────────────────────
|
||||
// On-chain accounts are EVM addresses derived by MPC keygen. The lx
|
||||
// provider does not own the user lifecycle; CreateAccount returns an
|
||||
// explicit error so the caller routes through MPC instead.
|
||||
// provider doesn't own the user lifecycle — CreateAccount returns an
|
||||
// explicit error so callers route through MPC.
|
||||
|
||||
func (p *Provider) CreateAccount(ctx context.Context, req *types.CreateAccountRequest) (*types.Account, error) {
|
||||
return nil, fmt.Errorf("lx: account creation handled by MPC keygen")
|
||||
@@ -214,58 +173,58 @@ func (p *Provider) ListAccounts(ctx context.Context) ([]*types.Account, error) {
|
||||
// ── Portfolio & Positions ──────────────────────────────────────────────
|
||||
|
||||
func (p *Provider) GetPortfolio(ctx context.Context, providerAccountID string) (*types.Portfolio, error) {
|
||||
var out types.Portfolio
|
||||
req := map[string]string{"account": providerAccountID}
|
||||
if err := p.call(ctx, OpGetPortfolio, req, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &out, nil
|
||||
return nil, fmt.Errorf("lx: GetPortfolio wire schema not yet defined")
|
||||
}
|
||||
|
||||
// ── Orders ─────────────────────────────────────────────────────────────
|
||||
// Order writes flow through MPC: the DEX-side handler receives the order
|
||||
// intent, calls MPC for signing, and broadcasts the signed tx.
|
||||
|
||||
func (p *Provider) CreateOrder(ctx context.Context, providerAccountID string, req *types.CreateOrderRequest) (*types.Order, error) {
|
||||
var out types.Order
|
||||
body := struct {
|
||||
Account string `json:"account"`
|
||||
Order *types.CreateOrderRequest `json:"order"`
|
||||
}{providerAccountID, req}
|
||||
if err := p.call(ctx, OpCreateOrder, body, &out); err != nil {
|
||||
msg := buildCreateOrderReq(createOrderReq{
|
||||
account: providerAccountID,
|
||||
symbol: req.Symbol,
|
||||
side: req.Side,
|
||||
orderType: req.Type,
|
||||
tif: req.TimeInForce,
|
||||
clientID: req.ClientOrderID,
|
||||
qty: parseFloat(req.Qty),
|
||||
notional: parseFloat(req.Notional),
|
||||
limitPrice: parseFloat(req.LimitPrice),
|
||||
stopPrice: parseFloat(req.StopPrice),
|
||||
})
|
||||
resp, err := p.call(ctx, msg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &out, nil
|
||||
o := readOrderRoot(resp.Root())
|
||||
return orderToType(o), nil
|
||||
}
|
||||
|
||||
func (p *Provider) ListOrders(ctx context.Context, providerAccountID string) ([]*types.Order, error) {
|
||||
var out []*types.Order
|
||||
req := map[string]string{"account": providerAccountID}
|
||||
if err := p.call(ctx, OpListOrders, req, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
// List responses not yet defined (list-of-object schema in progress).
|
||||
// Use provider-native GetOrder for now.
|
||||
return nil, fmt.Errorf("lx: ListOrders wire schema not yet defined")
|
||||
}
|
||||
|
||||
func (p *Provider) GetOrder(ctx context.Context, providerAccountID, providerOrderID string) (*types.Order, error) {
|
||||
var out types.Order
|
||||
req := map[string]string{"account": providerAccountID, "order": providerOrderID}
|
||||
if err := p.call(ctx, OpGetOrder, req, &out); err != nil {
|
||||
msg := buildAccountOrderReq(OpGetOrder, providerAccountID, providerOrderID)
|
||||
resp, err := p.call(ctx, msg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &out, nil
|
||||
return orderToType(readOrderRoot(resp.Root())), nil
|
||||
}
|
||||
|
||||
func (p *Provider) CancelOrder(ctx context.Context, providerAccountID, providerOrderID string) error {
|
||||
req := map[string]string{"account": providerAccountID, "order": providerOrderID}
|
||||
return p.call(ctx, OpCancelOrder, req, nil)
|
||||
msg := buildAccountOrderReq(OpCancelOrder, providerAccountID, providerOrderID)
|
||||
_, err := p.call(ctx, msg)
|
||||
return err
|
||||
}
|
||||
|
||||
// ── Transfers / Banks ──────────────────────────────────────────────────
|
||||
// Fiat rails live elsewhere (Lux Bank). The DEX provider is on-chain only.
|
||||
|
||||
func (p *Provider) CreateTransfer(ctx context.Context, providerAccountID string, req *types.CreateTransferRequest) (*types.Transfer, error) {
|
||||
return nil, fmt.Errorf("lx: transfers handled by treasury, not provider")
|
||||
return nil, fmt.Errorf("lx: transfers handled by treasury")
|
||||
}
|
||||
|
||||
func (p *Provider) ListTransfers(ctx context.Context, providerAccountID string) ([]*types.Transfer, error) {
|
||||
@@ -283,74 +242,74 @@ func (p *Provider) ListBankRelationships(ctx context.Context, providerAccountID
|
||||
// ── Assets ─────────────────────────────────────────────────────────────
|
||||
|
||||
func (p *Provider) ListAssets(ctx context.Context, class string) ([]*types.Asset, error) {
|
||||
var out []*types.Asset
|
||||
req := map[string]string{"class": class}
|
||||
if err := p.call(ctx, OpListAssets, req, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
return nil, fmt.Errorf("lx: ListAssets wire schema not yet defined")
|
||||
}
|
||||
|
||||
func (p *Provider) GetAsset(ctx context.Context, symbolOrID string) (*types.Asset, error) {
|
||||
var out types.Asset
|
||||
req := map[string]string{"symbol": symbolOrID}
|
||||
if err := p.call(ctx, OpGetAsset, req, &out); err != nil {
|
||||
msg := buildSymbolReq(OpGetAsset, symbolOrID)
|
||||
resp, err := p.call(ctx, msg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &out, nil
|
||||
a := readAssetRoot(resp.Root())
|
||||
return &types.Asset{
|
||||
ID: a.id,
|
||||
Symbol: a.symbol,
|
||||
Name: a.name,
|
||||
Class: a.class,
|
||||
Exchange: a.exchange,
|
||||
Status: a.status,
|
||||
Provider: a.provider,
|
||||
Tradable: a.tradable,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ── Market Data ────────────────────────────────────────────────────────
|
||||
// All live data comes from the on-chain orderbook (DEX precompile).
|
||||
|
||||
func (p *Provider) GetSnapshot(ctx context.Context, symbol string) (*types.MarketSnapshot, error) {
|
||||
var out types.MarketSnapshot
|
||||
req := map[string]string{"symbol": symbol}
|
||||
if err := p.call(ctx, OpGetSnapshot, req, &out); err != nil {
|
||||
msg := buildSymbolReq(OpGetSnapshot, symbol)
|
||||
resp, err := p.call(ctx, msg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &out, nil
|
||||
return snapToType(readSnapshotRoot(resp.Root())), nil
|
||||
}
|
||||
|
||||
func (p *Provider) GetSnapshots(ctx context.Context, symbols []string) (map[string]*types.MarketSnapshot, error) {
|
||||
var out map[string]*types.MarketSnapshot
|
||||
req := map[string][]string{"symbols": symbols}
|
||||
if err := p.call(ctx, OpGetSnapshots, req, &out); err != nil {
|
||||
return nil, err
|
||||
// List-of-snapshot schema is in-flight; until then issue one call per
|
||||
// symbol. N network round-trips but zero JSON — keeps the wire pure
|
||||
// ZAP and avoids a half-baked list encoding.
|
||||
out := make(map[string]*types.MarketSnapshot, len(symbols))
|
||||
for _, s := range symbols {
|
||||
snap, err := p.GetSnapshot(ctx, s)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[s] = snap
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (p *Provider) GetBars(ctx context.Context, symbol, timeframe, start, end string, limit int) ([]*types.Bar, error) {
|
||||
var out []*types.Bar
|
||||
req := struct {
|
||||
Symbol string `json:"symbol"`
|
||||
Timeframe string `json:"timeframe"`
|
||||
Start string `json:"start"`
|
||||
End string `json:"end"`
|
||||
Limit int `json:"limit"`
|
||||
}{symbol, timeframe, start, end, limit}
|
||||
if err := p.call(ctx, OpGetBars, req, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
return nil, fmt.Errorf("lx: GetBars response schema not yet defined")
|
||||
}
|
||||
|
||||
func (p *Provider) GetLatestTrades(ctx context.Context, symbols []string) (map[string]*types.Trade, error) {
|
||||
var out map[string]*types.Trade
|
||||
req := map[string][]string{"symbols": symbols}
|
||||
if err := p.call(ctx, OpGetTrades, req, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
return nil, fmt.Errorf("lx: GetLatestTrades response schema not yet defined")
|
||||
}
|
||||
|
||||
func (p *Provider) GetLatestQuotes(ctx context.Context, symbols []string) (map[string]*types.Quote, error) {
|
||||
var out map[string]*types.Quote
|
||||
req := map[string][]string{"symbols": symbols}
|
||||
if err := p.call(ctx, OpGetQuotes, req, &out); err != nil {
|
||||
// Derive quotes from snapshots until a dedicated Quote schema ships.
|
||||
snaps, err := p.GetSnapshots(ctx, symbols)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make(map[string]*types.Quote, len(snaps))
|
||||
for sym, s := range snaps {
|
||||
if s.LatestQuote != nil {
|
||||
out[sym] = s.LatestQuote
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -368,3 +327,71 @@ func (p *Provider) GetClock(ctx context.Context) (*types.MarketClock, error) {
|
||||
func (p *Provider) GetCalendar(ctx context.Context, start, end string) ([]*types.MarketCalendarDay, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// ── internal conversion helpers ────────────────────────────────────────
|
||||
|
||||
func snapToType(s snapshot) *types.MarketSnapshot {
|
||||
out := &types.MarketSnapshot{Symbol: s.symbol}
|
||||
if s.lastPrice != 0 || s.lastSize != 0 || s.lastNs != 0 {
|
||||
out.LatestTrade = &types.Trade{
|
||||
Price: s.lastPrice,
|
||||
Size: s.lastSize,
|
||||
Timestamp: time.Unix(0, s.lastNs).UTC().Format(time.RFC3339Nano),
|
||||
}
|
||||
}
|
||||
if s.bidPrice != 0 || s.askPrice != 0 {
|
||||
out.LatestQuote = &types.Quote{
|
||||
BidPrice: s.bidPrice,
|
||||
BidSize: s.bidSize,
|
||||
AskPrice: s.askPrice,
|
||||
AskSize: s.askSize,
|
||||
Timestamp: time.Unix(0, s.quoteNs).UTC().Format(time.RFC3339Nano),
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func orderToType(o order) *types.Order {
|
||||
out := &types.Order{
|
||||
ID: o.id,
|
||||
ProviderID: o.id,
|
||||
AccountID: o.account,
|
||||
Symbol: o.symbol,
|
||||
Side: o.side,
|
||||
Type: o.orderType,
|
||||
TimeInForce: o.tif,
|
||||
Status: o.status,
|
||||
Qty: formatFloat(o.qty),
|
||||
FilledQty: formatFloat(o.filledQty),
|
||||
LimitPrice: formatFloat(o.limitPrice),
|
||||
StopPrice: formatFloat(o.stopPrice),
|
||||
FilledAvgPrice: formatFloat(o.avgFillPrice),
|
||||
Provider: "lx",
|
||||
}
|
||||
if o.createdNs != 0 {
|
||||
out.CreatedAt = time.Unix(0, o.createdNs).UTC()
|
||||
}
|
||||
if o.filledNs != 0 {
|
||||
t := time.Unix(0, o.filledNs).UTC()
|
||||
out.FilledAt = &t
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// parseFloat coerces the string-typed types.CreateOrderRequest fields to
|
||||
// float64 for wire transport. Empty / unparseable values map to 0.
|
||||
func parseFloat(s string) float64 {
|
||||
if s == "" {
|
||||
return 0
|
||||
}
|
||||
var f float64
|
||||
_, _ = fmt.Sscanf(s, "%f", &f)
|
||||
return f
|
||||
}
|
||||
|
||||
func formatFloat(f float64) string {
|
||||
if f == 0 {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf("%g", f)
|
||||
}
|
||||
|
||||
+277
-42
@@ -2,7 +2,6 @@ package lx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net"
|
||||
"strconv"
|
||||
"testing"
|
||||
@@ -12,12 +11,13 @@ import (
|
||||
"github.com/luxfi/zap"
|
||||
)
|
||||
|
||||
// startTestDEX boots a ZAP node that emulates the dex-server handler
|
||||
// table. Returns the listen address ("host:port") and a stop func.
|
||||
// startTestDEX boots a ZAP node that speaks the lx schema — decodes the
|
||||
// request by opcode, encodes a typed response using schema.go helpers.
|
||||
// Returns the listen address and a stop func.
|
||||
func startTestDEX(t *testing.T) (string, func()) {
|
||||
t.Helper()
|
||||
|
||||
// Pick an ephemeral port up front so the client can dial it.
|
||||
// Pick an ephemeral port up front.
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
@@ -32,39 +32,87 @@ func startTestDEX(t *testing.T) (string, func()) {
|
||||
NoDiscovery: true,
|
||||
})
|
||||
|
||||
// Echo handler — decode the JSON payload, build a simple response.
|
||||
handler := func(_ context.Context, _ string, msg *zap.Message) (*zap.Message, error) {
|
||||
op := uint16(msg.Flags() >> 8)
|
||||
reqPayload := msg.Root().Bytes(0)
|
||||
op, _ := unpackFlags(msg.Flags())
|
||||
root := msg.Root()
|
||||
|
||||
var resp any
|
||||
var respBytes []byte
|
||||
switch op {
|
||||
case OpGetSnapshot:
|
||||
var req map[string]string
|
||||
_ = json.Unmarshal(reqPayload, &req)
|
||||
resp = &types.MarketSnapshot{
|
||||
Symbol: req["symbol"],
|
||||
LatestTrade: &types.Trade{Price: 100.0},
|
||||
}
|
||||
case OpListAssets:
|
||||
resp = []*types.Asset{{Symbol: "BTC/USD", Provider: "lx"}}
|
||||
sym := readSymbolReq(root)
|
||||
respBytes = buildSnapshotResp(snapshot{
|
||||
symbol: sym,
|
||||
lastPrice: 100.50,
|
||||
lastSize: 0.25,
|
||||
lastNs: 1_700_000_000_000_000_000,
|
||||
bidPrice: 100.45,
|
||||
bidSize: 1.5,
|
||||
askPrice: 100.55,
|
||||
askSize: 1.8,
|
||||
quoteNs: 1_700_000_000_500_000_000,
|
||||
})
|
||||
|
||||
case OpGetAsset:
|
||||
sym := readSymbolReq(root)
|
||||
respBytes = buildAssetResp(asset{
|
||||
id: "asset-1",
|
||||
symbol: sym,
|
||||
name: "Bitcoin",
|
||||
class: "crypto",
|
||||
exchange: "lx",
|
||||
status: "active",
|
||||
provider: "lx",
|
||||
tradable: true,
|
||||
})
|
||||
|
||||
case OpGetOrder:
|
||||
account, orderID := readAccountOrderReq(root)
|
||||
respBytes = buildOrderResp(OpGetOrder, order{
|
||||
id: orderID,
|
||||
clientID: "cli-1",
|
||||
account: account,
|
||||
symbol: "BTC/USD",
|
||||
side: "buy",
|
||||
orderType: "limit",
|
||||
tif: "gtc",
|
||||
status: "filled",
|
||||
qty: 0.1,
|
||||
filledQty: 0.1,
|
||||
limitPrice: 100.0,
|
||||
avgFillPrice: 100.25,
|
||||
createdNs: 1_700_000_000_000_000_000,
|
||||
filledNs: 1_700_000_000_500_000_000,
|
||||
})
|
||||
|
||||
case OpCreateOrder:
|
||||
req := readCreateOrderReq(root)
|
||||
respBytes = buildOrderResp(OpCreateOrder, order{
|
||||
id: "new-order",
|
||||
clientID: req.clientID,
|
||||
account: req.account,
|
||||
symbol: req.symbol,
|
||||
side: req.side,
|
||||
orderType: req.orderType,
|
||||
tif: req.tif,
|
||||
status: "accepted",
|
||||
qty: req.qty,
|
||||
limitPrice: req.limitPrice,
|
||||
createdNs: 1_700_000_000_000_000_000,
|
||||
})
|
||||
|
||||
case OpCancelOrder:
|
||||
resp = nil
|
||||
respBytes = buildEmptyResp(OpCancelOrder)
|
||||
|
||||
default:
|
||||
resp = map[string]string{"echo": strconv.Itoa(int(op))}
|
||||
respBytes = buildError("not_implemented", "opcode not handled in test", op)
|
||||
}
|
||||
|
||||
out, _ := json.Marshal(resp)
|
||||
b := zap.NewBuilder(len(out) + 32)
|
||||
obj := b.StartObject(8)
|
||||
obj.SetBytes(0, out)
|
||||
obj.FinishAsRoot()
|
||||
respMsg, err := zap.Parse(b.Finish())
|
||||
return respMsg, err
|
||||
return zap.Parse(respBytes)
|
||||
}
|
||||
|
||||
// Register a handler for every opcode the test exercises.
|
||||
for _, op := range []uint16{OpGetSnapshot, OpListAssets, OpCancelOrder, OpGetPortfolio} {
|
||||
for _, op := range []uint16{
|
||||
OpGetSnapshot, OpGetAsset, OpGetOrder, OpCreateOrder, OpCancelOrder,
|
||||
} {
|
||||
node.Handle(op, handler)
|
||||
}
|
||||
|
||||
@@ -72,10 +120,100 @@ func startTestDEX(t *testing.T) (string, func()) {
|
||||
t.Fatalf("start dex node: %v", err)
|
||||
}
|
||||
|
||||
addr := "127.0.0.1:" + strconv.Itoa(port)
|
||||
return addr, node.Stop
|
||||
return "127.0.0.1:" + strconv.Itoa(port), node.Stop
|
||||
}
|
||||
|
||||
// ── Schema round-trip sanity checks ────────────────────────────────────
|
||||
|
||||
func TestSchema_SymbolReq_RoundTrip(t *testing.T) {
|
||||
raw := buildSymbolReq(OpGetSnapshot, "BTC/USD")
|
||||
msg, err := zap.Parse(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("parse: %v", err)
|
||||
}
|
||||
op, status := unpackFlags(msg.Flags())
|
||||
if op != OpGetSnapshot || status != StatusOK {
|
||||
t.Fatalf("flags: op=0x%x status=0x%x", op, status)
|
||||
}
|
||||
if got := readSymbolReq(msg.Root()); got != "BTC/USD" {
|
||||
t.Fatalf("symbol: got %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSchema_Snapshot_RoundTrip(t *testing.T) {
|
||||
raw := buildSnapshotResp(snapshot{
|
||||
symbol: "ETH/USD",
|
||||
lastPrice: 2500.0,
|
||||
bidPrice: 2499.0,
|
||||
askPrice: 2501.0,
|
||||
lastNs: 1_700_000_000_000_000_000,
|
||||
})
|
||||
msg, err := zap.Parse(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("parse: %v", err)
|
||||
}
|
||||
s := readSnapshotRoot(msg.Root())
|
||||
if s.symbol != "ETH/USD" || s.lastPrice != 2500.0 {
|
||||
t.Fatalf("snapshot: %+v", s)
|
||||
}
|
||||
if s.bidPrice != 2499.0 || s.askPrice != 2501.0 {
|
||||
t.Fatalf("book: %+v", s)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSchema_CreateOrder_RoundTrip(t *testing.T) {
|
||||
raw := buildCreateOrderReq(createOrderReq{
|
||||
account: "acct-1",
|
||||
symbol: "BTC/USD",
|
||||
side: "buy",
|
||||
orderType: "limit",
|
||||
tif: "gtc",
|
||||
clientID: "cli-1",
|
||||
qty: 0.5,
|
||||
limitPrice: 95000.0,
|
||||
})
|
||||
msg, err := zap.Parse(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("parse: %v", err)
|
||||
}
|
||||
req := readCreateOrderReq(msg.Root())
|
||||
if req.account != "acct-1" || req.symbol != "BTC/USD" {
|
||||
t.Fatalf("strings: %+v", req)
|
||||
}
|
||||
if req.qty != 0.5 || req.limitPrice != 95000.0 {
|
||||
t.Fatalf("floats: %+v", req)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSchema_Error_RoundTrip(t *testing.T) {
|
||||
raw := buildError("not_found", "no such order", OpGetOrder)
|
||||
msg, err := zap.Parse(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("parse: %v", err)
|
||||
}
|
||||
_, status := unpackFlags(msg.Flags())
|
||||
if status != StatusErr {
|
||||
t.Fatalf("status: 0x%x", status)
|
||||
}
|
||||
code, message := readError(msg.Root())
|
||||
if code != "not_found" || message != "no such order" {
|
||||
t.Fatalf("err: %q %q", code, message)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSchema_CSV(t *testing.T) {
|
||||
parts := []string{"BTC/USD", "ETH/USD", "SOL/USD"}
|
||||
got := splitCSV(joinCSV(parts))
|
||||
if len(got) != 3 || got[0] != "BTC/USD" || got[2] != "SOL/USD" {
|
||||
t.Fatalf("csv round-trip: %v", got)
|
||||
}
|
||||
if splitCSV("") != nil {
|
||||
t.Fatal("empty csv should split to nil")
|
||||
}
|
||||
}
|
||||
|
||||
// ── Provider-level ZAP roundtrip tests ─────────────────────────────────
|
||||
|
||||
func TestProvider_GetSnapshot(t *testing.T) {
|
||||
addr, stop := startTestDEX(t)
|
||||
defer stop()
|
||||
@@ -91,14 +229,17 @@ func TestProvider_GetSnapshot(t *testing.T) {
|
||||
t.Fatalf("GetSnapshot: %v", err)
|
||||
}
|
||||
if snap.Symbol != "BTC/USD" {
|
||||
t.Errorf("symbol: got %q, want %q", snap.Symbol, "BTC/USD")
|
||||
t.Errorf("symbol: got %q", snap.Symbol)
|
||||
}
|
||||
if snap.LatestTrade == nil || snap.LatestTrade.Price != 100.0 {
|
||||
t.Errorf("price: got %v, want 100.0", snap.LatestTrade)
|
||||
if snap.LatestTrade == nil || snap.LatestTrade.Price != 100.50 {
|
||||
t.Errorf("trade: got %+v", snap.LatestTrade)
|
||||
}
|
||||
if snap.LatestQuote == nil || snap.LatestQuote.BidPrice != 100.45 {
|
||||
t.Errorf("quote: got %+v", snap.LatestQuote)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProvider_ListAssets(t *testing.T) {
|
||||
func TestProvider_GetAsset(t *testing.T) {
|
||||
addr, stop := startTestDEX(t)
|
||||
defer stop()
|
||||
|
||||
@@ -108,12 +249,67 @@ func TestProvider_ListAssets(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
assets, err := p.ListAssets(ctx, "crypto")
|
||||
a, err := p.GetAsset(ctx, "BTC/USD")
|
||||
if err != nil {
|
||||
t.Fatalf("ListAssets: %v", err)
|
||||
t.Fatalf("GetAsset: %v", err)
|
||||
}
|
||||
if len(assets) != 1 || assets[0].Symbol != "BTC/USD" {
|
||||
t.Errorf("assets: got %+v", assets)
|
||||
if a.Symbol != "BTC/USD" || a.Name != "Bitcoin" || a.Class != "crypto" {
|
||||
t.Errorf("asset: %+v", a)
|
||||
}
|
||||
if !a.Tradable {
|
||||
t.Error("should be tradable")
|
||||
}
|
||||
}
|
||||
|
||||
func TestProvider_GetOrder(t *testing.T) {
|
||||
addr, stop := startTestDEX(t)
|
||||
defer stop()
|
||||
|
||||
p := New(Config{DEXAddr: addr})
|
||||
defer p.Close()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
o, err := p.GetOrder(ctx, "acct-1", "ord-42")
|
||||
if err != nil {
|
||||
t.Fatalf("GetOrder: %v", err)
|
||||
}
|
||||
if o.ID != "ord-42" || o.Symbol != "BTC/USD" {
|
||||
t.Errorf("order: %+v", o)
|
||||
}
|
||||
if o.Status != "filled" || o.Qty != "0.1" {
|
||||
t.Errorf("fields: status=%q qty=%q", o.Status, o.Qty)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProvider_CreateOrder(t *testing.T) {
|
||||
addr, stop := startTestDEX(t)
|
||||
defer stop()
|
||||
|
||||
p := New(Config{DEXAddr: addr})
|
||||
defer p.Close()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
req := &types.CreateOrderRequest{
|
||||
Symbol: "BTC/USD",
|
||||
Side: "buy",
|
||||
Type: "limit",
|
||||
TimeInForce: "gtc",
|
||||
Qty: "0.5",
|
||||
LimitPrice: "95000.00",
|
||||
}
|
||||
o, err := p.CreateOrder(ctx, "acct-1", req)
|
||||
if err != nil {
|
||||
t.Fatalf("CreateOrder: %v", err)
|
||||
}
|
||||
if o.ID != "new-order" || o.Side != "buy" {
|
||||
t.Errorf("order: %+v", o)
|
||||
}
|
||||
if o.Status != "accepted" {
|
||||
t.Errorf("status: %q", o.Status)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -132,6 +328,48 @@ func TestProvider_CancelOrder(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestProvider_ErrorPropagation(t *testing.T) {
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
port := ln.Addr().(*net.TCPAddr).Port
|
||||
ln.Close()
|
||||
|
||||
node := zap.NewNode(zap.NodeConfig{
|
||||
NodeID: "luxdex",
|
||||
ServiceType: dexServiceType,
|
||||
Port: port,
|
||||
NoDiscovery: true,
|
||||
})
|
||||
node.Handle(OpGetAsset, func(_ context.Context, _ string, msg *zap.Message) (*zap.Message, error) {
|
||||
raw := buildError("not_found", "no such symbol", OpGetAsset)
|
||||
return zap.Parse(raw)
|
||||
})
|
||||
if err := node.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer node.Stop()
|
||||
|
||||
p := New(Config{DEXAddr: "127.0.0.1:" + strconv.Itoa(port)})
|
||||
defer p.Close()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
_, err = p.GetAsset(ctx, "XXX")
|
||||
if err == nil {
|
||||
t.Fatal("expected error")
|
||||
}
|
||||
rpcErr, ok := err.(*RPCError)
|
||||
if !ok {
|
||||
t.Fatalf("want *RPCError, got %T: %v", err, err)
|
||||
}
|
||||
if rpcErr.Code != "not_found" || rpcErr.Message != "no such symbol" {
|
||||
t.Errorf("err: %+v", rpcErr)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProvider_Name(t *testing.T) {
|
||||
if (&Provider{}).Name() != "lx" {
|
||||
t.Fatal("Name should be \"lx\"")
|
||||
@@ -141,12 +379,9 @@ func TestProvider_Name(t *testing.T) {
|
||||
func TestProvider_DefaultsApplied(t *testing.T) {
|
||||
p := New(Config{})
|
||||
if p.cfg.DEXAddr != DefaultDEXAddr {
|
||||
t.Errorf("DEXAddr: got %q, want %q", p.cfg.DEXAddr, DefaultDEXAddr)
|
||||
t.Errorf("DEXAddr: got %q", p.cfg.DEXAddr)
|
||||
}
|
||||
if p.cfg.MPCAddr != DefaultMPCAddr {
|
||||
t.Errorf("MPCAddr: got %q, want %q", p.cfg.MPCAddr, DefaultMPCAddr)
|
||||
}
|
||||
if p.cfg.NodeID != "lx-client" {
|
||||
t.Errorf("NodeID: got %q, want lx-client", p.cfg.NodeID)
|
||||
t.Errorf("MPCAddr: got %q", p.cfg.MPCAddr)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,638 @@
|
||||
// Schema definitions for Lux DEX ZAP messages. Every RPC has a dedicated
|
||||
// request/response struct with fixed field offsets; payload is zero-copy.
|
||||
//
|
||||
// Wire layout:
|
||||
//
|
||||
// Header (16 bytes):
|
||||
// magic (4): "ZAP\x00"
|
||||
// version (2): 1
|
||||
// flags (2): [opcode:8 | status:4 | reserved:4]
|
||||
// root off (4): offset into body where the root struct starts
|
||||
// size (4): total message size
|
||||
//
|
||||
// Body: one root struct, fields at fixed aligned offsets.
|
||||
//
|
||||
// Status encoding (flags lower nibble of upper byte):
|
||||
//
|
||||
// 0x0 OK
|
||||
// 0x1 ERR — root is ErrorResp{code, message}
|
||||
// 0x2 NOT_IMPL — root is ErrorResp with code "not_implemented"
|
||||
//
|
||||
// All float values use IEEE-754 binary64 little-endian. All timestamps are
|
||||
// int64 Unix nanoseconds. Strings are encoded via ZAP text fields
|
||||
// (4-byte relative offset + 4-byte length).
|
||||
package lx
|
||||
|
||||
import (
|
||||
"github.com/luxfi/zap"
|
||||
)
|
||||
|
||||
// Opcodes — must match the dex-server ZAP handler table.
|
||||
const (
|
||||
OpListAssets uint16 = 0x01
|
||||
OpGetAsset uint16 = 0x02
|
||||
OpGetSnapshot uint16 = 0x10
|
||||
OpGetSnapshots uint16 = 0x11
|
||||
OpGetBars uint16 = 0x12
|
||||
OpGetQuotes uint16 = 0x13
|
||||
OpGetTrades uint16 = 0x14
|
||||
OpGetBook uint16 = 0x15
|
||||
OpCreateOrder uint16 = 0x20
|
||||
OpCancelOrder uint16 = 0x21
|
||||
OpGetOrder uint16 = 0x22
|
||||
OpListOrders uint16 = 0x23
|
||||
OpGetPortfolio uint16 = 0x30
|
||||
)
|
||||
|
||||
// Status bits in the flags lower byte. The opcode occupies the upper byte.
|
||||
const (
|
||||
StatusOK uint16 = 0x0000
|
||||
StatusErr uint16 = 0x0001
|
||||
StatusNotImpl uint16 = 0x0002
|
||||
)
|
||||
|
||||
// packFlags encodes opcode + status into the 16-bit flags header field.
|
||||
func packFlags(op, status uint16) uint16 {
|
||||
return (op << 8) | (status & 0x00FF)
|
||||
}
|
||||
|
||||
// unpackFlags extracts opcode + status from the flags header field.
|
||||
func unpackFlags(flags uint16) (op, status uint16) {
|
||||
return flags >> 8, flags & 0x00FF
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// ErrorResp (flags.status != OK)
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// offset type name
|
||||
// ────── ────── ───────────
|
||||
// 0 text code (8 bytes: 4 off + 4 len)
|
||||
// 8 text message (8 bytes: 4 off + 4 len)
|
||||
// 16 (total)
|
||||
const (
|
||||
errOffCode = 0
|
||||
errOffMessage = 8
|
||||
errSize = 16
|
||||
)
|
||||
|
||||
func buildError(code, msg string, op uint16) []byte {
|
||||
b := zap.NewBuilder(64 + len(code) + len(msg))
|
||||
ob := b.StartObject(errSize)
|
||||
ob.SetText(errOffCode, code)
|
||||
ob.SetText(errOffMessage, msg)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(op, StatusErr))
|
||||
}
|
||||
|
||||
func readError(root zap.Object) (code, message string) {
|
||||
return root.Text(errOffCode), root.Text(errOffMessage)
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// SymbolReq — single-symbol request (GetSnapshot, GetAsset, GetBook)
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// 0 text symbol (8)
|
||||
// 8 (total)
|
||||
const (
|
||||
symReqOffSymbol = 0
|
||||
symReqSize = 8
|
||||
)
|
||||
|
||||
func buildSymbolReq(op uint16, symbol string) []byte {
|
||||
b := zap.NewBuilder(64 + len(symbol))
|
||||
ob := b.StartObject(symReqSize)
|
||||
ob.SetText(symReqOffSymbol, symbol)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(op, StatusOK))
|
||||
}
|
||||
|
||||
func readSymbolReq(root zap.Object) string {
|
||||
return root.Text(symReqOffSymbol)
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// ClassReq — asset class filter (ListAssets)
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// 0 text class (8)
|
||||
// 8 (total)
|
||||
const (
|
||||
classReqOffClass = 0
|
||||
classReqSize = 8
|
||||
)
|
||||
|
||||
func buildClassReq(op uint16, class string) []byte {
|
||||
b := zap.NewBuilder(64)
|
||||
ob := b.StartObject(classReqSize)
|
||||
ob.SetText(classReqOffClass, class)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(op, StatusOK))
|
||||
}
|
||||
|
||||
func readClassReq(root zap.Object) string {
|
||||
return root.Text(classReqOffClass)
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// SymbolsReq — multi-symbol request (GetSnapshots/Quotes/Trades)
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// Symbols are packed as a single comma-separated text field. This keeps the
|
||||
// wire layout compact without needing a heterogeneous list-of-text encoding.
|
||||
//
|
||||
// 0 text symbols (8) — CSV: "BTC/USD,ETH/USD,SOL/USD"
|
||||
// 8 (total)
|
||||
const (
|
||||
symsReqOffSymbols = 0
|
||||
symsReqSize = 8
|
||||
)
|
||||
|
||||
func buildSymbolsReq(op uint16, symbols []string) []byte {
|
||||
csv := joinCSV(symbols)
|
||||
b := zap.NewBuilder(64 + len(csv))
|
||||
ob := b.StartObject(symsReqSize)
|
||||
ob.SetText(symsReqOffSymbols, csv)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(op, StatusOK))
|
||||
}
|
||||
|
||||
func readSymbolsReq(root zap.Object) []string {
|
||||
return splitCSV(root.Text(symsReqOffSymbols))
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// BarsReq — historical bars query (GetBars)
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// 0 text symbol (8)
|
||||
// 8 text timeframe (8)
|
||||
// 16 text start (8)
|
||||
// 24 text end (8)
|
||||
// 32 int64 limit (8)
|
||||
// 40 (total)
|
||||
const (
|
||||
barsReqOffSymbol = 0
|
||||
barsReqOffTimeframe = 8
|
||||
barsReqOffStart = 16
|
||||
barsReqOffEnd = 24
|
||||
barsReqOffLimit = 32
|
||||
barsReqSize = 40
|
||||
)
|
||||
|
||||
type barsReq struct {
|
||||
symbol, timeframe, start, end string
|
||||
limit int64
|
||||
}
|
||||
|
||||
func buildBarsReq(r barsReq) []byte {
|
||||
b := zap.NewBuilder(128)
|
||||
ob := b.StartObject(barsReqSize)
|
||||
ob.SetText(barsReqOffSymbol, r.symbol)
|
||||
ob.SetText(barsReqOffTimeframe, r.timeframe)
|
||||
ob.SetText(barsReqOffStart, r.start)
|
||||
ob.SetText(barsReqOffEnd, r.end)
|
||||
ob.SetInt64(barsReqOffLimit, r.limit)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(OpGetBars, StatusOK))
|
||||
}
|
||||
|
||||
func readBarsReq(root zap.Object) barsReq {
|
||||
return barsReq{
|
||||
symbol: root.Text(barsReqOffSymbol),
|
||||
timeframe: root.Text(barsReqOffTimeframe),
|
||||
start: root.Text(barsReqOffStart),
|
||||
end: root.Text(barsReqOffEnd),
|
||||
limit: root.Int64(barsReqOffLimit),
|
||||
}
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// AccountReq — account-scoped request (GetPortfolio, ListOrders)
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// 0 text account (8)
|
||||
// 8 (total)
|
||||
const (
|
||||
acctReqOffAccount = 0
|
||||
acctReqSize = 8
|
||||
)
|
||||
|
||||
func buildAccountReq(op uint16, account string) []byte {
|
||||
b := zap.NewBuilder(64)
|
||||
ob := b.StartObject(acctReqSize)
|
||||
ob.SetText(acctReqOffAccount, account)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(op, StatusOK))
|
||||
}
|
||||
|
||||
func readAccountReq(root zap.Object) string {
|
||||
return root.Text(acctReqOffAccount)
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// AccountOrderReq — (account, order_id) (GetOrder, CancelOrder)
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// 0 text account (8)
|
||||
// 8 text order_id (8)
|
||||
// 16 (total)
|
||||
const (
|
||||
acctOrderReqOffAccount = 0
|
||||
acctOrderReqOffOrder = 8
|
||||
acctOrderReqSize = 16
|
||||
)
|
||||
|
||||
func buildAccountOrderReq(op uint16, account, order string) []byte {
|
||||
b := zap.NewBuilder(64)
|
||||
ob := b.StartObject(acctOrderReqSize)
|
||||
ob.SetText(acctOrderReqOffAccount, account)
|
||||
ob.SetText(acctOrderReqOffOrder, order)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(op, StatusOK))
|
||||
}
|
||||
|
||||
func readAccountOrderReq(root zap.Object) (account, order string) {
|
||||
return root.Text(acctOrderReqOffAccount), root.Text(acctOrderReqOffOrder)
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// CreateOrderReq — place an order
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// 0 text account (8)
|
||||
// 8 text symbol (8)
|
||||
// 16 text side (8) — "buy" | "sell"
|
||||
// 24 text order_type (8) — "market" | "limit" | "stop" | "stop_limit"
|
||||
// 32 text time_in_force (8) — "day" | "gtc" | "ioc" | "fok"
|
||||
// 40 text client_order_id (8)
|
||||
// 48 float64 qty (8)
|
||||
// 56 float64 notional (8)
|
||||
// 64 float64 limit_price (8)
|
||||
// 72 float64 stop_price (8)
|
||||
// 80 (total)
|
||||
const (
|
||||
coReqOffAccount = 0
|
||||
coReqOffSymbol = 8
|
||||
coReqOffSide = 16
|
||||
coReqOffOrderType = 24
|
||||
coReqOffTIF = 32
|
||||
coReqOffCliID = 40
|
||||
coReqOffQty = 48
|
||||
coReqOffNotional = 56
|
||||
coReqOffLimit = 64
|
||||
coReqOffStop = 72
|
||||
coReqSize = 80
|
||||
)
|
||||
|
||||
type createOrderReq struct {
|
||||
account, symbol, side, orderType, tif, clientID string
|
||||
qty, notional, limitPrice, stopPrice float64
|
||||
}
|
||||
|
||||
func buildCreateOrderReq(r createOrderReq) []byte {
|
||||
b := zap.NewBuilder(256)
|
||||
ob := b.StartObject(coReqSize)
|
||||
ob.SetText(coReqOffAccount, r.account)
|
||||
ob.SetText(coReqOffSymbol, r.symbol)
|
||||
ob.SetText(coReqOffSide, r.side)
|
||||
ob.SetText(coReqOffOrderType, r.orderType)
|
||||
ob.SetText(coReqOffTIF, r.tif)
|
||||
ob.SetText(coReqOffCliID, r.clientID)
|
||||
ob.SetFloat64(coReqOffQty, r.qty)
|
||||
ob.SetFloat64(coReqOffNotional, r.notional)
|
||||
ob.SetFloat64(coReqOffLimit, r.limitPrice)
|
||||
ob.SetFloat64(coReqOffStop, r.stopPrice)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(OpCreateOrder, StatusOK))
|
||||
}
|
||||
|
||||
func readCreateOrderReq(root zap.Object) createOrderReq {
|
||||
return createOrderReq{
|
||||
account: root.Text(coReqOffAccount),
|
||||
symbol: root.Text(coReqOffSymbol),
|
||||
side: root.Text(coReqOffSide),
|
||||
orderType: root.Text(coReqOffOrderType),
|
||||
tif: root.Text(coReqOffTIF),
|
||||
clientID: root.Text(coReqOffCliID),
|
||||
qty: root.Float64(coReqOffQty),
|
||||
notional: root.Float64(coReqOffNotional),
|
||||
limitPrice: root.Float64(coReqOffLimit),
|
||||
stopPrice: root.Float64(coReqOffStop),
|
||||
}
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// SnapshotResp — single quote snapshot
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// 0 text symbol (8)
|
||||
// 8 float64 last_price (8)
|
||||
// 16 float64 last_size (8)
|
||||
// 24 int64 last_time_ns (8)
|
||||
// 32 float64 bid_price (8)
|
||||
// 40 float64 bid_size (8)
|
||||
// 48 float64 ask_price (8)
|
||||
// 56 float64 ask_size (8)
|
||||
// 64 int64 quote_time_ns (8)
|
||||
// 72 (total)
|
||||
const (
|
||||
snapOffSymbol = 0
|
||||
snapOffLastPrice = 8
|
||||
snapOffLastSize = 16
|
||||
snapOffLastNS = 24
|
||||
snapOffBidPrice = 32
|
||||
snapOffBidSize = 40
|
||||
snapOffAskPrice = 48
|
||||
snapOffAskSize = 56
|
||||
snapOffQuoteNS = 64
|
||||
snapSize = 72
|
||||
)
|
||||
|
||||
type snapshot struct {
|
||||
symbol string
|
||||
lastPrice, lastSize, bidPrice, bidSize float64
|
||||
askPrice, askSize float64
|
||||
lastNs, quoteNs int64
|
||||
}
|
||||
|
||||
// writeSnapshot writes a snapshot struct into b and returns its abs offset.
|
||||
// Caller is responsible for referencing the offset from a parent object.
|
||||
func writeSnapshot(b *zap.Builder, s snapshot) int {
|
||||
ob := b.StartObject(snapSize)
|
||||
ob.SetText(snapOffSymbol, s.symbol)
|
||||
ob.SetFloat64(snapOffLastPrice, s.lastPrice)
|
||||
ob.SetFloat64(snapOffLastSize, s.lastSize)
|
||||
ob.SetInt64(snapOffLastNS, s.lastNs)
|
||||
ob.SetFloat64(snapOffBidPrice, s.bidPrice)
|
||||
ob.SetFloat64(snapOffBidSize, s.bidSize)
|
||||
ob.SetFloat64(snapOffAskPrice, s.askPrice)
|
||||
ob.SetFloat64(snapOffAskSize, s.askSize)
|
||||
ob.SetInt64(snapOffQuoteNS, s.quoteNs)
|
||||
return ob.Finish()
|
||||
}
|
||||
|
||||
func buildSnapshotResp(s snapshot) []byte {
|
||||
b := zap.NewBuilder(128 + len(s.symbol))
|
||||
ob := b.StartObject(snapSize)
|
||||
ob.SetText(snapOffSymbol, s.symbol)
|
||||
ob.SetFloat64(snapOffLastPrice, s.lastPrice)
|
||||
ob.SetFloat64(snapOffLastSize, s.lastSize)
|
||||
ob.SetInt64(snapOffLastNS, s.lastNs)
|
||||
ob.SetFloat64(snapOffBidPrice, s.bidPrice)
|
||||
ob.SetFloat64(snapOffBidSize, s.bidSize)
|
||||
ob.SetFloat64(snapOffAskPrice, s.askPrice)
|
||||
ob.SetFloat64(snapOffAskSize, s.askSize)
|
||||
ob.SetInt64(snapOffQuoteNS, s.quoteNs)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(OpGetSnapshot, StatusOK))
|
||||
}
|
||||
|
||||
func readSnapshotRoot(root zap.Object) snapshot {
|
||||
return snapshot{
|
||||
symbol: root.Text(snapOffSymbol),
|
||||
lastPrice: root.Float64(snapOffLastPrice),
|
||||
lastSize: root.Float64(snapOffLastSize),
|
||||
lastNs: root.Int64(snapOffLastNS),
|
||||
bidPrice: root.Float64(snapOffBidPrice),
|
||||
bidSize: root.Float64(snapOffBidSize),
|
||||
askPrice: root.Float64(snapOffAskPrice),
|
||||
askSize: root.Float64(snapOffAskSize),
|
||||
quoteNs: root.Int64(snapOffQuoteNS),
|
||||
}
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// AssetResp — single asset record
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// 0 text id (8)
|
||||
// 8 text symbol (8)
|
||||
// 16 text name (8)
|
||||
// 24 text class (8)
|
||||
// 32 text exchange (8)
|
||||
// 40 text status (8)
|
||||
// 48 text provider (8)
|
||||
// 56 bool tradable (1, padded to 8)
|
||||
// 64 (total, 8-aligned)
|
||||
const (
|
||||
assetOffID = 0
|
||||
assetOffSymbol = 8
|
||||
assetOffName = 16
|
||||
assetOffClass = 24
|
||||
assetOffExchange = 32
|
||||
assetOffStatus = 40
|
||||
assetOffProvider = 48
|
||||
assetOffTradable = 56
|
||||
assetSize = 64
|
||||
)
|
||||
|
||||
type asset struct {
|
||||
id, symbol, name, class, exchange, status, provider string
|
||||
tradable bool
|
||||
}
|
||||
|
||||
func buildAssetResp(a asset) []byte {
|
||||
b := zap.NewBuilder(256)
|
||||
ob := b.StartObject(assetSize)
|
||||
ob.SetText(assetOffID, a.id)
|
||||
ob.SetText(assetOffSymbol, a.symbol)
|
||||
ob.SetText(assetOffName, a.name)
|
||||
ob.SetText(assetOffClass, a.class)
|
||||
ob.SetText(assetOffExchange, a.exchange)
|
||||
ob.SetText(assetOffStatus, a.status)
|
||||
ob.SetText(assetOffProvider, a.provider)
|
||||
ob.SetBool(assetOffTradable, a.tradable)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(OpGetAsset, StatusOK))
|
||||
}
|
||||
|
||||
func readAssetRoot(root zap.Object) asset {
|
||||
return asset{
|
||||
id: root.Text(assetOffID),
|
||||
symbol: root.Text(assetOffSymbol),
|
||||
name: root.Text(assetOffName),
|
||||
class: root.Text(assetOffClass),
|
||||
exchange: root.Text(assetOffExchange),
|
||||
status: root.Text(assetOffStatus),
|
||||
provider: root.Text(assetOffProvider),
|
||||
tradable: root.Bool(assetOffTradable),
|
||||
}
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// OrderResp — single order record
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// 0 text id (8)
|
||||
// 8 text client_order_id (8)
|
||||
// 16 text account (8)
|
||||
// 24 text symbol (8)
|
||||
// 32 text side (8)
|
||||
// 40 text order_type (8)
|
||||
// 48 text time_in_force (8)
|
||||
// 56 text status (8)
|
||||
// 64 float64 qty (8)
|
||||
// 72 float64 filled_qty (8)
|
||||
// 80 float64 limit_price (8)
|
||||
// 88 float64 stop_price (8)
|
||||
// 96 float64 avg_fill_price (8)
|
||||
// 104 int64 created_ns (8)
|
||||
// 112 int64 updated_ns (8)
|
||||
// 120 int64 filled_ns (8)
|
||||
// 128 (total)
|
||||
const (
|
||||
ordOffID = 0
|
||||
ordOffCliID = 8
|
||||
ordOffAccount = 16
|
||||
ordOffSymbol = 24
|
||||
ordOffSide = 32
|
||||
ordOffOrderType = 40
|
||||
ordOffTIF = 48
|
||||
ordOffStatus = 56
|
||||
ordOffQty = 64
|
||||
ordOffFilledQty = 72
|
||||
ordOffLimit = 80
|
||||
ordOffStop = 88
|
||||
ordOffAvgFill = 96
|
||||
ordOffCreatedNS = 104
|
||||
ordOffUpdatedNS = 112
|
||||
ordOffFilledNS = 120
|
||||
ordSize = 128
|
||||
)
|
||||
|
||||
type order struct {
|
||||
id, clientID, account, symbol, side, orderType, tif, status string
|
||||
qty, filledQty, limitPrice, stopPrice, avgFillPrice float64
|
||||
createdNs, updatedNs, filledNs int64
|
||||
}
|
||||
|
||||
func writeOrder(b *zap.Builder, o order) int {
|
||||
ob := b.StartObject(ordSize)
|
||||
ob.SetText(ordOffID, o.id)
|
||||
ob.SetText(ordOffCliID, o.clientID)
|
||||
ob.SetText(ordOffAccount, o.account)
|
||||
ob.SetText(ordOffSymbol, o.symbol)
|
||||
ob.SetText(ordOffSide, o.side)
|
||||
ob.SetText(ordOffOrderType, o.orderType)
|
||||
ob.SetText(ordOffTIF, o.tif)
|
||||
ob.SetText(ordOffStatus, o.status)
|
||||
ob.SetFloat64(ordOffQty, o.qty)
|
||||
ob.SetFloat64(ordOffFilledQty, o.filledQty)
|
||||
ob.SetFloat64(ordOffLimit, o.limitPrice)
|
||||
ob.SetFloat64(ordOffStop, o.stopPrice)
|
||||
ob.SetFloat64(ordOffAvgFill, o.avgFillPrice)
|
||||
ob.SetInt64(ordOffCreatedNS, o.createdNs)
|
||||
ob.SetInt64(ordOffUpdatedNS, o.updatedNs)
|
||||
ob.SetInt64(ordOffFilledNS, o.filledNs)
|
||||
return ob.Finish()
|
||||
}
|
||||
|
||||
func buildOrderResp(op uint16, o order) []byte {
|
||||
b := zap.NewBuilder(512)
|
||||
ob := b.StartObject(ordSize)
|
||||
ob.SetText(ordOffID, o.id)
|
||||
ob.SetText(ordOffCliID, o.clientID)
|
||||
ob.SetText(ordOffAccount, o.account)
|
||||
ob.SetText(ordOffSymbol, o.symbol)
|
||||
ob.SetText(ordOffSide, o.side)
|
||||
ob.SetText(ordOffOrderType, o.orderType)
|
||||
ob.SetText(ordOffTIF, o.tif)
|
||||
ob.SetText(ordOffStatus, o.status)
|
||||
ob.SetFloat64(ordOffQty, o.qty)
|
||||
ob.SetFloat64(ordOffFilledQty, o.filledQty)
|
||||
ob.SetFloat64(ordOffLimit, o.limitPrice)
|
||||
ob.SetFloat64(ordOffStop, o.stopPrice)
|
||||
ob.SetFloat64(ordOffAvgFill, o.avgFillPrice)
|
||||
ob.SetInt64(ordOffCreatedNS, o.createdNs)
|
||||
ob.SetInt64(ordOffUpdatedNS, o.updatedNs)
|
||||
ob.SetInt64(ordOffFilledNS, o.filledNs)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(op, StatusOK))
|
||||
}
|
||||
|
||||
func readOrderRoot(root zap.Object) order {
|
||||
return order{
|
||||
id: root.Text(ordOffID),
|
||||
clientID: root.Text(ordOffCliID),
|
||||
account: root.Text(ordOffAccount),
|
||||
symbol: root.Text(ordOffSymbol),
|
||||
side: root.Text(ordOffSide),
|
||||
orderType: root.Text(ordOffOrderType),
|
||||
tif: root.Text(ordOffTIF),
|
||||
status: root.Text(ordOffStatus),
|
||||
qty: root.Float64(ordOffQty),
|
||||
filledQty: root.Float64(ordOffFilledQty),
|
||||
limitPrice: root.Float64(ordOffLimit),
|
||||
stopPrice: root.Float64(ordOffStop),
|
||||
avgFillPrice: root.Float64(ordOffAvgFill),
|
||||
createdNs: root.Int64(ordOffCreatedNS),
|
||||
updatedNs: root.Int64(ordOffUpdatedNS),
|
||||
filledNs: root.Int64(ordOffFilledNS),
|
||||
}
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// EmptyResp — acknowledgment payload (CancelOrder success, etc.)
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
//
|
||||
// 0 bool ok (1, padded to 8)
|
||||
// 8 (total)
|
||||
const (
|
||||
emptyOffOK = 0
|
||||
emptySize = 8
|
||||
)
|
||||
|
||||
func buildEmptyResp(op uint16) []byte {
|
||||
b := zap.NewBuilder(32)
|
||||
ob := b.StartObject(emptySize)
|
||||
ob.SetBool(emptyOffOK, true)
|
||||
ob.FinishAsRoot()
|
||||
return b.FinishWithFlags(packFlags(op, StatusOK))
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
// CSV helpers — compact encoding for list-of-string requests.
|
||||
// ───────────────────────────────────────────────────────────────────────
|
||||
|
||||
func joinCSV(parts []string) string {
|
||||
if len(parts) == 0 {
|
||||
return ""
|
||||
}
|
||||
n := len(parts) - 1 // commas
|
||||
for _, p := range parts {
|
||||
n += len(p)
|
||||
}
|
||||
buf := make([]byte, 0, n)
|
||||
for i, p := range parts {
|
||||
if i > 0 {
|
||||
buf = append(buf, ',')
|
||||
}
|
||||
buf = append(buf, p...)
|
||||
}
|
||||
return string(buf)
|
||||
}
|
||||
|
||||
func splitCSV(s string) []string {
|
||||
if s == "" {
|
||||
return nil
|
||||
}
|
||||
// Count commas first to size the output exactly.
|
||||
n := 1
|
||||
for i := 0; i < len(s); i++ {
|
||||
if s[i] == ',' {
|
||||
n++
|
||||
}
|
||||
}
|
||||
out := make([]string, 0, n)
|
||||
start := 0
|
||||
for i := 0; i < len(s); i++ {
|
||||
if s[i] == ',' {
|
||||
out = append(out, s[start:i])
|
||||
start = i + 1
|
||||
}
|
||||
}
|
||||
out = append(out, s[start:])
|
||||
return out
|
||||
}
|
||||
Reference in New Issue
Block a user