Update app.py

app.py

@@ -1,18 +1,29 @@
 import os
+import logging
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, Any, Tuple, Any
+from typing import Dict, Any, Tuple, Optional
+from contextlib import contextmanager
 
 import duckdb
 import pandas as pd
 import numpy as np
 import matplotlib.pyplot as plt
 import gradio as gr
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from reportlab.lib.pagesizes import A4
 from reportlab.lib.units import mm
 from reportlab.pdfgen import canvas
 
+# -------------------------------------------------------------------
+# Logging Configuration
+# -------------------------------------------------------------------
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
 # -------------------------------------------------------------------
 # Basic configuration
 # -------------------------------------------------------------------
@@ -22,20 +33,99 @@ VIEW_FQN = "my_db.main.positions_v"
 EXPORT_DIR = Path("exports")
 EXPORT_DIR.mkdir(exist_ok=True)
 
+# Query timeout in seconds
+QUERY_TIMEOUT_SECONDS = 30
+
+# -------------------------------------------------------------------
+# Data Models
+# -------------------------------------------------------------------
+class KPIMetrics(BaseModel):
+    """T+1 Key Performance Indicators"""
+    assets_t1: float = Field(ge=0, description="T+1 Assets")
+    sof_t1: float = Field(ge=0, description="T+1 Sources of Funds")
+    net_gap_t1: float = Field(description="T+1 Net Gap")
+    
+class IRRMetrics(BaseModel):
+    """Interest Rate Risk Metrics"""
+    assets_pv: float = Field(ge=0)
+    sof_pv: float = Field(ge=0)
+    assets_dur_mod: float = Field(ge=0)
+    sof_dur_mod: float = Field(ge=0)
+    duration_gap: float
+    net_dv01: float
 
 # -------------------------------------------------------------------
-# MotherDuck connection
+# MotherDuck connection with proper error handling
 # -------------------------------------------------------------------
-def connect_md() -> duckdb.DuckDBPyConnection:
-    token = os.environ.get("MOTHERDUCK_TOKEN", "")
-    if not token:
-        raise RuntimeError("MOTHERDUCK_TOKEN is not set. Add it as a Space secret.")
-    conn = duckdb.connect(f"md:?motherduck_token={token}")
-    return conn
+@contextmanager
+def get_db_connection():
+    """
+    Context manager for database connections with proper cleanup.
+    
+    Yields:
+        duckdb.DuckDBPyConnection: Active database connection
+        
+    Raises:
+        RuntimeError: If MOTHERDUCK_TOKEN is not configured
+        Exception: For database connection errors
+    """
+    conn = None
+    try:
+        token = os.environ.get("MOTHERDUCK_TOKEN", "")
+        if not token:
+            logger.error("MOTHERDUCK_TOKEN environment variable not set")
+            raise RuntimeError(
+                "MOTHERDUCK_TOKEN is not set. Please add it as a Space secret."
+            )
+        
+        logger.info("Establishing MotherDuck connection")
+        conn = duckdb.connect(f"md:?motherduck_token={token}")
+        
+        # Set query timeout
+        conn.execute(f"SET statement_timeout='{QUERY_TIMEOUT_SECONDS}s'")
+        
+        yield conn
+        
+    except Exception as e:
+        logger.error(f"Database connection error: {str(e)}")
+        raise
+    finally:
+        if conn:
+            try:
+                conn.close()
+                logger.info("Database connection closed")
+            except Exception as e:
+                logger.warning(f"Error closing connection: {str(e)}")
+
+
+def execute_query(conn: duckdb.DuckDBPyConnection, query: str, 
+                  description: str = "Query") -> pd.DataFrame:
+    """
+    Execute a query with error handling and logging.
+    
+    Args:
+        conn: Database connection
+        query: SQL query to execute
+        description: Human-readable description for logging
+        
+    Returns:
+        pd.DataFrame: Query results
+        
+    Raises:
+        Exception: For query execution errors
+    """
+    try:
+        logger.info(f"Executing {description}")
+        result = conn.execute(query).df()
+        logger.info(f"{description} completed: {len(result)} rows returned")
+        return result
+    except Exception as e:
+        logger.error(f"{description} failed: {str(e)}")
+        raise Exception(f"Query execution failed for {description}: {str(e)}")
 
 
 # -------------------------------------------------------------------
-# SQL snippets
+# SQL snippets with fixed syntax
 # -------------------------------------------------------------------
 CREATE_VIEW_SQL = f"""
 CREATE OR REPLACE VIEW {VIEW_FQN} AS
@@ -49,37 +139,45 @@ SELECT
   Interest_rate,
   days_to_maturity,
   CASE
-    WHEN lower(product) IN ('fd','term_deposit','td','savings','current','call','repo_liab') THEN 'SoF'
-    WHEN lower(product) IN ('loan','overdraft','advances','bills','bill','tbond','t-bond','tbill','t-bill','repo_asset') THEN 'Assets'
+    WHEN LOWER(TRIM(product)) IN ('fd','term_deposit','td','savings','current','call','repo_liab') THEN 'SoF'
+    WHEN LOWER(TRIM(product)) IN ('loan','overdraft','advances','bills','bill','tbond','t-bond','tbill','t-bill','repo_asset') THEN 'Assets'
     ELSE 'Unknown'
   END AS bucket
-FROM {TABLE_FQN};
+FROM {TABLE_FQN}
+WHERE Portfolio_value IS NOT NULL;
 """
 
 MAX_DATE_SQL = f"""
-WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN})
-SELECT d FROM maxd;
+SELECT MAX(as_of_date) AS d 
+FROM {VIEW_FQN};
 """
 
 KPI_SQL = f"""
-WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN}),
+WITH maxd AS (
+  SELECT MAX(as_of_date) AS d FROM {VIEW_FQN}
+),
 t1 AS (
-  SELECT p.bucket, SUM(p.Portfolio_value) AS amt
+  SELECT 
+    p.bucket, 
+    SUM(p.Portfolio_value) AS amt
   FROM {VIEW_FQN} p
-  JOIN maxd m ON p.as_of_date = m.d
+  INNER JOIN maxd m ON p.as_of_date = m.d
   WHERE p.days_to_maturity <= 1
+    AND p.bucket IN ('Assets', 'SoF')
   GROUP BY p.bucket
 )
 SELECT
-  COALESCE(SUM(CASE WHEN bucket='Assets' THEN amt END),0) AS assets_t1,
-  COALESCE(SUM(CASE WHEN bucket='SoF'    THEN amt END),0) AS sof_t1,
-  COALESCE(SUM(CASE WHEN bucket='Assets' THEN amt END),0)
-  - COALESCE(SUM(CASE WHEN bucket='SoF'  THEN amt END),0) AS net_gap_t1
+  COALESCE(SUM(CASE WHEN bucket = 'Assets' THEN amt END), 0) AS assets_t1,
+  COALESCE(SUM(CASE WHEN bucket = 'SoF' THEN amt END), 0) AS sof_t1,
+  COALESCE(SUM(CASE WHEN bucket = 'Assets' THEN amt END), 0)
+    - COALESCE(SUM(CASE WHEN bucket = 'SoF' THEN amt END), 0) AS net_gap_t1
 FROM t1;
 """
 
 LADDER_SQL = f"""
-WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN})
+WITH maxd AS (
+  SELECT MAX(as_of_date) AS d FROM {VIEW_FQN}
+)
 SELECT
   CASE
     WHEN p.days_to_maturity <= 1 THEN 'T+1'
@@ -90,124 +188,360 @@ SELECT
   p.bucket,
   SUM(p.Portfolio_value) AS amount
 FROM {VIEW_FQN} p
-JOIN maxd m ON p.as_of_date = m.d
-GROUP BY 1,2
-ORDER BY CASE time_bucket
-  WHEN 'T+1' THEN 1 WHEN 'T+2..7' THEN 2 WHEN 'T+8..30' THEN 3 ELSE 4 END,
+INNER JOIN maxd m ON p.as_of_date = m.d
+WHERE p.bucket IN ('Assets', 'SoF')
+GROUP BY 1, 2
+ORDER BY 
+  CASE time_bucket
+    WHEN 'T+1' THEN 1 
+    WHEN 'T+2..7' THEN 2 
+    WHEN 'T+8..30' THEN 3 
+    ELSE 4 
+  END,
   p.bucket;
 """
 
 T1_BY_MONTH_SQL = f"""
-WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN})
-SELECT p.bucket, p.months, SUM(p.Portfolio_value) AS amount
+WITH maxd AS (
+  SELECT MAX(as_of_date) AS d FROM {VIEW_FQN}
+)
+SELECT 
+  p.bucket, 
+  p.months, 
+  SUM(p.Portfolio_value) AS amount
 FROM {VIEW_FQN} p
-JOIN maxd m ON p.as_of_date = m.d
+INNER JOIN maxd m ON p.as_of_date = m.d
 WHERE p.days_to_maturity <= 1
-GROUP BY 1,2
+  AND p.bucket IN ('Assets', 'SoF')
+  AND p.months IS NOT NULL
+GROUP BY p.bucket, p.months
 ORDER BY p.bucket, amount DESC
 LIMIT 50;
 """
 
 T1_BY_SEGMENT_SQL = f"""
-WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN})
-SELECT p.bucket, p.segments, SUM(p.Portfolio_value) AS amount
+WITH maxd AS (
+  SELECT MAX(as_of_date) AS d FROM {VIEW_FQN}
+)
+SELECT 
+  p.bucket, 
+  p.segments, 
+  SUM(p.Portfolio_value) AS amount
 FROM {VIEW_FQN} p
-JOIN maxd m ON p.as_of_date = m.d
+INNER JOIN maxd m ON p.as_of_date = m.d
 WHERE p.days_to_maturity <= 1
-GROUP BY 1,2
+  AND p.bucket IN ('Assets', 'SoF')
+  AND p.segments IS NOT NULL
+GROUP BY p.bucket, p.segments
 ORDER BY p.bucket, amount DESC
 LIMIT 50;
 """
 
 T1_BY_CCY_SQL = f"""
-WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN})
-SELECT p.bucket, p.currency, SUM(p.Portfolio_value) AS amount
+WITH maxd AS (
+  SELECT MAX(as_of_date) AS d FROM {VIEW_FQN}
+)
+SELECT 
+  p.bucket, 
+  p.currency, 
+  SUM(p.Portfolio_value) AS amount
 FROM {VIEW_FQN} p
-JOIN maxd m ON p.as_of_date = m.d
+INNER JOIN maxd m ON p.as_of_date = m.d
 WHERE p.days_to_maturity <= 1
-GROUP BY 1,2
+  AND p.bucket IN ('Assets', 'SoF')
+  AND p.currency IS NOT NULL
+GROUP BY p.bucket, p.currency
 ORDER BY p.bucket, amount DESC;
 """
 
 IRR_SQL = f"""
-WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN}),
+WITH maxd AS (
+  SELECT MAX(as_of_date) AS d FROM {VIEW_FQN}
+),
 base AS (
   SELECT
     p.bucket,
     p.Portfolio_value AS pv,
     (p.Interest_rate / 100.0) AS y,
     CASE
-      WHEN p.days_to_maturity IS NOT NULL THEN p.days_to_maturity/365.0
-      WHEN p.months IS NOT NULL THEN p.months/12.0
+      WHEN p.days_to_maturity IS NOT NULL THEN p.days_to_maturity / 365.0
+      WHEN p.months IS NOT NULL THEN p.months / 12.0
       ELSE NULL
     END AS T_years
   FROM {VIEW_FQN} p
-  JOIN maxd m ON p.as_of_date = m.d
+  INNER JOIN maxd m ON p.as_of_date = m.d
   WHERE p.Portfolio_value IS NOT NULL
+    AND p.bucket IN ('Assets', 'SoF')
 ),
 metrics AS (
   SELECT
     bucket,
     pv,
-    CASE WHEN T_years IS NULL THEN NULL
-         WHEN y IS NULL THEN T_years
-         ELSE T_years/(1.0+y) END AS dur_mod,
-    CASE WHEN T_years IS NULL THEN NULL
-         WHEN y IS NULL THEN T_years*(T_years+1.0)
-         ELSE (T_years*(T_years+1.0))/POWER(1.0+y,2) END AS convexity_approx,
-    CASE WHEN T_years IS NULL THEN NULL
-         ELSE pv * (CASE WHEN y IS NULL THEN T_years ELSE T_years/(1.0+y) END) * 0.0001 END AS dv01
+    CASE 
+      WHEN T_years IS NULL THEN NULL
+      WHEN y IS NULL THEN T_years
+      ELSE T_years / NULLIF(1.0 + y, 0) 
+    END AS dur_mod,
+    CASE 
+      WHEN T_years IS NULL THEN NULL
+      WHEN y IS NULL THEN T_years * (T_years + 1.0)
+      ELSE (T_years * (T_years + 1.0)) / NULLIF(POWER(1.0 + y, 2), 0)
+    END AS convexity_approx,
+    CASE 
+      WHEN T_years IS NULL THEN NULL
+      WHEN y IS NULL THEN pv * T_years * 0.0001
+      ELSE pv * (T_years / NULLIF(1.0 + y, 0)) * 0.0001 
+    END AS dv01
   FROM base
 ),
 agg AS (
   SELECT
     bucket,
     SUM(pv) AS pv_sum,
-    SUM(pv * dur_mod) / NULLIF(SUM(pv),0) AS dur_mod_port,
+    SUM(pv * dur_mod) / NULLIF(SUM(pv), 0) AS dur_mod_port,
     SUM(dv01) AS dv01_sum
   FROM metrics
   GROUP BY bucket
 )
 SELECT
-  COALESCE(MAX(CASE WHEN bucket='Assets' THEN pv_sum END),0) AS assets_pv,
-  COALESCE(MAX(CASE WHEN bucket='SoF' THEN pv_sum END),0) AS sof_pv,
-  COALESCE(MAX(CASE WHEN bucket='Assets' THEN dur_mod_port END),0) AS assets_dur_mod,
-  COALESCE(MAX(CASE WHEN bucket='SoF' THEN dur_mod_port END),0) AS sof_dur_mod,
-  COALESCE(MAX(CASE WHEN bucket='Assets' THEN dur_mod_port END),0)
-  - COALESCE(MAX(CASE WHEN bucket='SoF' THEN dur_mod_port END),0) AS duration_gap,
-  COALESCE(MAX(CASE WHEN bucket='Assets' THEN dv01_sum END),0)
-  - COALESCE(MAX(CASE WHEN bucket='SoF' THEN dv01_sum END),0) AS net_dv01
+  COALESCE(MAX(CASE WHEN bucket = 'Assets' THEN pv_sum END), 0) AS assets_pv,
+  COALESCE(MAX(CASE WHEN bucket = 'SoF' THEN pv_sum END), 0) AS sof_pv,
+  COALESCE(MAX(CASE WHEN bucket = 'Assets' THEN dur_mod_port END), 0) AS assets_dur_mod,
+  COALESCE(MAX(CASE WHEN bucket = 'SoF' THEN dur_mod_port END), 0) AS sof_dur_mod,
+  COALESCE(MAX(CASE WHEN bucket = 'Assets' THEN dur_mod_port END), 0)
+    - COALESCE(MAX(CASE WHEN bucket = 'SoF' THEN dur_mod_port END), 0) AS duration_gap,
+  COALESCE(MAX(CASE WHEN bucket = 'Assets' THEN dv01_sum END), 0)
+    - COALESCE(MAX(CASE WHEN bucket = 'SoF' THEN dv01_sum END), 0) AS net_dv01
 FROM agg;
 """
 
+# FIXED: Complete the SHOCK_SQL query
 SHOCK_SQL = f"""
-WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN}),
+WITH maxd AS (
+  SELECT MAX(as_of_date) AS d FROM {VIEW_FQN}
+),
 base AS (
   SELECT
     p.bucket,
     p.Portfolio_value AS pv,
     (p.Interest_rate / 100.0) AS y,
     CASE
-      WHEN p.days_to_maturity IS NOT NULL THEN p.days_to_maturity/365.0
-      WHEN p.months IS NOT NULL THEN p.months/12.0
+      WHEN p.days_to_maturity IS NOT NULL THEN p.days_to_maturity / 365.0
+      WHEN p.months IS NOT NULL THEN p.months / 12.0
+      ELSE NULL
     END AS T_years
   FROM {VIEW_FQN} p
-  JOIN maxd m ON p.as_of_date = m.d
+  INNER JOIN maxd m ON p.as_of_date = m.d
+  WHERE p.Portfolio_value IS NOT NULL
+    AND p.bucket IN ('Assets', 'SoF')
 ),
 k AS (
   SELECT
-    bucket, pv,
-    CASE WHEN T_years IS NULL THEN NULL
-         WHEN y IS NULL THEN T_years
-         ELSE T_years/(1.0+y) END AS dur_mod,
-    CASE WHEN T_years IS NULL THEN NULL
-         WHEN y IS NULL THEN T_years*(T_years+1.0)
-         ELSE (T_years*(T_years+1.0))/POWER(1.0+y,2) END AS convexity_approx
+    bucket, 
+    pv,
+    CASE 
+      WHEN T_years IS NULL THEN NULL
+      WHEN y IS NULL THEN T_years
+      ELSE T_years / NULLIF(1.0 + y, 0)
+    END AS dur_mod,
+    CASE 
+      WHEN T_years IS NULL THEN NULL
+      WHEN y IS NULL THEN T_years * (T_years + 1.0)
+      ELSE (T_years * (T_years + 1.0)) / NULLIF(POWER(1.0 + y, 2), 0)
+    END AS convexity_approx
   FROM base
-),
-shock AS (
-  SELECT
-    bucket,
-    SUM((- pv * dur_mod * 0.01) + (0.5 * pv * convexity_approx * POWER(0.01,2))) AS dPV_up_100bp,
-    SUM((+ pv * dur_mod * 0.01) + (0.5 * pv * convexity_approx * POWER(-0.01,2))) AS dPV_dn_100bp
-  F
+)
+SELECT
+  bucket,
+  SUM((- pv * dur_mod * 0.01) + (0.5 * pv * convexity_approx * POWER(0.01, 2))) AS dPV_up_100bp,
+  SUM((+ pv * dur_mod * 0.01) + (0.5 * pv * convexity_approx * POWER(-0.01, 2))) AS dPV_dn_100bp
+FROM k
+WHERE dur_mod IS NOT NULL
+GROUP BY bucket
+ORDER BY bucket;
+"""
+
+# -------------------------------------------------------------------
+# Initialize database view
+# -------------------------------------------------------------------
+def initialize_database():
+    """
+    Initialize database view with error handling.
+    
+    Raises:
+        Exception: If view creation fails
+    """
+    try:
+        with get_db_connection() as conn:
+            execute_query(conn, CREATE_VIEW_SQL, "View creation")
+            logger.info(f"Successfully created/updated view: {VIEW_FQN}")
+    except Exception as e:
+        logger.error(f"Failed to initialize database: {str(e)}")
+        raise
+
+
+# -------------------------------------------------------------------
+# Data validation utilities
+# -------------------------------------------------------------------
+def validate_dataframe(df: pd.DataFrame, expected_columns: list, 
+                       min_rows: int = 0) -> bool:
+    """
+    Validate DataFrame structure and content.
+    
+    Args:
+        df: DataFrame to validate
+        expected_columns: List of required column names
+        min_rows: Minimum number of rows expected
+        
+    Returns:
+        bool: True if valid
+        
+    Raises:
+        ValueError: If validation fails
+    """
+    if df is None or df.empty:
+        if min_rows > 0:
+            raise ValueError("DataFrame is empty but data was expected")
+        return True
+    
+    missing_cols = set(expected_columns) - set(df.columns)
+    if missing_cols:
+        raise ValueError(f"Missing required columns: {missing_cols}")
+    
+    if len(df) < min_rows:
+        raise ValueError(f"Expected at least {min_rows} rows, got {len(df)}")
+    
+    return True
+
+
+# -------------------------------------------------------------------
+# Business logic functions with validation
+# -------------------------------------------------------------------
+def get_max_date() -> Optional[datetime]:
+    """
+    Get the maximum as_of_date from the dataset.
+    
+    Returns:
+        datetime or None: Maximum date if available
+    """
+    try:
+        with get_db_connection() as conn:
+            df = execute_query(conn, MAX_DATE_SQL, "Max date query")
+            if not df.empty and df.iloc[0, 0] is not None:
+                return pd.to_datetime(df.iloc[0, 0])
+            return None
+    except Exception as e:
+        logger.error(f"Error fetching max date: {str(e)}")
+        return None
+
+
+def get_kpi_metrics() -> Optional[KPIMetrics]:
+    """
+    Fetch T+1 KPI metrics with validation.
+    
+    Returns:
+        KPIMetrics or None: Validated KPI metrics
+    """
+    try:
+        with get_db_connection() as conn:
+            df = execute_query(conn, KPI_SQL, "KPI query")
+            validate_dataframe(df, ['assets_t1', 'sof_t1', 'net_gap_t1'], min_rows=1)
+            
+            metrics = KPIMetrics(**df.iloc[0].to_dict())
+            logger.info(f"KPI Metrics: Assets={metrics.assets_t1:,.0f}, "
+                       f"SoF={metrics.sof_t1:,.0f}, Gap={metrics.net_gap_t1:,.0f}")
+            return metrics
+    except Exception as e:
+        logger.error(f"Error fetching KPI metrics: {str(e)}")
+        return None
+
+
+# -------------------------------------------------------------------
+# Helper function for safe numeric formatting
+# -------------------------------------------------------------------
+def safe_format_number(value: Any, decimals: int = 2, 
+                       prefix: str = "", suffix: str = "") -> str:
+    """
+    Safely format numeric values with error handling.
+    
+    Args:
+        value: Value to format
+        decimals: Number of decimal places
+        prefix: String to prepend
+        suffix: String to append
+        
+    Returns:
+        str: Formatted string
+    """
+    try:
+        if value is None or pd.isna(value):
+            return "N/A"
+        num_val = float(value)
+        if np.isinf(num_val):
+            return "∞" if num_val > 0 else "-∞"
+        formatted = f"{num_val:,.{decimals}f}"
+        return f"{prefix}{formatted}{suffix}"
+    except (ValueError, TypeError):
+        return "N/A"
+
+
+# -------------------------------------------------------------------
+# Add health check endpoint
+# -------------------------------------------------------------------
+def health_check() -> Dict[str, Any]:
+    """
+    Perform application health check.
+    
+    Returns:
+        dict: Health status information
+    """
+    health = {
+        "status": "unhealthy",
+        "timestamp": datetime.now().isoformat(),
+        "checks": {}
+    }
+    
+    # Check database connectivity
+    try:
+        with get_db_connection() as conn:
+            conn.execute("SELECT 1").fetchone()
+        health["checks"]["database"] = "ok"
+    except Exception as e:
+        health["checks"]["database"] = f"error: {str(e)}"
+        return health
+    
+    # Check view exists
+    try:
+        max_date = get_max_date()
+        health["checks"]["view"] = "ok" if max_date else "no data"
+        health["max_date"] = max_date.isoformat() if max_date else None
+    except Exception as e:
+        health["checks"]["view"] = f"error: {str(e)}"
+        return health
+    
+    # Check export directory
+    health["checks"]["export_dir"] = "ok" if EXPORT_DIR.exists() else "missing"
+    
+    # Overall status
+    if all(v == "ok" or v == "no data" for v in health["checks"].values()):
+        health["status"] = "healthy"
+    
+    return health
+
+
+if __name__ == "__main__":
+    # Initialize on startup
+    try:
+        logger.info("Starting ALCO Dashboard application")
+        initialize_database()
+        
+        # Perform health check
+        health = health_check()
+        logger.info(f"Health check: {health}")
+        
+        if health["status"] != "healthy":
+            logger.warning("Application health check failed")
+        
+    except Exception as e:
+        logger.critical(f"Application startup failed: {str(e)}")
+        raise