1.1.1

src/main.rs

@@ -1,6 +1,7 @@
 use clokwerk::{AsyncScheduler, Job, TimeUnits};
 use futures::future::join_all;
 use lazy_static::lazy_static;
+use polars::prelude::DataFrame;
 use rand::Rng;
 use std::env;
 use std::error::Error;
@@ -16,19 +17,29 @@ use util::*;
 
 lazy_static! {
     static ref SOURCE: Source = match env::var("ARK_SOURCE") {
-        Ok(val) => Source::from_str(val.as_str()).expect("Env string SOURCE is not in enum Source"),
-        Err(_e) => Source::Ark,
+        Ok(val) =>
+            Source::from_str(val.as_str()).expect("Env string ARK_SOURCE is not in enum Source"),
+        Err(_e) => Source::ApiIncremental,
     };
 }
 
+fn print_df(ticker: &Ticker, df: &DataFrame) {
+    println!(
+        "Ticker: {:#?}\nShape: {:?}\n{:#?}",
+        ticker,
+        df.shape(),
+        df.tail(Some(1))
+    );
+}
+
 fn csv_merge() -> Result<(), Box<dyn Error>> {
     for ticker in Ticker::iter() {
         let df = Ark::merge_old_csv_to_parquet(ticker, None)?
             .format()?
             .sort()?
             .write_parquet()?
-            .collect();
-        println!("Ticker: {:#?}\n{:#?}", ticker, df);
+            .collect()?;
+        print_df(&ticker, &df);
     }
     Ok(())
 }
@@ -44,7 +55,7 @@ fn ark_plan(ticker: Ticker) -> Result<(), Box<dyn Error>> {
         .write_parquet()?
         .collect()?;
 
-    println!("Ticker: {:#?}\n{:#?}", ticker, df.tail(Some(1)));
+    print_df(&ticker, &df);
     Ok(())
 }
 
@@ -73,7 +84,7 @@ async fn main() {
         .map(|v| v == "true")
         .unwrap_or(false)
     {
-        print!("Merging CSVs to Parquet...");
+        println!("Merging CSVs to Parquet");
         csv_merge().unwrap();
     }
 
@@ -84,7 +95,7 @@ async fn main() {
         ark_etf().await;
     }
 
-    scheduler.every(1.day()).at("11:30 pm").run(ark_etf);
+    scheduler.every(1.day()).at("10:00 am").run(ark_etf);
 
     scheduler
         .every(5.day())

src/util.rs

@@ -82,9 +82,13 @@ impl DFS for Vec<DF> {
 
 #[derive(EnumString, Clone, Copy)]
 pub enum Source {
+    // Reads Parquet file if exists
     Read,
+    // From ARK Invest
     Ark,
+    // From api.NexVeridian.com
     ApiIncremental,
+    // From api.NexVeridian.com, not usually nessisary, use ApiIncremental
     ApiFull,
 }
 pub struct Ark {
@@ -259,25 +263,34 @@ impl Ark {
         }
 
         if !df.fields().contains(&Field::new("date", DataType::Date)) {
-            let date_format = |df: DataFrame, format: &str| -> Result<DataFrame, Box<dyn Error>> {
-                Ok(df
+            let date_format = |mut df: DataFrame, format:Option<String>| -> Result<DataFrame, Box<dyn Error>> {
+                df = df
                     .lazy()
                     .with_column(col("date").str().strptime(
                         DataType::Date,
                         StrptimeOptions {
-                            format: Some(format.into()),
+                            format,
                             strict: false,
                             exact: true,
                             cache: true,
                         },
                     ))
-                    .collect()?)
+                    .collect()?;
+                
+                if df.column("date").unwrap().null_count() > df.height() / 10 {
+                    return Err("wrong date format".into());
+                }
+
+                Ok(df)
             };
 
-            if let Ok(x) = date_format(df.clone(), "%m/%d/%Y") {
+            if let Ok(x) = date_format(df.clone(), Some("%m/%d/%Y".into())) {
                 df = x
             }
-            if let Ok(x) = date_format(df.clone(), "%Y/%m/%d") {
+            else if let Ok(x) = date_format(df.clone(), Some("%Y/%m/%d".into())) {
+                df = x
+            }
+            else if let Ok(x) = date_format(df.clone(), None) {
                 df = x
             }
         }
@@ -397,9 +410,8 @@ impl Ark {
 
         if Self::read_parquet(ticker, path.clone()).is_ok() {
             let df_old = Self::read_parquet(ticker, path.clone())?;
-            df = Self::concat_df(vec![Self::df_format(df_old)?, Self::df_format(df)?])?
+            df = Self::concat_df(vec![Self::df_format(df_old)?, Self::df_format(df)?])?;
         }
-
         Ok(Self { df, ticker, path })
     }
 }