sample_app/
sample_app.rs

1//! Sample app: a miniature market-data store read entirely through merge
2//! iterators. One session of data is declared up front; every section below is
3//! a different query against it.
4//!
5//! Run the base with `cargo run --example sample_app`; add `--features full`
6//! to light up the opt-in sections.
7//!
8//! The store holds three things, the way an LSM-backed tick store does:
9//! a trade tape per venue, a bid ladder per venue, and reference and last-price
10//! rows spread across levels (oldest flushed level first, live memtable last).
11//!
12//! * base       - consolidate the per-venue tapes into one chronological tape
13//! * seek-to    - read one half-open session window out of that tape
14//! * reverse    - walk the consolidated bid ladder down from the top of book
15//! * tombstones - resolve the reference rows, honouring a delisting
16//! * dedup      - collapse the last-price rows to the freshest per symbol
17//! * priority   - the same rows, with the memtable stated as authoritative
18
19use subms_merge_iterator::MergeIterator;
20
21/// Trade timestamps in ns since the session epoch, one ascending tape per
22/// venue. 9_300 is the open and 9_800 the close.
23const VENUE_TAPES: [&[i64]; 3] = [
24    &[8_000, 9_100, 9_400, 9_800],
25    &[8_500, 9_300, 9_600],
26    &[9_050, 9_450, 9_900],
27];
28
29/// Bid price levels in ticks, one descending ladder per venue - the order a
30/// depth feed already publishes them in.
31const BID_LADDERS: [&[i64]; 2] = [&[10_120, 10_105, 10_101, 10_095], &[10_118, 10_110, 10_099]];
32
33/// Instrument reference rows, oldest level first, sorted by symbol within a
34/// level. `None` is a tombstone: the delisting written to the newest level.
35const REFERENCE_LEVELS: [&[(&str, Option<&str>)]; 3] = [
36    &[
37        ("AAPL", Some("listed")),
38        ("ENRN", Some("listed")),
39        ("MSFT", Some("listed")),
40    ],
41    &[("AAPL", Some("listed-adr"))],
42    &[("ENRN", None)],
43];
44
45/// Last-price rows. The flushed level is stale for AAPL; the memtable holds the
46/// write that has not reached disk yet.
47#[cfg(any(feature = "dedup", feature = "priority"))]
48const PRICE_FLUSHED: &[(&str, i64)] = &[("AAPL", 150), ("MSFT", 300)];
49#[cfg(any(feature = "dedup", feature = "priority"))]
50const PRICE_MEMTABLE: &[(&str, i64)] = &[("AAPL", 152)];
51
52fn main() {
53    println!(
54        "market-data store: {} venue tapes, {} bid ladders, {} reference levels",
55        VENUE_TAPES.len(),
56        BID_LADDERS.len(),
57        REFERENCE_LEVELS.len()
58    );
59
60    base_consolidated_tape();
61
62    #[cfg(feature = "seek-to")]
63    session_window_scan();
64
65    #[cfg(feature = "reverse")]
66    walk_bid_ladder_down();
67
68    #[cfg(feature = "tombstones")]
69    resolve_reference_rows();
70
71    #[cfg(feature = "dedup")]
72    compact_last_prices();
73
74    #[cfg(feature = "priority")]
75    memtable_wins_the_read();
76}
77
78fn tapes() -> Vec<std::iter::Copied<std::slice::Iter<'static, i64>>> {
79    VENUE_TAPES.iter().map(|t| t.iter().copied()).collect()
80}
81
82/// Base API: each venue publishes trades already sorted by exchange timestamp.
83/// Merging their heads on a min-heap gives one chronological consolidated tape
84/// without materialising and re-sorting the union.
85fn base_consolidated_tape() {
86    println!("\n== base: consolidated trade tape ==");
87    let merge = MergeIterator::new(tapes());
88    println!("  live venues: {}", merge.live_streams());
89    println!("  earliest trade: {:?}", merge.peek());
90    let tape: Vec<i64> = merge.collect();
91    println!("  {} trades in order: {tape:?}", tape.len());
92    assert_eq!(tape.len(), 10, "every trade appears once");
93    assert!(
94        tape.windows(2).all(|w| w[0] <= w[1]),
95        "the tape stays chronological"
96    );
97}
98
99/// `seek-to`: a regular-session query wants `[open, close)` and nothing else.
100/// `seek(open)` advances every venue past its pre-market ticks in one bounded
101/// reposition; `set_upper_bound(close)` ends the scan, so the caller pulls
102/// `next()` until it stops rather than testing each element itself.
103#[cfg(feature = "seek-to")]
104fn session_window_scan() {
105    use subms_merge_iterator::SeekableMergeIterator;
106    println!("\n== seek-to: one session window out of the tape ==");
107    let (open, close) = (9_300, 9_800);
108
109    let mut scan = SeekableMergeIterator::new(tapes());
110    scan.seek(&open);
111    scan.set_upper_bound(close);
112
113    let window: Vec<i64> = scan.collect();
114    println!("  window [{open}, {close}): {window:?}");
115    assert_eq!(
116        window,
117        vec![9_300, 9_400, 9_450, 9_600],
118        "half-open: the close tick is excluded"
119    );
120}
121
122/// `reverse`: a bid ladder is quoted best-price-first, so it arrives sorted
123/// descending already. Merging the ladders descending gives one consolidated
124/// book. Pricing a marketable sell only needs the levels between the touch and
125/// a limit, so `seek_for_prev` starts the walk and `set_lower_bound` ends it -
126/// the rest of the book is never read.
127#[cfg(feature = "reverse")]
128fn walk_bid_ladder_down() {
129    use subms_merge_iterator::ReverseMergeIterator;
130    println!("\n== reverse: walk the consolidated bid ladder down ==");
131    let ladders: Vec<_> = BID_LADDERS.iter().map(|l| l.iter().copied()).collect();
132
133    let mut book = ReverseMergeIterator::new(ladders);
134    println!("  best bid across venues: {:?}", book.peek());
135
136    let limit = 10_100;
137    book.seek_for_prev(&10_110);
138    book.set_lower_bound(limit);
139
140    let fillable: Vec<i64> = book.collect();
141    println!("  levels from 10110 down to the {limit} limit: {fillable:?}");
142    assert_eq!(
143        fillable,
144        vec![10_110, 10_105, 10_101],
145        "descending, and the lower bound is inclusive"
146    );
147}
148
149/// `tombstones`: a reference read across three levels. The newest level's
150/// delisting shadows the same symbol everywhere below it, so the key leaves the
151/// result entirely; AAPL takes its newer status from the middle level.
152#[cfg(feature = "tombstones")]
153fn resolve_reference_rows() {
154    use subms_merge_iterator::{TombstoneEntry, TombstoneMergeIterator};
155    println!("\n== tombstones: resolve the reference rows ==");
156    let levels: Vec<std::vec::IntoIter<TombstoneEntry<&str, &str>>> = REFERENCE_LEVELS
157        .iter()
158        .map(|rows| {
159            rows.iter()
160                .map(|&(sym, status)| match status {
161                    Some(s) => TombstoneEntry::live(sym, s),
162                    None => TombstoneEntry::tombstone(sym),
163                })
164                .collect::<Vec<_>>()
165                .into_iter()
166        })
167        .collect();
168
169    let resolved: Vec<(&str, &str)> = TombstoneMergeIterator::new(levels)
170        .map(|e| (e.key, e.value.unwrap()))
171        .collect();
172    println!("  live instruments: {resolved:?}");
173    assert_eq!(
174        resolved,
175        vec![("AAPL", "listed-adr"), ("MSFT", "listed")],
176        "the delisted symbol is shadowed out, AAPL takes the newer row"
177    );
178}
179
180/// `dedup`: the same symbol appears in the flushed level and the memtable.
181/// Latest-source-wins collapses each symbol to one row, which is the compaction
182/// output. Registration order carries the recency here.
183#[cfg(feature = "dedup")]
184fn compact_last_prices() {
185    use subms_merge_iterator::{DedupEntry, DedupMergeIterator};
186    println!("\n== dedup: compact the last-price rows ==");
187    let flushed = PRICE_FLUSHED
188        .iter()
189        .map(|&(k, v)| DedupEntry::new(k, v))
190        .collect::<Vec<_>>();
191    let memtable = PRICE_MEMTABLE
192        .iter()
193        .map(|&(k, v)| DedupEntry::new(k, v))
194        .collect::<Vec<_>>();
195
196    let compacted: Vec<(&str, i64)> =
197        DedupMergeIterator::new([flushed.into_iter(), memtable.into_iter()])
198            .map(|e| (e.key, e.value))
199            .collect();
200    println!("  compacted last prices: {compacted:?}");
201    assert_eq!(compacted, vec![("AAPL", 152), ("MSFT", 300)]);
202}
203
204/// `priority`: the same two sources, registered the other way round - a read
205/// path holds the memtable first. Registration order now says the wrong thing,
206/// so authority is stated explicitly and the merge still resolves AAPL to the
207/// unflushed write.
208#[cfg(feature = "priority")]
209fn memtable_wins_the_read() {
210    use subms_merge_iterator::{PriorityEntry, PriorityMergeIterator, PrioritySource};
211    println!("\n== priority: the memtable is authoritative ==");
212    let memtable = PrioritySource::new(
213        100,
214        PRICE_MEMTABLE
215            .iter()
216            .map(|&(k, v)| PriorityEntry::new(k, v))
217            .collect::<Vec<_>>()
218            .into_iter(),
219    );
220    let flushed = PrioritySource::new(
221        10,
222        PRICE_FLUSHED
223            .iter()
224            .map(|&(k, v)| PriorityEntry::new(k, v))
225            .collect::<Vec<_>>()
226            .into_iter(),
227    );
228
229    let view: Vec<(&str, i64)> = PriorityMergeIterator::new([memtable, flushed])
230        .map(|e| (e.key, e.value))
231        .collect();
232    println!("  resolved read view: {view:?}");
233    assert_eq!(
234        view,
235        vec![("AAPL", 152), ("MSFT", 300)],
236        "the memtable wins AAPL despite being registered first"
237    );
238}
sample_app/sample_app.rs

sample_app/
sample_app.rs
