opendeviationbar-streaming 13.78.1

Real-time streaming engine for open deviation bar processing
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

//! Integration tests for gap-fill command channel dispatch (Issue #257).
//!
//! Verifies that `GapFillCommand` can be sent via `GapFillSender` and
//! received + responded to via the `GapFillReceiver`.

use std::sync::Arc;
use tokio::sync::oneshot;

use opendeviationbar_streaming::gap::{GapFillCommand, GapFillResult, gap_fill_channel};

#[tokio::test]
async fn test_command_channel_roundtrip() {
    let (tx, mut rx) = gap_fill_channel(16);

    // Simulate Python fill_gap() call
    let (response_tx, response_rx) = oneshot::channel();
    let cmd = GapFillCommand {
        symbol: Arc::from("BTCUSDT"),
        from_tid: 1000,
        to_tid: 2000,
        response_tx,
    };
    tx.send(cmd).await.unwrap();

    // Simulate engine receiving + processing
    let received = rx.recv().await.unwrap();
    assert_eq!(&*received.symbol, "BTCUSDT");
    assert_eq!(received.from_tid, 1000);
    assert_eq!(received.to_tid, 2000);

    // Send response back (as engine would after fill)
    received
        .response_tx
        .send(GapFillResult {
            trades_fetched: 1000,
            bars_produced: 42,
            duration_ms: 850,
            partial: false,
        })
        .unwrap();

    // Verify caller receives the result
    let result = response_rx.await.unwrap();
    assert_eq!(result.trades_fetched, 1000);
    assert_eq!(result.bars_produced, 42);
    assert_eq!(result.duration_ms, 850);
    assert!(!result.partial);
}

#[tokio::test]
async fn test_command_channel_multiple_symbols() {
    // Verify per-symbol command channels are independent
    let (btc_tx, mut btc_rx) = gap_fill_channel(4);
    let (eth_tx, mut eth_rx) = gap_fill_channel(4);

    // Send to BTC
    let (resp_tx, _) = oneshot::channel();
    btc_tx
        .send(GapFillCommand {
            symbol: Arc::from("BTCUSDT"),
            from_tid: 100,
            to_tid: 200,
            response_tx: resp_tx,
        })
        .await
        .unwrap();

    // Send to ETH
    let (resp_tx, _) = oneshot::channel();
    eth_tx
        .send(GapFillCommand {
            symbol: Arc::from("ETHUSDT"),
            from_tid: 500,
            to_tid: 600,
            response_tx: resp_tx,
        })
        .await
        .unwrap();

    // Each channel receives only its own command
    let btc_cmd = btc_rx.recv().await.unwrap();
    assert_eq!(&*btc_cmd.symbol, "BTCUSDT");

    let eth_cmd = eth_rx.recv().await.unwrap();
    assert_eq!(&*eth_cmd.symbol, "ETHUSDT");
}

#[tokio::test]
async fn test_command_channel_sender_cloneable() {
    // Verify sender can be cloned (needed for multiple callers)
    let (tx, mut rx) = gap_fill_channel(4);
    let tx2 = tx.clone();

    let (resp_tx, _) = oneshot::channel();
    tx2.send(GapFillCommand {
        symbol: Arc::from("BTCUSDT"),
        from_tid: 1,
        to_tid: 10,
        response_tx: resp_tx,
    })
    .await
    .unwrap();

    let cmd = rx.recv().await.unwrap();
    assert_eq!(cmd.from_tid, 1);
}

#[tokio::test]
async fn test_command_channel_caller_dropped() {
    // When the caller drops the oneshot receiver, sending the response
    // should not panic (just return Err, which we ignore)
    let (tx, mut rx) = gap_fill_channel(4);

    let (resp_tx, resp_rx) = oneshot::channel();
    tx.send(GapFillCommand {
        symbol: Arc::from("BTCUSDT"),
        from_tid: 1,
        to_tid: 10,
        response_tx: resp_tx,
    })
    .await
    .unwrap();

    // Caller drops their receiver
    drop(resp_rx);

    // Engine still receives command
    let cmd = rx.recv().await.unwrap();
    // Sending response should not panic, just returns Err
    let send_result = cmd.response_tx.send(GapFillResult {
        trades_fetched: 0,
        bars_produced: 0,
        duration_ms: 0,
        partial: false,
    });
    assert!(send_result.is_err());
}