use super::{BatchQuotesResponse, Tickers};
use crate::error::{FinanceError, Result};
use crate::format::Both;
use crate::models::format::Format;
use crate::models::quote::{Quote, QuoteSummaryResponse};
use crate::providers::Capability;
use futures::stream::{self, StreamExt};
use std::collections::HashMap;
use std::sync::Arc;
impl Tickers {
pub async fn quotes(&self) -> Result<BatchQuotesResponse> {
{
let cache = self.quote_cache.read().await;
if self.all_cached(&cache, self.symbols.iter().cloned()) {
let mut response = BatchQuotesResponse::with_capacity(self.symbols.len());
for symbol in &self.symbols {
if let Some(entry) = cache.get(symbol) {
response
.quotes
.insert(symbol.to_string(), entry.value.clone());
}
}
return Ok(response);
}
}
let _fetch_guard = self.quotes_fetch.lock().await;
{
let cache = self.quote_cache.read().await;
if self.all_cached(&cache, self.symbols.iter().cloned()) {
let mut response = BatchQuotesResponse::with_capacity(self.symbols.len());
for symbol in &self.symbols {
if let Some(entry) = cache.get(symbol) {
response
.quotes
.insert(symbol.to_string(), entry.value.clone());
}
}
return Ok(response);
}
}
let symbol_strings: Vec<String> = self.symbols.iter().map(|s| s.to_string()).collect();
let mut response = BatchQuotesResponse::with_capacity(self.symbols.len());
let (quote_data, logos) = if self.include_logo {
let providers_logo = Arc::clone(&self.providers);
let syms_logo = symbol_strings.clone();
let logo_future = async move {
if let Ok(client) = providers_logo.first_yahoo() {
let syms_ref: Vec<&str> = syms_logo.iter().map(String::as_str).collect();
crate::adapters::yahoo::quote::quotes::fetch_with_fields(
&client,
&syms_ref,
Some(&["logoUrl", "companyLogoUrl"]),
true,
true,
)
.await
.ok()
} else {
None
}
};
let providers_quote = Arc::clone(&self.providers);
let syms_quote = symbol_strings.clone();
let quote_future = async move {
providers_quote
.fetch(Capability::QUOTE, |p| {
let syms = syms_quote.clone();
let p = p.clone();
async move {
let syms_ref: Vec<&str> = syms.iter().map(String::as_str).collect();
p.as_quote()
.ok_or_else(|| {
p.not_supported(crate::providers::Operation::QuotesBatch)
})?
.fetch_quotes_batch(&syms_ref)
.await
}
})
.await
};
let (batch_result, logo_result) = tokio::join!(quote_future, logo_future);
let quote_data = match batch_result {
Ok(data) => data,
Err(_) => {
self.fetch_quotes_per_symbol(&symbol_strings, &mut response)
.await
}
};
(quote_data, logo_result)
} else {
let providers = Arc::clone(&self.providers);
let syms = symbol_strings.clone();
let batch_result = providers
.fetch(Capability::QUOTE, |p| {
let syms = syms.clone();
let p = p.clone();
async move {
let syms_ref: Vec<&str> = syms.iter().map(String::as_str).collect();
p.as_quote()
.ok_or_else(|| {
p.not_supported(crate::providers::Operation::QuotesBatch)
})?
.fetch_quotes_batch(&syms_ref)
.await
}
})
.await;
let data = match batch_result {
Ok(data) => data,
Err(_) => {
self.fetch_quotes_per_symbol(&symbol_strings, &mut response)
.await
}
};
(data, None)
};
let logo_map: HashMap<String, (Option<String>, Option<String>)> = logos
.and_then(|l| l.get("quoteResponse")?.get("result")?.as_array().cloned())
.map(|results| {
results
.iter()
.filter_map(|r| {
let symbol = r.get("symbol")?.as_str()?.to_string();
let logo_url = r.get("logoUrl").and_then(|v| v.as_str()).map(String::from);
let company_logo_url = r
.get("companyLogoUrl")
.and_then(|v| v.as_str())
.map(String::from);
Some((symbol, (logo_url, company_logo_url)))
})
.collect()
})
.unwrap_or_default();
let mut parsed_quotes: Vec<(String, Quote)> = Vec::new();
for (symbol, summary) in quote_data {
let logo_url = logo_map.get(&symbol).and_then(|(l, _)| l.clone());
let company_logo_url = logo_map.get(&symbol).and_then(|(_, c)| c.clone());
let quote = Quote::from_response(&summary, logo_url, company_logo_url);
parsed_quotes.push((symbol, quote));
}
for (symbol, quote) in parsed_quotes {
response.quotes.insert(symbol, quote);
}
#[cfg(feature = "translation")]
self.translate_response(&mut response).await?;
if self.cache_mode.enabled() {
let mut cache = self.quote_cache.write().await;
for (symbol, quote) in &response.quotes {
self.cache_insert(&mut cache, symbol.as_str().into(), quote.clone());
}
}
for symbol in &self.symbols {
let s = &**symbol;
if !response.quotes.contains_key(s) && !response.errors.contains_key(s) {
response.errors.insert(
symbol.to_string(),
"Symbol not found in response".to_string(),
);
}
}
Ok(response)
}
async fn fetch_quotes_per_symbol(
&self,
symbols: &[String],
response: &mut BatchQuotesResponse,
) -> Vec<(String, QuoteSummaryResponse)> {
let futures: Vec<_> = symbols
.iter()
.map(|sym| {
let providers = Arc::clone(&self.providers);
let sym = sym.clone();
async move {
let result = providers
.fetch(Capability::QUOTE, |p| {
let sym = sym.clone();
let p = p.clone();
async move {
p.as_quote()
.ok_or_else(|| {
p.not_supported(crate::providers::Operation::Quote)
})?
.fetch_quote(&sym)
.await
}
})
.await;
(sym, result)
}
})
.collect();
let results: Vec<_> = stream::iter(futures)
.buffer_unordered(self.max_concurrency)
.collect()
.await;
let mut successes = Vec::new();
for (sym, result) in results {
match result {
Ok(resp) => successes.push((sym, resp)),
Err(e) => {
response.errors.insert(sym, e.to_string());
}
}
}
successes
}
pub async fn quote<F>(&self, symbol: &str) -> Result<Quote<F>>
where
F: Format,
Quote<Both>: Into<Quote<F>>,
{
{
let cache = self.quote_cache.read().await;
if let Some(entry) = cache.get(symbol)
&& self.is_cache_fresh(Some(entry))
{
return Ok(entry.value.clone().into());
}
}
let response = self.quotes().await?;
response
.quotes
.get(symbol)
.cloned()
.map(Into::into)
.ok_or_else(|| FinanceError::SymbolNotFound {
symbol: Some(symbol.to_string()),
context: response
.errors
.get(symbol)
.cloned()
.unwrap_or_else(|| "Symbol not found".to_string()),
})
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
#[ignore = "requires network access"]
async fn test_tickers_quotes() {
let tickers = Tickers::new(["AAPL", "MSFT", "GOOGL"]).await.unwrap();
let result = tickers.quotes().await.unwrap();
assert!(result.success_count() > 0);
}
}