[ Feat ]: Implementation Server Side Caching \u Redis (Aiven Cloud)

.github/workflows/gru_pipeline.yaml

@@ -186,10 +186,12 @@ jobs:
         run: |
           rm models.zip
           rm pickles.zip
+          rm datasets.zip
           rm posttrained.zip
 
           rm -rf models
           rm -rf pickles
+          rm -rf datasets
           rm -rf posttrained
 
       - name: Commit changes
@@ -205,7 +207,15 @@ jobs:
         uses: ad-m/github-push-action@master
         with:
           github_token: ${{ secrets.GH_TOKEN }}
-          branch: main 
+          branch: main
+
+      - name: Flush All Caching Data (Redis)
+        if: env.match != 'true'
+        run: |
+          echo "FLUSHALL" | redis-cli --tls \
+            -h ${{ secrets.REDIS_HOST }} \
+            -p ${{ secrets.REDIS_PORT }} \
+            -a ${{ secrets.REDIS_TOKEN }}
 
 
   tebakaja_crypto_space-0:

.github/workflows/lstm_gru_pipeline.yaml

@@ -186,10 +186,12 @@ jobs:
         run: |
           rm models.zip
           rm pickles.zip
+          rm datasets.zip
           rm posttrained.zip
 
           rm -rf models
           rm -rf pickles
+          rm -rf datasets
           rm -rf posttrained
 
       - name: Commit changes

.github/workflows/lstm_pipeline.yaml

@@ -186,10 +186,12 @@ jobs:
         run: |
           rm models.zip
           rm pickles.zip
+          rm datasets.zip
           rm posttrained.zip
 
           rm -rf models
           rm -rf pickles
+          rm -rf datasets
           rm -rf posttrained
 
       - name: Commit changes

.gitignore

@@ -9,6 +9,8 @@
 /Include
 /Scripts
 
+.env
+
 # Pycache
 /__pycache__
 /restful/__pycache__

app.py

@@ -2,6 +2,7 @@ from fastapi import FastAPI
 from restful.routes import route
 from fastapi.responses import RedirectResponse
 from fastapi.middleware.cors import CORSMiddleware
+from restful.cachings import close_caching_connector
 
 app = FastAPI(
     title = "Cryptocurency Prediction Service",
@@ -24,5 +25,10 @@ app.include_router(
 )
 
 @app.get("/", tags = ['Main'])
-def root():
+def root() -> None:
     return RedirectResponse(url="/docs")
+
+
+@app.on_event("shutdown")
+async def on_shutdown() -> None:
+    await close_caching_connector()

requirements.txt

@@ -7,7 +7,9 @@ pandas==2.2.2
 protobuf==4.25.3
 pydantic==2.7.2
 pydantic_core==2.18.3
+redis==3.5.3
 scikit-learn==1.5.0
 scipy==1.13.1
 tensorflow==2.16.1
 uvicorn==0.30.1
+valkey==6.0.0b1

restful/cachings.py

@@ -0,0 +1,39 @@
+from redis import Redis
+from valkey import Valkey
+from typing import _ProtocolMeta
+from restful.configs import settings, Settings
+
+
+""" caching_connector """
+def _caching_connector(db_type: str, env_var: Settings) -> _ProtocolMeta:
+  if db_type == 'redis':
+    return Redis(
+      host     = env_var.CACHING_HOST,
+      port     = env_var.CACHING_PORT,
+      password = env_var.CACHING_PASS,
+      ssl      = True
+    )
+  
+  elif db_type == 'valkey':
+    return Valkey(
+      host     = env_var.CACHING_HOST,
+      port     = env_var.CACHING_PORT,
+      password = env_var.CACHING_PASS,
+      ssl      = True
+    )
+
+  # default
+  else:
+    return Redis(
+      host     = env_var.CACHING_HOST,
+      port     = env_var.CACHING_PORT,
+      password = env_var.CACHING_PASS,
+      ssl      = True
+    )
+
+connector: _ProtocolMeta = _caching_connector(
+  db_type = settings.CACHING_TYPE, env_var = settings)
+
+async def close_caching_connector() -> None:
+    global connector
+    if connector: await connector.close()

restful/configs.py

@@ -0,0 +1,13 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    CACHING_TYPE: str
+    CACHING_HOST: str
+    CACHING_PORT: int
+    CACHING_PASS: str
+
+    class Config:
+        env_file = ".env"
+
+settings = Settings()

restful/controllers.py

@@ -1,22 +1,23 @@
 import os
 from http import HTTPStatus
+from typing import _ProtocolMeta
 from fastapi.responses import JSONResponse
 from restful.services import ForecastingService
 from restful.schemas import ForecastingServiceSchema
 
 
-""" Forecasting Controller """
+# Forecasting Controller
 class ForecastingControllers:
 
     __SERVICE: ForecastingService = ForecastingService()
 
 
-    """
-        Algorithms Controller
-    """
+    # Algorithms Controller
     async def algorithms_controller(self) -> JSONResponse:
         try:
-            algorithms: list = sorted(os.listdir("resources/algorithms"))
+            algorithms: list = sorted(
+                os.listdir("resources/algorithms"))
+
             return JSONResponse(
                 content = {
                     'message': 'Success',
@@ -39,16 +40,15 @@ class ForecastingControllers:
 
 
 
-    """
-        Currency Controller
-    """
+    # Currency Controller
     async def currencies_controller(self) -> JSONResponse:
         try:
             path: str = 'resources/datasets'
             datasets: list = sorted(
                 [
                     item.replace(".csv", "") for item in os.listdir(path)
-                    if os.path.isfile(os.path.join(path, item)) and item.endswith('.csv')
+                    if os.path.isfile(os.path.join(path, item)) 
+                        and item.endswith('.csv')
                 ]
             )
 
@@ -74,10 +74,9 @@ class ForecastingControllers:
 
 
 
-    """ 
-        Forecasting Controller
-    """
-    async def forecasting_controller(self, payload: ForecastingServiceSchema) -> JSONResponse:
+    # Forecasting Controller
+    async def forecasting_controller(self, payload: ForecastingServiceSchema,
+        caching: _ProtocolMeta) -> JSONResponse:
         try:
             path: str = 'resources/datasets'
             datasets: list = sorted(
@@ -120,7 +119,8 @@ class ForecastingControllers:
                 )
 
 
-            prediction: dict = await self.__SERVICE.forecasting(payload)
+            prediction: dict = await self.__SERVICE.forecasting(
+                payload = payload, caching = caching)
 
             if not prediction :
                 return JSONResponse(

restful/cutils/utilities.pyx

@@ -1,13 +1,14 @@
 import os
+import cython
 from joblib import load
 from numpy import append, expand_dims
 from pandas import read_json, to_datetime, Timedelta
 from tensorflow.keras.models import load_model
-import cython
+
 
 cdef class Utilities:
     async def forecasting_utils(self, int sequence_length,
-        int days, str model_name, str algorithm) -> tuple:
+        int days, str model_name, str algorithm, bint with_pred) -> tuple:
         cdef str model_path = os.path.join(f'./resources/algorithms/{algorithm}/models',
             f'{model_name}.keras')
         model = load_model(model_path)
@@ -21,27 +22,30 @@ cdef class Utilities:
             f'{model_name}_minmax_scaler.pickle'))
         standard_scaler = load(os.path.join(f'./resources/algorithms/{algorithm}/pickles',
             f'{model_name}_standard_scaler.pickle'))
-        
-        # Prediction
-        lst_seq = dataframe[-sequence_length:].values
-        lst_seq = expand_dims(lst_seq, axis=0)
 
-        cdef dict predicted_prices = {}
-        last_date = to_datetime(dataframe.index[-1])
+        if with_pred:
+            # Prediction
+            lst_seq = dataframe[-sequence_length:].values
+            lst_seq = expand_dims(lst_seq, axis=0)
 
-        for _ in range(days):
-            predicted_price = model.predict(lst_seq)
-            last_date = last_date + Timedelta(days=1)
+            predicted_prices = {}
+            last_date = to_datetime(dataframe.index[-1])
 
-            predicted_prices[last_date] = minmax_scaler.inverse_transform(predicted_price)
-            predicted_prices[last_date] = standard_scaler.inverse_transform(predicted_prices[last_date])
+            for _ in range(days):
+                predicted_price = model.predict(lst_seq)
+                last_date = last_date + Timedelta(days=1)
 
-            lst_seq = append(lst_seq[:, 1:, :], [predicted_price], axis=1)
+                predicted_prices[last_date] = minmax_scaler.inverse_transform(predicted_price)
+                predicted_prices[last_date] = standard_scaler.inverse_transform(predicted_prices[last_date])
 
-        predictions = [
-            {'date': date.strftime('%Y-%m-%d'), 'price': float(price)} \
-                for date, price in predicted_prices.items()
-        ]
+                lst_seq = append(lst_seq[:, 1:, :], [predicted_price], axis=1)
+
+            predictions = [
+                {'date': date.strftime('%Y-%m-%d'), 'price': float(price)} \
+                    for date, price in predicted_prices.items()
+            ]
+
+        else: predictions = []
 
         # Actual
         df_date = dataframe.index[-sequence_length:].values

restful/routes.py

@@ -1,13 +1,22 @@
+from typing import _ProtocolMeta
 from fastapi import APIRouter, Body
 from fastapi.responses import JSONResponse
-from restful.controllers import ForecastingControllers
+
+from restful.cachings import connector
 from restful.schemas import ForecastingServiceSchema
+from restful.controllers import ForecastingControllers
+
 
 """ API Router """
 route = APIRouter()
 
+
 """ Forecasting Controller """
-__CONTROLLER = ForecastingControllers()
+__CONTROLLER: ForecastingControllers = ForecastingControllers()
+
+
+""" Caching Connector """
+__CONNECTOR: _ProtocolMeta = connector
 
 
 """ Algorithms Route """
@@ -27,5 +36,5 @@ async def currencies_route() -> JSONResponse:
 async def forecasting_route(
     payload: ForecastingServiceSchema = Body(...)
 ) -> JSONResponse:
-    return await __CONTROLLER.forecasting_controller(payload = payload)
+    return await __CONTROLLER.forecasting_controller(payload = payload, caching = __CONNECTOR)
 

restful/services.py

@@ -1,3 +1,5 @@
+import json
+from typing import _ProtocolMeta
 from restful.cutils.utilities import Utilities
 from restful.schemas import ForecastingServiceSchema
 
@@ -7,17 +9,27 @@ class ForecastingService:
 
 	__FORECAST_UTILS = Utilities()
 
-	async def forecasting(self, payload: ForecastingServiceSchema) -> dict:
-		days: int      = payload.days
-		currency: str  = payload.currency
-		algorithm: str = payload.algorithm
+	async def forecasting(self, payload: ForecastingServiceSchema, caching: _ProtocolMeta) -> dict:
+		caching_data = caching.get(
+			f'{payload.algorithm}_{payload.currency}_{payload.days}')
 
 		actuals, predictions = await self.__FORECAST_UTILS.forecasting_utils(
-			days            = days,
-			algorithm       = algorithm,
-			model_name      = currency,
+			days        = payload.days,
+			algorithm   = payload.algorithm,
+			model_name  = payload.currency,
 
+			with_pred   = (caching_data == None),
 			sequence_length = 60
 		)
 
+		if caching_data != None:
+			predictions = json.loads(caching_data.decode('utf-8'))
+
+		else:
+			caching.set(
+				f'{payload.algorithm}_{payload.currency}_{payload.days}',
+				json.dumps(predictions)
+			)
+
+			
 		return {'actuals': actuals, 'predictions': predictions}