Added API to update NAV table

buid_requirements.txt

@@ -3,3 +3,4 @@ fastapi==0.141.1
 python-dotenv==1.2.2
 python-multipart==0.0.32
 uvicorn==0.52.1
+psycopg2-binary==2.9.12

main.py

@@ -5,6 +5,8 @@ from fastapi import Depends, FastAPI, HTTPException, UploadFile, Response
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from dotenv import load_dotenv
 from pathlib import Path
+import psycopg2
+from psycopg2.extras import execute_values
 
 load_dotenv()
 
@@ -15,6 +17,16 @@ UPLOAD_PATH = "uploads"
 
 security = HTTPBearer()
 
+DB_HOST = os.getenv('DB_HOST')
+DB_USER = os.getenv('DB_USER')
+DB_PWD  = os.getenv('DB_PWD')
+DB_PORT = os.getenv('DB_PORT')
+DB_NAME = os.getenv('DB_NAME')
+
+
+def connect_db():
+    return psycopg2.connect(f"dbname=mf user={DB_USER} password={DB_PWD} host={DB_HOST} port={DB_PORT}")
+
 
 def authenticate(
     credentials: HTTPAuthorizationCredentials = Depends(security),
@@ -45,3 +57,53 @@ async def upload(file: UploadFile):
     await file.close()
 
     return {"status": "success"}
+
+
+@app.post("/update-nav", dependencies=[Depends(authenticate)])
+async def update_nav(data: list[dict]):
+    """Upsert the NAV data into the nav_history table"""
+
+    columns = ['amfi_code', 'date', 'nav']
+
+    query = f"""
+        INSERT INTO nav_history ({', '.join(columns)})
+        VALUES %s
+    """
+
+    values = [
+        tuple(row[col] for col in columns)
+        for row in data
+    ]
+    conn = connect_db()
+    with conn.cursor() as cursor:
+        execute_values(cursor, query, values)
+        conn.commit()
+    
+
+@app.post('/update-navff', dependencies=[Depends(authenticate)])
+async def update_nav_ff():
+    conn = connect_db()
+    update_query = """
+        insert into nav_history_ff 
+        with cte1 as (
+            select date_value, amfi_code
+            from daily_date_series dds 
+            cross join (select amfi_code from latest_nav)
+            where date_value between current_date - '7 days'::interval and current_date - '1 day'::interval
+        ), cte2 as (
+            select cte1.*, nh.date, nh.nav, count(nav) over (partition by cte1.amfi_code order by date_value) as grouper
+            from cte1
+            left join nav_history nh on cte1.amfi_code = nh.amfi_code and cte1.date_value = nh.date
+        )
+        select date_value as date_ff, amfi_code, first_value(date) over (partition by amfi_code, grouper order by date_value) as actual_date,
+                first_value( nav) over (partition by amfi_code, grouper order by date_value) as nav_ff
+        from cte2
+        ON CONFLICT (amfi_code, date_ff) 
+        DO UPDATE SET 
+            nav_ff = EXCLUDED.nav_ff,
+            actual_date = EXCLUDED.actual_date;
+    """
+
+    with conn.cursor() as cur:
+        cur.execute(update_query)
+        conn.commit()