finance_query_core/streaming/
mod.rs

1//! Streaming functionality for real-time data.
2//!
3//! This module provides async streams for continuously fetching financial data
4//! at configurable intervals. These streams can be used with any async runtime
5//! and integrated into WebSocket servers or other streaming applications.
6
7use crate::client::error::YahooError;
8use crate::client::YahooFinanceClient;
9use crate::models::SimpleQuote;
10use crate::websocket::QuotesUpdate;
11use async_stream::stream;
12use chrono::Utc;
13use futures_util::Stream;
14use serde_json::Value;
15use std::pin::Pin;
16use std::sync::Arc;
17use std::time::Duration;
18use tokio::time::interval;
19use tracing::{debug, error};
20
21/// A stream that yields quote updates at regular intervals.
22pub struct QuoteStream;
23
24impl QuoteStream {
25    /// Create a new quote stream that fetches quotes for the given symbols
26    /// at the specified interval.
27    ///
28    /// # Arguments
29    /// * `client` - The Yahoo Finance client to use for fetching quotes
30    /// * `symbols` - List of stock symbols to track
31    /// * `poll_interval` - How often to fetch new quotes
32    ///
33    /// # Example
34    /// ```rust,ignore
35    /// use finance_query_core::{QuoteStream, YahooFinanceClient};
36    /// use std::time::Duration;
37    /// use futures_util::StreamExt;
38    ///
39    /// let stream = QuoteStream::new(&client, vec!["AAPL", "GOOGL"], Duration::from_secs(5));
40    /// 
41    /// while let Some(result) = stream.next().await {
42    ///     match result {
43    ///         Ok(update) => println!("Got {} quotes", update.len()),
44    ///         Err(e) => eprintln!("Error: {}", e),
45    ///     }
46    /// }
47    /// ```
48    pub fn new(
49        client: Arc<YahooFinanceClient>,
50        symbols: Vec<String>,
51        poll_interval: Duration,
52    ) -> Pin<Box<dyn Stream<Item = Result<QuotesUpdate, YahooError>> + Send>> {
53        Box::pin(stream! {
54            let mut ticker = interval(poll_interval);
55            
56            loop {
57                ticker.tick().await;
58                debug!("Fetching quotes for {:?}", symbols);
59                
60                let symbol_refs: Vec<&str> = symbols.iter().map(|s| s.as_str()).collect();
61                
62                match client.get_simple_quotes(&symbol_refs).await {
63                    Ok(data) => {
64                        let quotes = parse_simple_quotes(&data);
65                        let update = QuotesUpdate::with_timestamp(quotes, Utc::now());
66                        yield Ok(update);
67                    }
68                    Err(e) => {
69                        error!("Failed to fetch quotes: {}", e);
70                        yield Err(e);
71                    }
72                }
73            }
74        })
75    }
76
77    /// Create a quote stream with a default 5-second interval.
78    pub fn with_default_interval(
79        client: Arc<YahooFinanceClient>,
80        symbols: Vec<String>,
81    ) -> Pin<Box<dyn Stream<Item = Result<QuotesUpdate, YahooError>> + Send>> {
82        Self::new(client, symbols, Duration::from_secs(5))
83    }
84}
85
86/// A stream that yields a single quote update for one symbol.
87pub struct SingleQuoteStream;
88
89impl SingleQuoteStream {
90    /// Create a stream for a single symbol.
91    pub fn new(
92        client: Arc<YahooFinanceClient>,
93        symbol: String,
94        poll_interval: Duration,
95    ) -> Pin<Box<dyn Stream<Item = Result<SimpleQuote, YahooError>> + Send>> {
96        Box::pin(stream! {
97            let mut ticker = interval(poll_interval);
98            
99            loop {
100                ticker.tick().await;
101                debug!("Fetching quote for {}", symbol);
102                
103                match client.get_simple_quotes(&[symbol.as_str()]).await {
104                    Ok(data) => {
105                        let quotes = parse_simple_quotes(&data);
106                        if let Some(quote) = quotes.into_iter().next() {
107                            yield Ok(quote);
108                        }
109                    }
110                    Err(e) => {
111                        error!("Failed to fetch quote for {}: {}", symbol, e);
112                        yield Err(e);
113                    }
114                }
115            }
116        })
117    }
118}
119
120/// Parse simple quotes from Yahoo Finance API response.
121fn parse_simple_quotes(data: &Value) -> Vec<SimpleQuote> {
122    let mut quotes = Vec::new();
123    
124    if let Some(results) = data
125        .get("quoteResponse")
126        .and_then(|qr| qr.get("result"))
127        .and_then(|r| r.as_array())
128    {
129        for result in results {
130            let quote = SimpleQuote {
131                symbol: result.get("symbol")
132                    .and_then(|s| s.as_str())
133                    .unwrap_or("")
134                    .to_string(),
135                name: result.get("longName")
136                    .or_else(|| result.get("shortName"))
137                    .and_then(|n| n.as_str())
138                    .unwrap_or("")
139                    .to_string(),
140                price: result.get("regularMarketPrice")
141                    .and_then(|p| p.as_f64())
142                    .map(|p| format!("{:.2}", p))
143                    .unwrap_or_else(|| "0.00".to_string()),
144                pre_market_price: result.get("preMarketPrice")
145                    .and_then(|p| p.as_f64())
146                    .map(|p| format!("{:.2}", p)),
147                after_hours_price: result.get("postMarketPrice")
148                    .and_then(|p| p.as_f64())
149                    .map(|p| format!("{:.2}", p)),
150                change: result.get("regularMarketChange")
151                    .and_then(|c| c.as_f64())
152                    .map(|c| format!("{:+.2}", c))
153                    .unwrap_or_else(|| "0.00".to_string()),
154                percent_change: result.get("regularMarketChangePercent")
155                    .and_then(|p| p.as_f64())
156                    .map(|p| format!("{:+.2}%", p))
157                    .unwrap_or_else(|| "0.00%".to_string()),
158                logo: None,
159            };
160            quotes.push(quote);
161        }
162    }
163    
164    quotes
165}
166
167#[cfg(test)]
168mod tests {
169    use super::*;
170
171    #[test]
172    fn test_parse_simple_quotes() {
173        let data = serde_json::json!({
174            "quoteResponse": {
175                "result": [
176                    {
177                        "symbol": "AAPL",
178                        "longName": "Apple Inc.",
179                        "regularMarketPrice": 175.50,
180                        "regularMarketChange": 2.50,
181                        "regularMarketChangePercent": 1.45
182                    }
183                ]
184            }
185        });
186
187        let quotes = parse_simple_quotes(&data);
188        assert_eq!(quotes.len(), 1);
189        assert_eq!(quotes[0].symbol, "AAPL");
190        assert_eq!(quotes[0].name, "Apple Inc.");
191        assert_eq!(quotes[0].price, "175.50");
192        assert_eq!(quotes[0].change, "+2.50");
193        assert_eq!(quotes[0].percent_change, "+1.45%");
194    }
195}
finance_query_core/streaming/mod.rs

finance_query_core/streaming/
mod.rs
