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 {
49            symbol: "BTCUSDT".to_string(),
50            ticker: Ticker {
51                last_price: price,
52                bid_price: Some(price - 1.0),
53                ask_price: Some(price + 1.0),
54                volume_24h: Some(1000.0),
55                quote_volume_24h: None,
56                price_change_24h: None,
57                price_change_percent_24h: None,
58                high_24h: None,
59                low_24h: None,
60                timestamp: ts_ms,
61            },
62        };
63        let payload = serde_json::to_vec(&ev)?;
64        storage.append(&key, ts_ms, &payload).await?;
65    }
66    storage.flush_all().await?;
67    println!("Wrote {n_events} synthetic ticker events to {}", tmp_dir.display());
68
69    // ── 2. Replay at 10× speed ────────────────────────────────────────────────
70
71    let config = ReplayConfig {
72        storage_root: PathBuf::from(&tmp_dir),
73        rate: ReplayRate::Accelerated(10.0),
74        from_ms: Some(base_ms),
75        to_ms: Some(base_ms + (n_events as i64) * 100),
76    };
77    let hub = ReplayHub::new(config).await?;
78    hub.connect_full(ExchangeId::Binance, &[AccountType::Spot], false)
79        .await?;
80
81    let ws = hub.ws(ExchangeId::Binance, AccountType::Spot).unwrap();
82    ws.subscribe(SubscriptionRequest::ticker(Symbol::with_raw(
83        "BTC",
84        "USDT",
85        "BTCUSDT".into(),
86    )))
87    .await?;
88
89    let mut stream = ws.event_stream();
90    let mut count = 0usize;
91    let start = std::time::Instant::now();
92
93    while let Some(ev) = stream.next().await {
94        match ev {
95            Ok(StreamEvent::Ticker { .. }) => {
96                count += 1;
97                if count % 50 == 0 {
98                    println!("  {count} events received…");
99                }
100                if count >= n_events {
101                    break;
102                }
103            }
104            Ok(_other) => {}
105            Err(e) => eprintln!("stream error: {e}"),
106        }
107    }
108
109    let elapsed = start.elapsed();
110    println!(
111        "Replay done: {count} events in {:.2}s  (simulated 20s at 10×)",
112        elapsed.as_secs_f64()
113    );
114
115    // ── 3. Cleanup ────────────────────────────────────────────────────────────
116    std::fs::remove_dir_all(&tmp_dir).ok();
117
118    Ok(())
119}
replay_demo/replay_demo.rs

replay_demo/
replay_demo.rs
