Merge pull request #73 from Codykilpatrick/feat/create-clean-database…

…-action

feat: clean model predictions daily

.github/workflows/purge-predictions.yaml

@@ -0,0 +1,35 @@
+name: Daily Purge Predict Average Increase
+on:
+ schedule:
+    - cron: "0 8 * * *"
+ workflow_dispatch:
+
+
+jobs:
+  Predict-average-increase:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+      - name: Install Poetry
+        run: pipx install poetry
+      - name: Setup Python
+        uses: actions/[email protected]
+        with:
+          python-version: 3.11
+          cache: poetry
+      - name: Install project dependencies
+        run: |
+          poetry install
+        working-directory: packages/celestia-models
+      - name: Run Purge Predict Average Increase
+        env:
+          NEON_CONNECTION_STRING: ${{ secrets.NEON_CONNECTION_STRING }}
+          DATABASE: ${{ secrets.DATABASE }}
+          NEON_USER: ${{ secrets.NEON_USER }}
+          NEON_PASSWORD: ${{ secrets.NEON_PASSWORD }}
+          NEON_HOST: ${{ secrets.NEON_HOST }}
+          NEON_PORT: ${{ secrets.NEON_PORT }}
+        run: |
+          poetry run python clean_model_predictions.py
+        working-directory: packages/celestia-models/pipeline

packages/celestia-data-pipeline/pipeline/clean_model_predictions.py

@@ -0,0 +1,46 @@
+import psycopg2
+import os
+from dotenv import load_dotenv
+from datetime import datetime, timedelta, timezone
+
+
+load_dotenv()
+
+# Get the current date and time in UTC
+now_utc = datetime.now(timezone.utc)
+
+# Get yesterday's date in UTC
+
+purge_day = now_utc - timedelta(days=4)
+purge_day = purge_day.strftime("%Y-%m-%d")
+
+# Connect to the PostgreSQL database
+db_host = os.getenv('NEON_HOST')
+db_user = os.getenv('NEON_USER')
+db_password = os.getenv('NEON_PASSWORD')
+db_name = os.getenv('DATABASE')
+db_params = {'host': db_host, 'database': db_name, 'user': db_user, 'password': db_password, }
+print(db_params)
+
+conn = psycopg2.connect(**db_params)
+# Create a cursor object
+cur = conn.cursor()
+
+print(purge_day)
+
+# Write the DELETE query
+delete_query = "DELETE FROM celestia_public.model_predict_average_increase WHERE date_predicted != %s;"
+
+# Execute the DELETE query
+cur.execute(delete_query, (purge_day,))
+
+# Get the number of rows deleted
+rows_deleted = cur.rowcount
+print(f"{rows_deleted} rows deleted.")
+
+# Commit the changes
+conn.commit()
+
+# Close the cursor and connection
+cur.close()
+conn.close()