feat: ✨ Adding ability to insert multiple

.pre-commit-config.yaml

@@ -10,7 +10,6 @@ repos:
       - id: check-json
       - id: check-merge-conflict
       - id: check-toml
-      - id: detect-aws-credentials
       - id: name-tests-test
 
   - repo: https://github.com/PyCQA/flake8

foresight/utils/database.py

@@ -1,6 +1,7 @@
 """Provides a singleton class to interact with the TimescaleDB database."""
 
 import os
+from typing import Union
 
 import dotenv
 import psycopg2
@@ -74,12 +75,16 @@ def create_table(self, query, table_name=None, column_name=None):
         else:
             raise Exception("Database connection not established.")
 
-    def execute(self, query, params=None):
+    def execute(self, query, params: Union[tuple, list] = None):
         """Execute a query on the database."""
         if self.connection is not None:
             try:
                 with self.connection.cursor() as cursor:
-                    cursor.execute(query, params)
+                    if params is None or isinstance(params, tuple):
+                        cursor.execute(query, params)
+                    elif isinstance(params, list):
+                        psycopg2.extras.execute_values(cursor, query, params)
+
                     if cursor.description is not None:
                         data = cursor.fetchall()
                         return [dict(row) for row in data]

foresight/utils/models/forex_data.py

@@ -2,7 +2,6 @@
 
 from datetime import datetime
 
-import pandas as pd
 from pydantic import BaseModel
 
 from foresight.utils.database import TimeScaleService
@@ -11,7 +10,15 @@
 
 logger = generate_logger(name=__name__)
 
-timeMap: dict = {"S": "second", "M": "minute", "H": "hour", "D": "day"}
+time_map: dict = {"S": "second", "M": "minute", "H": "hour", "D": "day"}
+
+# Generate the interval based on the timescale
+interval_map: dict = {
+    "S": "60 minute",
+    "M": "24 hour",
+    "H": "14 day",
+    "D": "120 day",
+}
 
 
 class ForexData(BaseModel):
@@ -68,6 +75,23 @@ def insert_forex_data(self, table_name: str = "forex_data"):
             ),
         )
 
+    @staticmethod
+    def insert_multiple(data: list["ForexData"], table_name: str = "forex_data"):
+        """Insert list of multiple forex data efficiently."""
+        if len(data) > 0:
+            TimeScaleService().execute(
+                query=f"""INSERT INTO {table_name} (instrument, time, bid, ask) VALUES %s""",
+                params=[
+                    (
+                        row.instrument,
+                        row.time,
+                        row.bid,
+                        row.ask,
+                    )
+                    for row in data
+                ],
+            )
+
     @staticmethod
     def drop_table(table_name: str = "forex_data"):
         """Drop a table in the data store.
@@ -84,6 +108,8 @@ def fetch(instrument: str = "EUR_USD", timescale: str = "M") -> list["ForexData"
         """
         Fetch all data from the database and return a DataFrame.
 
+        The goal is to great the moving average at the timescale granularity.
+
         Parameters:
             instrument (str): The instrument to fetch
             timescale (str): The timescale to fetch (S = Second, M = Minute)
@@ -92,38 +118,19 @@ def fetch(instrument: str = "EUR_USD", timescale: str = "M") -> list["ForexData"
             dict: The data from the database
         """
         try:
-            df = pd.DataFrame(
-                TimeScaleService().execute(
-                    query=f"""
-                        SELECT
-                        TO_CHAR(
-                            date_trunc(
-                                '{timeMap[timescale.lower()]}', time
-                            ),
-                            'YYYY-MM-DD HH24:MI:SS'
-                        ) as time,
-                        AVG(ask) as ask, AVG(bid) as bid
-                        FROM forex_data
-                        WHERE instrument = '{instrument}'
-                        AND time >= NOW() - INTERVAL '60 minute'
-                        GROUP BY time
-                        ORDER BY time ASC
-                    """,
-                ),
-            )
+            query = f"""SELECT
+                instrument,
+                time_bucket('{interval_map[timescale]}', time) as time,
+                AVG(bid) as bid,
+                AVG(ask) as ask
+            FROM forex_data
+            WHERE instrument = '{instrument}'
+            GROUP BY instrument, time
+            ORDER BY time ASC"""
+            results = TimeScaleService().execute(query=query)
+
+            return [ForexData(**row) for row in results]
 
-            df = (
-                df.groupby(df["time"]).agg({"ask": "mean", "bid": "mean"}).reset_index()
-            )
-            return [
-                ForexData(
-                    instrument=instrument,
-                    time=datetime.strptime(row["time"], "%Y-%m-%d %H:%M:%S"),
-                    bid=row["bid"],
-                    ask=row["ask"],
-                )
-                for _, row in df.iterrows()
-            ]
         except Exception as fetch_exception:  # pylint: disable=broad-except
             logger.error("Error fetching data: %s", fetch_exception)
 

tests/conftest.py

@@ -0,0 +1,13 @@
+"""Test configuration file for fixtures."""
+
+import pytest
+
+from foresight.utils.models.forex_data import ForexData
+
+
+@pytest.fixture()
+def setup_forex_data():
+    """Setup forex data for testing."""
+    ForexData.create_table()
+    yield
+    ForexData.drop_table()

tests/utils/models/forex_data_test.py

@@ -79,6 +79,47 @@ def test_create_table():
         assert column["data_type"] == expected_columns[column["column_name"]]
 
 
+@pytest.mark.usefixtures("setup_forex_data")
+@pytest.mark.parametrize(
+    "timescale",
+    [
+        "S",
+        "M",
+        "H",
+        "D",
+    ],
+)
+def test_insert_and_fetch(timescale):
+    """Insert and fetch forex data."""
+
+    # ! TODO Better Test of Aggregate (across timescales)
+
+    # ARRANGE
+    dt_end = datetime.datetime.now()
+
+    # Create random data for the last 1000 seconds
+    data_array = []
+    for i in range(1000):
+        dt = dt_end - datetime.timedelta(seconds=i)
+        dt_end = dt
+        data_array.append(
+            ForexData(
+                instrument="EUR_USD",
+                time=dt,
+                bid=1.0 + i,
+                ask=2.0 + i,
+            ),
+        )
+
+    # ACT
+    ForexData.insert_multiple(data=data_array)
+
+    data = ForexData.fetch(timescale=timescale)
+
+    # ASSERT
+    assert len(data) == 1000
+
+
 def test_to_price_json():
     """Test the to_price_json method."""