working with recursive

src/main.rs

@@ -1,8 +1,11 @@
+use log;
 use macrosynergy_dataquery::oauth_client::OAuthClient;
 use macrosynergy_dataquery::requester;
 use macrosynergy_dataquery::requester::*;
 use macrosynergy_dataquery::timeseries::*;
 use rand::seq::SliceRandom;
+use std::collections::HashSet;
+use std::time::Instant;
 
 fn main() {
     let mut requester = requester::DQRequester::new(OAuthClient::default());
@@ -10,7 +13,6 @@ fn main() {
 
     let responses = requester.get_catalogue("JPMAQS", 1000).unwrap();
 
-
     let mut tickers_vec = vec![];
     for rsp in responses {
         let response: DQCatalogueResponse = serde_json::from_str(&rsp.to_string()).unwrap();
@@ -41,12 +43,31 @@ fn main() {
         }
     }
 
-    let rsp = requester.get_timeseries(DQTimeseriesRequestArgs {
-        expressions: expressions_vec[0..40].to_vec(),
-        ..Default::default()
-    });
+    // let rsp = requester.get_timeseries(DQTimeseriesRequestArgs {
+    //     expressions: expressions_vec[0..40].to_vec(),
+    //     ..Default::default()
+    // });
 
-    for response in rsp {
+    // for response in rsp {
+    //     let dq_response: DQTimeSeriesResponse =
+    //         serde_json::from_str(&response.text().unwrap()).unwrap();
+    //     for ts_group in dq_response.get_timeseries_by_ticker() {
+    //         let jpmi = JPMaQSIndicator::new(ts_group).unwrap();
+    //         println!("{:?}", jpmi.as_qdf().unwrap());
+    //     }
+    // }
+
+    // fetch first 50 expressions
+    let start = Instant::now();
+
+    let responses = requester
+        .get_timeseries(DQTimeseriesRequestArgs {
+            expressions: expressions_vec[0..50].to_vec(),
+            ..Default::default()
+        })
+        .unwrap();
+
+    for response in responses {
         let dq_response: DQTimeSeriesResponse =
             serde_json::from_str(&response.text().unwrap()).unwrap();
         for ts_group in dq_response.get_timeseries_by_ticker() {

src/requester.rs

@@ -167,10 +167,14 @@ impl DQRequester {
         Ok(responses)
     }
 
-    pub fn get_single_timeseries_batch(
+    pub fn _fetch_single_timeseries_batch(
         &mut self,
         args: DQTimeseriesRequestArgs,
     ) -> Result<reqwest::blocking::Response, Box<dyn Error>> {
+        log::info!(
+            "Fetching timeseries batch with {} expressions",
+            args.expressions.len()
+        );
         if args.expressions.len() < 1 || args.expressions.len() > 20 {
             return Err("Number of expressions must be between 1 and 20".into());
         }
@@ -179,49 +183,94 @@ impl DQRequester {
         let endpoint = format!("{}?{}", TIMESERIES_ENDPOINT, query_string);
         let response = self._request(reqwest::Method::GET, &endpoint)?;
 
+        log::info!("Got response: {:?}", response.status());
         Ok(response)
     }
 
-    pub fn get_single_timeseries_with_defaults(
-        &mut self,
-        expressions: Vec<&str>,
-    ) -> Result<reqwest::blocking::Response, Box<dyn Error>> {
-        let args = DQTimeseriesRequestArgs {
-            expressions: expressions.into_iter().map(|s| s.to_string()).collect(),
-            ..Default::default()
-        };
+    // pub fn get_timeseries(
+    //     &mut self,
+    //     args: DQTimeseriesRequestArgs,
+    // ) -> Vec<reqwest::blocking::Response> {
+    //     let expressions_chunks: Vec<Vec<String>> = args
+    //         .expressions
+    //         .chunks(20)
+    //         .map(|chunk| chunk.to_vec())
+    //         .collect();
 
-        self.get_single_timeseries_batch(args)
-    }
+    //     let mut responses: Vec<Result<reqwest::blocking::Response, Box<dyn Error>>> = Vec::new();
 
-    pub fn get_timeseries(
+    //     for expressions in expressions_chunks {
+    //         let mut args = args.clone();
+    //         args.update_expressions(expressions);
+    //         let response = self.get_single_timeseries_batch(args);
+    //         responses.push(response);
+    //         // just sleep for a bit not threadsleep
+    //         // println!("Sleeping for 500ms");
+    //         std::thread::sleep(std::time::Duration::from_millis(500));
+    //     }
+
+    //     responses.into_iter().map(|r| r.unwrap()).collect()
+    // }
+
+    fn _fetch_timeseries_recursive(
         &mut self,
         args: DQTimeseriesRequestArgs,
-    ) -> Vec<reqwest::blocking::Response> {
-        let expressions_chunks: Vec<Vec<String>> = args
+        max_retries: u32,
+    ) -> Result<Vec<reqwest::blocking::Response>, Box<dyn Error>> {
+        let expression_batches: Vec<Vec<String>> = args
             .expressions
             .chunks(20)
             .map(|chunk| chunk.to_vec())
             .collect();
 
-        let mut responses: Vec<Result<reqwest::blocking::Response, Box<dyn Error>>> = Vec::new();
+        let mut okay_responses: Vec<reqwest::blocking::Response> = Vec::new();
+        let mut failed_batches: Vec<Vec<String>> = Vec::new();
 
-        for expressions in expressions_chunks {
+        for expr_batch in expression_batches {
             let mut args = args.clone();
-            args.update_expressions(expressions);
-            println!(
-                "Requesting timeseries for expressions: {:?}",
-                args.expressions
-            );
-            let response = self.get_single_timeseries_batch(args);
-            responses.push(response);
-            // just sleep for a bit not threadsleep
-            println!("Sleeping for 500ms");
-            std::thread::sleep(std::time::Duration::from_millis(500));
+            args.update_expressions(expr_batch.clone());
+            let response = self._fetch_single_timeseries_batch(args);
+            std::thread::sleep(std::time::Duration::from_millis(200));
+            match response {
+                Ok(r) => {
+                    okay_responses.push(r);
 
+                    log::info!("Got batch: {:?}", expr_batch);
+                }
+                Err(e) => {
+                    failed_batches.push(expr_batch.clone());
+                    log::error!("Failed to get batch: {:?} : {:?}", expr_batch, e);
+                }
+            }
         }
 
-        responses.into_iter().map(|r| r.unwrap()).collect()
+        if !failed_batches.is_empty() && max_retries == 0 {
+            return Err("Max retries reached".into());
+        }
+
+        if !failed_batches.is_empty() && max_retries > 0 {
+            log::info!("Retrying failed batches");
+            let mut new_args = args.clone();
+            new_args.update_expressions(failed_batches.concat());
+            log::info!("Retrying with {} failed expressions", failed_batches.len());
+            let mut retry_responses =
+                self._fetch_timeseries_recursive(new_args, max_retries - 1)?;
+            okay_responses.append(&mut retry_responses);
+        }
+        log::info!("Returning {} responses", okay_responses.len());
+        Ok(okay_responses)
+    }
+
+    pub fn get_timeseries(
+        &mut self,
+        args: DQTimeseriesRequestArgs,
+    ) -> Result<Vec<reqwest::blocking::Response>, Box<dyn Error>> {
+        let max_retries = 5;
+        println!(
+            "Invoking recursive function for {:?} expressions",
+            args.expressions.len()
+        );
+        self._fetch_timeseries_recursive(args, max_retries)
     }
 }
 

src/timeseries.rs

@@ -108,6 +108,18 @@ impl DQTimeSeries {
 }
 
 impl DQTimeSeriesResponse {
+    pub fn list_expressions(&self) -> Vec<String> {
+        self.instruments
+            .iter()
+            .flat_map(|instrument| {
+                instrument
+                    .attributes
+                    .iter()
+                    .map(|attribute| attribute.expression.clone())
+            })
+            .collect()
+    }
+
     /// Return a list of all DQTimeSeries in the response
     pub fn get_all_timeseries(&self) -> Vec<DQTimeSeries> {
         self.instruments