FixAlchemy

A FIX engine for Elixir that decodes only what your handlers read.

FixAlchemy frames the stream, runs the session layer, and routes each application message to your code by MsgType. A message you subscribe to is decoded at its subscriber, over the fields that subscriber reads. A type nobody subscribes to is framed, routed, and dropped undecoded.

Features

• Any dialect. Point it at the dictionary XML — FIX44.xml, your venue's variant, or the FIXT1.1.xml + FIX50SP2.xml pair a FIX 5.0 session runs on — and it generates the field, message, and repeating-group parsers for it. All begin strings, including FIXT.1.1.
• Per-message decoding. Decoding is the subscriber's, over the fields it reads. Unsubscribed types are never decoded.
• Keyed dispatch bus. Subscribe to a MsgType, to a MsgType where a tag equals a value (W for EUR/USD), or where a tag satisfies a predicate (fills over a size). A broker adapter is a set of subscribers on this bus.
• Adapter-defined readiness. Login-dependent sends are gated on named milestones — :ready on Logon by default, plus adapter milestones like :account for a post-logon handshake. Messages registered with register_logon_send/3 are re-established on every reconnect.
• Named or numeric outbound. Build a body from raw tags, from the dictionary's field names, or a mix; names resolve through the spec, and required fields are checked against it.
• Initiator or acceptor. Dial a venue with FixAlchemy.Client, or listen for one with FixAlchemy.Server. Both run the same session protocol, so an accepted session behaves exactly like an initiated one once logon completes.
• Permissioned market data fan-out. FixAlchemy.Feed publishes one price to many sessions through a tree of venue transforms — stream, spread, markup, commission — applying each transform once however many sessions sit below it.
• Optional, off by default. TLS, inbound checksum validation, inbound message validation, sequence recovery — gap detection, ResendRequest, SequenceReset, and replay — and sequence numbers that outlive a connection.

Installation

def deps do
  [
    {:fix_alchemy, "~> 0.2"}
  ]
end

Use the dictionary your counterparty publishes, not a generic one. Put it wherever you keep application data (priv/specs/ is a common choice) and point the engine at it:

# config/config.exs
config :fix_alchemy, spec_file: "priv/specs/my_broker.xml"

or pass :spec_file per connection, which is what a setup with several venues does. examples/specs/FIX44.xml in this repository is a stock FIX 4.4 dictionary to test against; it is not shipped with the package.

FIX 5.0 and later: two dictionaries

FIX 5.0 splits the dictionary in two. FIXT1.1.xml is the transport dictionary: it defines the session layer — Logon, Logout, Heartbeat, TestRequest, ResendRequest, SequenceReset, Reject — and nothing else. The business messages live in an application dictionary, FIX50SP2.xml or one of its siblings. A session on 8=FIXT.1.1 needs both, so give the engine both:

config :fix_alchemy,
  spec_file: "priv/specs/FIXT1.1.xml",
  app_spec_file: "priv/specs/FIX50SP2.xml"

or per connection:

FixAlchemy.connect(
  # ...
  fix_version: "FIXT.1.1",
  default_appl_ver_id: "9",
  spec_file: "priv/specs/FIXT1.1.xml",
  app_spec_file: "priv/specs/FIX50SP2.xml"
)

Both are also read from the FIX_SPEC_FILE and FIX_APP_SPEC_FILE environment variables, and a backend takes them as the :fix_spec_path and :app_spec_path configuration fields.

The two are merged into one dictionary and one generated module set, named after both files. The transport dictionary defines the session layer: every field, message type, component and group it declares is kept as declared, and the application dictionary supplies everything it leaves undefined. A field both declare keeps the transport name and type and carries the values of both. So 35=A decodes against FIXT.1.1, 35=8 against FIX.5.0SP2, and named outbound fields resolve across both. examples/specs/FIXT1.1.xml is the stock FIXT.1.1 transport dictionary and examples/specs/FIX50SP2.xml a FIX 5.0 SP2 application dictionary cut down to orders and market data, both to test against.

A session on a single dictionary — FIX 4.4 and earlier — needs :spec_file alone.

Quick start

{:ok, conn} =
  FixAlchemy.connect(
    host: "fixdemo.example.com",
    port: 9043,
    username: "demo",
    sender_comp_id: "CLIENT1",
    target_comp_id: "SERVER",
    spec_file: "priv/specs/my_broker.xml"
  )

FixAlchemy.subscribe(conn, "W", "EUR/USD")
FixAlchemy.send_message(conn, {"V", [{262, "req1"}, {263, "1"}, {55, "EUR/USD"}]})

receive do
  {:fix, "W", _raw, meta} ->
    {"55", symbol} = List.keyfind(meta.fields, "55", 0)
    symbol
end

FixAlchemy.disconnect(conn)

The calling process receives {:fix, type, raw, meta} for each matching message; meta.fields is the tag/value split framing already produced, and the full decode via FixAlchemy.Parser.process_message/2 is yours to call where you need it. send_message/2 is the outbound primitive:

FixAlchemy.send_message(conn, {"D", [{55, "EUR/USD"}, {54, "1"}, {38, "1000"}]})

A broker adapter (for example fxcm_alchemy) is a set of subscriber processes supplied via :handlers, supervised alongside the session.

Session lifecycle

The engine owns the session layer: Logon, Heartbeat, TestRequest, Logout, sequence numbers, and reconnect. What counts as usable beyond Logon is the adapter's to define, through milestones:

• A session reaches :ready on Logon by default. An adapter with a post-logon handshake starts with defer_ready: true and calls FixAlchemy.Client.ready/1 when its own flow completes.
• Further milestones are marked with FixAlchemy.Client.reach(conn, :account) and gated on with milestone_reached?/2.
• FixAlchemy.Client.register_logon_send(conn, key, message) keeps a message established: sent on :ready, re-sent on every reconnect, never before Logon. send_when(conn, milestone, message) queues a one-off send until a milestone is reached.

So a MarketDataRequest registered before Logon does not reach the wire until the session is ready, and it re-subscribes itself after a reconnect.

Outbound fields

Build a message body with numeric tags, the dictionary's field names, or a mix:

FixAlchemy.send_message(conn, {"D", [{55, "EUR/USD"}, {54, "1"}, {38, "1000"}]})
FixAlchemy.send_message(conn, {"D", [{:symbol, "EUR/USD"}, {:side, "1"}, {:order_qty, "1000"}]})

Named atoms resolve through the generated dictionary, and the engine reports any required field a named message omits. Numeric tags go straight to the wire with no lookup — use them on hot paths and for proprietary tags the dictionary doesn't name.

Handling a message the engine doesn't

The baseline covers the session layer, ExecutionReport, and PositionReport. Anything else your venue sends — CollateralReport, TradeCaptureReport, SecurityDefinition — is a handler you write. There are three levels of involvement.

1. A subscriber process

Any process can take delivery of a message type. This is the whole contract:

defmodule MyApp.CollateralWatcher do
  use GenServer

  alias FixAlchemy.Parser

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts)

  @impl true
  def init(opts) do
    connection_id = Keyword.fetch!(opts, :connection_id)
    FixAlchemy.Session.subscribe_types(connection_id, :trading, self(), ["BA", "BG"])
    {:ok, %{spec_name: opts |> Parser.init() |> elem(0)}}
  end

  @impl true
  def handle_info({:fix, "BA", raw, _meta}, state) do
    report = Parser.process_message(raw, state.spec_name)
    Logger.info("collateral: #{report[:account]} #{report[:end_cash]} #{report[:currency]}")
    {:noreply, state}
  end

  def handle_info(_other, state), do: {:noreply, state}
end

Decoding happens in your process, off the socket loop, so a malformed message or a slow handler cannot stall or kill the FIX session.

2. Message as portfolio state

To make a message part of the account the platform renders, extend the portfolio base instead. Overriding two callbacks is enough to turn CollateralReports into a live account panel:

defmodule MyBroker.Portfolio do
  use FixAlchemy.Portfolio

  alias FixAlchemy.Portfolio, as: Base

  @impl FixAlchemy.Portfolio
  def subscribed_types, do: super() ++ ["BA", "BG"]

  @impl FixAlchemy.Portfolio
  def handle_message("BA", raw, _meta, state) do
    report = Base.decode(raw, state)
    %{state | account: report[:account] || state.account, collateral: Map.delete(report, :raw)}
  end

  def handle_message(type, raw, meta, state), do: super(type, raw, meta, state)

  @impl FixAlchemy.Portfolio
  def build_account_summary(%{collateral: collateral}) when is_map(collateral) do
    %{
      account_id: collateral[:account] || "N/A",
      balance: Base.parse_float(collateral[:end_cash]),
      equity: Base.parse_float(collateral[:total_net_value]),
      margin_used: 0.0,
      margin_available: Base.parse_float(collateral[:margin_excess]),
      currency: collateral[:currency],
      unrealized_pnl: 0.0
    }
  end

  def build_account_summary(state), do: super(state)
end

The base supplies everything else: the GenServer and its registered name, bus subscription, spec-driven decode, the get_positions/get_orders/ get_account_summary reads, and supervision. It also broadcasts for you — after each message it compares positions, orders, and build_account_summary/1 against their previous values and publishes {:position_update, …}, {:orders_update, …}, and {:account_update, …} on position_raw:, orders:, and account:<connection_id> only when something actually changed. Handling a new message type wires it to the UI without writing a single broadcast.

State the base does not model goes in :extra, read back with FixAlchemy.Portfolio.get_extra/2,3:

def handle_message("AE", raw, _meta, state) do
  trade = raw |> Base.decode(state) |> Map.delete(:raw)
  update_in(state.extra, &Map.update(&1, :trades, [trade], fn ts -> [trade | ts] end))
end

:extra sits outside change detection, so it costs nothing per message; publish your own updates with FixAlchemy.Portfolio.broadcast/3 when you want them.

3. Requesting it, and shipping the backend

Messages that must be solicited belong in the session handler, which owns the login flow and its milestones:

defmodule MyBroker.Session do
  use FixAlchemy.Session

  alias FixAlchemy.Client

  @impl FixAlchemy.Session
  def message_types, do: ["A", "BA"]

  @impl FixAlchemy.Session
  def handle_fix("A", _raw, meta, state) do
    Client.send_custom_message(meta.client, {"BB", [{:coll_inquiry_id, "ci_1"}, {263, "1"}]})
    Client.ready(meta.client)
    state
  end

  def handle_fix("BA", _raw, meta, state) do
    Client.reach(meta.client, :account)
    state
  end

  def handle_fix(type, raw, meta, state), do: super(type, raw, meta, state)
end

A backend then names its handlers and is auto-discovered by the platform:

defmodule MyBroker.TradingBackend do
  use FixAlchemy.Backend

  @impl FixAlchemy.Backend
  def backend_config do
    %{FixAlchemy.Backend.default_config() | id: :mybroker, name: "MyBroker (FIX)"}
  end

  @impl FixAlchemy.Backend
  def handlers, do: [MyBroker.Session, FixAlchemy.MarketData.Plain, MyBroker.Portfolio]
end

Every callback has a working default, so an adapter overrides only what its venue does differently — handlers/0, defer_ready?/0, trading_module/0, start_extras/2, get_historical_candles/3, get_instrument_precision/2.

Publishing events

Session status changes and the position, order and account updates the subscriber bases produce are published through FixAlchemy.PubSub. The default implementation forwards to Phoenix.PubSub:

config :fix_alchemy, pubsub_server: MyApp.PubSub

With no server configured, publishing is a no-op and the engine runs unchanged — phoenix_pubsub is not a dependency of this package. To publish some other way, implement the FixAlchemy.PubSub behaviour and point the engine at it with config :fix_alchemy, pubsub: MyApp.FixEvents.

Connection options

Connections of several sessions

A venue rarely offers one session. FixAlchemy.Backend takes a connection as shared settings plus a list of sessions, which is the shape of a QuickFIX initiator file: a block of defaults and N session blocks overriding what each one needs.

MyBroker.TradingBackend.connect("conn-1",
  host: "fix.example.com",
  port: 5001,
  target_comp_id: "VENUE",
  sessions: [
    %{name: "order", roles: [:trading], sender_comp_id: "ACME_ORD"},
    %{name: "quote", roles: [:market_data], sender_comp_id: "ACME_MD", overrides: %{port: 5002}},
    %{name: "dropcopy", roles: [:drop_copy], sender_comp_id: "ACME_DC"}
  ]
)

Each session carries its own SenderCompID — a venue rejects two sessions logging on under the same one — and inherits every other shared setting until it names an override. Its name identifies it within the connection and is the second element of every registry key it owns: {connection_id, name} for its client, {connection_id, name, :portfolio | :market_data | :handshake | :supervisor} for the processes around it. Its roles say what it is for, and are what callers ask by:

FixAlchemy.Backend.with_client("conn-1", :market_data, &FixAlchemy.Trading.list_instruments/1)

A role no session declares falls back to the session serving :trading, so a connection of one session answers for everything. See FixAlchemy.SessionConfig for how a config is read and FixAlchemy.SessionDirectory for how a role is resolved at runtime.

Accepting connections

FixAlchemy.Client dials a venue. FixAlchemy.Server is the other side: it listens, and every connection it accepts becomes a session once the counterparty identifies itself.

{:ok, _server} =
  FixAlchemy.Server.start_link(
    server_id: "acme",
    port: 5001,
    spec_file: "priv/specs/FIX44.xml",
    validate_inbound: true,
    sessions: [
      [
        connection_id: "acme_client",
        sender_comp_id: "ACME",
        target_comp_id: "CLIENT",
        handlers: [MyApp.OrderHandler]
      ]
    ]
  )

A connection is anonymous until its first message, which must be a Logon. FixAlchemy.Server.Logon matches it to a configured session by comp id pair — the Logon's SenderCompID(49) is the session's :target_comp_id, its TargetCompID(56) the session's :sender_comp_id — and only then does the process register in FixAlchemy.Registry under {connection_id, session_name}. From that point everything works as it does for an initiated session: subscribers receive {:fix, type, raw, meta}, and :handlers are supervised alongside.

A Logon is refused with a Logout carrying the reason when it matches no configured session, fails the session's declared credentials, names a session already connected, or is not a Logon at all. A connection that sends nothing is closed after ten seconds.

The heartbeat interval is the initiator's: the acceptor adopts HeartBtInt(108) from the Logon rather than its own configuration.

Server options

Anything a matched session declares overrides the server-wide value, so one server can accept sessions with different handlers, roles, or storage.

Configuration errors are returned, not raised: start_link/1 answers {:error, {:invalid_config, reason}} and starts nothing. Once listening, no single connection can stop the server — a failed TLS handshake, a peer that aborts mid-accept, or a session that fails to start is logged, and accepting continues.

Inbound validation

With validate_inbound: true every inbound message is checked before it is processed, and a failing one is answered with a Reject (35=3) naming the SessionRejectReason: unknown tag, tag with no value, tag not defined for that message type, missing required tag, undefined MsgType, comp id mismatch, and SendingTime outside tolerance. A BeginString the session does not speak is answered with a Logout. Comp id and SendingTime failures end the session after the Reject, as the session protocol requires.

An initiator talking to a venue it trusts does not need this. An acceptor exposed to a counterparty it does not control should turn it on.

Before you point it at a venue

The acceptor has been tested against FixAlchemy.Client and against raw sockets, not against a third-party FIX engine. Because both sides run the same FixAlchemy.Engine, any place the engine misreads the session specification is invisible in those tests — both ends make the same assumption and agree. Certification against the counterparty's engine is what would find it.

Sequence recovery

By default FixAlchemy sends ResetSeqNumFlag=Y. For counterparties that persist sequence numbers across sessions and expect gap-fill recovery, enable it:

FixAlchemy.connect(
  # ...
  reset_seq_number: "N",
  sequence_recovery: true
)

FixAlchemy then tracks inbound MsgSeqNum and sends a ResendRequest on a gap, holding later messages until it is filled so handlers see them in order. It answers inbound ResendRequests by replaying stored messages with PossDupFlag, gap-filling admin messages, and honours SequenceReset in both modes. Sent messages are retained in a FixAlchemy.MessageStore — in-memory by default, swap in a module for durable storage.

Numbering that outlives a connection

Sequence recovery works within one connection. A session that must resume its numbering after a reconnect, a restart, or a crash needs storage keyed by session identity rather than by process, which is FixAlchemy.SessionStore:

FixAlchemy.Server.start_link(
  session_store: FixAlchemy.SessionStore.Ets,
  session_store_opts: [table: :my_fix_sequences],
  # ...
)

The key is the FIX session identity — BeginString, SenderCompID, TargetCompID — so a session finds its own numbers whichever process is running it. Without a store, numbering starts at 1 on every connection and nothing is written.

FixAlchemy.SessionStore.Ets ships with the engine and survives the process, its supervisor, and the connection, but not the node. For storage that outlives the node, implement the behaviour over whatever you already run — an embedded key/value store, a relational table, or a file. record_outbound/2 and record_inbound/2 are called for every message, and what gets written and when is entirely the implementation's choice: write through, batch and flush on close/1, or reserve a block of numbers at load/1 and write once per block. A load/1 returning a number higher than any actually used is safe — the gap is answered with SequenceReset-GapFill — but one lower ends the session.

Market data fan-out

A client subscribes to a symbol, but the symbol does not identify the price it receives. A subscription resolves against that client's entitlements to a particular source, depth, and markup, so one price reaches different sessions as different numbers, and reaches some sessions not at all.

FixAlchemy.Feed models that as an ordered path of permission stages. Subscriptions sharing a prefix share the computation of that prefix: the stages form a tree, a published price walks it from the root, and each stage is applied once however many sessions lie beneath it.

# in the session process entitled to the raw tier
FixAlchemy.Feed.subscribe("AUD/USD", [{:stream, "PRIME"}, {:spread, :tier_a}], md_req_id)

# in a session process entitled to the same tier, marked up
FixAlchemy.Feed.subscribe(
  "AUD/USD",
  [{:stream, "PRIME"}, {:spread, :tier_a}, {:markup, 12}],
  other_md_req_id
)

# from wherever prices arrive
FixAlchemy.Feed.publish("AUD/USD", {:stream, "PRIME"}, price_id, price, MyVenue.Prices)

The spread is applied once and serves both branches; the markup is applied once below it. However many sessions sit on either node, the arithmetic runs twice. Each subscriber receives the value as it reached its own node:

{:feed, %{symbol: symbol, price_id: price_id, price: price, ref: ref}}

price_id identifies the source price and is identical for every session a publish reaches, whatever transforms were applied. Given it you can recover the source and re-apply any client's path to reproduce exactly what that client should have seen. ref is the term given at subscribe time — the MDReqID(262), or whatever you track. The path is not delivered: it names the markup and commission applied, which is yours and not the client's.

A stage is a comparable term, not a function. {:markup, 12} and not fn price -> price * 1.0012 end, which produces an unequal term each time it is written and would split a node that should be shared. The venue turns a stage into a computation:

defmodule MyVenue.Prices do
  @behaviour FixAlchemy.Feed.Pipeline

  @impl true
  def apply_stage({:spread, :tier_a}, price) do
    %{price | bid: price.bid - 1, ask: price.ask + 1}
  end

  def apply_stage({:markup, points}, price) do
    %{price | bid: price.bid - points, ask: price.ask + points}
  end
end

Stages must appear in a consistent order across subscriptions. [a, b] and [b, a] are different paths sharing no work beyond the root — correctly, since their values differ — but a venue that lets clients order stages freely fragments its own tree. Subscribers are monitored, so a session that exits drops its subscriptions and prunes the branches it was holding open.

Feed is single-node: a session on one node does not receive a price published on another.

Architecture

FixAlchemy.Engine is the session protocol for one connection, independent of which side dialed: framing, sequence numbers in both directions, heartbeats and TestRequests, ResendRequests and replay, inbound validation, and delivery to subscribers. It does not open, accept, or re-open sockets — a session process does that and hands it a connected socket.

Two session processes drive an engine. FixAlchemy.Client dials, sends the Logon, and reconnects with backoff. FixAlchemy.Server.Session is handed an accepted socket and validates the Logon it receives. The engine handles neither Logon nor Logout itself; it reports each as an action for its session process to decide, because what they mean depends on which side you are.

An initiated connection is a supervision tree — the session process first, then the subscribers you supply:

SessionSupervisor          # rest_for_one
├── Client        # the socket: dial, Logon, reconnect; drives an Engine
└── subscribers…  # register on the dispatch bus; fed matching messages

An acceptor is a listener and a session per connection:

Server                     # rest_for_one
├── DynamicSupervisor      # one Server.Session per accepted connection
└── Server.Acceptor        # owns the listening socket

Application messages reach subscribers by MsgType through FixAlchemy.Dispatch. Field decoding, and any decode failure, happens in a subscriber rather than on the socket loop. FixAlchemy.Parser does the framing and, on a subscriber's request, the field decode for a single message.

Performance

Single core, FIX 4.4, hardware-dependent:

A Feed publish is dominated by delivery, not by the transforms: a stage costs around 0.014µs, a delivery around 0.45µs. Sharing a prefix across sessions saves transform work, which matters when a venue's transforms are expensive and rounds to nothing when they are arithmetic. Publishing runs in the calling process and parallelises its fan-out across schedulers, so several symbols published from several processes scale across cores.

Reproduce with mix run bench/feed_bench.exs and the other scripts in bench/.

Running on the BEAM

Each session is a supervised process: a crash or a slow handler on one connection does not affect another, and sessions run concurrently across cores. Reconnection, backpressure, and hot code upgrades are provided by the runtime.

Testing

mix test

License

MIT