1use super::macros::{batch_fetch_cached, define_batch_response};
6use crate::adapters::yahoo::client::ClientConfig;
7#[cfg(feature = "backtesting")]
8use crate::backtesting;
9use crate::constants::{Frequency, Interval, Region, StatementType, TimeRange};
10use crate::error::{FinanceError, Result};
11use crate::format::Both;
12#[cfg(any(feature = "backtesting", feature = "indicators"))]
13use crate::indicators;
14use crate::models::chart::events::ChartEvents;
15use crate::models::chart::spark::Spark;
16use crate::models::chart::{CapitalGain, Chart, Dividend, Split};
17use crate::models::corporate::news::News;
18use crate::models::corporate::recommendation::Recommendation;
19use crate::models::format::Format;
20use crate::models::fundamentals::FinancialStatement;
21use crate::models::options::Options;
22use crate::models::quote::{Quote, QuoteSummaryResponse};
23
24use crate::providers::types::recommendation_from_similar;
25use crate::providers::yahoo::YahooProvider;
26use crate::providers::{
27 Capability, Fetch, Provider, ProviderAdapter, ProviderSet, Routes, build_providers,
28};
29use crate::ticker::ClientHandle;
30use crate::utils::{CacheEntry, EVICTION_THRESHOLD, filter_by_range};
31use futures::stream::{self, StreamExt};
32use std::collections::HashMap;
33use std::sync::Arc;
34use std::time::Duration;
35use tokio::sync::RwLock;
36
37type MapCache<K, V> = Arc<RwLock<HashMap<K, CacheEntry<V>>>>;
39type ChartCacheKey = (Arc<str>, Interval, TimeRange);
40type QuoteCache = MapCache<Arc<str>, Quote>;
41type ChartCache = MapCache<ChartCacheKey, Chart>;
42type EventsCache = MapCache<Arc<str>, ChartEvents>;
43type FinancialsCache = MapCache<(Arc<str>, StatementType, Frequency), FinancialStatement>;
44type NewsCache = MapCache<Arc<str>, Vec<News>>;
45type RecommendationsCache = MapCache<(Arc<str>, u32), Recommendation>;
46type OptionsCache = MapCache<(Arc<str>, Option<i64>), Options>;
47type SparkCacheKey = (Arc<str>, Interval, TimeRange);
48type SparkCache = MapCache<SparkCacheKey, Spark>;
49#[cfg(feature = "indicators")]
50type IndicatorsCache = MapCache<(Arc<str>, Interval, TimeRange), indicators::IndicatorsSummary>;
51
52type FetchGuard = Arc<tokio::sync::Mutex<()>>;
54type FetchGuardMap<K> = Arc<RwLock<HashMap<K, FetchGuard>>>;
55
56define_batch_response! {
58 BatchQuotesResponse => quotes: Quote
60}
61
62define_batch_response! {
63 BatchChartsResponse => charts: Chart
65}
66
67define_batch_response! {
68 BatchSparksResponse => sparks: Spark
73}
74
75define_batch_response! {
76 BatchDividendsResponse => dividends: Vec<Dividend>
78}
79
80define_batch_response! {
81 BatchSplitsResponse => splits: Vec<Split>
83}
84
85define_batch_response! {
86 BatchCapitalGainsResponse => capital_gains: Vec<CapitalGain>
88}
89
90define_batch_response! {
91 BatchFinancialsResponse => financials: FinancialStatement
93}
94
95define_batch_response! {
96 BatchNewsResponse => news: Vec<News>
98}
99
100define_batch_response! {
101 BatchRecommendationsResponse => recommendations: Recommendation
103}
104
105define_batch_response! {
106 BatchOptionsResponse => options: Options
108}
109
110#[cfg(feature = "indicators")]
111define_batch_response! {
112 BatchIndicatorsResponse => indicators: indicators::IndicatorsSummary
114}
115
116const DEFAULT_MAX_CONCURRENCY: usize = 10;
118
119pub struct TickersBuilder {
121 symbols: Vec<Arc<str>>,
122 config: ClientConfig,
123 shared_client: Option<ClientHandle>,
124 injected_providers: Option<Arc<ProviderSet>>,
125 max_concurrency: usize,
126 cache_ttl: Option<Duration>,
127 include_logo: bool,
128}
129
130impl TickersBuilder {
131 fn new<S, I>(symbols: I) -> Self
132 where
133 S: Into<String>,
134 I: IntoIterator<Item = S>,
135 {
136 Self {
137 symbols: symbols.into_iter().map(|s| s.into().into()).collect(),
138 config: ClientConfig::default(),
139 shared_client: None,
140 injected_providers: None,
141 max_concurrency: DEFAULT_MAX_CONCURRENCY,
142 cache_ttl: None,
143 include_logo: false,
144 }
145 }
146
147 pub fn region(mut self, region: Region) -> Self {
149 self.config.lang = region.lang().to_string();
150 self.config.region = region.region().to_string();
151 self
152 }
153
154 pub fn lang(mut self, lang: impl Into<String>) -> Self {
156 self.config.lang = lang.into();
157 self
158 }
159
160 pub fn region_code(mut self, region: impl Into<String>) -> Self {
162 self.config.region = region.into();
163 self
164 }
165
166 pub fn timeout(mut self, timeout: Duration) -> Self {
168 self.config.timeout = timeout;
169 self
170 }
171
172 pub fn proxy(mut self, proxy: impl Into<String>) -> Self {
174 self.config.proxy = Some(proxy.into());
175 self
176 }
177
178 pub fn config(mut self, config: ClientConfig) -> Self {
180 self.config = config;
181 self
182 }
183
184 pub fn max_concurrency(mut self, n: usize) -> Self {
193 self.max_concurrency = n.max(1);
194 self
195 }
196
197 pub fn cache(mut self, ttl: Duration) -> Self {
218 self.cache_ttl = Some(ttl);
219 self
220 }
221
222 pub fn logo(mut self) -> Self {
227 self.include_logo = true;
228 self
229 }
230
231 pub(crate) fn with_provider_set(mut self, set: Arc<ProviderSet>) -> Self {
233 self.injected_providers = Some(set);
234 self
235 }
236
237 pub fn client(mut self, handle: ClientHandle) -> Self {
243 self.shared_client = Some(handle);
244 self
245 }
246
247 pub async fn build(self) -> Result<Tickers> {
249 #[cfg(feature = "translation")]
250 let translate_lang = {
251 let lang = crate::translation::Lang::parse(&self.config.lang)?;
252 (!lang.is_english()).then_some(lang)
253 };
254 let providers = if let Some(set) = self.injected_providers {
255 set
256 } else if let Some(handle) = self.shared_client {
257 let yahoo = YahooProvider::from_client(handle.0);
258 let client = yahoo.client_arc();
259 Arc::new(ProviderSet::new(
260 vec![Arc::new(yahoo) as Arc<dyn ProviderAdapter>],
261 Some(client),
262 Routes::new(Fetch::Sequential),
263 ))
264 } else {
265 Arc::new(
266 build_providers(
267 &[Provider::Yahoo],
268 &self.config,
269 Routes::new(Fetch::Sequential),
270 )
271 .await?,
272 )
273 };
274
275 Ok(Tickers {
276 symbols: self.symbols,
277 providers,
278 max_concurrency: self.max_concurrency,
279 cache_ttl: self.cache_ttl,
280 include_logo: self.include_logo,
281 #[cfg(feature = "translation")]
282 translate_lang,
283 quote_cache: Default::default(),
284 chart_cache: Default::default(),
285 events_cache: Default::default(),
286 financials_cache: Default::default(),
287 news_cache: Default::default(),
288 recommendations_cache: Default::default(),
289 options_cache: Default::default(),
290 spark_cache: Default::default(),
291 #[cfg(feature = "indicators")]
292 indicators_cache: Default::default(),
293
294 quotes_fetch: Arc::new(tokio::sync::Mutex::new(())),
296 charts_fetch: Default::default(),
297 financials_fetch: Default::default(),
298 news_fetch: Arc::new(tokio::sync::Mutex::new(())),
299 recommendations_fetch: Default::default(),
300 options_fetch: Default::default(),
301 spark_fetch: Default::default(),
302 #[cfg(feature = "indicators")]
303 indicators_fetch: Default::default(),
304 })
305 }
306}
307
308pub struct Tickers {
339 symbols: Vec<Arc<str>>,
340 providers: Arc<ProviderSet>,
341 max_concurrency: usize,
342 cache_ttl: Option<Duration>,
343 include_logo: bool,
344 #[cfg(feature = "translation")]
345 translate_lang: Option<crate::translation::Lang>,
346 quote_cache: QuoteCache,
347 chart_cache: ChartCache,
348 events_cache: EventsCache,
349 financials_cache: FinancialsCache,
350 news_cache: NewsCache,
351 recommendations_cache: RecommendationsCache,
352 options_cache: OptionsCache,
353 spark_cache: SparkCache,
354 #[cfg(feature = "indicators")]
355 indicators_cache: IndicatorsCache,
356
357 quotes_fetch: FetchGuard,
359 charts_fetch: FetchGuardMap<(Interval, TimeRange)>,
360 financials_fetch: FetchGuardMap<(StatementType, Frequency)>,
361 news_fetch: FetchGuard,
362 recommendations_fetch: FetchGuardMap<u32>,
363 options_fetch: FetchGuardMap<Option<i64>>,
364 spark_fetch: FetchGuardMap<(Interval, TimeRange)>,
365 #[cfg(feature = "indicators")]
366 indicators_fetch: FetchGuardMap<(Interval, TimeRange)>,
367}
368
369impl std::fmt::Debug for Tickers {
370 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
371 f.debug_struct("Tickers")
372 .field("symbols", &self.symbols)
373 .field("max_concurrency", &self.max_concurrency)
374 .field("cache_ttl", &self.cache_ttl)
375 .finish_non_exhaustive()
376 }
377}
378
379impl Tickers {
380 pub async fn new<S, I>(symbols: I) -> Result<Self>
397 where
398 S: Into<String>,
399 I: IntoIterator<Item = S>,
400 {
401 Self::builder(symbols).build().await
402 }
403
404 pub fn builder<S, I>(symbols: I) -> TickersBuilder
406 where
407 S: Into<String>,
408 I: IntoIterator<Item = S>,
409 {
410 TickersBuilder::new(symbols)
411 }
412
413 pub fn symbols(&self) -> Vec<&str> {
415 self.symbols.iter().map(|s| &**s).collect()
416 }
417
418 pub fn len(&self) -> usize {
420 self.symbols.len()
421 }
422
423 pub fn is_empty(&self) -> bool {
425 self.symbols.is_empty()
426 }
427
428 pub fn client_handle(&self) -> ClientHandle {
434 ClientHandle(
435 self.providers
436 .first_yahoo()
437 .expect("Tickers always uses a Yahoo session"),
438 )
439 }
440
441 #[inline]
443 fn is_cache_fresh<T>(&self, entry: Option<&CacheEntry<T>>) -> bool {
444 CacheEntry::is_fresh_with_ttl(entry, self.cache_ttl)
445 }
446
447 #[cfg(feature = "translation")]
450 pub(crate) async fn translate_response<T: crate::translation::Translatable>(
451 &self,
452 value: &mut T,
453 ) -> Result<()> {
454 if let Some(lang) = &self.translate_lang {
455 crate::translation::translate_with(value, lang).await?;
456 }
457 Ok(())
458 }
459
460 fn all_cached<K: Eq + std::hash::Hash, V>(
462 &self,
463 map: &HashMap<K, CacheEntry<V>>,
464 keys: impl Iterator<Item = K>,
465 ) -> bool {
466 let Some(ttl) = self.cache_ttl else {
467 return false;
468 };
469 keys.into_iter()
470 .all(|k| map.get(&k).map(|e| e.is_fresh(ttl)).unwrap_or(false))
471 }
472
473 #[inline]
478 fn cache_insert<K: Eq + std::hash::Hash, V>(
479 &self,
480 map: &mut HashMap<K, CacheEntry<V>>,
481 key: K,
482 value: V,
483 ) {
484 if let Some(ttl) = self.cache_ttl {
485 if map.len() >= EVICTION_THRESHOLD {
486 map.retain(|_, entry| entry.is_fresh(ttl));
487 }
488 map.insert(key, CacheEntry::new(value));
489 }
490 }
491
492 pub async fn quotes(&self) -> Result<BatchQuotesResponse> {
500 {
502 let cache = self.quote_cache.read().await;
503 if self.all_cached(&cache, self.symbols.iter().cloned()) {
504 let mut response = BatchQuotesResponse::with_capacity(self.symbols.len());
505 for symbol in &self.symbols {
506 if let Some(entry) = cache.get(symbol) {
507 response
508 .quotes
509 .insert(symbol.to_string(), entry.value.clone());
510 }
511 }
512 return Ok(response);
513 }
514 }
515
516 let _fetch_guard = self.quotes_fetch.lock().await;
517
518 {
520 let cache = self.quote_cache.read().await;
521 if self.all_cached(&cache, self.symbols.iter().cloned()) {
522 let mut response = BatchQuotesResponse::with_capacity(self.symbols.len());
523 for symbol in &self.symbols {
524 if let Some(entry) = cache.get(symbol) {
525 response
526 .quotes
527 .insert(symbol.to_string(), entry.value.clone());
528 }
529 }
530 return Ok(response);
531 }
532 }
533
534 let symbol_strings: Vec<String> = self.symbols.iter().map(|s| s.to_string()).collect();
535 let mut response = BatchQuotesResponse::with_capacity(self.symbols.len());
536
537 let (quote_data, logos) = if self.include_logo {
538 let providers_logo = Arc::clone(&self.providers);
540 let syms_logo = symbol_strings.clone();
541 let logo_future = async move {
542 if let Ok(client) = providers_logo.first_yahoo() {
543 let syms_ref: Vec<&str> = syms_logo.iter().map(String::as_str).collect();
544 crate::adapters::yahoo::quote::quotes::fetch_with_fields(
545 &client,
546 &syms_ref,
547 Some(&["logoUrl", "companyLogoUrl"]),
548 true,
549 true,
550 )
551 .await
552 .ok()
553 } else {
554 None
555 }
556 };
557
558 let providers_quote = Arc::clone(&self.providers);
559 let syms_quote = symbol_strings.clone();
560 let quote_future = async move {
561 providers_quote
562 .fetch(Capability::QUOTE, |p| {
563 let syms = syms_quote.clone();
564 let p = p.clone();
565 async move {
566 let syms_ref: Vec<&str> = syms.iter().map(String::as_str).collect();
567 p.fetch_quotes_batch(&syms_ref).await
568 }
569 })
570 .await
571 };
572
573 let (batch_result, logo_result) = tokio::join!(quote_future, logo_future);
574 let quote_data = match batch_result {
575 Ok(data) => data,
576 Err(_) => {
577 self.fetch_quotes_per_symbol(&symbol_strings, &mut response)
578 .await
579 }
580 };
581 (quote_data, logo_result)
582 } else {
583 let providers = Arc::clone(&self.providers);
584 let syms = symbol_strings.clone();
585 let batch_result = providers
586 .fetch(Capability::QUOTE, |p| {
587 let syms = syms.clone();
588 let p = p.clone();
589 async move {
590 let syms_ref: Vec<&str> = syms.iter().map(String::as_str).collect();
591 p.fetch_quotes_batch(&syms_ref).await
592 }
593 })
594 .await;
595 let data = match batch_result {
596 Ok(data) => data,
597 Err(_) => {
598 self.fetch_quotes_per_symbol(&symbol_strings, &mut response)
599 .await
600 }
601 };
602 (data, None)
603 };
604
605 let logo_map: HashMap<String, (Option<String>, Option<String>)> = logos
606 .and_then(|l| l.get("quoteResponse")?.get("result")?.as_array().cloned())
607 .map(|results| {
608 results
609 .iter()
610 .filter_map(|r| {
611 let symbol = r.get("symbol")?.as_str()?.to_string();
612 let logo_url = r.get("logoUrl").and_then(|v| v.as_str()).map(String::from);
613 let company_logo_url = r
614 .get("companyLogoUrl")
615 .and_then(|v| v.as_str())
616 .map(String::from);
617 Some((symbol, (logo_url, company_logo_url)))
618 })
619 .collect()
620 })
621 .unwrap_or_default();
622
623 let mut parsed_quotes: Vec<(String, Quote)> = Vec::new();
624
625 for (symbol, summary) in quote_data {
626 let logo_url = logo_map.get(&symbol).and_then(|(l, _)| l.clone());
627 let company_logo_url = logo_map.get(&symbol).and_then(|(_, c)| c.clone());
628 let quote = Quote::from_response(&summary, logo_url, company_logo_url);
629 parsed_quotes.push((symbol, quote));
630 }
631
632 for (symbol, quote) in parsed_quotes {
633 response.quotes.insert(symbol, quote);
634 }
635
636 #[cfg(feature = "translation")]
639 self.translate_response(&mut response).await?;
640
641 if self.cache_ttl.is_some() {
642 let mut cache = self.quote_cache.write().await;
643 for (symbol, quote) in &response.quotes {
644 self.cache_insert(&mut cache, symbol.as_str().into(), quote.clone());
645 }
646 }
647
648 for symbol in &self.symbols {
650 let s = &**symbol;
651 if !response.quotes.contains_key(s) && !response.errors.contains_key(s) {
652 response.errors.insert(
653 symbol.to_string(),
654 "Symbol not found in response".to_string(),
655 );
656 }
657 }
658
659 Ok(response)
660 }
661
662 async fn fetch_quotes_per_symbol(
665 &self,
666 symbols: &[String],
667 response: &mut BatchQuotesResponse,
668 ) -> Vec<(String, QuoteSummaryResponse)> {
669 let futures: Vec<_> = symbols
670 .iter()
671 .map(|sym| {
672 let providers = Arc::clone(&self.providers);
673 let sym = sym.clone();
674 async move {
675 let result = providers
676 .fetch(Capability::QUOTE, |p| {
677 let sym = sym.clone();
678 let p = p.clone();
679 async move { p.fetch_quote(&sym).await }
680 })
681 .await;
682 (sym, result)
683 }
684 })
685 .collect();
686
687 let results: Vec<_> = stream::iter(futures)
688 .buffer_unordered(self.max_concurrency)
689 .collect()
690 .await;
691
692 let mut successes = Vec::new();
693 for (sym, result) in results {
694 match result {
695 Ok(resp) => successes.push((sym, resp)),
696 Err(e) => {
697 response.errors.insert(sym, e.to_string());
698 }
699 }
700 }
701 successes
702 }
703
704 pub async fn quote<F>(&self, symbol: &str) -> Result<Quote<F>>
706 where
707 F: Format,
708 Quote<Both>: Into<Quote<F>>,
709 {
710 {
711 let cache = self.quote_cache.read().await;
712 if let Some(entry) = cache.get(symbol)
713 && self.is_cache_fresh(Some(entry))
714 {
715 return Ok(entry.value.clone().into());
716 }
717 }
718
719 let response = self.quotes().await?;
720
721 response
722 .quotes
723 .get(symbol)
724 .cloned()
725 .map(Into::into)
726 .ok_or_else(|| FinanceError::SymbolNotFound {
727 symbol: Some(symbol.to_string()),
728 context: response
729 .errors
730 .get(symbol)
731 .cloned()
732 .unwrap_or_else(|| "Symbol not found".to_string()),
733 })
734 }
735
736 async fn get_fetch_guard<K: Clone + Eq + std::hash::Hash>(
741 guard_map: &FetchGuardMap<K>,
742 key: K,
743 ) -> FetchGuard {
744 {
745 let guards = guard_map.read().await;
746 if let Some(guard) = guards.get(&key) {
747 return Arc::clone(guard);
748 }
749 }
750
751 let mut guards = guard_map.write().await;
752 Arc::clone(
753 guards
754 .entry(key)
755 .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))),
756 )
757 }
758
759 pub async fn charts(
764 &self,
765 interval: Interval,
766 range: TimeRange,
767 ) -> Result<BatchChartsResponse> {
768 {
770 let cache = self.chart_cache.read().await;
771 if self.all_cached(
772 &cache,
773 self.symbols.iter().map(|s| (s.clone(), interval, range)),
774 ) {
775 let mut response = BatchChartsResponse::with_capacity(self.symbols.len());
776 for symbol in &self.symbols {
777 if let Some(entry) = cache.get(&(symbol.clone(), interval, range)) {
778 response
779 .charts
780 .insert(symbol.to_string(), entry.value.clone());
781 }
782 }
783 return Ok(response);
784 }
785 }
786
787 let fetch_guard = Self::get_fetch_guard(&self.charts_fetch, (interval, range)).await;
789 let _guard = fetch_guard.lock().await;
790
791 {
793 let cache = self.chart_cache.read().await;
794 if self.all_cached(
795 &cache,
796 self.symbols.iter().map(|s| (s.clone(), interval, range)),
797 ) {
798 let mut response = BatchChartsResponse::with_capacity(self.symbols.len());
799 for symbol in &self.symbols {
800 if let Some(entry) = cache.get(&(symbol.clone(), interval, range)) {
801 response
802 .charts
803 .insert(symbol.to_string(), entry.value.clone());
804 }
805 }
806 return Ok(response);
807 }
808 }
809
810 let futures: Vec<_> = self
812 .symbols
813 .iter()
814 .map(|symbol| {
815 let providers = Arc::clone(&self.providers);
816 let symbol = Arc::clone(symbol);
817 async move {
818 let sym = symbol.to_string();
819 let result = providers
820 .fetch(Capability::CHART, |p| {
821 let sym = sym.clone();
822 let p = p.clone();
823 async move { p.fetch_chart(&sym, interval, range).await }
824 })
825 .await;
826 (symbol, result)
827 }
828 })
829 .collect();
830
831 let results: Vec<_> = stream::iter(futures)
832 .buffer_unordered(self.max_concurrency)
833 .collect()
834 .await;
835
836 let mut response = BatchChartsResponse::with_capacity(self.symbols.len());
837 let mut parsed_charts: Vec<(Arc<str>, Chart)> = Vec::new();
838
839 for (symbol, result) in results {
840 match result {
841 Ok(data) => {
842 let chart = data;
843 parsed_charts.push((symbol, chart));
844 }
845 Err(e) => {
846 response.errors.insert(symbol.to_string(), e.to_string());
847 }
848 }
849 }
850
851 if self.cache_ttl.is_some() {
853 let mut cache = self.chart_cache.write().await;
854 let cache_keys: Vec<_> = parsed_charts
855 .into_iter()
856 .map(|(symbol, chart)| {
857 self.cache_insert(&mut cache, (symbol.clone(), interval, range), chart);
858 symbol
859 })
860 .collect();
861 for symbol in cache_keys {
862 if let Some(cached) = cache.get(&(symbol.clone(), interval, range)) {
863 response
864 .charts
865 .insert(symbol.to_string(), cached.value.clone());
866 }
867 }
868 } else {
869 for (symbol, chart) in parsed_charts {
870 response.charts.insert(symbol.to_string(), chart);
871 }
872 }
873
874 Ok(response)
875 }
876
877 pub async fn chart(&self, symbol: &str, interval: Interval, range: TimeRange) -> Result<Chart> {
879 {
880 let cache = self.chart_cache.read().await;
881 let key: Arc<str> = symbol.into();
882 if let Some(entry) = cache.get(&(key, interval, range))
883 && self.is_cache_fresh(Some(entry))
884 {
885 return Ok(entry.value.clone());
886 }
887 }
888
889 let response = self.charts(interval, range).await?;
890
891 response
892 .charts
893 .get(symbol)
894 .cloned()
895 .ok_or_else(|| FinanceError::SymbolNotFound {
896 symbol: Some(symbol.to_string()),
897 context: response
898 .errors
899 .get(symbol)
900 .cloned()
901 .unwrap_or_else(|| "Symbol not found".to_string()),
902 })
903 }
904
905 pub async fn charts_range(
917 &self,
918 interval: Interval,
919 start: i64,
920 end: i64,
921 ) -> Result<BatchChartsResponse> {
922 let futures: Vec<_> = self
923 .symbols
924 .iter()
925 .map(|symbol| {
926 let providers = Arc::clone(&self.providers);
927 let symbol = Arc::clone(symbol);
928 async move {
929 let sym = symbol.to_string();
930 let result = providers
931 .fetch(Capability::CHART, |p| {
932 let sym = sym.clone();
933 let p = p.clone();
934 async move { p.fetch_chart_range(&sym, interval, start, end).await }
935 })
936 .await;
937 (symbol, result)
938 }
939 })
940 .collect();
941
942 let results: Vec<_> = stream::iter(futures)
943 .buffer_unordered(self.max_concurrency)
944 .collect()
945 .await;
946
947 let mut response = BatchChartsResponse::with_capacity(self.symbols.len());
948
949 for (symbol, result) in results {
950 match result {
951 Ok(data) => {
952 let chart = data;
953 response.charts.insert(symbol.to_string(), chart);
954 }
955 Err(e) => {
956 response.errors.insert(symbol.to_string(), e.to_string());
957 }
958 }
959 }
960
961 Ok(response)
962 }
963
964 async fn ensure_events_loaded(&self) -> Result<()> {
974 let symbols_to_fetch: Vec<Arc<str>> = {
976 let cache = self.events_cache.read().await;
977 self.symbols
978 .iter()
979 .filter(|sym| !cache.contains_key(*sym))
980 .cloned()
981 .collect()
982 };
983
984 if symbols_to_fetch.is_empty() {
985 return Ok(());
986 }
987
988 let futures: Vec<_> = symbols_to_fetch
990 .iter()
991 .map(|symbol| {
992 let providers = Arc::clone(&self.providers);
993 let symbol = Arc::clone(symbol);
994 async move {
995 let sym = symbol.to_string();
996 let result = providers
997 .fetch(Capability::CORPORATE, |p| {
998 let sym = sym.clone();
999 let p = p.clone();
1000 async move { p.fetch_events(&sym).await }
1001 })
1002 .await;
1003 (symbol, result)
1004 }
1005 })
1006 .collect();
1007
1008 let results: Vec<_> = stream::iter(futures)
1009 .buffer_unordered(self.max_concurrency)
1010 .collect()
1011 .await;
1012
1013 let mut parsed_events: Vec<(Arc<str>, ChartEvents)> = Vec::new();
1014
1015 for (symbol, result) in results {
1016 if let Ok(events_data) = result {
1017 parsed_events.push((symbol, events_data));
1018 }
1019 }
1020
1021 if !parsed_events.is_empty() {
1023 let mut events_cache = self.events_cache.write().await;
1024 for (symbol, events) in parsed_events {
1025 events_cache.insert(symbol, CacheEntry::new(events));
1026 }
1027 }
1028
1029 Ok(())
1030 }
1031
1032 pub async fn spark(&self, interval: Interval, range: TimeRange) -> Result<BatchSparksResponse> {
1061 {
1063 let cache = self.spark_cache.read().await;
1064 if self.all_cached(
1065 &cache,
1066 self.symbols.iter().map(|s| (s.clone(), interval, range)),
1067 ) {
1068 let mut response = BatchSparksResponse::with_capacity(self.symbols.len());
1069 for symbol in &self.symbols {
1070 if let Some(entry) = cache.get(&(symbol.clone(), interval, range)) {
1071 response
1072 .sparks
1073 .insert(symbol.to_string(), entry.value.clone());
1074 }
1075 }
1076 return Ok(response);
1077 }
1078 }
1079
1080 let fetch_guard = Self::get_fetch_guard(&self.spark_fetch, (interval, range)).await;
1082 let _guard = fetch_guard.lock().await;
1083
1084 {
1086 let cache = self.spark_cache.read().await;
1087 if self.all_cached(
1088 &cache,
1089 self.symbols.iter().map(|s| (s.clone(), interval, range)),
1090 ) {
1091 let mut response = BatchSparksResponse::with_capacity(self.symbols.len());
1092 for symbol in &self.symbols {
1093 if let Some(entry) = cache.get(&(symbol.clone(), interval, range)) {
1094 response
1095 .sparks
1096 .insert(symbol.to_string(), entry.value.clone());
1097 }
1098 }
1099 return Ok(response);
1100 }
1101 }
1102
1103 let providers = Arc::clone(&self.providers);
1106 let syms: Vec<String> = self.symbols.iter().map(|s| s.to_string()).collect();
1107 let spark_result = providers
1108 .fetch(Capability::CHART, |p| {
1109 let syms = syms.clone();
1110 let p = p.clone();
1111 async move {
1112 let syms_ref: Vec<&str> = syms.iter().map(String::as_str).collect();
1113 p.fetch_spark(&syms_ref, interval, range).await
1114 }
1115 })
1116 .await;
1117
1118 let mut response = BatchSparksResponse::with_capacity(self.symbols.len());
1119
1120 match spark_result {
1121 Ok(parsed_sparks) => {
1122 if self.cache_ttl.is_some() {
1124 let mut cache = self.spark_cache.write().await;
1125 for (symbol, spark) in &parsed_sparks {
1126 let key: Arc<str> = symbol.as_str().into();
1127 self.cache_insert(&mut cache, (key, interval, range), spark.clone());
1128 }
1129 }
1130
1131 for (symbol, spark) in parsed_sparks {
1133 response.sparks.insert(symbol, spark);
1134 }
1135
1136 for symbol in &self.symbols {
1138 let symbol_str = &**symbol;
1139 if !response.sparks.contains_key(symbol_str)
1140 && !response.errors.contains_key(symbol_str)
1141 {
1142 response.errors.insert(
1143 symbol.to_string(),
1144 "Symbol not found in response".to_string(),
1145 );
1146 }
1147 }
1148 }
1149 Err(e) => {
1150 for symbol in &self.symbols {
1151 response.errors.insert(symbol.to_string(), e.to_string());
1152 }
1153 }
1154 }
1155
1156 Ok(response)
1157 }
1158
1159 pub async fn dividends(&self, range: TimeRange) -> Result<BatchDividendsResponse> {
1184 let mut response = BatchDividendsResponse::with_capacity(self.symbols.len());
1185
1186 self.ensure_events_loaded().await?;
1188
1189 let events_cache = self.events_cache.read().await;
1190
1191 for symbol in &self.symbols {
1192 if let Some(entry) = events_cache.get(symbol) {
1193 let all_dividends = entry.value.to_dividends();
1194 let filtered = filter_by_range(all_dividends, range);
1195 response.dividends.insert(symbol.to_string(), filtered);
1196 } else {
1197 response
1198 .errors
1199 .insert(symbol.to_string(), "No events data available".to_string());
1200 }
1201 }
1202
1203 Ok(response)
1204 }
1205
1206 pub async fn splits(&self, range: TimeRange) -> Result<BatchSplitsResponse> {
1233 let mut response = BatchSplitsResponse::with_capacity(self.symbols.len());
1234
1235 self.ensure_events_loaded().await?;
1237
1238 let events_cache = self.events_cache.read().await;
1239
1240 for symbol in &self.symbols {
1241 if let Some(entry) = events_cache.get(symbol) {
1242 let all_splits = entry.value.to_splits();
1243 let filtered = filter_by_range(all_splits, range);
1244 response.splits.insert(symbol.to_string(), filtered);
1245 } else {
1246 response
1247 .errors
1248 .insert(symbol.to_string(), "No events data available".to_string());
1249 }
1250 }
1251
1252 Ok(response)
1253 }
1254
1255 pub async fn capital_gains(&self, range: TimeRange) -> Result<BatchCapitalGainsResponse> {
1281 let mut response = BatchCapitalGainsResponse::with_capacity(self.symbols.len());
1282
1283 self.ensure_events_loaded().await?;
1285
1286 let events_cache = self.events_cache.read().await;
1287
1288 for symbol in &self.symbols {
1289 if let Some(entry) = events_cache.get(symbol) {
1290 let all_gains = entry.value.to_capital_gains();
1291 let filtered = filter_by_range(all_gains, range);
1292 response.capital_gains.insert(symbol.to_string(), filtered);
1293 } else {
1294 response
1295 .errors
1296 .insert(symbol.to_string(), "No events data available".to_string());
1297 }
1298 }
1299
1300 Ok(response)
1301 }
1302
1303 pub async fn financials(
1331 &self,
1332 statement_type: StatementType,
1333 frequency: Frequency,
1334 ) -> Result<BatchFinancialsResponse> {
1335 batch_fetch_cached!(self;
1336 cache: financials_cache,
1337 guard: map(financials_fetch, (statement_type, frequency)),
1338 key: |s| (s.clone(), statement_type, frequency),
1339 response: BatchFinancialsResponse.financials,
1340 fetch: |providers, symbol| {
1341 let sym = symbol.to_string();
1342 providers.fetch(Capability::FUNDAMENTALS, move |p| {
1343 let sym = sym.clone();
1344 let p = p.clone();
1345 async move {
1346 p.fetch_financials(&sym, statement_type, frequency)
1347 .await
1348 }
1349 }).await
1350 },
1351 )
1352 }
1353
1354 pub async fn news(&self) -> Result<BatchNewsResponse> {
1378 batch_fetch_cached!(self;
1379 cache: news_cache,
1380 guard: simple(news_fetch),
1381 key: |s| s.clone(),
1382 response: BatchNewsResponse.news,
1383 fetch: |providers, symbol| {
1384 let sym = symbol.to_string();
1385 providers.fetch(Capability::CORPORATE, move |p| {
1386 let sym = sym.clone();
1387 let p = p.clone();
1388 async move {
1389 p.fetch_news(&sym)
1390 .await
1391 .map(|data| data.into_iter().collect::<Vec<News>>())
1392 }
1393 }).await
1394 },
1395 )
1396 }
1397
1398 pub async fn recommendations(&self, limit: u32) -> Result<BatchRecommendationsResponse> {
1427 batch_fetch_cached!(self;
1428 cache: recommendations_cache,
1429 guard: map(recommendations_fetch, limit),
1430 key: |s| (s.clone(), limit),
1431 response: BatchRecommendationsResponse.recommendations,
1432 fetch: |providers, symbol| {
1433 let sym = symbol.to_string();
1434 providers.fetch(Capability::CORPORATE, move |p| {
1435 let sym = sym.clone();
1436 let p = p.clone();
1437 async move {
1438 let items = p.fetch_similar_symbols(&sym, limit).await?;
1439 Ok(recommendation_from_similar(
1440 sym,
1441 Some(Provider::from_id_str(p.id()).ok_or_else(|| {
1442 FinanceError::InternalError(format!("unknown provider id: {}", p.id()))
1443 })?),
1444 items,
1445 Some(limit),
1446 ))
1447 }
1448 }).await
1449 },
1450 )
1451 }
1452
1453 pub async fn options(&self, date: Option<i64>) -> Result<BatchOptionsResponse> {
1478 batch_fetch_cached!(self;
1479 cache: options_cache,
1480 guard: map(options_fetch, date),
1481 key: |s| (s.clone(), date),
1482 response: BatchOptionsResponse.options,
1483 fetch: |providers, symbol| {
1484 let sym = symbol.to_string();
1485 providers.fetch(Capability::OPTIONS, move |p| {
1486 let sym = sym.clone();
1487 let p = p.clone();
1488 async move {
1489 p.fetch_options(&sym, date).await
1490 }
1491 }).await
1492 },
1493 )
1494 }
1495
1496 #[cfg(feature = "indicators")]
1522 pub async fn indicators(
1523 &self,
1524 interval: Interval,
1525 range: TimeRange,
1526 ) -> Result<BatchIndicatorsResponse> {
1527 let cache_key_for = |symbol: &Arc<str>| (symbol.clone(), interval, range);
1528
1529 {
1531 let cache = self.indicators_cache.read().await;
1532 if self.all_cached(&cache, self.symbols.iter().map(&cache_key_for)) {
1533 let mut response = BatchIndicatorsResponse::with_capacity(self.symbols.len());
1534 for symbol in &self.symbols {
1535 if let Some(entry) = cache.get(&cache_key_for(symbol)) {
1536 response
1537 .indicators
1538 .insert(symbol.to_string(), entry.value.clone());
1539 }
1540 }
1541 return Ok(response);
1542 }
1543 }
1544
1545 let fetch_guard = Self::get_fetch_guard(&self.indicators_fetch, (interval, range)).await;
1547 let _guard = fetch_guard.lock().await;
1548
1549 {
1551 let cache = self.indicators_cache.read().await;
1552 if self.all_cached(&cache, self.symbols.iter().map(&cache_key_for)) {
1553 let mut response = BatchIndicatorsResponse::with_capacity(self.symbols.len());
1554 for symbol in &self.symbols {
1555 if let Some(entry) = cache.get(&cache_key_for(symbol)) {
1556 response
1557 .indicators
1558 .insert(symbol.to_string(), entry.value.clone());
1559 }
1560 }
1561 return Ok(response);
1562 }
1563 }
1564
1565 let charts_response = self.charts(interval, range).await?;
1567
1568 let mut response = BatchIndicatorsResponse::with_capacity(self.symbols.len());
1569
1570 let mut calculated_indicators: Vec<(String, indicators::IndicatorsSummary)> = Vec::new();
1572
1573 for (symbol, chart) in &charts_response.charts {
1574 let indicators = indicators::summary::calculate_indicators(&chart.candles);
1575 calculated_indicators.push((symbol.to_string(), indicators));
1576 }
1577
1578 if self.cache_ttl.is_some() {
1580 let mut cache = self.indicators_cache.write().await;
1581 for (symbol, indicators) in &calculated_indicators {
1582 let key: Arc<str> = symbol.as_str().into();
1583 self.cache_insert(&mut cache, cache_key_for(&key), indicators.clone());
1584 }
1585 }
1586
1587 for (symbol, indicators) in calculated_indicators {
1589 response.indicators.insert(symbol, indicators);
1590 }
1591
1592 for (symbol, error) in &charts_response.errors {
1594 response.errors.insert(symbol.to_string(), error.clone());
1595 }
1596
1597 Ok(response)
1598 }
1599
1600 pub fn add_symbols(&mut self, symbols: &[impl AsRef<str>]) {
1621 use std::collections::HashSet;
1623
1624 let existing: HashSet<&str> = self.symbols.iter().map(|s| &**s).collect();
1625 let to_add: Vec<Arc<str>> = symbols
1626 .iter()
1627 .map(|s| s.as_ref())
1628 .filter(|s| !existing.contains(s))
1629 .map(|s| s.into())
1630 .collect();
1631
1632 self.symbols.extend(to_add);
1633 }
1634
1635 #[cfg(feature = "backtesting")]
1674 pub async fn backtest<S, F>(
1675 &self,
1676 interval: Interval,
1677 range: TimeRange,
1678 config: Option<backtesting::portfolio::PortfolioConfig>,
1679 factory: F,
1680 ) -> backtesting::Result<backtesting::portfolio::PortfolioResult>
1681 where
1682 S: backtesting::Strategy,
1683 F: Fn(&str) -> S,
1684 {
1685 use crate::backtesting::portfolio::{PortfolioEngine, SymbolData};
1686
1687 let config = config.unwrap_or_default();
1688 config.validate(self.symbols.len())?;
1689
1690 let charts = self
1692 .charts(interval, range)
1693 .await
1694 .map_err(|e| backtesting::BacktestError::ChartError(e.to_string()))?;
1695
1696 let dividends_map = self
1699 .dividends(range)
1700 .await
1701 .map(|b| b.dividends)
1702 .unwrap_or_default();
1703
1704 let symbol_data: Vec<SymbolData> = self
1706 .symbols
1707 .iter()
1708 .filter_map(|sym| {
1709 charts.charts.get(sym.as_ref()).map(|chart| {
1710 let divs = dividends_map.get(sym.as_ref()).cloned().unwrap_or_default();
1711 SymbolData::new(sym.as_ref(), chart.candles.clone()).with_dividends(divs)
1712 })
1713 })
1714 .collect();
1715
1716 let engine = PortfolioEngine::new(config);
1717 engine.run(&symbol_data, factory)
1718 }
1719
1720 pub async fn remove_symbols(&mut self, symbols: &[impl AsRef<str>]) {
1737 use std::collections::HashSet;
1738 let to_remove: HashSet<&str> = symbols.iter().map(|s| s.as_ref()).collect();
1739
1740 self.symbols.retain(|s| !to_remove.contains(&**s));
1742
1743 let (
1745 mut quote_cache,
1746 mut chart_cache,
1747 mut events_cache,
1748 mut financials_cache,
1749 mut news_cache,
1750 mut recommendations_cache,
1751 mut options_cache,
1752 mut spark_cache,
1753 ) = tokio::join!(
1754 self.quote_cache.write(),
1755 self.chart_cache.write(),
1756 self.events_cache.write(),
1757 self.financials_cache.write(),
1758 self.news_cache.write(),
1759 self.recommendations_cache.write(),
1760 self.options_cache.write(),
1761 self.spark_cache.write(),
1762 );
1763
1764 for symbol in &to_remove {
1766 let key: Arc<str> = (*symbol).into();
1767 quote_cache.remove(&key);
1768 events_cache.remove(&key);
1769 news_cache.remove(&key);
1770 }
1771
1772 chart_cache.retain(|(sym, _, _), _| !to_remove.contains(&**sym));
1774 financials_cache.retain(|(sym, _, _), _| !to_remove.contains(&**sym));
1775 recommendations_cache.retain(|(sym, _), _| !to_remove.contains(&**sym));
1776 options_cache.retain(|(sym, _), _| !to_remove.contains(&**sym));
1777 spark_cache.retain(|(sym, _, _), _| !to_remove.contains(&**sym));
1778
1779 drop((
1781 quote_cache,
1782 chart_cache,
1783 events_cache,
1784 financials_cache,
1785 news_cache,
1786 recommendations_cache,
1787 options_cache,
1788 spark_cache,
1789 ));
1790
1791 #[cfg(feature = "indicators")]
1792 self.indicators_cache
1793 .write()
1794 .await
1795 .retain(|(sym, _, _), _| !to_remove.contains(&**sym));
1796 }
1797
1798 pub async fn clear_cache(&self) {
1803 tokio::join!(
1804 async { self.quote_cache.write().await.clear() },
1806 async { self.chart_cache.write().await.clear() },
1807 async { self.events_cache.write().await.clear() },
1808 async { self.financials_cache.write().await.clear() },
1809 async { self.news_cache.write().await.clear() },
1810 async { self.recommendations_cache.write().await.clear() },
1811 async { self.options_cache.write().await.clear() },
1812 async { self.spark_cache.write().await.clear() },
1813 async {
1814 #[cfg(feature = "indicators")]
1815 self.indicators_cache.write().await.clear();
1816 },
1817 async { self.charts_fetch.write().await.clear() },
1819 async { self.financials_fetch.write().await.clear() },
1820 async { self.recommendations_fetch.write().await.clear() },
1821 async { self.options_fetch.write().await.clear() },
1822 async { self.spark_fetch.write().await.clear() },
1823 async {
1824 #[cfg(feature = "indicators")]
1825 self.indicators_fetch.write().await.clear();
1826 },
1827 );
1828 }
1829
1830 pub async fn clear_quote_cache(&self) {
1834 self.quote_cache.write().await.clear();
1835 }
1836
1837 pub async fn clear_chart_cache(&self) {
1842 tokio::join!(
1843 async { self.chart_cache.write().await.clear() },
1844 async { self.events_cache.write().await.clear() },
1845 async { self.spark_cache.write().await.clear() },
1846 async {
1847 #[cfg(feature = "indicators")]
1848 self.indicators_cache.write().await.clear();
1849 },
1850 );
1851 }
1852}
1853
1854#[cfg(test)]
1855mod tests {
1856 use super::*;
1857
1858 #[tokio::test]
1859 #[ignore = "requires network access"]
1860 async fn test_tickers_quotes() {
1861 let tickers = Tickers::new(["AAPL", "MSFT", "GOOGL"]).await.unwrap();
1862 let result = tickers.quotes().await.unwrap();
1863
1864 assert!(result.success_count() > 0);
1865 }
1866
1867 #[tokio::test]
1868 #[ignore = "requires network access"]
1869 async fn test_tickers_charts() {
1870 let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
1871 let result = tickers
1872 .charts(Interval::OneDay, TimeRange::FiveDays)
1873 .await
1874 .unwrap();
1875
1876 assert!(result.success_count() > 0);
1877 }
1878
1879 #[tokio::test]
1880 #[ignore = "requires network access"]
1881 async fn test_tickers_spark() {
1882 let tickers = Tickers::new(["AAPL", "MSFT", "GOOGL"]).await.unwrap();
1883 let result = tickers
1884 .spark(Interval::FiveMinutes, TimeRange::OneDay)
1885 .await
1886 .unwrap();
1887
1888 assert!(result.success_count() > 0);
1889
1890 if let Some(spark) = result.sparks.get("AAPL") {
1892 assert!(!spark.closes.is_empty());
1893 assert_eq!(spark.symbol, "AAPL");
1894 assert!(spark.percent_change().is_some());
1896 }
1897 }
1898
1899 #[tokio::test]
1900 #[ignore = "requires network access"]
1901 async fn test_tickers_dividends() {
1902 let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
1903 let result = tickers.dividends(TimeRange::OneYear).await.unwrap();
1904
1905 assert!(result.success_count() > 0);
1906
1907 if let Some(dividends) = result.dividends.get("AAPL")
1909 && !dividends.is_empty()
1910 {
1911 let div = ÷nds[0];
1912 assert!(div.timestamp > 0);
1913 assert!(div.amount > 0.0);
1914 }
1915 }
1916
1917 #[tokio::test]
1918 #[ignore = "requires network access"]
1919 async fn test_tickers_splits() {
1920 let tickers = Tickers::new(["NVDA", "TSLA"]).await.unwrap();
1921 let result = tickers.splits(TimeRange::FiveYears).await.unwrap();
1922
1923 assert!(result.success_count() > 0);
1925
1926 for splits in result.splits.values() {
1928 for split in splits {
1929 assert!(split.timestamp > 0);
1930 assert!(split.numerator > 0.0);
1931 assert!(split.denominator > 0.0);
1932 assert!(!split.ratio.is_empty());
1933 }
1934 }
1935 }
1936
1937 #[tokio::test]
1938 #[ignore = "requires network access"]
1939 async fn test_tickers_capital_gains() {
1940 let tickers = Tickers::new(["VFIAX", "VTI"]).await.unwrap();
1941 let result = tickers.capital_gains(TimeRange::TwoYears).await.unwrap();
1942
1943 assert!(result.success_count() > 0);
1945
1946 for gains in result.capital_gains.values() {
1948 for gain in gains {
1949 assert!(gain.timestamp > 0);
1950 assert!(gain.amount >= 0.0);
1951 }
1952 }
1953 }
1954
1955 #[tokio::test]
1956 #[ignore = "requires network access"]
1957 async fn test_tickers_financials() {
1958 let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
1959 let result = tickers
1960 .financials(StatementType::Income, Frequency::Annual)
1961 .await
1962 .unwrap();
1963
1964 assert!(result.success_count() > 0);
1965
1966 for (symbol, stmt) in &result.financials {
1968 assert_eq!(stmt.symbol, *symbol);
1969 assert_eq!(stmt.statement_type, "income");
1970 assert_eq!(stmt.frequency, "annual");
1971 assert!(!stmt.statement.is_empty());
1972
1973 if let Some(revenue) = stmt.statement.get("TotalRevenue") {
1975 assert!(!revenue.is_empty());
1976 }
1977 }
1978 }
1979
1980 #[tokio::test]
1981 #[ignore = "requires network access"]
1982 async fn test_tickers_news() {
1983 let tickers = Tickers::new(["AAPL", "TSLA"]).await.unwrap();
1984 let result = tickers.news().await.unwrap();
1985
1986 assert!(result.success_count() > 0);
1987
1988 for articles in result.news.values() {
1990 if !articles.is_empty() {
1991 let article = &articles[0];
1992 assert!(!article.title.is_empty());
1993 assert!(!article.link.is_empty());
1994 assert!(!article.source.is_empty());
1995 }
1996 }
1997 }
1998
1999 #[tokio::test]
2000 #[ignore = "requires network access"]
2001 async fn test_tickers_recommendations() {
2002 let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
2003 let result = tickers.recommendations(5).await.unwrap();
2004
2005 assert!(result.success_count() > 0);
2006
2007 for (symbol, rec) in &result.recommendations {
2009 assert_eq!(rec.symbol, *symbol);
2010 assert!(rec.count() > 0);
2011 for similar in &rec.recommendations {
2012 assert!(!similar.symbol.is_empty());
2013 }
2014 }
2015 }
2016
2017 #[tokio::test]
2018 #[ignore = "requires network access"]
2019 async fn test_tickers_options() {
2020 let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
2021 let result = tickers.options(None).await.unwrap();
2022
2023 assert!(result.success_count() > 0);
2024
2025 for opts in result.options.values() {
2027 assert!(!opts.expiration_dates().is_empty());
2028 }
2029 }
2030
2031 #[tokio::test]
2032 #[ignore = "requires network access"]
2033 #[cfg(feature = "indicators")]
2034 async fn test_tickers_indicators() {
2035 let tickers = Tickers::new(["AAPL", "MSFT"]).await.unwrap();
2036 let result = tickers
2037 .indicators(Interval::OneDay, TimeRange::ThreeMonths)
2038 .await
2039 .unwrap();
2040
2041 assert!(result.success_count() > 0);
2042
2043 for ind in result.indicators.values() {
2045 assert!(ind.rsi_14.is_some() || ind.sma_20.is_some());
2047 }
2048 }
2049
2050 #[tokio::test]
2051 async fn test_tickers_add_symbols() {
2052 let mut tickers = Tickers::new(["AAPL"]).await.unwrap();
2053 assert_eq!(tickers.len(), 1);
2054 assert_eq!(tickers.symbols(), &["AAPL"]);
2055
2056 tickers.add_symbols(&["MSFT", "GOOGL"]);
2057 assert_eq!(tickers.len(), 3);
2058 assert!(tickers.symbols().contains(&"AAPL"));
2059 assert!(tickers.symbols().contains(&"MSFT"));
2060 assert!(tickers.symbols().contains(&"GOOGL"));
2061
2062 tickers.add_symbols(&["AAPL"]);
2064 assert_eq!(tickers.len(), 3);
2065 }
2066
2067 #[tokio::test]
2068 #[ignore = "requires network access"]
2069 async fn test_tickers_remove_symbols() {
2070 let mut tickers = Tickers::new(["AAPL", "MSFT", "GOOGL"]).await.unwrap();
2071 assert_eq!(tickers.len(), 3);
2072
2073 let _ = tickers.quotes().await;
2075
2076 tickers.remove_symbols(&["MSFT"]).await;
2078 assert_eq!(tickers.len(), 2);
2079 assert!(tickers.symbols().contains(&"AAPL"));
2080 assert!(!tickers.symbols().contains(&"MSFT"));
2081 assert!(tickers.symbols().contains(&"GOOGL"));
2082
2083 let quotes = tickers.quotes().await.unwrap();
2085 assert!(!quotes.quotes.contains_key("MSFT"));
2086 assert_eq!(quotes.quotes.len(), 2);
2087 }
2088}