Update app.py

app.py

@@ -1,603 +1,573 @@
 import os
-import logging
+import sys
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, Any, Tuple, Optional
-from contextlib import contextmanager
-import signal
-from functools import wraps
+from typing import Tuple, Any, Dict, List
 
 import duckdb
 import pandas as pd
 import numpy as np
 import matplotlib.pyplot as plt
 import gradio as gr
-from pydantic import BaseModel, Field
+from pydantic import BaseModel
 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
 # -------------------------------------------------------------------
 APP_TITLE = "ALCO Liquidity & Interest-Rate Risk Dashboard"
-TABLE_FQN = "my_db.main.masterdataset_v"
-VIEW_FQN = "my_db.main.positions_v"
+TABLE_FQN = "my_db.main.masterdataset_v"   # <- your source
+VIEW_FQN = "my_db.main.positions_v"        # <- normalized view we create
 EXPORT_DIR = Path("exports")
 EXPORT_DIR.mkdir(exist_ok=True)
 
-# Query timeout in seconds
-QUERY_TIMEOUT_SECONDS = 30
 
 # -------------------------------------------------------------------
-# Query Timeout Handler (since DuckDB doesn't support statement_timeout)
+# MotherDuck connection
 # -------------------------------------------------------------------
-class QueryTimeoutError(Exception):
-    """Raised when a query exceeds the timeout limit"""
-    pass
-
+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.")
+    try:
+        conn = duckdb.connect(f"md:?motherduck_token={token}")
+        return conn
+    except Exception as e:
+        print("ERROR: Unable to connect to MotherDuck:", e, file=sys.stderr)
+        raise
 
-def with_timeout(timeout_seconds: int = QUERY_TIMEOUT_SECONDS):
-    """
-    Decorator to add timeout to query functions.
-    Note: This is a simplified version. For production, consider using
-    concurrent.futures or multiprocessing for better timeout handling.
-    
-    Args:
-        timeout_seconds: Maximum execution time in seconds
-    """
-    def decorator(func):
-        @wraps(func)
-        def wrapper(*args, **kwargs):
-            # For now, we'll log the timeout but not enforce it
-            # since signal.alarm doesn't work well with threads
-            # and DuckDB doesn't support native query timeouts
-            logger.debug(f"Starting {func.__name__} with {timeout_seconds}s timeout")
-            start_time = datetime.now()
-            
-            try:
-                result = func(*args, **kwargs)
-                elapsed = (datetime.now() - start_time).total_seconds()
-                
-                if elapsed > timeout_seconds:
-                    logger.warning(
-                        f"{func.__name__} exceeded timeout: {elapsed:.2f}s > {timeout_seconds}s"
-                    )
-                else:
-                    logger.debug(f"{func.__name__} completed in {elapsed:.2f}s")
-                
-                return result
-            except Exception as e:
-                elapsed = (datetime.now() - start_time).total_seconds()
-                logger.error(f"{func.__name__} failed after {elapsed:.2f}s: {str(e)}")
-                raise
-        
-        return wrapper
-    return decorator
 
 # -------------------------------------------------------------------
-# Data Models
+# Column discovery & dynamic SQL builders
 # -------------------------------------------------------------------
-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
+WANTED_COLS = [
+    "as_of_date",
+    "product",
+    "months",
+    "segments",
+    "currency",
+    "Portfolio_value",
+    "Interest_rate",
+    "days_to_maturity",
+]
+
+PRODUCT_ASSETS = (
+    "loan", "overdraft", "advances", "bills", "bill", "tbond", "t-bond", "tbill",
+    "t-bill", "repo_asset", "assets"
+)
+PRODUCT_SOF = (
+    "fd", "term_deposit", "td", "savings", "current", "call", "repo_liab"
+)
 
-# -------------------------------------------------------------------
-# MotherDuck connection with proper error handling
-# -------------------------------------------------------------------
-@contextmanager
-def get_db_connection():
+
+def discover_columns(conn: duckdb.DuckDBPyConnection, table_fqn: str) -> List[str]:
+    q = f"""
+    SELECT lower(column_name) AS col
+    FROM information_schema.columns
+    WHERE table_schema = split_part('{table_fqn}', '.', 2)
+      AND table_name   = split_part('{table_fqn}', '.', 3)
     """
-    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
+    df = conn.execute(q).fetchdf()
+    return [c for c in df["col"].tolist()]
+
+
+def build_view_sql(existing_cols: List[str]) -> str:
+    # Build a SELECT list that only references columns that exist; others are NULLs.
+    parts = []
+    for c in WANTED_COLS:
+        if c.lower() in existing_cols:
+            parts.append(c)
+        else:
+            # use sensible defaults for types
+            if c in ("Portfolio_value", "Interest_rate", "days_to_maturity", "months"):
+                parts.append(f"CAST(NULL AS DOUBLE) AS {c}")
+            else:
+                parts.append(f"CAST(NULL AS VARCHAR) AS {c}")
+
+    # Add bucket derived from product (which must exist; we hard-require product & Portfolio_value & days_to_maturity)
+    # If 'product' doesn't exist, the app can't work; guard above (we will assert later).
+    bucket_case = (
+        "CASE "
+        f"WHEN lower(product) IN ({','.join([f\"'{p}'\" for p in PRODUCT_SOF])}) THEN 'SoF' "
+        f"WHEN lower(product) IN ({','.join([f\"'{p}'\" for p in PRODUCT_ASSETS])}) THEN 'Assets' "
+        "ELSE 'Unknown' END AS bucket"
+    )
+
+    select_list = ",\n  ".join(parts + [bucket_case])
+    return f"""
+    CREATE OR REPLACE VIEW {VIEW_FQN} AS
+    SELECT
+      {select_list}
+    FROM {TABLE_FQN};
     """
-    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}")
-        
-        # Note: DuckDB/MotherDuck doesn't support statement_timeout like PostgreSQL
-        # Query timeouts should be handled at application level with threading/async
-        
-        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
-    """
-    start_time = datetime.now()
-    try:
-        logger.info(f"Executing {description}")
-        result = conn.execute(query).df()
-        elapsed = (datetime.now() - start_time).total_seconds()
-        logger.info(f"{description} completed: {len(result)} rows in {elapsed:.2f}s")
-        
-        # Warn if query is slow
-        if elapsed > QUERY_TIMEOUT_SECONDS:
-            logger.warning(f"{description} exceeded timeout threshold: {elapsed:.2f}s")
-        
-        return result
-    except Exception as e:
-        elapsed = (datetime.now() - start_time).total_seconds()
-        logger.error(f"{description} failed after {elapsed:.2f}s: {str(e)}")
-        raise Exception(f"Query execution failed for {description}: {str(e)}")
+def make_max_date_sql(has_asof: bool) -> str:
+    if not has_asof:
+        # No as_of_date column -> return N/A row
+        return "SELECT 'N/A'::VARCHAR AS d;"
+    return f"WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN}) SELECT d FROM maxd;"
 
 
-# -------------------------------------------------------------------
-# SQL snippets with fixed syntax
-# -------------------------------------------------------------------
-CREATE_VIEW_SQL = f"""
-CREATE OR REPLACE VIEW {VIEW_FQN} AS
-SELECT
-  as_of_date,
-  product,
-  months,
-  segments,
-  currency,
-  Portfolio_value,
-  Interest_rate,
-  days_to_maturity,
-  CASE
-    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', 'assets') THEN 'Assets'
-    ELSE 'Unknown'
-  END AS bucket
-FROM {TABLE_FQN}
-WHERE Portfolio_value IS NOT NULL;
-"""
-
-MAX_DATE_SQL = f"""
-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}
-),
-t1 AS (
-  SELECT 
-    p.bucket, 
-    SUM(p.Portfolio_value) AS amt
-  FROM {VIEW_FQN} p
-  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
-FROM t1;
-"""
-
-LADDER_SQL = f"""
-WITH maxd AS (
-  SELECT MAX(as_of_date) AS d FROM {VIEW_FQN}
-)
-SELECT
-  CASE
-    WHEN p.days_to_maturity <= 1 THEN 'T+1'
-    WHEN p.days_to_maturity BETWEEN 2 AND 7 THEN 'T+2..7'
-    WHEN p.days_to_maturity BETWEEN 8 AND 30 THEN 'T+8..30'
-    ELSE 'T+31+'
-  END AS time_bucket,
-  p.bucket,
-  SUM(p.Portfolio_value) AS amount
-FROM {VIEW_FQN} p
-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
-FROM {VIEW_FQN} p
-INNER JOIN maxd m ON p.as_of_date = m.d
-WHERE p.days_to_maturity <= 1
-  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
-FROM {VIEW_FQN} p
-INNER JOIN maxd m ON p.as_of_date = m.d
-WHERE p.days_to_maturity <= 1
-  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
-FROM {VIEW_FQN} p
-INNER JOIN maxd m ON p.as_of_date = m.d
-WHERE p.days_to_maturity <= 1
-  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}
-),
+def wrap_latest_date(sql_body: str, has_asof: bool) -> str:
+    """
+    If as_of_date exists, pin to latest date using a CTE and JOIN.
+    Otherwise, return the body directly from VIEW_FQN (no date pinning).
+    The sql_body must reference the view as 'p'.
+    """
+    if not has_asof:
+        # Remove any JOIN to maxd; just select from the view
+        return f"SELECT * FROM ({sql_body})"
+    else:
+        return f"SELECT * FROM ({sql_body})"
+
+
+def build_kpi_sql(has_asof: bool) -> str:
+    if has_asof:
+        return f"""
+        WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN}),
+        t1 AS (
+          SELECT p.bucket, SUM(p.Portfolio_value) AS amt
+          FROM {VIEW_FQN} p
+          JOIN maxd m ON p.as_of_date = m.d
+          WHERE p.days_to_maturity <= 1
+          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
+        FROM t1;
+        """
+    else:
+        return f"""
+        WITH t1 AS (
+          SELECT p.bucket, SUM(p.Portfolio_value) AS amt
+          FROM {VIEW_FQN} p
+          WHERE p.days_to_maturity <= 1
+          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
+        FROM t1;
+        """
+
+
+def build_ladder_sql(has_asof: bool) -> str:
+    if has_asof:
+        return f"""
+        WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN})
+        SELECT
+          CASE
+            WHEN p.days_to_maturity <= 1 THEN 'T+1'
+            WHEN p.days_to_maturity BETWEEN 2 AND 7  THEN 'T+2..7'
+            WHEN p.days_to_maturity BETWEEN 8 AND 30 THEN 'T+8..30'
+            ELSE 'T+31+'
+          END AS time_bucket,
+          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, p.bucket;
+        """
+    else:
+        return f"""
+        SELECT
+          CASE
+            WHEN p.days_to_maturity <= 1 THEN 'T+1'
+            WHEN p.days_to_maturity BETWEEN 2 AND 7  THEN 'T+2..7'
+            WHEN p.days_to_maturity BETWEEN 8 AND 30 THEN 'T+8..30'
+            ELSE 'T+31+'
+          END AS time_bucket,
+          p.bucket,
+          SUM(p.Portfolio_value) AS amount
+        FROM {VIEW_FQN} p
+        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;
+        """
+
+
+def build_t1_group_sql(group_col: str, has_asof: bool) -> str:
+    if has_asof:
+        return f"""
+        WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN})
+        SELECT p.bucket, p.{group_col} AS grp, SUM(p.Portfolio_value) AS amount
+        FROM {VIEW_FQN} p
+        JOIN maxd m ON p.as_of_date = m.d
+        WHERE p.days_to_maturity <= 1
+        GROUP BY 1,2
+        ORDER BY p.bucket, amount DESC
+        LIMIT 50;
+        """
+    else:
+        return f"""
+        SELECT p.bucket, p.{group_col} AS grp, SUM(p.Portfolio_value) AS amount
+        FROM {VIEW_FQN} p
+        WHERE p.days_to_maturity <= 1
+        GROUP BY 1,2
+        ORDER BY p.bucket, amount DESC
+        LIMIT 50;
+        """
+
+
+def build_irr_sql(has_asof: bool, has_months: bool, has_ir: bool) -> str:
+    # T_years uses days_to_maturity OR months (if present). y uses Interest_rate (if present).
+    t_years_expr = "CASE WHEN p.days_to_maturity IS NOT NULL THEN p.days_to_maturity/365.0"
+    if has_months:
+        t_years_expr += " WHEN p.months IS NOT NULL THEN p.months/12.0"
+    t_years_expr += " ELSE NULL END"
+
+    y_expr = "(p.Interest_rate / 100.0)" if has_ir else "NULL"
+
+    if has_asof:
+        base_from = f"FROM {VIEW_FQN} p JOIN maxd m ON p.as_of_date = m.d"
+        max_cte = f"WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN}),"
+    else:
+        base_from = f"FROM {VIEW_FQN} p"
+        max_cte = "WITH"
+
+    return f"""
+{max_cte}
 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
-      ELSE NULL
-    END AS T_years
-  FROM {VIEW_FQN} p
-  INNER JOIN maxd m ON p.as_of_date = m.d
+    {y_expr} AS y,
+    {t_years_expr} AS T_years
+  {base_from}
   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 / 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
+    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
   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
-FROM agg;
-"""
-
-# FIXED: Complete the SHOCK_SQL query
-SHOCK_SQL = f"""
-WITH maxd AS (
-  SELECT MAX(as_of_date) AS d FROM {VIEW_FQN}
-),
+  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;
+    """
+
+
+def build_shock_sql(has_asof: bool, has_months: bool, has_ir: bool) -> str:
+    t_years_expr = "CASE WHEN p.days_to_maturity IS NOT NULL THEN p.days_to_maturity/365.0"
+    if has_months:
+        t_years_expr += " WHEN p.months IS NOT NULL THEN p.months/12.0"
+    t_years_expr += " ELSE NULL END"
+
+    y_expr = "(p.Interest_rate / 100.0)" if has_ir else "NULL"
+
+    if has_asof:
+        base_from = f"FROM {VIEW_FQN} p JOIN maxd m ON p.as_of_date = m.d"
+        max_cte = f"WITH maxd AS (SELECT max(as_of_date) AS d FROM {VIEW_FQN}),"
+    else:
+        base_from = f"FROM {VIEW_FQN} p"
+        max_cte = "WITH"
+
+    return f"""
+{max_cte}
 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
-      ELSE NULL
-    END AS T_years
-  FROM {VIEW_FQN} p
-  INNER JOIN maxd m ON p.as_of_date = m.d
-  WHERE p.Portfolio_value IS NOT NULL
-    AND p.bucket IN ('Assets', 'SoF')
+    {y_expr} AS y,
+    {t_years_expr} AS T_years
+  {base_from}
 ),
 k AS (
   SELECT
-    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
+    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
   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
+  FROM k
+  GROUP BY bucket
 )
-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
+SELECT * FROM shock ORDER BY bucket;
     """
-    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
+# Data class
 # -------------------------------------------------------------------
-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
+class DashboardResult(BaseModel):
+    as_of_date: str
+    assets_t1: float
+    sof_t1: float
+    net_gap_t1: float
+    ladder: pd.DataFrame
+    t1_by_month: pd.DataFrame
+    t1_by_segment: pd.DataFrame
+    t1_by_ccy: pd.DataFrame
+    irr: pd.DataFrame
+    shocks: pd.DataFrame
 
 
 # -------------------------------------------------------------------
-# Business logic functions with validation
+# Query helpers
 # -------------------------------------------------------------------
-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 ensure_view(conn: duckdb.DuckDBPyConnection, existing_cols: List[str]) -> None:
+    # sanity: ensure mandatory columns exist in source table
+    mandatory = {"product", "portfolio_value", "days_to_maturity"}
+    if not mandatory.issubset(set(existing_cols)):
+        raise RuntimeError(
+            f"Source table {TABLE_FQN} must contain {mandatory}, "
+            f"found only: {existing_cols}"
+        )
+    conn.execute(build_view_sql(existing_cols))
+
+
+def fetch_all(conn: duckdb.DuckDBPyConnection) -> DashboardResult:
+    existing = discover_columns(conn, TABLE_FQN)
+    ensure_view(conn, existing)
+
+    has_asof = "as_of_date" in existing
+    has_months = "months" in existing
+    has_segments = "segments" in existing
+    has_currency = "currency" in existing
+    has_ir = "interest_rate" in existing
+
+    # As-of date (or N/A)
+    asof_df = conn.execute(make_max_date_sql(has_asof)).fetchdf()
+    as_of = asof_df["d"].iloc[0]
+    as_of_str = (
+        pd.to_datetime(as_of).strftime("%Y-%m-%d")
+        if has_asof and not pd.isna(as_of)
+        else "N/A"
+    )
+
+    # KPIs & ladder
+    kpis = conn.execute(build_kpi_sql(has_asof)).fetchdf()
+    ladder = conn.execute(build_ladder_sql(has_asof)).fetchdf()
+
+    # Contributors (only if columns exist)
+    if has_months:
+        t1_m = conn.execute(build_t1_group_sql("months", has_asof)).fetchdf()
+        t1_m = t1_m.rename(columns={"grp": "months"})
+    else:
+        t1_m = pd.DataFrame(columns=["bucket", "months", "amount"])
+
+    if has_segments:
+        t1_s = conn.execute(build_t1_group_sql("segments", has_asof)).fetchdf()
+        t1_s = t1_s.rename(columns={"grp": "segments"})
+    else:
+        t1_s = pd.DataFrame(columns=["bucket", "segments", "amount"])
+
+    if has_currency:
+        t1_c = conn.execute(build_t1_group_sql("currency", has_asof)).fetchdf()
+        t1_c = t1_c.rename(columns={"grp": "currency"})
+    else:
+        t1_c = pd.DataFrame(columns=["bucket", "currency", "amount"])
+
+    # IRR & shocks (works even if Interest_rate/months are missing)
+    irr = conn.execute(build_irr_sql(has_asof, has_months, has_ir)).fetchdf()
+    shocks = conn.execute(build_shock_sql(has_asof, has_months, has_ir)).fetchdf()
+
+    return DashboardResult(
+        as_of_date=as_of_str,
+        assets_t1=float(kpis["assets_t1"].iloc[0]),
+        sof_t1=float(kpis["sof_t1"].iloc[0]),
+        net_gap_t1=float(kpis["net_gap_t1"].iloc[0]),
+        ladder=ladder,
+        t1_by_month=t1_m,
+        t1_by_segment=t1_s,
+        t1_by_ccy=t1_c,
+        irr=irr,
+        shocks=shocks,
+    )
 
 
-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
+# -------------------------------------------------------------------
+# Plotting
+# -------------------------------------------------------------------
+def plot_ladder(df: pd.DataFrame):
+    pivot = df.pivot(index="time_bucket", columns="bucket", values="amount").fillna(0)
+    order = ["T+1", "T+2..7", "T+8..30", "T+31+"]
+    pivot = pivot.reindex(order)
+    fig, ax = plt.subplots(figsize=(7, 4))
+    assets = pivot.get("Assets", pd.Series([0] * len(pivot), index=pivot.index))
+    sof = pivot.get("SoF", pd.Series([0] * len(pivot), index=pivot.index))
+    ax.bar(pivot.index, assets, label="Assets")
+    ax.bar(pivot.index, -sof, bottom=0, label="SoF")
+    ax.axhline(0, linewidth=1)
+    ax.set_ylabel("LKR")
+    ax.set_title("Maturity Ladder (Assets vs SoF)")
+    ax.legend()
+    fig.tight_layout()
+    return fig
 
 
 # -------------------------------------------------------------------
-# Helper function for safe numeric formatting
+# Exports
 # -------------------------------------------------------------------
-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"
+def export_excel(res: DashboardResult) -> Path:
+    out = EXPORT_DIR / f"alco_report_{res.as_of_date}.xlsx"
+    with pd.ExcelWriter(out, engine="xlsxwriter") as xw:
+        pd.DataFrame({
+            "as_of_date": [res.as_of_date],
+            "assets_t1": [res.assets_t1],
+            "sof_t1": [res.sof_t1],
+            "net_gap_t1": [res.net_gap_t1],
+        }).to_excel(xw, index=False, sheet_name="kpis")
+        res.ladder.to_excel(xw, index=False, sheet_name="ladder")
+        res.t1_by_month.to_excel(xw, index=False, sheet_name="t1_by_month")
+        res.t1_by_segment.to_excel(xw, index=False, sheet_name="t1_by_segment")
+        res.t1_by_ccy.to_excel(xw, index=False, sheet_name="t1_by_ccy")
+        res.irr.to_excel(xw, index=False, sheet_name="irr")
+        res.shocks.to_excel(xw, index=False, sheet_name="shocks")
+    return out
+
+
+def export_pdf(res: DashboardResult) -> Path:
+    out = EXPORT_DIR / f"alco_report_{res.as_of_date}.pdf"
+    c = canvas.Canvas(str(out), pagesize=A4)
+    W, H = A4
+    y = H - 20 * mm
+
+    def line(txt, size=11, dy=6 * mm):
+        nonlocal y
+        c.setFont("Helvetica", size)
+        c.drawString(20 * mm, y, txt)
+        y -= dy
+
+    line(APP_TITLE, 14, dy=8 * mm)
+    line(f"As of: {res.as_of_date}")
+    line(f"Assets T+1: {res.assets_t1:,.0f} LKR")
+    line(f"SoF T+1: {res.sof_t1:,.0f} LKR")
+    line(f"Net Gap T+1: {res.net_gap_t1:,.0f} LKR  (negative = shortfall)")
+    y -= 4 * mm
+
+    if not res.irr.empty:
+        irr = res.irr.iloc[0]
+        line("Interest-Rate Risk (approx)", 12, dy=7 * mm)
+        line(f"Assets ModDur: {irr['assets_dur_mod']:.2f}  |  SoF ModDur: {irr['sof_dur_mod']:.2f}")
+        line(f"Duration Gap: {irr['duration_gap']:.2f}")
+        line(f"Net DV01: {irr['net_dv01']:,.0f} LKR/bp")
+
+    if not res.shocks.empty:
+        net_up = res.shocks["dPV_up_100bp"].sum()
+        net_dn = res.shocks["dPV_dn_100bp"].sum()
+        y -= 2 * mm
+        line(f"+100bp net ΔPV: {net_up:,.0f} LKR  |  -100bp net ΔPV: {net_dn:,.0f} LKR")
+
+    c.showPage()
+    c.save()
+    return out
 
 
 # -------------------------------------------------------------------
-# Add health check endpoint
+# Gradio UI
 # -------------------------------------------------------------------
-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
-
+def run_dashboard() -> Tuple[str, float, float, float, Any, Any, Any, Any, Any, Any, Any]:
+    conn = connect_md()
+    res = fetch_all(conn)
+    fig = plot_ladder(res.ladder)
+    excel_path = export_excel(res)
+    pdf_path = export_pdf(res)
+    return (
+        res.as_of_date,
+        res.assets_t1,
+        res.sof_t1,
+        res.net_gap_t1,
+        fig,
+        res.t1_by_month,
+        res.t1_by_segment,
+        res.t1_by_ccy,
+        res.irr,
+        res.shocks,
+        str(excel_path),
+        str(pdf_path),
+    )
+
+
+with gr.Blocks(title=APP_TITLE) as demo:
+    gr.Markdown(
+        f"# {APP_TITLE}\n"
+        "*Source:* `my_db.main.masterdataset_v` → `positions_v`  |  *Sign:* Assets=+  SoF=–"
+    )
+
+    with gr.Row():
+        btn = gr.Button("🔄 Refresh", variant="primary")
+
+    with gr.Row():
+        as_of = gr.Textbox(label="As of date", interactive=False)
+
+    with gr.Row():
+        k1 = gr.Number(label="Assets T+1 (LKR)", precision=0)
+        k2 = gr.Number(label="SoF T+1 (LKR)", precision=0)
+        k3 = gr.Number(label="Net Gap T+1 (LKR)", precision=0)
+
+    chart = gr.Plot(label="Maturity Ladder")
+
+    with gr.Row():
+        t1m = gr.Dataframe(label="T+1 by Tenor (months)")
+        t1s = gr.Dataframe(label="T+1 by Segment")
+
+    t1c = gr.Dataframe(label="T+1 by Currency")
+    irr = gr.Dataframe(label="Interest-Rate Risk (bucketed)")
+    shocks = gr.Dataframe(label="Parallel Shock ±100bp (bucketed)")
+
+    with gr.Row():
+        excel_file = gr.File(label="Excel export", interactive=False)
+        pdf_file = gr.File(label="PDF export", interactive=False)
+
+    btn.click(
+        fn=run_dashboard,
+        outputs=[
+            as_of,
+            k1,
+            k2,
+            k3,
+            chart,
+            t1m,
+            t1s,
+            t1c,
+            irr,
+            shocks,
+            excel_file,
+            pdf_file,
+        ],
+    )
 
 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
+    demo.launch()