Module Algostream_data_ingestion.Data_quality

Real-time market-data quality monitors.

Sequence-gap detection is parser-side: it runs BEFORE the bus publish so a backpressure-driven try_publish drop cannot cause a false-positive gap alert downstream. Stale-tick and crossed-book checks are evaluated on the parsed payload and instruct the connector whether to publish, drop, or emit an additional Risk_alert.

type stats = {
  1. total_observed : int;
  2. sequence_gaps : int;
  3. dropped_to_gap : int;
  4. stale_ticks : int;
  5. crossed_books : int;
  6. out_of_order_trades : int;
}
type t
val create : exchange:string -> ?stale_threshold_ns:int64 -> unit -> t
type verdict =
  1. | Ok_publish
  2. | Drop_stale of {
    1. age_ns : int64;
    }
  3. | Drop_crossed of {
    1. bid : float;
    2. ask : float;
    }
  4. | Out_of_order
  5. | Gap_then_publish of {
    1. expected : int64;
    2. received : int64;
    3. dropped : int;
    }

Verdict for a single tick or trade-print observation.

What sequence must be

Some n enables gap detection, and commits the caller to a specific guarantee: n is a per-symbol counter that increments by exactly one for each message this connector actually receives. Anything else makes dropped_to_gap meaningless, and it will be meaningless quietly — every message will look like a gap and the counter will read in the millions.

Two feeds already got this wrong. Coinbase's sequence field counts the product's whole full channel while the connector subscribes only to matches, and Binance was passing the trade timestamp in milliseconds, so the reported "dropped" count was the elapsed time between trades. Both now pass the exchange's per-symbol trade id, which is dense over the messages received.

None disables gap detection for that message. Use it when the feed offers no such counter — which is the case for every tick stream here, since neither exchange numbers its quote updates.

val check_market_tick : t -> symbol:string -> exchange_ts_ns:int64 -> ingest_ts_ns:int64 -> bid:float -> ask:float -> sequence:int64 option -> verdict

Inspect a market-tick payload prior to publishing.

val check_trade_print : t -> symbol:string -> exchange_ts_ns:int64 -> ingest_ts_ns:int64 -> sequence:int64 option -> verdict

Inspect a trade-print payload prior to publishing.

val exchange_name : t -> string
val stats : t -> stats