OCaml for High-Frequency Trading: Production Patterns

1open Core
2open Async
3
4module Market_data = struct
5  type t = {
6    symbol: string;
7    price: int64;  (* Fixed-point, e.g., cents *)
8    quantity: int;
9    timestamp: Time_ns.t;
10  }
11  [@@deriving sexp, bin_io]
12end
13
14(* Async UDP receiver *)
15let receive_market_data ~port =
16  let%bind socket = 
17    Udp.bind (Udp.Bind_to_port port) 
18  in
19  
20  let rec loop () =
21    let%bind (buf, addr) = Udp.recvfrom socket in
22    
23    (* Parse binary message *)
24    let data = 
25      Market_data.bin_read_t 
26        (Bigstring.of_string buf) 
27        ~pos_ref:(ref 0)
28    in
29    
30    (* Process asynchronously *)
31    don't_wait_for (process_market_data data);
32    
33    loop ()
34  in
35  loop ()
36
37and process_market_data data =
38  (* Update order book, calculate signals, etc. *)
39  Log.Global.info "Received: %s @ %Ld" 
40    data.symbol 
41    data.price;
42  return ()
43
44(* Start receiver *)
45let () =
46  Command.async
47    ~summary:"Market data receiver"
48    Command.Let_syntax.(
49      let%map_open port = 
50        flag "port" (required int) ~doc:"PORT UDP port"
51      in
52      fun () -> receive_market_data ~port
53    )
54  |> Command_unix.run
55

1open Core
2open Incr.Let_syntax
3
4module Order = struct
5  type t = {
6    price: int64;
7    quantity: int;
8    order_id: int64;
9  }
10  [@@deriving compare, sexp]
11end
12
13module Order_book = struct
14  type side = Bid | Ask
15  
16  type t = {
17    bids: Order.t list Incr.t;
18    asks: Order.t list Incr.t;
19  }
20
21  (* Incremental best bid calculation *)
22  let best_bid t =
23    let%map bids = t.bids in
24    List.hd bids
25
26  (* Incremental mid price *)
27  let mid_price t =
28    let%map best_bid = best_bid t
29    and best_ask = 
30      let%map asks = t.asks in
31      List.hd asks
32    in
33    match best_bid, best_ask with
34    | Some bid, Some ask ->
35        Some (Int64.((bid.price + ask.price) / 2L))
36    | _ -> None
37
38  (* Incremental spread *)
39  let spread t =
40    let%map best_bid = best_bid t
41    and best_ask =
42      let%map asks = t.asks in
43      List.hd asks
44    in
45    match best_bid, best_ask with
46    | Some bid, Some ask ->
47        Some Int64.(ask.price - bid.price)
48    | _ -> None
49
50  (* Add order incrementally *)
51  let add_order t side order =
52    match side with
53    | Bid ->
54        let bids = 
55          Incr.map t.bids ~f:(fun bids ->
56            List.merge bids [order] ~compare:(fun a b ->
57              Int64.compare b.price a.price  (* Descending *)
58            )
59          )
60        in
61        { t with bids }
62    | Ask ->
63        let asks =
64          Incr.map t.asks ~f:(fun asks ->
65            List.merge asks [order] ~compare:(fun a b ->
66              Int64.compare a.price b.price  (* Ascending *)
67            )
68          )
69        in
70        { t with asks }
71end
72
73(* Usage: only recompute affected values *)
74let book = {
75  Order_book.bids = Incr.Var.create [] |> Incr.Var.watch;
76  asks = Incr.Var.create [] |> Incr.Var.watch;
77}
78
79let mid = Order_book.mid_price book
80let spread = Order_book.spread book
81
82(* Stabilize to propagate changes *)
83let () = Incr.stabilize ()
84

1open Core
2
3(* GADT for FIX message types *)
4type _ fix_msg =
5  | NewOrderSingle : {
6      cl_ord_id: string;
7      symbol: string;
8      side: [`Buy | `Sell];
9      quantity: int;
10      price: int64 option;
11    } -> [`NewOrderSingle] fix_msg
12  
13  | ExecutionReport : {
14      order_id: string;
15      exec_id: string;
16      exec_type: [`New | `PartialFill | `Fill];
17      ord_status: [`New | `PartiallyFilled | `Filled];
18      symbol: string;
19      side: [`Buy | `Sell];
20      leaves_qty: int;
21      cum_qty: int;
22    } -> [`ExecutionReport] fix_msg
23  
24  | OrderCancelRequest : {
25      orig_cl_ord_id: string;
26      cl_ord_id: string;
27      symbol: string;
28      side: [`Buy | `Sell];
29    } -> [`OrderCancelRequest] fix_msg
30
31(* Type-safe message handling *)
32let handle_new_order 
33  (msg : [`NewOrderSingle] fix_msg) 
34  : [`ExecutionReport] fix_msg Async.Deferred.t 
35=
36  let NewOrderSingle { cl_ord_id; symbol; side; quantity; price } = msg in
37  
38  (* Validate and execute order *)
39  let%bind.Async order_id = execute_order ~symbol ~side ~quantity ~price in
40  
41  return (ExecutionReport {
42    order_id;
43    exec_id = generate_exec_id ();
44    exec_type = `New;
45    ord_status = `New;
46    symbol;
47    side;
48    leaves_qty = quantity;
49    cum_qty = 0;
50  })
51
52(* Serialize with type safety *)
53let serialize : type a. a fix_msg -> string = function
54  | NewOrderSingle { cl_ord_id; symbol; side; quantity; price } ->
55      sprintf "35=D\x0111=%s\x0155=%s\x0154=%s\x0138=%d\x01"
56        cl_ord_id
57        symbol
58        (match side with `Buy -> "1" | `Sell -> "2")
59        quantity
60      ^ (match price with
61         | Some p -> sprintf "44=%Ld\x01" p
62         | None -> "")
63  
64  | ExecutionReport { order_id; exec_id; exec_type; ord_status; _ } ->
65      sprintf "35=8\x0137=%s\x0117=%s\x01150=%s\x0139=%s\x01"
66        order_id
67        exec_id
68        (match exec_type with `New -> "0" | `PartialFill -> "1" | `Fill -> "2")
69        (match ord_status with `New -> "0" | `PartiallyFilled -> "1" | `Filled -> "2")
70  
71  | OrderCancelRequest { orig_cl_ord_id; cl_ord_id; symbol; side } ->
72      sprintf "35=F\x0141=%s\x0111=%s\x0155=%s\x0154=%s\x01"
73        orig_cl_ord_id
74        cl_ord_id
75        symbol
76        (match side with `Buy -> "1" | `Sell -> "2")
77

1open Core
2
3let calculate_vwap orders =
4  let total_value = 
5    List.fold orders ~init:0L ~f:(fun acc order ->
6      Int64.(acc + (of_int order.quantity * order.price))
7    )
8  in
9  let total_qty =
10    List.fold orders ~init:0 ~f:(fun acc order ->
11      acc + order.quantity
12    )
13  in
14  if total_qty = 0 then None
15  else Some Int64.(total_value / of_int total_qty)
16
17let%expect_test "vwap calculation" =
18  let orders = [
19    { Order.price = 10000L; quantity = 100; order_id = 1L };
20    { price = 10050L; quantity = 200; order_id = 2L };
21    { price = 10100L; quantity = 100; order_id = 3L };
22  ] in
23  
24  let vwap = calculate_vwap orders in
25  print_s [%sexp (vwap : int64 option)];
26  
27  [%expect {| (10050) |}]
28
29let%expect_test "empty order list" =
30  let vwap = calculate_vwap [] in
31  print_s [%sexp (vwap : int64 option)];
32  
33  [%expect {| () |}]
34

1open Core
2open Landmarks
3
4let process_order order =
5  Landmark.enter "process_order";
6  
7  (* Validate order *)
8  Landmark.enter "validate";
9  let valid = validate_order order in
10  Landmark.exit "validate";
11  
12  if valid then begin
13    (* Execute order *)
14    Landmark.enter "execute";
15    let result = execute_order order in
16    Landmark.exit "execute";
17    result
18  end else
19    Error "Invalid order"
20  
21  |> Landmark.exit "process_order"
22
23(* Generate profiling report *)
24let () =
25  at_exit (fun () ->
26    Landmark_graph.export "profile.json"
27  )
28

1open Core
2open Bigstring
3
4module Binary_parser = struct
5  type t = {
6    buf: Bigstring.t;
7    mutable pos: int;
8  }
9
10  let create buf = { buf; pos = 0 }
11
12  let read_u64_le t =
13    let v = Bigstring.unsafe_get_int64_le t.buf ~pos:t.pos in
14    t.pos <- t.pos + 8;
15    v
16
17  let read_u32_le t =
18    let v = Bigstring.unsafe_get_int32_le t.buf ~pos:t.pos in
19    t.pos <- t.pos + 4;
20    Int32.to_int_exn v
21
22  let read_bytes t len =
23    let v = Bigstring.To_string.sub t.buf ~pos:t.pos ~len in
24    t.pos <- t.pos + len;
25    v
26end
27
28(* Parse market data message *)
29let parse_market_data buf =
30  let parser = Binary_parser.create buf in
31  
32  let symbol = Binary_parser.read_bytes parser 8 in
33  let price = Binary_parser.read_u64_le parser in
34  let quantity = Binary_parser.read_u32_le parser in
35  let timestamp = 
36    Binary_parser.read_u64_le parser 
37    |> Time_ns.of_int63_ns_since_epoch 
38    |> Time_ns.Span.of_int63_ns
39  in
40  
41  { Market_data.symbol; price; quantity; timestamp }
42

1(* dune file *)
2(executable
3 (name trading_engine)
4 (libraries core async incremental)
5 (preprocess (pps ppx_jane ppx_bin_prot))
6 (modes native)
7 (flags (:standard -O3 -unbox-closures -inline 100)))
8
9(* Dockerfile *)
10FROM ocaml/opam:debian-11-ocaml-4.14 as builder
11
12WORKDIR /app
13COPY . .
14
15RUN opam install . --deps-only -y
16RUN eval $(opam env) && dune build --release
17
18FROM debian:11-slim
19COPY --from=builder /app/_build/default/trading_engine.exe /usr/local/bin/
20CMD ["/usr/local/bin/trading_engine.exe"]
21

1(* Set GC parameters at startup *)
2let () =
3  Gc.set {
4    (Gc.get ()) with
5    minor_heap_size = 2 * 1024 * 1024;  (* 2 MB *)
6    major_heap_increment = 8 * 1024 * 1024;  (* 8 MB *)
7    space_overhead = 120;  (* Trigger major GC at 120% overhead *)
8    max_overhead = 500;
9    allocation_policy = 2;  (* Best-fit *)
10  };
11  
12  (* Enable incremental major GC *)
13  Gc.major_slice 0 |> ignore
14