Streaming Orders into NATS JetStream

Overview

This document explains how to configure and use NATS JetStream as a data warehouse for Ember.

Install and Configure NATS

To learn how to install NATS server with JetStream enabled, see the NATS documentation. JetStream must be enabled on the server (for example, nats-server -js).

Configure Ember

Streams

Since Ember version 1.16, you can set up a warehouse service to export data into NATS. The service creates the following objects if they are missing:

• EmberMessages stream: All trading flow messages, including trading requests (order submissions, cancellations, modifications) and trading events (order ACKs/NACKs, cancellation ACKs/NACKs, fills, etc.). You can configure a filter to reduce this stream to trade events only.
• EmberOrders stream: Order snapshots. By default it contains one record with the final state of each order (see Updatable Orders).
• EmberMetadata-<region>-<tenant> key-value bucket: The term key holds the current Ember journal term.

Configuration

Add the following section to $EMBER_HOME/ember.conf. The data export mode is controlled by the live parameter.

warehouse {
  nats {                   # unit id, you will use it when you run the app, it might be any
    live = true            # the process does not wait for new data and exits if false

    messages = [
      ${template.warehouse.nats.messages} {  # loader which loads order messages
        filter.settings.inclusion = null     # EVENTS or TRADES, null means no filter

        loader.settings {
          servers = "nats://nats-1:4222,nats://nats-2:4222"  # comma-separated NATS servers
          region = EU                                        # US, EU, AP or AU
          streamMaxAge = 30d                                 # stream retention
          streamMaxBytes = 100G                              # stream size
        }
      }
    ]

    orders = [
      ${template.warehouse.nats.orders} {    # loader which loads orders
        loader.settings {
          servers = "nats://nats-1:4222,nats://nats-2:4222"
          region = EU
          streamMaxAge = 30d
          streamMaxBytes = 10G
        }
      }
    ]
  }
}

Loader settings:

Setting	Default	Description
servers	nats://localhost:4222	Comma-separated list of NATS servers
region	not set	Region of this Ember instance: US, EU, AP or AU. When not set, _ is used
username	not set	User name for user/password authentication
password	not set	Password for user/password authentication
streamMaxAge	unlimited	Maximum age of records in the stream, for example 30d
streamMaxBytes	unlimited	Maximum size of the stream, for example 100G
maxPendingAcks	1024	Number of asynchronous publications awaiting acknowledgement before the loader waits for them

Stream limits are applied when a stream is created and to an existing stream on startup. Limits that are not specified keep their current values (or NATS defaults).

Updatable Orders

By default the EmberOrders stream receives one snapshot per order when the order reaches its final state (completely filled, canceled, expired, or rejected).

Set storeActiveOrders = true to publish a snapshot on every order change: order submission, acknowledgement, each fill, replace, restatement, and the final state. NATS streams are append-only, so every change is appended as a new record and the stream contains the complete history of each order.

warehouse {
  nats {
    storeActiveOrders = true   # publish every order change

    orders = [
      ${template.warehouse.nats.orders} {
        loader.settings {
          servers = "nats://nats-1:4222"
          region = EU
        }
      }
    ]
  }
}

note

This mode applies to orders only. The EmberMessages stream always contains every trading message.

Start Exporting Data

Run Ember's data-warehouse script with a single argument that specifies the NATS data warehouse:

export EMBER_HOME=/deltix/emberhome
/deltix/ember/bin/data-warehouse nats

In Batch Mode, when the live parameter is set to false, this script exits as soon as all recent data is exported. If you re-run the script, it appends any new data that Ember accumulated since the last export: on startup each loader reads the last record of its stream and continues from the next journal sequence.

You can inspect the exported data using the NATS CLI:

nats stream ls
nats stream subjects EmberMessages
nats stream view EmberOrders
nats kv get EmberMetadata-EU-_ term

Data Format

Subjects

EmberMessages.term.<term>.r.<region>.t.<tenant>.<MessageType>
EmberOrders.term.<term>.r.<region>.t.<tenant>

Token	Description
term.<term>	Ember journal term (journal creation time). A new journal starts a new term and restarts sequence numbers
r.<region>	Configured region, or _ when the region is not set
t.<tenant>	Tenant. Tenants are not supported yet, so the value is always _
<MessageType>	Ember message type of the record, for example OrderNewRequest

Stream names match the subject root. Subject keys follow the conventions of the C2 protocol: r for region, t for tenant, and _ for no value.

Headers

Header	Description
Ember-Sequence	Ember journal sequence of the record. For orders, the sequence of the last order event (close sequence)
Nats-Msg-Id	<region>:<term>:<sequence> (with a :<n> suffix for records that share a sequence). JetStream discards duplicates within its duplicate window
Nats-Expected-Stream	Target stream name. NATS rejects the record if its subject belongs to another stream

Record Format

Records are encoded with Protocol Buffers (proto3):

• EmberMessages records are EmberMessage messages.
• EmberOrders records are EmberOrder messages.

The schema file deltix/ember/warehouse/nats/warehouse.proto (package deltix.ember.warehouse.nats) is included in the deltix-ember-data-warehouse-nats-<version>.jar library of the Ember distribution. Use it to generate a client for your language. The schema imports google/type/decimal.proto from the Google common protos.

Field encoding:

Ember type	Protobuf type	Description
Decimal	google.type.Decimal	Exact decimal value as a string, for example "5000.25". Values are not rounded
Alphanumeric	string	Decoded text, for example "CME" for sourceId
Timestamp	int64	Unix epoch time. Nanoseconds for timestamp and original_timestamp of messages, milliseconds for all other timestamps
Enumeration	enum	Ember value with an enum-name prefix, for example SIDE_BUY for Side.BUY.
Text	string
Integer	int32, int64

Messages Stream

Each record contains one trading message. The message type is the last token of the subject and the type field of the record:

• Requests: OrderNewRequest, OrderReplaceRequest, OrderCancelRequest, OrderStatusRequest.
• Events: OrderPendingNewEvent, OrderNewEvent, OrderRejectEvent, OrderPendingCancelEvent, OrderCancelEvent, OrderCancelRejectEvent, OrderPendingReplaceEvent, OrderReplaceEvent, OrderReplaceRejectEvent, OrderTradeReportEvent, OrderTradeCancelEvent, OrderTradeCorrectEvent, OrderStatusEvent, OrderRestateEvent.

OrderDiscardRequest and OrderTradeAckEvent messages are not exported.

EmberMessage fields:

Name	Type	Description
type	MessageType	Message type, for example MESSAGE_TYPE_ORDER_NEW_REQUEST
sequence	int64	Ember journal sequence of the message
sequence_number	int32	Sequence number of the event reported by execution venue
timestamp	int64	Message timestamp, nanoseconds
original_timestamp	int64	Timestamp reported by execution venue, nanoseconds
source_id	string	Message source
destination_id	string	Message destination
order_id	string	Order ID assigned by order source
original_order_id	string	Original order ID in cancel-replace chain
correlation_order_id	string	ID of the first order in order-replace chain
parent_source_id	string	Source of parent order
parent_order_id	string	ID of parent order
request_id	string	Request ID
external_order_id	string	Optional order ID assigned by execution venue
event_id	string	Event ID
reference_event_id	string	Used to identify previously communicated event
quote_id	string	Optional ID of quote targeted by order
order_status	OrderStatus	Order status
symbol	string	Instrument symbol
instrument_type	InstrumentType	Instrument type
currency	string	Order currency
exchange_id	string	Optional destination exchange for outbound message and source exchange for inbound messages
trader_id	string	ID of trader
trader_group	string	Group of trader
account	string	Order account
clearing_account	string	Order clearing account
side	Side	Order side
time_in_force	TimeInForce	Order time in force
expire_time	int64	Order expiration time
quantity	Decimal	Order quantity
min_quantity	Decimal	Minimum quantity to execute
display_quantity	Decimal	Minimum quantity to display on exchange floor
order_type	OrderType	Order type
limit_price	Decimal	Limit price
stop_price	Decimal	Stop price
peg_difference	Decimal	Peg offset for pegged order types
average_price	Decimal	Average execution price
cumulative_quantity	Decimal	Cumulative executed quantity
remaining_quantity	Decimal	Remaining order quantity
trade_price	Decimal	Order trade price
trade_quantity	Decimal	Order trade quantity
commission	Decimal	Trade commission
commission_currency	string	Trade commission currency
discretion_offset	Decimal	Discretion offset
counter_party_source_id	string	Source of the other side of the trade
counter_party_order_id	string	Order ID of the other side of the trade
settlement_date	int64	Trade settlement date
settlement_type	string	Settlement type
trade_date	int64	Trade date
reason	string	Reason of cancel and reject events
vendor_reject_code	int32	Vendor specific reject code
deltix_reject_code	int32	Reject code in Deltix classification
multi_leg_reporting_type	MultiLegReportingType	Identifies single-leg trade or whole contract trade of multi-legged security
aggressor_side	AggressorIndicator	Trade aggressor side
order_unknown	bool	Flag used by order cancel reject events
cancel_type	CancelType	Enum used by cancel events
exec_restatement_reason	OrderRestateReasonType	Used by order restate events to classify restate type
flags	int32	Bitmask containing various order flags
user_data	string	User-provided order tag
module_key	string	Used to identify strategy that placed the order
portfolio_key	string	Used to identify portfolio this order belongs to
party	string	Describes business entity involved in trade
clearing_broker	string	Clearing broker
attributes	repeated Attribute	Custom attributes of the message: int32 key and optional string value

Orders Stream

Each record contains one order snapshot (see Updatable Orders).

EmberOrder fields:

Name	Type	Description
open_sequence	int64	Sequence of new order request
close_sequence	int64	Sequence of the last order event
source_id	string	Order source
destination_id	string	Order destination
order_id	string	Order ID assigned by order source
correlation_order_id	string	ID of the first order in order-replace chain
parent_source_id	string	Source of parent order
parent_order_id	string	ID of parent order
external_order_id	string	Optional order ID assigned by execution venue
account	string	Order account
clearing_account	string	Clearing account
trader_id	string	ID of trader
trader_group	string	Group of trader
quote_id	string	Optional ID of quote targeted by order
symbol	string	Instrument symbol
instrument_type	InstrumentType	Instrument type
exchange_id	string	Optional destination exchange for outbound message and source exchange for inbound messages
currency	string	Order currency
side	Side	Order side
time_in_force	TimeInForce	Order time in force
expire_time	int64	Order expiration time
order_status	OrderStatus	Order state at the time of the snapshot
open_time	int64	Order submission time
close_time	int64	Order completion time
order_type	OrderType	Order type
limit_price	Decimal	Limit price
stop_price	Decimal	Stop price
peg_difference	Decimal	Peg offset for pegged order types
quantity	Decimal	Order quantity
display_quantity	Decimal	Minimum quantity to display on exchange floor
min_quantity	Decimal	Minimum quantity to execute
cumulative_quantity	Decimal	Cumulative executed quantity
total_commission	Decimal	Total trade commission
commission_currency	string	Trade commission currency
average_price	Decimal	Average execution price
user_data	string	User-provided order tag
module_key	string	Used to identify strategy that placed the order
portfolio_key	string	Used to identify portfolio this order belongs to
party	string	Describes business entity involved in trade
clearing_broker	string	Clearing broker
settlement_date	int64	Settlement date
reason	string	Reason for cancel or reject events
deltix_reject_code	int32	Reject code in Deltix classification
vendor_reject_code	int32	Vendor specific reject code
attributes	repeated Attribute	Custom order attributes: int32 key and optional string value. Written only with storeCustomAttributes = true, see How the Attributes field is formed