Fix #685 - [Do Not Merge] - DuckDB - E2E Integration

pdr_backend/analytics/predictoor_stats.py

@@ -34,16 +34,16 @@ class PredictoorStat(TypedDict):
 def get_feed_summary_stats(predictions_df: pl.DataFrame) -> pl.DataFrame:
     # 1 - filter from lake only the rows that you're looking for
     df = predictions_df.filter(
-        ~((pl.col("trueval").is_null()) | (pl.col("payout").is_null()))
+        ~((pl.col("truevalue").is_null()) | (pl.col("payout").is_null()))
     )
 
     # Group by pair
     df = df.group_by(["pair", "timeframe"]).agg(
         pl.col("source").first().alias("source"),
         pl.col("payout").sum().alias("sum_payout"),
         pl.col("stake").sum().alias("sum_stake"),
-        pl.col("prediction").count().alias("num_predictions"),
-        (pl.col("prediction").sum() / pl.col("pair").count() * 100).alias("accuracy"),
+        pl.col("predvalue").count().alias("num_predictions"),
+        (pl.col("predvalue").sum() / pl.col("pair").count() * 100).alias("accuracy"),
     )
 
     return df
@@ -53,16 +53,16 @@ def get_feed_summary_stats(predictions_df: pl.DataFrame) -> pl.DataFrame:
 def get_predictoor_summary_stats(predictions_df: pl.DataFrame) -> pl.DataFrame:
     # 1 - filter from lake only the rows that you're looking for
     df = predictions_df.filter(
-        ~((pl.col("trueval").is_null()) | (pl.col("payout").is_null()))
+        ~((pl.col("truevalue").is_null()) | (pl.col("payout").is_null()))
     )
 
     # Group by pair
     df = df.group_by(["user", "pair", "timeframe"]).agg(
         pl.col("source").first().alias("source"),
         pl.col("payout").sum().alias("sum_payout"),
         pl.col("stake").sum().alias("sum_stake"),
-        pl.col("prediction").count().alias("num_predictions"),
-        (pl.col("prediction").sum() / pl.col("pair").count() * 100).alias("accuracy"),
+        pl.col("predvalue").count().alias("num_predictions"),
+        (pl.col("predvalue").sum() / pl.col("pair").count() * 100).alias("accuracy"),
     )
 
     return df

pdr_backend/lake/base_data_store.py

@@ -0,0 +1,44 @@
+from hashlib import md5
+from abc import abstractmethod
+from typing import Optional, Literal
+
+import duckdb
+from enforce_typing import enforce_types
+
+
+class BaseDataStore:
+    @enforce_types
+    def __init__(self, base_directory=str):
+        """
+        Initialize a PartitionedDataStore instance.
+        @arguments:
+            base_directory - The base directory to store the partitioned Parquet files.
+        """
+
+        self.base_directory = base_directory
+        self.duckdb_conn = duckdb.connect(
+            database=f"{self.base_directory}/duckdb.db"
+        )  # Keep a persistent connection
+
+    @enforce_types
+    def _generate_view_name(self, base_path=str) -> str:
+        """
+        Generate a unique view name for a given base path.
+        @arguments:
+            base_path - The base path to generate a view name for.
+        @returns:
+            str - A unique view name.
+        """
+
+        path = f"{self.base_directory}/{base_path}"
+        hash_object = md5(path.encode())
+        return f"dataset_{hash_object.hexdigest()}"
+
+    @abstractmethod
+    def query_data(
+        self,
+        dataset_identifier: str,
+        query: str,
+        partition_type: Optional[Literal["date", "address"]] = None,
+    ):
+        pass