Update tools/ts_preprocess.py

tools/ts_preprocess.py

@@ -1,108 +1,120 @@
-# space/tools/ts_preprocess.py
+# tools/ts_preprocess.py
 import pandas as pd
 import numpy as np
-from typing import List
+from datetime import date
+from typing import Tuple
 
-MONTH = "MS"  # month-start frequency
+# -------------------------
+# Cash-flow projections
+# -------------------------
+def estimated_monthly_interest(principal: float, annual_rate: float, n_months: int) -> float:
+    return principal * (annual_rate / 12.0) * n_months
 
+def project_deposit(principal: float, annual_rate: float, months: int) -> pd.DataFrame:
+    start = pd.Timestamp(date.today()).normalize().replace(day=1)
+    periods = pd.period_range(start=start, periods=max(int(months), 1), freq="M")
+    df = pd.DataFrame({"period": periods.to_timestamp()})
+    df["interest"] = principal * (annual_rate / 12.0)
+    df["principal_cf"] = 0.0
+    df.loc[df.index[-1], "principal_cf"] = principal
+    df["cash_flow"] = df["interest"] + df["principal_cf"]
+    return df
 
-def _emi(principal: float, annual_rate: float, n_months: int) -> float:
-    """
-    EMI formula with monthly compounding.
-    r_m = annual_rate / 12
-    EMI = P * r_m * (1+r_m)^n / ((1+r_m)^n - 1)
-    """
-    if n_months <= 0 or principal <= 0:
-        return 0.0
+def project_asset(principal: float, annual_rate: float, months: int) -> pd.DataFrame:
+    m = max(int(months), 1)
     r = annual_rate / 12.0
-    if r <= 0:
-        return principal / n_months
-    fac = (1.0 + r) ** n_months
-    return principal * r * fac / (fac - 1.0)
-
+    if r == 0.0:
+        pmt = principal / m
+    else:
+        pmt = principal * (r * (1 + r) ** m) / ((1 + r) ** m - 1)
 
-def _project_deposit(principal: float, annual_rate: float, months: int) -> pd.DataFrame:
-    """
-    Monthly path for a deposit. Value compounds monthly.
-    """
-    r = annual_rate / 12.0
-    data = []
+    rows = []
+    start = pd.Timestamp(date.today()).normalize().replace(day=1)
     bal = principal
-    for m in range(months + 1):
-        data.append({"step": m, "portfolio_value": bal})
-        bal = bal * (1.0 + r)
-    return pd.DataFrame(data)
-
+    for i in range(1, m + 1):
+        interest = bal * r
+        principal_cf = pmt - interest
+        bal = max(0.0, bal - principal_cf)
+        rows.append({
+            "period": start + pd.offsets.MonthEnd(i),
+            "interest": float(interest),
+            "principal_cf": float(principal_cf),
+            "cash_flow": float(pmt),
+            "balance": float(bal)
+        })
+    return pd.DataFrame(rows)
 
-def _project_asset(principal: float, annual_rate: float, tenor_months: int) -> pd.DataFrame:
+def liquidity_gap(cf: pd.DataFrame) -> pd.DataFrame:
     """
-    Monthly amortization schedule for an asset/loan using EMI.
+    Input columns: period, product, cash_flow
     """
-    emi = _emi(principal, annual_rate, tenor_months)
-    r = annual_rate / 12.0
-    data = []
-    bal = principal
-    for m in range(tenor_months + 1):
-        interest = bal * r
-        principal_pay = max(0.0, emi - interest)
-        next_bal = max(0.0, bal - principal_pay)
-        data.append({
-            "step": m,
-            "portfolio_value": bal,
-            "emi": emi,
-            "interest_component": interest,
-            "principal_component": principal_pay,
-            "remaining_balance": next_bal
-        })
-        bal = next_bal
-    return pd.DataFrame(data)
+    cf = cf.copy()
+    cf["bucket"] = pd.PeriodIndex(cf["period"], freq="M").astype(str)
+    piv = cf.pivot_table(index="bucket", columns="product", values="cash_flow",
+                         aggfunc="sum", fill_value=0.0)
+    # assets - liabilities(FD)
+    piv["gap"] = piv.get("assets", 0.0) - piv.get("fd", 0.0)
+    piv["cumulative_gap"] = piv["gap"].cumsum()
+    return piv.reset_index()
 
+# -------------------------
+# Adapter + main builder
+# -------------------------
+REQUIRED_MASTER_COLS = {"Portfolio_value", "Interest_rate", "months", "product"}
 
-def build_timeseries(df: pd.DataFrame) -> pd.DataFrame:
+def _adapt_masterdataset(df: pd.DataFrame) -> pd.DataFrame:
     """
-    Input df columns (example):
-      - portfolio_date (datetime or str)
-      - instrument_type: 'Deposit' or 'Asset'
-      - balance: float
-      - interest_rate: annual rate (e.g., 0.12)
-      - time_to_maturity: months (int)
-      - tenor_months: months (for Assets; if missing, fallback to time_to_maturity)
-    Output:
-      Long time-series with monthly timestamps, projected 'portfolio_value'
-      (and EMI breakdown for Assets).
+    Accepts rows from main.masterdataset_v and returns a normalized frame
+    with a monthly projection per row.
     """
     df = df.copy()
-    if "timestamp" not in df.columns:
-        df["timestamp"] = pd.to_datetime(df["portfolio_date"])
-
-    out_frames: List[pd.DataFrame] = []
-    for _, row in df.iterrows():
-        itype = str(row.get("instrument_type", "")).strip().lower()
-        start = pd.to_datetime(row["timestamp"])
-        months = int(row.get("time_to_maturity", 0) or 0)
-        principal = float(row.get("balance", 0.0) or 0.0)
-        annual_rate = float(row.get("interest_rate", 0.0) or 0.0)
+    assert REQUIRED_MASTER_COLS.issubset(df.columns), (
+        f"DataFrame must include {REQUIRED_MASTER_COLS}, got {set(df.columns)}"
+    )
+    all_cf = []
+    for _, r in df.iterrows():
+        principal = float(r["Portfolio_value"])
+        rate = float(r["Interest_rate"])
+        months = int(r["months"]) if pd.notna(r["months"]) else max(int((r.get("days_to_maturity", 0) or 0) // 30), 1)
+        prod = str(r["product"]).lower().strip()
 
-        if itype == "deposit":
-            sched = _project_deposit(principal, annual_rate, months)
-        elif itype == "asset":
-            tenor = int(row.get("tenor_months", months) or months or 0)
-            sched = _project_asset(principal, annual_rate, tenor)
+        if prod == "fd":
+            cf = project_deposit(principal, rate, months)
         else:
-            # unknown types: keep flat
-            sched = pd.DataFrame({"step": range(months + 1), "portfolio_value": principal})
+            cf = project_asset(principal, rate, months)
 
-        # Build timestamps: month-start frequency
-        sched["timestamp"] = pd.date_range(start=start, periods=len(sched), freq=MONTH)
-        # Carry identifiers
-        for col in ["instrument_type", "interest_rate"]:
-            if col in df.columns:
-                sched[col] = row.get(col)
-        sched["origin_portfolio_date"] = start
-        sched["origin_balance"] = principal
+        # carry attributes
+        cf["product"] = prod
+        cf["contract_number"] = r.get("contract_number", None)
+        cf["segments"] = r.get("segments", None)
+        cf["currency"] = r.get("currency", None)
+        cf["board_currency"] = r.get("board_currency", None)
+        cf["Portfolio_value"] = principal
+        cf["Interest_rate"] = rate
+        all_cf.append(cf)
 
-        out_frames.append(sched)
+    return pd.concat(all_cf, ignore_index=True) if all_cf else pd.DataFrame(
+        columns=["period", "interest", "principal_cf", "cash_flow", "product",
+                 "contract_number", "segments", "currency", "board_currency",
+                 "Portfolio_value", "Interest_rate"]
+    )
+
+def build_timeseries(df: pd.DataFrame) -> Tuple[pd.DataFrame, pd.DataFrame]:
+    """
+    Main entrypoint used by app.py:
+      - If frame matches masterdataset_v, generate cashflows
+      - Compute liquidity gap on the result
+    Returns: (cashflows_df, gap_df)
+    """
+    if REQUIRED_MASTER_COLS.issubset(df.columns):
+        cf = _adapt_masterdataset(df)
+    else:
+        # if already a projected cashflow frame (period/product/cash_flow), pass-through
+        if {"period", "product", "cash_flow"}.issubset(df.columns):
+            cf = df.copy()
+        else:
+            # unsupported schema
+            return pd.DataFrame(), pd.DataFrame()
 
-    ts = pd.concat(out_frames, ignore_index=True)
-    ts = ts.sort_values(["timestamp", "instrument_type"]).reset_index(drop=True)
-    return ts
+    gap = liquidity_gap(cf[["period", "product", "cash_flow"]])
+    return cf, gap