replay_demo/
replay_demo.rs

1//! Phase ν — replay demo.
2//!
3//! Writes 200 synthetic ticker events to a temp storage directory, then
4//! replays them at 10× speed and counts received events.
5//!
6//! Run with:
7//!   cargo run --example replay_demo --release
8
9use std::path::PathBuf;
10
11use digdigdig3::core::types::{
12    AccountType, ExchangeId, StreamEvent, SubscriptionRequest, Symbol, Ticker,
13};
14use digdigdig3_station::{ReplayConfig, ReplayHub, ReplayRate, StorageConfig, StorageManager, StreamKey};
15use futures_util::StreamExt;
16
17#[tokio::main]
18async fn main() -> Result<(), Box<dyn std::error::Error>> {
19    // ── 1. Write synthetic events ─────────────────────────────────────────────
20
21    let tmp_dir = {
22        let mut d = std::env::temp_dir();
23        d.push("dig3_replay_demo");
24        d
25    };
26    std::fs::remove_dir_all(&tmp_dir).ok();
27    std::fs::create_dir_all(&tmp_dir)?;
28
29    let storage = StorageManager::new(StorageConfig {
30        root: tmp_dir.clone(),
31        default_retention_days: 365,
32        orderbook_snapshot_interval_secs: 0,
33    })?;
34
35    let key = StreamKey {
36        exchange: "binance".into(),
37        account: "spot".into(),
38        symbol: "BTCUSDT".into(),
39        stream_kind: "ticker".into(),
40    };
41
42    let base_ms = chrono::Utc::now().timestamp_millis();
43    let n_events = 200usize;
44
45    for i in 0..n_events {
46        let ts_ms = base_ms + (i as i64) * 100; // 100 ms apart → 20 s simulated
47        let price = 50_000.0 + i as f64;
48        let ev = StreamEvent::Ticker(Ticker {
49            symbol: "BTCUSDT".into(),
50            last_price: price,
51            bid_price: Some(price - 1.0),
52            ask_price: Some(price + 1.0),
53            volume_24h: Some(1000.0),
54            quote_volume_24h: None,
55            price_change_24h: None,
56            price_change_percent_24h: None,
57            high_24h: None,
58            low_24h: None,
59            timestamp: ts_ms,
60        });
61        let payload = serde_json::to_vec(&ev)?;
62        storage.append(&key, ts_ms, &payload).await?;
63    }
64    storage.flush_all().await?;
65    println!("Wrote {n_events} synthetic ticker events to {}", tmp_dir.display());
66
67    // ── 2. Replay at 10× speed ────────────────────────────────────────────────
68
69    let config = ReplayConfig {
70        storage_root: PathBuf::from(&tmp_dir),
71        rate: ReplayRate::Accelerated(10.0),
72        from_ms: Some(base_ms),
73        to_ms: Some(base_ms + (n_events as i64) * 100),
74    };
75    let hub = ReplayHub::new(config).await?;
76    hub.connect_full(ExchangeId::Binance, &[AccountType::Spot], false)
77        .await?;
78
79    let ws = hub.ws(ExchangeId::Binance, AccountType::Spot).unwrap();
80    ws.subscribe(SubscriptionRequest::ticker(Symbol::with_raw(
81        "BTC",
82        "USDT",
83        "BTCUSDT".into(),
84    )))
85    .await?;
86
87    let mut stream = ws.event_stream();
88    let mut count = 0usize;
89    let start = std::time::Instant::now();
90
91    while let Some(ev) = stream.next().await {
92        match ev {
93            Ok(StreamEvent::Ticker(_t)) => {
94                count += 1;
95                if count % 50 == 0 {
96                    println!("  {count} events received…");
97                }
98                if count >= n_events {
99                    break;
100                }
101            }
102            Ok(_other) => {}
103            Err(e) => eprintln!("stream error: {e}"),
104        }
105    }
106
107    let elapsed = start.elapsed();
108    println!(
109        "Replay done: {count} events in {:.2}s  (simulated 20s at 10×)",
110        elapsed.as_secs_f64()
111    );
112
113    // ── 3. Cleanup ────────────────────────────────────────────────────────────
114    std::fs::remove_dir_all(&tmp_dir).ok();
115
116    Ok(())
117}
replay_demo/replay_demo.rs

replay_demo/
replay_demo.rs
