finance_query/tickers/

core.rs

//! Tickers implementation for batch operations on multiple symbols.
//!
//! Optimizes data fetching by using batch endpoints and concurrent requests.

use super::macros::{batch_fetch_cached, define_batch_response};
use crate::adapters::yahoo::client::ClientConfig;
#[cfg(feature = "backtesting")]
use crate::backtesting;
use crate::constants::{Frequency, Interval, Region, StatementType, TimeRange};
use crate::error::{FinanceError, Result};
use crate::format::Both;
#[cfg(any(feature = "backtesting", feature = "indicators"))]
use crate::indicators;
use crate::models::chart::events::ChartEvents;
use crate::models::chart::spark::Spark;
use crate::models::chart::{CapitalGain, Chart, Dividend, Split};
use crate::models::corporate::news::News;
use crate::models::corporate::recommendation::Recommendation;
use crate::models::format::Format;
use crate::models::fundamentals::FinancialStatement;
use crate::models::options::Options;
use crate::models::quote::{Quote, QuoteSummaryResponse};

use crate::providers::types::recommendation_from_similar;
use crate::providers::yahoo::YahooProvider;
use crate::providers::{
    Capability, Fetch, Provider, ProviderAdapter, ProviderSet, Routes, build_providers,
};
use crate::ticker::ClientHandle;
use crate::utils::{CacheEntry, EVICTION_THRESHOLD, filter_by_range};
use futures::stream::{self, StreamExt};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;

// Type aliases — MapCache wraps values in CacheEntry for TTL support.
type MapCache<K, V> = Arc<RwLock<HashMap<K, CacheEntry<V>>>>;
type ChartCacheKey = (Arc<str>, Interval, TimeRange);
type QuoteCache = MapCache<Arc<str>, Quote>;
type ChartCache = MapCache<ChartCacheKey, Chart>;
type EventsCache = MapCache<Arc<str>, ChartEvents>;
type FinancialsCache = MapCache<(Arc<str>, StatementType, Frequency), FinancialStatement>;
type NewsCache = MapCache<Arc<str>, Vec<News>>;
type RecommendationsCache = MapCache<(Arc<str>, u32), Recommendation>;
type OptionsCache = MapCache<(Arc<str>, Option<i64>), Options>;
type SparkCacheKey = (Arc<str>, Interval, TimeRange);
type SparkCache = MapCache<SparkCacheKey, Spark>;
#[cfg(feature = "indicators")]
type IndicatorsCache = MapCache<(Arc<str>, Interval, TimeRange), indicators::IndicatorsSummary>;

// Fetch guards for request deduplication — prevent concurrent duplicate fetches
type FetchGuard = Arc<tokio::sync::Mutex<()>>;
type FetchGuardMap<K> = Arc<RwLock<HashMap<K, FetchGuard>>>;

// Generate all batch response types
define_batch_response! {
    /// Response containing quotes for multiple symbols.
    BatchQuotesResponse => quotes: Quote
}

define_batch_response! {
    /// Response containing charts for multiple symbols.
    BatchChartsResponse => charts: Chart
}

define_batch_response! {
    /// Response containing spark data for multiple symbols.
    ///
    /// Spark data is optimized for sparkline rendering with only close prices.
    /// Unlike charts, spark data is fetched in a single batch request.
    BatchSparksResponse => sparks: Spark
}

define_batch_response! {
    /// Response containing dividends for multiple symbols.
    BatchDividendsResponse => dividends: Vec<Dividend>
}

define_batch_response! {
    /// Response containing splits for multiple symbols.
    BatchSplitsResponse => splits: Vec<Split>
}

define_batch_response! {
    /// Response containing capital gains for multiple symbols.
    BatchCapitalGainsResponse => capital_gains: Vec<CapitalGain>
}

define_batch_response! {
    /// Response containing financial statements for multiple symbols.
    BatchFinancialsResponse => financials: FinancialStatement
}

define_batch_response! {
    /// Response containing news articles for multiple symbols.
    BatchNewsResponse => news: Vec<News>
}

define_batch_response! {
    /// Response containing recommendations for multiple symbols.
    BatchRecommendationsResponse => recommendations: Recommendation
}

define_batch_response! {
    /// Response containing options chains for multiple symbols.
    BatchOptionsResponse => options: Options
}

#[cfg(feature = "indicators")]
define_batch_response! {
    /// Response containing technical indicators for multiple symbols.
    BatchIndicatorsResponse => indicators: indicators::IndicatorsSummary
}

/// Default maximum concurrent requests for batch operations.
const DEFAULT_MAX_CONCURRENCY: usize = 10;

/// Builder for Tickers
pub struct TickersBuilder {
    symbols: Vec<Arc<str>>,
    config: ClientConfig,
    shared_client: Option<ClientHandle>,
    injected_providers: Option<Arc<ProviderSet>>,
    max_concurrency: usize,
    cache_ttl: Option<Duration>,
    include_logo: bool,
}

impl TickersBuilder {
    fn new<S, I>(symbols: I) -> Self
    where
        S: Into<String>,
        I: IntoIterator<Item = S>,
    {
        Self {
            symbols: symbols.into_iter().map(|s| s.into().into()).collect(),
            config: ClientConfig::default(),
            shared_client: None,
            injected_providers: None,
            max_concurrency: DEFAULT_MAX_CONCURRENCY,
            cache_ttl: None,
            include_logo: false,
        }
    }

    /// Set the region (automatically sets correct lang and region code)
    pub fn region(mut self, region: Region) -> Self {
        self.config.lang = region.lang().to_string();
        self.config.region = region.region().to_string();
        self
    }

    /// Set the language code (e.g., "en-US", "ja-JP", "de-DE")
    pub fn lang(mut self, lang: impl Into<String>) -> Self {
        self.config.lang = lang.into();
        self
    }

    /// Set the region code (e.g., "US", "JP", "DE")
    pub fn region_code(mut self, region: impl Into<String>) -> Self {
        self.config.region = region.into();
        self
    }

    /// Set the HTTP request timeout
    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.config.timeout = timeout;
        self
    }

    /// Set the proxy URL
    pub fn proxy(mut self, proxy: impl Into<String>) -> Self {
        self.config.proxy = Some(proxy.into());
        self
    }

    #[allow(dead_code)]
    pub(crate) fn config(mut self, config: ClientConfig) -> Self {
        self.config = config;
        self
    }

    /// Set the maximum number of concurrent requests for batch operations.
    ///
    /// Controls how many HTTP requests run in parallel when methods like
    /// `charts()`, `financials()`, or `news()` fetch data for each symbol.
    /// Default is 10.
    ///
    /// Lower values reduce the risk of rate limiting from Yahoo Finance.
    /// Higher values increase throughput for large symbol lists.
    pub fn max_concurrency(mut self, n: usize) -> Self {
        self.max_concurrency = n.max(1);
        self
    }

    /// Enable response caching with a time-to-live.
    ///
    /// By default caching is **disabled** — every call fetches fresh data.
    /// When enabled, responses are reused until the TTL expires. Stale
    /// entries are evicted on the next write.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// use finance_query::Tickers;
    /// use std::time::Duration;
    ///
    /// let tickers = Tickers::builder(["AAPL", "MSFT"])
    ///     .cache(Duration::from_secs(30))
    ///     .build()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn cache(mut self, ttl: Duration) -> Self {
        self.cache_ttl = Some(ttl);
        self
    }

    /// Include company logo URLs in quote responses.
    ///
    /// When enabled, `quotes()` will fetch logo URLs in parallel with the
    /// quote batch request, adding a small extra request.
    pub fn logo(mut self) -> Self {
        self.include_logo = true;
        self
    }

    /// Pre-inject a shared provider set (used by [`Providers::tickers`]).
    pub(crate) fn with_provider_set(mut self, set: Arc<ProviderSet>) -> Self {
        self.injected_providers = Some(set);
        self
    }

    /// Share an existing authenticated session instead of creating a new one.
    ///
    /// Avoids redundant auth handshakes when combining `Tickers` with other
    /// `Ticker` instances. Obtain a handle from any existing `Ticker` or
    /// `Tickers` via `.client_handle()`.
    pub fn client(mut self, handle: ClientHandle) -> Self {
        self.shared_client = Some(handle);
        self
    }

    /// Build the Tickers instance
    pub async fn build(self) -> Result<Tickers> {
        #[cfg(feature = "translation")]
        let translate_lang = {
            let lang = crate::translation::Lang::parse(&self.config.lang)?;
            (!lang.is_english()).then_some(lang)
        };
        let providers = if let Some(set) = self.injected_providers {
            set
        } else if let Some(handle) = self.shared_client {
            let yahoo = YahooProvider::from_client(handle.0);
            let client = yahoo.client_arc();
            Arc::new(ProviderSet::new(
                vec![Arc::new(yahoo) as Arc<dyn ProviderAdapter>],
                Some(client),
                Routes::new(Fetch::Sequential),
            ))
        } else {
            Arc::new(
                build_providers(
                    &[Provider::Yahoo],
                    &self.config,
                    Routes::new(Fetch::Sequential),
                )
                .await?,
            )
        };

        Ok(Tickers {
            symbols: self.symbols,
            providers,
            max_concurrency: self.max_concurrency,
            cache_ttl: self.cache_ttl,
            include_logo: self.include_logo,
            #[cfg(feature = "translation")]
            translate_lang,
            quote_cache: Default::default(),
            chart_cache: Default::default(),
            events_cache: Default::default(),
            financials_cache: Default::default(),
            news_cache: Default::default(),
            recommendations_cache: Default::default(),
            options_cache: Default::default(),
            spark_cache: Default::default(),
            #[cfg(feature = "indicators")]
            indicators_cache: Default::default(),

            // Initialize fetch guards for request deduplication
            quotes_fetch: Arc::new(tokio::sync::Mutex::new(())),
            charts_fetch: Default::default(),
            financials_fetch: Default::default(),
            news_fetch: Arc::new(tokio::sync::Mutex::new(())),
            recommendations_fetch: Default::default(),
            options_fetch: Default::default(),
            spark_fetch: Default::default(),
            #[cfg(feature = "indicators")]
            indicators_fetch: Default::default(),
        })
    }
}

/// Multi-symbol ticker for efficient batch operations.
///
/// `Tickers` optimizes data fetching for multiple symbols by:
/// - Using batch endpoints where available (e.g., /v7/finance/quote)
/// - Fetching concurrently when batch endpoints don't exist
/// - Sharing a single authenticated client across all symbols
/// - Caching results per symbol
///
/// # Example
///
/// ```no_run
/// use finance_query::Tickers;
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// // Create tickers for multiple symbols
/// let tickers = Tickers::new(["AAPL", "MSFT", "GOOGL"]).await?;
///
/// // Batch fetch all quotes (single API call)
/// let quotes = tickers.quotes().await?;
/// for (symbol, quote) in &quotes.quotes {
///     let price = quote.regular_market_price.as_ref().and_then(|v| v.raw).unwrap_or(0.0);
///     println!("{}: ${:.2}", symbol, price);
/// }
///
/// // Fetch charts concurrently
/// use finance_query::{Interval, TimeRange};
/// let charts = tickers.charts(Interval::OneDay, TimeRange::OneMonth).await?;
/// # Ok(())
/// # }
/// ```
pub struct Tickers {
    symbols: Vec<Arc<str>>,
    providers: Arc<ProviderSet>,
    max_concurrency: usize,
    cache_ttl: Option<Duration>,
    include_logo: bool,
    #[cfg(feature = "translation")]
    translate_lang: Option<crate::translation::Lang>,
    quote_cache: QuoteCache,
    chart_cache: ChartCache,
    events_cache: EventsCache,
    financials_cache: FinancialsCache,
    news_cache: NewsCache,
    recommendations_cache: RecommendationsCache,
    options_cache: OptionsCache,
    spark_cache: SparkCache,
    #[cfg(feature = "indicators")]
    indicators_cache: IndicatorsCache,

    // Fetch guards prevent duplicate concurrent requests
    quotes_fetch: FetchGuard,
    charts_fetch: FetchGuardMap<(Interval, TimeRange)>,
    financials_fetch: FetchGuardMap<(StatementType, Frequency)>,
    news_fetch: FetchGuard,
    recommendations_fetch: FetchGuardMap<u32>,
    options_fetch: FetchGuardMap<Option<i64>>,
    spark_fetch: FetchGuardMap<(Interval, TimeRange)>,
    #[cfg(feature = "indicators")]
    indicators_fetch: FetchGuardMap<(Interval, TimeRange)>,
}

impl std::fmt::Debug for Tickers {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Tickers")
            .field("symbols", &self.symbols)
            .field("max_concurrency", &self.max_concurrency)
            .field("cache_ttl", &self.cache_ttl)
            .finish_non_exhaustive()
    }
}

impl Tickers {
    /// Creates new tickers with default configuration
    ///
    /// # Arguments
    ///
    /// * `symbols` - Iterable of stock symbols (e.g., `["AAPL", "MSFT"]`)
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::Tickers;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["AAPL", "MSFT", "GOOGL"]).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn new<S, I>(symbols: I) -> Result<Self>
    where
        S: Into<String>,
        I: IntoIterator<Item = S>,
    {
        Self::builder(symbols).build().await
    }

    /// Creates a new builder for Tickers
    pub fn builder<S, I>(symbols: I) -> TickersBuilder
    where
        S: Into<String>,
        I: IntoIterator<Item = S>,
    {
        TickersBuilder::new(symbols)
    }

    /// Returns the symbols this tickers instance manages
    pub fn symbols(&self) -> Vec<&str> {
        self.symbols.iter().map(|s| &**s).collect()
    }

    /// Number of symbols
    pub fn len(&self) -> usize {
        self.symbols.len()
    }

    /// Check if empty
    pub fn is_empty(&self) -> bool {
        self.symbols.is_empty()
    }

    /// Returns a handle to the underlying Yahoo Finance session.
    ///
    /// Pass to [`Ticker::builder`](crate::Ticker::builder) or other
    /// [`Tickers::builder`] calls via `.client(handle)` to share the
    /// authenticated session without a new auth handshake.
    ///
    /// # Panics
    ///
    /// Panics if these tickers were created via [`Providers`](crate::Providers) with
    /// no Yahoo provider configured. For session sharing across multiple tickers,
    /// prefer [`Providers::tickers`](crate::Providers::tickers) instead.
    pub fn client_handle(&self) -> ClientHandle {
        ClientHandle(
            self.providers
                .first_yahoo()
                .expect("Tickers always uses a Yahoo session"),
        )
    }

    /// Returns `true` if a cache entry exists and has not exceeded the TTL.
    #[inline]
    fn is_cache_fresh<T>(&self, entry: Option<&CacheEntry<T>>) -> bool {
        CacheEntry::is_fresh_with_ttl(entry, self.cache_ttl)
    }

    /// Translate a response value when a non-English language is configured
    /// (no-op otherwise).
    #[cfg(feature = "translation")]
    pub(crate) async fn translate_response<T: crate::translation::Translatable>(
        &self,
        value: &mut T,
    ) -> Result<()> {
        if let Some(lang) = &self.translate_lang {
            crate::translation::translate_with(value, lang).await?;
        }
        Ok(())
    }

    /// Returns `true` if all keys are present and fresh in a map cache.
    fn all_cached<K: Eq + std::hash::Hash, V>(
        &self,
        map: &HashMap<K, CacheEntry<V>>,
        keys: impl Iterator<Item = K>,
    ) -> bool {
        let Some(ttl) = self.cache_ttl else {
            return false;
        };
        keys.into_iter()
            .all(|k| map.get(&k).map(|e| e.is_fresh(ttl)).unwrap_or(false))
    }

    /// Insert into a map cache, amortizing stale-entry eviction.
    ///
    /// Only sweeps stale entries when the map exceeds [`EVICTION_THRESHOLD`],
    /// avoiding O(n) scans on every write.
    #[inline]
    fn cache_insert<K: Eq + std::hash::Hash, V>(
        &self,
        map: &mut HashMap<K, CacheEntry<V>>,
        key: K,
        value: V,
    ) {
        if let Some(ttl) = self.cache_ttl {
            if map.len() >= EVICTION_THRESHOLD {
                map.retain(|_, entry| entry.is_fresh(ttl));
            }
            map.insert(key, CacheEntry::new(value));
        }
    }

    /// Batch fetch quotes for all symbols.
    ///
    /// Dispatches through the configured provider set. When logos are enabled,
    /// fetches logo URLs in parallel via the Yahoo client.
    ///
    /// Use [`TickersBuilder::logo()`](TickersBuilder::logo) to enable logo fetching
    /// for this tickers instance.
    pub async fn quotes(&self) -> Result<BatchQuotesResponse> {
        // Fast path: check if all symbols are cached
        {
            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;

        // Double-check: another task may have fetched while we waited
        {
            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 {
            // Fire logo fetch in parallel with quote fetch; logos are Yahoo-only
            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.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.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);
        }

        // Translate before caching so cached quotes are already localized
        // and repeat reads don't re-run the translation backend.
        #[cfg(feature = "translation")]
        self.translate_response(&mut response).await?;

        if self.cache_ttl.is_some() {
            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());
            }
        }

        // Track missing symbols
        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)
    }

    /// Fallback for when no provider supports `fetch_quotes_batch`.
    /// Fetches each symbol individually; failures go into `response.errors`.
    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.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
    }

    /// Get a specific quote by symbol (from cache or fetch all)
    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()),
            })
    }

    /// Helper to get or create a fetch guard for a given key.
    ///
    /// Returns the guard from the map, never a locally-created copy that
    /// could diverge under contention.
    async fn get_fetch_guard<K: Clone + Eq + std::hash::Hash>(
        guard_map: &FetchGuardMap<K>,
        key: K,
    ) -> FetchGuard {
        {
            let guards = guard_map.read().await;
            if let Some(guard) = guards.get(&key) {
                return Arc::clone(guard);
            }
        }

        let mut guards = guard_map.write().await;
        Arc::clone(
            guards
                .entry(key)
                .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))),
        )
    }

    /// Batch fetch charts for all symbols concurrently
    ///
    /// Chart data cannot be batched in a single request, so this fetches
    /// all charts concurrently using tokio for maximum performance.
    pub async fn charts(
        &self,
        interval: Interval,
        range: TimeRange,
    ) -> Result<BatchChartsResponse> {
        // Fast path: check if all symbols are cached
        {
            let cache = self.chart_cache.read().await;
            if self.all_cached(
                &cache,
                self.symbols.iter().map(|s| (s.clone(), interval, range)),
            ) {
                let mut response = BatchChartsResponse::with_capacity(self.symbols.len());
                for symbol in &self.symbols {
                    if let Some(entry) = cache.get(&(symbol.clone(), interval, range)) {
                        response
                            .charts
                            .insert(symbol.to_string(), entry.value.clone());
                    }
                }
                return Ok(response);
            }
        }

        // Slow path: acquire fetch guard to prevent duplicate concurrent requests
        let fetch_guard = Self::get_fetch_guard(&self.charts_fetch, (interval, range)).await;
        let _guard = fetch_guard.lock().await;

        // Double-check: another task may have fetched while we waited
        {
            let cache = self.chart_cache.read().await;
            if self.all_cached(
                &cache,
                self.symbols.iter().map(|s| (s.clone(), interval, range)),
            ) {
                let mut response = BatchChartsResponse::with_capacity(self.symbols.len());
                for symbol in &self.symbols {
                    if let Some(entry) = cache.get(&(symbol.clone(), interval, range)) {
                        response
                            .charts
                            .insert(symbol.to_string(), entry.value.clone());
                    }
                }
                return Ok(response);
            }
        }

        // Fetch all charts concurrently via provider dispatch (no lock held during I/O)
        let futures: Vec<_> = self
            .symbols
            .iter()
            .map(|symbol| {
                let providers = Arc::clone(&self.providers);
                let symbol = Arc::clone(symbol);
                async move {
                    let sym = symbol.to_string();
                    let result = providers
                        .fetch(Capability::CHART, |p| {
                            let sym = sym.clone();
                            let p = p.clone();
                            async move { p.fetch_chart(&sym, interval, range).await }
                        })
                        .await;
                    (symbol, result)
                }
            })
            .collect();

        let results: Vec<_> = stream::iter(futures)
            .buffer_unordered(self.max_concurrency)
            .collect()
            .await;

        let mut response = BatchChartsResponse::with_capacity(self.symbols.len());
        let mut parsed_charts: Vec<(Arc<str>, Chart)> = Vec::new();

        for (symbol, result) in results {
            match result {
                Ok(data) => {
                    let chart = data;
                    parsed_charts.push((symbol, chart));
                }
                Err(e) => {
                    response.errors.insert(symbol.to_string(), e.to_string());
                }
            }
        }

        // Move into cache, then clone for response — avoids double-clone
        if self.cache_ttl.is_some() {
            let mut cache = self.chart_cache.write().await;
            let cache_keys: Vec<_> = parsed_charts
                .into_iter()
                .map(|(symbol, chart)| {
                    self.cache_insert(&mut cache, (symbol.clone(), interval, range), chart);
                    symbol
                })
                .collect();
            for symbol in cache_keys {
                if let Some(cached) = cache.get(&(symbol.clone(), interval, range)) {
                    response
                        .charts
                        .insert(symbol.to_string(), cached.value.clone());
                }
            }
        } else {
            for (symbol, chart) in parsed_charts {
                response.charts.insert(symbol.to_string(), chart);
            }
        }

        Ok(response)
    }

    /// Get a specific chart by symbol
    pub async fn chart(&self, symbol: &str, interval: Interval, range: TimeRange) -> Result<Chart> {
        {
            let cache = self.chart_cache.read().await;
            let key: Arc<str> = symbol.into();
            if let Some(entry) = cache.get(&(key, interval, range))
                && self.is_cache_fresh(Some(entry))
            {
                return Ok(entry.value.clone());
            }
        }

        let response = self.charts(interval, range).await?;

        response
            .charts
            .get(symbol)
            .cloned()
            .ok_or_else(|| FinanceError::SymbolNotFound {
                symbol: Some(symbol.to_string()),
                context: response
                    .errors
                    .get(symbol)
                    .cloned()
                    .unwrap_or_else(|| "Symbol not found".to_string()),
            })
    }

    /// Batch fetch chart data for a custom date range for all symbols concurrently.
    ///
    /// Unlike [`charts()`](Self::charts) which uses predefined time ranges,
    /// this method accepts absolute start/end timestamps. Results are **not cached**
    /// since custom ranges have unbounded key space.
    ///
    /// # Arguments
    ///
    /// * `interval` - Time interval between data points
    /// * `start` - Start date as Unix timestamp (seconds since epoch)
    /// * `end` - End date as Unix timestamp (seconds since epoch)
    pub async fn charts_range(
        &self,
        interval: Interval,
        start: i64,
        end: i64,
    ) -> Result<BatchChartsResponse> {
        let futures: Vec<_> = self
            .symbols
            .iter()
            .map(|symbol| {
                let providers = Arc::clone(&self.providers);
                let symbol = Arc::clone(symbol);
                async move {
                    let sym = symbol.to_string();
                    let result = providers
                        .fetch(Capability::CHART, |p| {
                            let sym = sym.clone();
                            let p = p.clone();
                            async move { p.fetch_chart_range(&sym, interval, start, end).await }
                        })
                        .await;
                    (symbol, result)
                }
            })
            .collect();

        let results: Vec<_> = stream::iter(futures)
            .buffer_unordered(self.max_concurrency)
            .collect()
            .await;

        let mut response = BatchChartsResponse::with_capacity(self.symbols.len());

        for (symbol, result) in results {
            match result {
                Ok(data) => {
                    let chart = data;
                    response.charts.insert(symbol.to_string(), chart);
                }
                Err(e) => {
                    response.errors.insert(symbol.to_string(), e.to_string());
                }
            }
        }

        Ok(response)
    }

    /// Ensures events are loaded for all symbols using chart requests.
    ///
    /// Fetches events concurrently for symbols that don't have cached events.
    /// Uses `TimeRange::Max` to get full event history (Yahoo returns all
    /// dividends/splits/capital gains regardless of chart range).
    ///
    /// Events are always stored regardless of `cache_ttl` because they are
    /// derived data (not a TTL-bounded cache). When `cache_ttl` is `None`,
    /// events persist for the lifetime of the `Tickers` instance.
    async fn ensure_events_loaded(&self) -> Result<()> {
        // Check which symbols need event data (existence check, not TTL-based)
        let symbols_to_fetch: Vec<Arc<str>> = {
            let cache = self.events_cache.read().await;
            self.symbols
                .iter()
                .filter(|sym| !cache.contains_key(*sym))
                .cloned()
                .collect()
        };

        if symbols_to_fetch.is_empty() {
            return Ok(());
        }

        // Fetch events concurrently for all symbols that need it via provider dispatch
        let futures: Vec<_> = symbols_to_fetch
            .iter()
            .map(|symbol| {
                let providers = Arc::clone(&self.providers);
                let symbol = Arc::clone(symbol);
                async move {
                    let sym = symbol.to_string();
                    let result = providers
                        .fetch(Capability::CORPORATE, |p| {
                            let sym = sym.clone();
                            let p = p.clone();
                            async move { p.fetch_events(&sym).await }
                        })
                        .await;
                    (symbol, result)
                }
            })
            .collect();

        let results: Vec<_> = stream::iter(futures)
            .buffer_unordered(self.max_concurrency)
            .collect()
            .await;

        let mut parsed_events: Vec<(Arc<str>, ChartEvents)> = Vec::new();

        for (symbol, result) in results {
            if let Ok(events_data) = result {
                parsed_events.push((symbol, events_data));
            }
        }

        // Always store events — they are derived data, not TTL-bounded cache
        if !parsed_events.is_empty() {
            let mut events_cache = self.events_cache.write().await;
            for (symbol, events) in parsed_events {
                events_cache.insert(symbol, CacheEntry::new(events));
            }
        }

        Ok(())
    }

    /// Batch fetch spark data for all symbols in a single request.
    ///
    /// Spark data is optimized for sparkline rendering, returning only close prices.
    /// Unlike `charts()`, this fetches all symbols in ONE API call, making it
    /// much more efficient for displaying price trends on dashboards or watchlists.
    ///
    /// # Arguments
    ///
    /// * `interval` - Time interval between data points (e.g., `Interval::FiveMinutes`)
    /// * `range` - Time range to fetch (e.g., `TimeRange::OneDay`)
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::{Tickers, Interval, TimeRange};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["AAPL", "MSFT", "GOOGL"]).await?;
    /// let sparks = tickers.spark(Interval::FiveMinutes, TimeRange::OneDay).await?;
    ///
    /// for (symbol, spark) in &sparks.sparks {
    ///     if let Some(change) = spark.percent_change() {
    ///         println!("{}: {:.2}%", symbol, change);
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn spark(&self, interval: Interval, range: TimeRange) -> Result<BatchSparksResponse> {
        // Fast path: check if all symbols are cached
        {
            let cache = self.spark_cache.read().await;
            if self.all_cached(
                &cache,
                self.symbols.iter().map(|s| (s.clone(), interval, range)),
            ) {
                let mut response = BatchSparksResponse::with_capacity(self.symbols.len());
                for symbol in &self.symbols {
                    if let Some(entry) = cache.get(&(symbol.clone(), interval, range)) {
                        response
                            .sparks
                            .insert(symbol.to_string(), entry.value.clone());
                    }
                }
                return Ok(response);
            }
        }

        // Slow path: acquire fetch guard
        let fetch_guard = Self::get_fetch_guard(&self.spark_fetch, (interval, range)).await;
        let _guard = fetch_guard.lock().await;

        // Double-check after guard
        {
            let cache = self.spark_cache.read().await;
            if self.all_cached(
                &cache,
                self.symbols.iter().map(|s| (s.clone(), interval, range)),
            ) {
                let mut response = BatchSparksResponse::with_capacity(self.symbols.len());
                for symbol in &self.symbols {
                    if let Some(entry) = cache.get(&(symbol.clone(), interval, range)) {
                        response
                            .sparks
                            .insert(symbol.to_string(), entry.value.clone());
                    }
                }
                return Ok(response);
            }
        }

        // Dispatch through the provider set under the CHART capability so spark
        // honors routing like every other chart path (Yahoo is the default).
        let providers = Arc::clone(&self.providers);
        let syms: Vec<String> = self.symbols.iter().map(|s| s.to_string()).collect();
        let spark_result = providers
            .fetch(Capability::CHART, |p| {
                let syms = syms.clone();
                let p = p.clone();
                async move {
                    let syms_ref: Vec<&str> = syms.iter().map(String::as_str).collect();
                    p.fetch_spark(&syms_ref, interval, range).await
                }
            })
            .await;

        let mut response = BatchSparksResponse::with_capacity(self.symbols.len());

        match spark_result {
            Ok(parsed_sparks) => {
                // Cache all parsed sparks
                if self.cache_ttl.is_some() {
                    let mut cache = self.spark_cache.write().await;
                    for (symbol, spark) in &parsed_sparks {
                        let key: Arc<str> = symbol.as_str().into();
                        self.cache_insert(&mut cache, (key, interval, range), spark.clone());
                    }
                }

                // Build response
                for (symbol, spark) in parsed_sparks {
                    response.sparks.insert(symbol, spark);
                }

                // Track missing symbols
                for symbol in &self.symbols {
                    let symbol_str = &**symbol;
                    if !response.sparks.contains_key(symbol_str)
                        && !response.errors.contains_key(symbol_str)
                    {
                        response.errors.insert(
                            symbol.to_string(),
                            "Symbol not found in response".to_string(),
                        );
                    }
                }
            }
            Err(e) => {
                for symbol in &self.symbols {
                    response.errors.insert(symbol.to_string(), e.to_string());
                }
            }
        }

        Ok(response)
    }

    /// Batch fetch dividends for all symbols
    ///
    /// Returns dividend history for all symbols, filtered by the specified time range.
    /// Dividends are cached per symbol after the first chart fetch.
    ///
    /// # Arguments
    ///
    /// * `range` - Time range to filter dividends
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::{Tickers, TimeRange};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["AAPL", "MSFT"]).await?;
    /// let dividends = tickers.dividends(TimeRange::OneYear).await?;
    ///
    /// for (symbol, divs) in &dividends.dividends {
    ///     println!("{}: {} dividends", symbol, divs.len());
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn dividends(&self, range: TimeRange) -> Result<BatchDividendsResponse> {
        let mut response = BatchDividendsResponse::with_capacity(self.symbols.len());

        // Fetch events efficiently (1-day chart request per symbol)
        self.ensure_events_loaded().await?;

        let events_cache = self.events_cache.read().await;

        for symbol in &self.symbols {
            if let Some(entry) = events_cache.get(symbol) {
                let all_dividends = entry.value.to_dividends();
                let filtered = filter_by_range(all_dividends, range);
                response.dividends.insert(symbol.to_string(), filtered);
            } else {
                response
                    .errors
                    .insert(symbol.to_string(), "No events data available".to_string());
            }
        }

        Ok(response)
    }

    /// Batch fetch stock splits for all symbols
    ///
    /// Returns stock split history for all symbols, filtered by the specified time range.
    /// Splits are cached per symbol after the first chart fetch.
    ///
    /// # Arguments
    ///
    /// * `range` - Time range to filter splits
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::{Tickers, TimeRange};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["NVDA", "TSLA"]).await?;
    /// let splits = tickers.splits(TimeRange::FiveYears).await?;
    ///
    /// for (symbol, sp) in &splits.splits {
    ///     for split in sp {
    ///         println!("{}: {}", symbol, split.ratio);
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn splits(&self, range: TimeRange) -> Result<BatchSplitsResponse> {
        let mut response = BatchSplitsResponse::with_capacity(self.symbols.len());

        // Fetch events efficiently (1-day chart request per symbol)
        self.ensure_events_loaded().await?;

        let events_cache = self.events_cache.read().await;

        for symbol in &self.symbols {
            if let Some(entry) = events_cache.get(symbol) {
                let all_splits = entry.value.to_splits();
                let filtered = filter_by_range(all_splits, range);
                response.splits.insert(symbol.to_string(), filtered);
            } else {
                response
                    .errors
                    .insert(symbol.to_string(), "No events data available".to_string());
            }
        }

        Ok(response)
    }

    /// Batch fetch capital gains for all symbols
    ///
    /// Returns capital gain distribution history for all symbols, filtered by the
    /// specified time range. This is primarily relevant for mutual funds and ETFs.
    /// Capital gains are cached per symbol after the first chart fetch.
    ///
    /// # Arguments
    ///
    /// * `range` - Time range to filter capital gains
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::{Tickers, TimeRange};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["VFIAX", "VTI"]).await?;
    /// let gains = tickers.capital_gains(TimeRange::TwoYears).await?;
    ///
    /// for (symbol, cg) in &gains.capital_gains {
    ///     println!("{}: {} distributions", symbol, cg.len());
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn capital_gains(&self, range: TimeRange) -> Result<BatchCapitalGainsResponse> {
        let mut response = BatchCapitalGainsResponse::with_capacity(self.symbols.len());

        // Fetch events efficiently (1-day chart request per symbol)
        self.ensure_events_loaded().await?;

        let events_cache = self.events_cache.read().await;

        for symbol in &self.symbols {
            if let Some(entry) = events_cache.get(symbol) {
                let all_gains = entry.value.to_capital_gains();
                let filtered = filter_by_range(all_gains, range);
                response.capital_gains.insert(symbol.to_string(), filtered);
            } else {
                response
                    .errors
                    .insert(symbol.to_string(), "No events data available".to_string());
            }
        }

        Ok(response)
    }

    /// Batch fetch financial statements for all symbols
    ///
    /// Fetches the specified financial statement type for all symbols concurrently.
    /// Financial statements are cached per (symbol, statement_type, frequency) tuple.
    ///
    /// # Arguments
    ///
    /// * `statement_type` - Type of statement (Income, Balance, CashFlow)
    /// * `frequency` - Annual or Quarterly
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::{Tickers, StatementType, Frequency};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["AAPL", "MSFT", "GOOGL"]).await?;
    /// let financials = tickers.financials(StatementType::Income, Frequency::Annual).await?;
    ///
    /// for (symbol, stmt) in &financials.financials {
    ///     if let Some(revenue) = stmt.statement.get("TotalRevenue") {
    ///         println!("{}: {:?}", symbol, revenue);
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn financials(
        &self,
        statement_type: StatementType,
        frequency: Frequency,
    ) -> Result<BatchFinancialsResponse> {
        batch_fetch_cached!(self;
            cache: financials_cache,
            guard: map(financials_fetch, (statement_type, frequency)),
            key: |s| (s.clone(), statement_type, frequency),
            response: BatchFinancialsResponse.financials,
            fetch: |providers, symbol| {
                let sym = symbol.to_string();
                providers.fetch(Capability::FUNDAMENTALS, move |p| {
                    let sym = sym.clone();
                    let p = p.clone();
                    async move {
                        p.fetch_financials(&sym, statement_type, frequency)
                            .await
                    }
                }).await
            },
        )
    }

    /// Batch fetch news articles for all symbols
    ///
    /// Fetches recent news articles for all symbols concurrently using scrapers.
    /// News articles are cached per symbol.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::Tickers;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["AAPL", "MSFT"]).await?;
    /// let news = tickers.news().await?;
    ///
    /// for (symbol, articles) in &news.news {
    ///     println!("{}: {} articles", symbol, articles.len());
    ///     for article in articles.iter().take(3) {
    ///         println!("  - {}", article.title);
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn news(&self) -> Result<BatchNewsResponse> {
        batch_fetch_cached!(self;
            cache: news_cache,
            guard: simple(news_fetch),
            key: |s| s.clone(),
            response: BatchNewsResponse.news,
            fetch: |providers, symbol| {
                let sym = symbol.to_string();
                providers.fetch(Capability::CORPORATE, move |p| {
                    let sym = sym.clone();
                    let p = p.clone();
                    async move {
                        p.fetch_news(&sym)
                            .await
                            .map(|data| data.into_iter().collect::<Vec<News>>())
                    }
                }).await
            },
        )
    }

    /// Batch fetch recommendations for all symbols
    ///
    /// Fetches analyst recommendations and similar stocks for all symbols concurrently.
    /// Recommendations are cached per (symbol, limit) tuple — different limits
    /// produce different API responses and are cached independently.
    ///
    /// # Arguments
    ///
    /// * `limit` - Maximum number of similar stocks to return per symbol
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::Tickers;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["AAPL", "MSFT"]).await?;
    /// let recommendations = tickers.recommendations(10).await?;
    ///
    /// for (symbol, rec) in &recommendations.recommendations {
    ///     println!("{}: {} recommendations", symbol, rec.count());
    ///     for similar in &rec.recommendations {
    ///         println!("  - {}: score {}", similar.symbol, similar.score);
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn recommendations(&self, limit: u32) -> Result<BatchRecommendationsResponse> {
        batch_fetch_cached!(self;
            cache: recommendations_cache,
            guard: map(recommendations_fetch, limit),
            key: |s| (s.clone(), limit),
            response: BatchRecommendationsResponse.recommendations,
            fetch: |providers, symbol| {
                let sym = symbol.to_string();
                providers.fetch(Capability::CORPORATE, move |p| {
                    let sym = sym.clone();
                    let p = p.clone();
                    async move {
                        let items = p.fetch_similar_symbols(&sym, limit).await?;
                        Ok(recommendation_from_similar(
                            sym,
                            Some(p.id()),
                            items,
                            Some(limit),
                        ))
                    }
                }).await
            },
        )
    }

    /// Batch fetch options chains for all symbols
    ///
    /// Fetches options chains for the specified expiration date for all symbols concurrently.
    /// Options are cached per (symbol, date) tuple.
    ///
    /// # Arguments
    ///
    /// * `date` - Optional expiration date (Unix timestamp). If None, fetches nearest expiration.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::Tickers;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["AAPL", "MSFT"]).await?;
    /// let options = tickers.options(None).await?;
    ///
    /// for (symbol, opts) in &options.options {
    ///     println!("{}: {} expirations", symbol, opts.expiration_dates().len());
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn options(&self, date: Option<i64>) -> Result<BatchOptionsResponse> {
        batch_fetch_cached!(self;
            cache: options_cache,
            guard: map(options_fetch, date),
            key: |s| (s.clone(), date),
            response: BatchOptionsResponse.options,
            fetch: |providers, symbol| {
                let sym = symbol.to_string();
                providers.fetch(Capability::OPTIONS, move |p| {
                    let sym = sym.clone();
                    let p = p.clone();
                    async move {
                        p.fetch_options(&sym, date).await
                    }
                }).await
            },
        )
    }

    /// Aggregate upcoming financial events across all symbols into a single
    /// time-sorted list.
    ///
    /// Merges earnings, dividend, and standard-monthly options-expiration events
    /// for every symbol — plus, with the `fred` feature, major economic releases
    /// (CPI, NFP, GDP, …) — within the forward window `[now, now + range]`,
    /// sorted ascending by timestamp.
    ///
    /// Best-effort per symbol: a symbol whose quote or options fetch fails
    /// simply contributes no events rather than failing the whole call.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::{Tickers, TimeRange};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["AAPL", "MSFT", "TSLA"]).await?;
    /// let events = tickers.calendar(TimeRange::OneMonth).await?;
    /// for e in &events {
    ///     println!("{} {:?} {:?}", e.date, e.symbol, e.event);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn calendar(
        &self,
        range: TimeRange,
    ) -> Result<Vec<crate::models::calendar::CalendarEvent>> {
        let now = chrono::Utc::now().timestamp();
        let window = (now, now + range.approx_duration_secs());

        let symbol_strings: Vec<String> = self.symbols.iter().map(|s| s.to_string()).collect();
        let providers = Arc::clone(&self.providers);

        let per_symbol = symbol_strings.into_iter().map(|sym| {
            let providers = Arc::clone(&providers);
            async move {
                let quote_fut = {
                    let sym = sym.clone();
                    providers.fetch(Capability::QUOTE, move |p| {
                        let sym = sym.clone();
                        let p = p.clone();
                        async move { p.fetch_quote(&sym).await }
                    })
                };
                let opts_fut = {
                    let sym = sym.clone();
                    providers.fetch(Capability::OPTIONS, move |p| {
                        let sym = sym.clone();
                        let p = p.clone();
                        async move { p.fetch_options(&sym, None).await }
                    })
                };
                let (quote, options) = tokio::join!(quote_fut, opts_fut);
                let calendar_events = quote.ok().and_then(|q| q.calendar_events);
                let options = options.ok();
                crate::models::calendar::build_symbol_events(
                    &sym,
                    calendar_events.as_ref(),
                    options.as_ref(),
                    window,
                )
            }
        });

        let per_symbol_fut = stream::iter(per_symbol)
            .buffer_unordered(self.max_concurrency)
            .collect::<Vec<_>>();

        // The FRED economic-release fetch is independent of the per-symbol work,
        // so run it concurrently with the per-symbol stream.
        #[cfg(feature = "fred")]
        let (per_symbol_events, releases) =
            tokio::join!(per_symbol_fut, crate::adapters::fred::release_dates());
        #[cfg(not(feature = "fred"))]
        let per_symbol_events = per_symbol_fut.await;

        let mut events: Vec<crate::models::calendar::CalendarEvent> =
            per_symbol_events.into_iter().flatten().collect();

        #[cfg(feature = "fred")]
        if let Ok(releases) = releases {
            events.extend(crate::models::calendar::build_economic_events(
                releases, window,
            ));
        }

        crate::models::calendar::sort_events(&mut events);
        Ok(events)
    }

    /// Batch calculate all technical indicators for all symbols
    ///
    /// Calculates complete indicator summaries for all symbols from their chart data.
    /// Indicators are cached per (symbol, interval, range) tuple.
    ///
    /// # Arguments
    ///
    /// * `interval` - The time interval for each candle
    /// * `range` - The time range to fetch data for
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::{Tickers, Interval, TimeRange};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["AAPL", "MSFT"]).await?;
    /// let indicators = tickers.indicators(Interval::OneDay, TimeRange::ThreeMonths).await?;
    ///
    /// for (symbol, ind) in &indicators.indicators {
    ///     println!("{}: RSI(14) = {:?}, SMA(20) = {:?}", symbol, ind.rsi_14, ind.sma_20);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "indicators")]
    pub async fn indicators(
        &self,
        interval: Interval,
        range: TimeRange,
    ) -> Result<BatchIndicatorsResponse> {
        let cache_key_for = |symbol: &Arc<str>| (symbol.clone(), interval, range);

        // Fast path: check if all symbols are cached
        {
            let cache = self.indicators_cache.read().await;
            if self.all_cached(&cache, self.symbols.iter().map(&cache_key_for)) {
                let mut response = BatchIndicatorsResponse::with_capacity(self.symbols.len());
                for symbol in &self.symbols {
                    if let Some(entry) = cache.get(&cache_key_for(symbol)) {
                        response
                            .indicators
                            .insert(symbol.to_string(), entry.value.clone());
                    }
                }
                return Ok(response);
            }
        }

        // Slow path: acquire fetch guard to prevent duplicate concurrent calculations
        let fetch_guard = Self::get_fetch_guard(&self.indicators_fetch, (interval, range)).await;
        let _guard = fetch_guard.lock().await;

        // Double-check: another task may have computed while we waited
        {
            let cache = self.indicators_cache.read().await;
            if self.all_cached(&cache, self.symbols.iter().map(&cache_key_for)) {
                let mut response = BatchIndicatorsResponse::with_capacity(self.symbols.len());
                for symbol in &self.symbols {
                    if let Some(entry) = cache.get(&cache_key_for(symbol)) {
                        response
                            .indicators
                            .insert(symbol.to_string(), entry.value.clone());
                    }
                }
                return Ok(response);
            }
        }

        // Fetch charts first (which may already be cached, has its own deduplication)
        let charts_response = self.charts(interval, range).await?;

        let mut response = BatchIndicatorsResponse::with_capacity(self.symbols.len());

        // Calculate all indicators first (no lock held)
        let mut calculated_indicators: Vec<(String, indicators::IndicatorsSummary)> = Vec::new();

        for (symbol, chart) in &charts_response.charts {
            let indicators = indicators::summary::calculate_indicators(&chart.candles);
            calculated_indicators.push((symbol.to_string(), indicators));
        }

        // Now acquire write lock briefly for batch cache insertion
        if self.cache_ttl.is_some() {
            let mut cache = self.indicators_cache.write().await;
            for (symbol, indicators) in &calculated_indicators {
                let key: Arc<str> = symbol.as_str().into();
                self.cache_insert(&mut cache, cache_key_for(&key), indicators.clone());
            }
        }

        // Populate response (no lock needed)
        for (symbol, indicators) in calculated_indicators {
            response.indicators.insert(symbol, indicators);
        }

        // Add errors from chart fetch
        for (symbol, error) in &charts_response.errors {
            response.errors.insert(symbol.to_string(), error.clone());
        }

        Ok(response)
    }

    // ========================================================================
    // Dynamic Symbol Management
    // ========================================================================

    /// Add symbols to the watch list
    ///
    /// Adds new symbols to track without affecting existing cached data.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::Tickers;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut tickers = Tickers::new(["AAPL"]).await?;
    /// tickers.add_symbols(["MSFT", "GOOGL"]);
    /// assert_eq!(tickers.len(), 3);
    /// # Ok(())
    /// # }
    /// ```
    pub fn add_symbols<S, I>(&mut self, symbols: I)
    where
        S: Into<String>,
        I: IntoIterator<Item = S>,
    {
        // Use HashSet for O(n+m) deduplication instead of O(n*m) linear search
        use std::collections::HashSet;

        let existing: HashSet<&str> = self.symbols.iter().map(|s| &**s).collect();
        let to_add: Vec<Arc<str>> = symbols
            .into_iter()
            .map(Into::into)
            .filter(|s| !existing.contains(s.as_str()))
            .map(|s| s.into())
            .collect();

        self.symbols.extend(to_add);
    }

    // ========================================================================
    // Portfolio Backtesting
    // ========================================================================

    /// Run a multi-symbol portfolio backtest across all tracked symbols.
    ///
    /// Fetches charts and dividends for each symbol concurrently, then runs
    /// the portfolio engine with the given strategy factory. Capital is shared
    /// across all symbols according to the [`PortfolioConfig`] allocation rules.
    ///
    /// `factory` is called once per symbol to produce an independent strategy
    /// instance:
    ///
    /// ```no_run
    /// use finance_query::{Tickers, Interval, TimeRange};
    /// use finance_query::backtesting::{SmaCrossover, BacktestConfig};
    /// use finance_query::backtesting::portfolio::{PortfolioConfig, RebalanceMode};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let tickers = Tickers::new(["AAPL", "MSFT", "NVDA"]).await?;
    ///
    /// let config = PortfolioConfig::new(BacktestConfig::default())
    ///     .max_total_positions(2)
    ///     .rebalance(RebalanceMode::EqualWeight);
    ///
    /// let result = tickers.backtest(
    ///     Interval::OneDay,
    ///     TimeRange::TwoYears,
    ///     Some(config),
    ///     |_sym| SmaCrossover::new(10, 50),
    /// ).await?;
    ///
    /// println!("Portfolio return: {:.2}%", result.portfolio_metrics.total_return_pct);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// [`PortfolioConfig`]: backtesting::portfolio::PortfolioConfig
    #[cfg(feature = "backtesting")]
    pub async fn backtest<S, F>(
        &self,
        interval: Interval,
        range: TimeRange,
        config: Option<backtesting::portfolio::PortfolioConfig>,
        factory: F,
    ) -> backtesting::Result<backtesting::portfolio::PortfolioResult>
    where
        S: backtesting::Strategy,
        F: Fn(&str) -> S,
    {
        use crate::backtesting::portfolio::{PortfolioEngine, SymbolData};

        let config = config.unwrap_or_default();
        config.validate(self.symbols.len())?;

        // Fetch charts for all symbols (uses the batch chart cache)
        let charts = self
            .charts(interval, range)
            .await
            .map_err(|e| backtesting::BacktestError::ChartError(e.to_string()))?;

        // Fetch dividends for all symbols (events cache is already warm after charts())
        // Treat errors as "no dividends" — dividend processing is best-effort
        let dividends_map = self
            .dividends(range)
            .await
            .map(|b| b.dividends)
            .unwrap_or_default();

        // Assemble SymbolData slices — skip symbols with no chart data
        let symbol_data: Vec<SymbolData> = self
            .symbols
            .iter()
            .filter_map(|sym| {
                charts.charts.get(sym.as_ref()).map(|chart| {
                    let divs = dividends_map.get(sym.as_ref()).cloned().unwrap_or_default();
                    SymbolData::new(sym.as_ref(), chart.candles.clone()).with_dividends(divs)
                })
            })
            .collect();

        let engine = PortfolioEngine::new(config);
        engine.run(&symbol_data, factory)
    }

    /// Remove symbols from the watch list
    ///
    /// Removes symbols and clears their cached data to free memory.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use finance_query::Tickers;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut tickers = Tickers::new(["AAPL", "MSFT", "GOOGL"]).await?;
    /// tickers.remove_symbols(["MSFT"]);
    /// assert_eq!(tickers.len(), 2);
    /// # Ok(())
    /// # }
    /// ```
    pub async fn remove_symbols<S, I>(&mut self, symbols: I)
    where
        S: Into<String>,
        I: IntoIterator<Item = S>,
    {
        use std::collections::HashSet;
        let owned: Vec<String> = symbols.into_iter().map(Into::into).collect();
        let to_remove: HashSet<&str> = owned.iter().map(|s| s.as_str()).collect();

        // Remove from symbol list — O(1) lookup per element
        self.symbols.retain(|s| !to_remove.contains(&**s));

        // Acquire all independent write locks in parallel
        let (
            mut quote_cache,
            mut chart_cache,
            mut events_cache,
            mut financials_cache,
            mut news_cache,
            mut recommendations_cache,
            mut options_cache,
            mut spark_cache,
        ) = tokio::join!(
            self.quote_cache.write(),
            self.chart_cache.write(),
            self.events_cache.write(),
            self.financials_cache.write(),
            self.news_cache.write(),
            self.recommendations_cache.write(),
            self.options_cache.write(),
            self.spark_cache.write(),
        );

        // Simple key caches — O(1) per removal
        for symbol in &to_remove {
            let key: Arc<str> = (*symbol).into();
            quote_cache.remove(&key);
            events_cache.remove(&key);
            news_cache.remove(&key);
        }

        // Composite key caches — O(n) retain but O(1) contains check
        chart_cache.retain(|(sym, _, _), _| !to_remove.contains(&**sym));
        financials_cache.retain(|(sym, _, _), _| !to_remove.contains(&**sym));
        recommendations_cache.retain(|(sym, _), _| !to_remove.contains(&**sym));
        options_cache.retain(|(sym, _), _| !to_remove.contains(&**sym));
        spark_cache.retain(|(sym, _, _), _| !to_remove.contains(&**sym));

        // Drop all guards before cfg-gated lock
        drop((
            quote_cache,
            chart_cache,
            events_cache,
            financials_cache,
            news_cache,
            recommendations_cache,
            options_cache,
            spark_cache,
        ));

        #[cfg(feature = "indicators")]
        self.indicators_cache
            .write()
            .await
            .retain(|(sym, _, _), _| !to_remove.contains(&**sym));
    }

    /// Clear all cached data and fetch guards, forcing fresh fetches on next access.
    ///
    /// Use this when you need up-to-date data from a long-lived `Tickers` instance.
    /// Also clears fetch guard maps to prevent unbounded growth.
    pub async fn clear_cache(&self) {
        tokio::join!(
            // Data caches
            async { self.quote_cache.write().await.clear() },
            async { self.chart_cache.write().await.clear() },
            async { self.events_cache.write().await.clear() },
            async { self.financials_cache.write().await.clear() },
            async { self.news_cache.write().await.clear() },
            async { self.recommendations_cache.write().await.clear() },
            async { self.options_cache.write().await.clear() },
            async { self.spark_cache.write().await.clear() },
            async {
                #[cfg(feature = "indicators")]
                self.indicators_cache.write().await.clear();
            },
            // Fetch guard maps (prevent unbounded growth)
            async { self.charts_fetch.write().await.clear() },
            async { self.financials_fetch.write().await.clear() },
            async { self.recommendations_fetch.write().await.clear() },
            async { self.options_fetch.write().await.clear() },
            async { self.spark_fetch.write().await.clear() },
            async {
                #[cfg(feature = "indicators")]
                self.indicators_fetch.write().await.clear();
            },
        );
    }

    /// Clear only the cached quote data.
    ///
    /// The next call to `quotes()` or `quote()` will re-fetch from the API.
    pub async fn clear_quote_cache(&self) {
        self.quote_cache.write().await.clear();
    }

    /// Clear only the cached chart, spark, and events data.
    ///
    /// The next call to `charts()`, `spark()`, `dividends()`, `splits()`,
    /// or `capital_gains()` will re-fetch from the API.
    pub async fn clear_chart_cache(&self) {
        tokio::join!(
            async { self.chart_cache.write().await.clear() },
            async { self.events_cache.write().await.clear() },
            async { self.spark_cache.write().await.clear() },
            async {
                #[cfg(feature = "indicators")]
                self.indicators_cache.write().await.clear();
            },
        );
    }
}

#[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);
    }

    #[tokio::test]
    #[ignore = "requires network access"]
    async fn test_tickers_charts() {
        let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
        let result = tickers
            .charts(Interval::OneDay, TimeRange::FiveDays)
            .await
            .unwrap();

        assert!(result.success_count() > 0);
    }

    #[tokio::test]
    #[ignore = "requires network access"]
    async fn test_tickers_spark() {
        let tickers = Tickers::new(["AAPL", "MSFT", "GOOGL"]).await.unwrap();
        let result = tickers
            .spark(Interval::FiveMinutes, TimeRange::OneDay)
            .await
            .unwrap();

        assert!(result.success_count() > 0);

        // Verify spark data structure
        if let Some(spark) = result.sparks.get("AAPL") {
            assert!(!spark.closes.is_empty());
            assert_eq!(spark.symbol, "AAPL");
            // Verify helper methods work
            assert!(spark.percent_change().is_some());
        }
    }

    #[tokio::test]
    #[ignore = "requires network access"]
    async fn test_tickers_dividends() {
        let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
        let result = tickers.dividends(TimeRange::OneYear).await.unwrap();

        assert!(result.success_count() > 0);

        // Verify dividend data structure
        if let Some(dividends) = result.dividends.get("AAPL")
            && !dividends.is_empty()
        {
            let div = &dividends[0];
            assert!(div.timestamp > 0);
            assert!(div.amount > 0.0);
        }
    }

    #[tokio::test]
    #[ignore = "requires network access"]
    async fn test_tickers_splits() {
        let tickers = Tickers::new(["NVDA", "TSLA"]).await.unwrap();
        let result = tickers.splits(TimeRange::FiveYears).await.unwrap();

        // Note: Not all symbols have splits, so we just check for successful response
        assert!(result.success_count() > 0);

        // If there are splits, verify structure
        for splits in result.splits.values() {
            for split in splits {
                assert!(split.timestamp > 0);
                assert!(split.numerator > 0.0);
                assert!(split.denominator > 0.0);
                assert!(!split.ratio.is_empty());
            }
        }
    }

    #[tokio::test]
    #[ignore = "requires network access"]
    async fn test_tickers_capital_gains() {
        let tickers = Tickers::new(["VFIAX", "VTI"]).await.unwrap();
        let result = tickers.capital_gains(TimeRange::TwoYears).await.unwrap();

        // Note: Not all symbols have capital gains distributions
        assert!(result.success_count() > 0);

        // If there are capital gains, verify structure
        for gains in result.capital_gains.values() {
            for gain in gains {
                assert!(gain.timestamp > 0);
                assert!(gain.amount >= 0.0);
            }
        }
    }

    #[tokio::test]
    #[ignore = "requires network access"]
    async fn test_tickers_financials() {
        let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
        let result = tickers
            .financials(StatementType::Income, Frequency::Annual)
            .await
            .unwrap();

        assert!(result.success_count() > 0);

        // Verify financial statement structure
        for (symbol, stmt) in &result.financials {
            assert_eq!(stmt.symbol, *symbol);
            assert_eq!(stmt.statement_type, "income");
            assert_eq!(stmt.frequency, "annual");
            assert!(!stmt.statement.is_empty());

            // Common income statement fields
            if let Some(revenue) = stmt.statement.get("TotalRevenue") {
                assert!(!revenue.is_empty());
            }
        }
    }

    #[tokio::test]
    #[ignore = "requires network access"]
    async fn test_tickers_news() {
        let tickers = Tickers::new(["AAPL", "TSLA"]).await.unwrap();
        let result = tickers.news().await.unwrap();

        assert!(result.success_count() > 0);

        // Verify news structure
        for articles in result.news.values() {
            if !articles.is_empty() {
                let article = &articles[0];
                assert!(!article.title.is_empty());
                assert!(!article.link.is_empty());
                assert!(!article.source.is_empty());
            }
        }
    }

    #[tokio::test]
    #[ignore = "requires network access"]
    async fn test_tickers_recommendations() {
        let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
        let result = tickers.recommendations(5).await.unwrap();

        assert!(result.success_count() > 0);

        // Verify recommendations structure
        for (symbol, rec) in &result.recommendations {
            assert_eq!(rec.symbol, *symbol);
            assert!(rec.count() > 0);
            for similar in &rec.recommendations {
                assert!(!similar.symbol.is_empty());
            }
        }
    }

    #[tokio::test]
    #[ignore = "requires network access"]
    async fn test_tickers_options() {
        let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
        let result = tickers.options(None).await.unwrap();

        assert!(result.success_count() > 0);

        // Verify options structure
        for opts in result.options.values() {
            assert!(!opts.expiration_dates().is_empty());
        }
    }

    #[tokio::test]
    #[ignore = "requires network access"]
    #[cfg(feature = "indicators")]
    async fn test_tickers_indicators() {
        let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
        let result = tickers
            .indicators(Interval::OneDay, TimeRange::ThreeMonths)
            .await
            .unwrap();

        assert!(result.success_count() > 0);

        // Verify indicators structure
        for ind in result.indicators.values() {
            // Check that at least some indicators are present
            assert!(ind.rsi_14.is_some() || ind.sma_20.is_some());
        }
    }

    #[tokio::test]
    async fn test_tickers_add_symbols() {
        let mut tickers = Tickers::new(["AAPL"]).await.unwrap();
        assert_eq!(tickers.len(), 1);
        assert_eq!(tickers.symbols(), &["AAPL"]);

        tickers.add_symbols(["MSFT", "GOOGL"]);
        assert_eq!(tickers.len(), 3);
        assert!(tickers.symbols().contains(&"AAPL"));
        assert!(tickers.symbols().contains(&"MSFT"));
        assert!(tickers.symbols().contains(&"GOOGL"));

        // Adding duplicate shouldn't increase count
        tickers.add_symbols(["AAPL"]);
        assert_eq!(tickers.len(), 3);
    }

    #[tokio::test]
    #[ignore = "requires network access"]
    async fn test_tickers_remove_symbols() {
        let mut tickers = Tickers::new(["AAPL", "MSFT", "GOOGL"]).await.unwrap();
        assert_eq!(tickers.len(), 3);

        // Fetch some data to populate caches
        let _ = tickers.quotes().await;

        // Remove one symbol
        tickers.remove_symbols(["MSFT"]).await;
        assert_eq!(tickers.len(), 2);
        assert!(tickers.symbols().contains(&"AAPL"));
        assert!(!tickers.symbols().contains(&"MSFT"));
        assert!(tickers.symbols().contains(&"GOOGL"));

        // Verify cache was cleared
        let quotes = tickers.quotes().await.unwrap();
        assert!(!quotes.quotes.contains_key("MSFT"));
        assert_eq!(quotes.quotes.len(), 2);
    }
}