Made data retreiver faster

data_updater/find_markets.py

@@ -216,20 +216,27 @@ def process_single_row(row, client):
     return ret
 
 
-def get_all_results(all_df, client):
+def get_all_results(all_df, client, max_workers=5):
     all_results = []
 
-    for idx, row in all_df.iterrows():
+    def process_with_progress(args):
-
+        idx, row = args
         try:
-            if idx % 10 == 0:
+            return process_single_row(row, client)
-                print(f'{idx} of {len(all_df)}')
-
-            time.sleep(1)
-            result = process_single_row(row, client)
-            all_results.append(result)
         except:
             print("error fetching market")
+            return None
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+        futures = [executor.submit(process_with_progress, (idx, row)) for idx, row in all_df.iterrows()]
+        
+        for future in concurrent.futures.as_completed(futures):
+            result = future.result()
+            if result is not None:
+                all_results.append(result)
+
+            if len(all_results) % (max_workers * 2) == 0:
+                print(f'{len(all_results)} of {len(all_df)}')
 
     return all_results
 
@@ -283,21 +290,30 @@ def add_volatility(row):
     new_dict = {**row_dict, **stats}
     return new_dict
 
-def add_volatility_to_df(df):
+def add_volatility_to_df(df, max_workers=3):
     
     results = []
     df = df.reset_index(drop=True)
 
-    for idx, row in df.iterrows():
+    def process_volatility_with_progress(args):
+        idx, row = args
         try:
-            if idx % 10 == 0:
-                print(f'{idx} of {len(df)}')
-                
             ret = add_volatility(row.to_dict())
-            time.sleep(1)
+            return ret
-            results.append(ret)
         except:
             print("Error fetching volatility")
+            return None
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+        futures = [executor.submit(process_volatility_with_progress, (idx, row)) for idx, row in df.iterrows()]
+        
+        for future in concurrent.futures.as_completed(futures):
+            result = future.result()
+            if result is not None:
+                results.append(result)
+                
+            if len(results) % (max_workers * 2) == 0:
+                print(f'{len(results)} of {len(df)}')
             
     return pd.DataFrame(results)
 

update_markets.py

@@ -88,8 +88,11 @@ def fetch_and_process_data():
 
 
     all_df = get_all_markets(client)
+    print("Got all Markets")
     all_results = get_all_results(all_df, client)
+    print("Got all Results")
     m_data, all_markets = get_markets(all_results, sel_df, maker_reward=0.75)
+    print("Got all orderbook")
 
     print(f'{pd.to_datetime("now")}: Fetched all markets data of length {len(all_markets)}.')
     new_df = add_volatility_to_df(all_markets)
@@ -124,7 +127,7 @@ if __name__ == "__main__":
     while True:
         try:
             fetch_and_process_data()
-            time.sleep(60 * 5)  # Sleep for 5 minutes (300 seconds)
+            time.sleep(60 * 60)  # Sleep for an hour
         except Exception as e:
             traceback.print_exc()
             print(str(e))