src/streamlit_app.py

# app.py
# Streamlit front-end for the Liquidity Decision Map
# ---------------------------------------------------
# pip install streamlit numpy pandas seaborn matplotlib

# --- bootstrap: install missing deps automatically ---------------------------
def _ensure_packages(pkgs):
    """
    pkgs: sequence of dicts like
      {"pip": "streamlit", "import": "streamlit", "spec": ">=1.30"}
    - "pip":   name used with pip install
    - "import":module name used in 'import ...' (defaults to pip name)
    - "spec":  optional version spec (e.g., '==1.26.4' or '>=1.26')
    """
    import importlib, subprocess, sys

    for meta in pkgs:
        pip_name = meta["pip"]
        import_name = meta.get("import", pip_name)
        spec = meta.get("spec", "")
        try:
            importlib.import_module(import_name)
        except ImportError:
            pkg_spec = pip_name + (spec or "")
            print(f"[bootstrap] Installing {pkg_spec} …")
            try:
                subprocess.check_call([sys.executable, "-m", "pip", "install", pkg_spec])
            except subprocess.CalledProcessError:
                # Fallback: try --user (useful on locked-down machines)
                subprocess.check_call([sys.executable, "-m", "pip", "install", "--user", pkg_spec])
            # try import again (module just installed)
            importlib.invalidate_caches()
            importlib.import_module(import_name)

# call it for your app's deps
_ensure_packages([
    {"pip": "streamlit"},
    {"pip": "numpy"},
    {"pip": "pandas"},
    {"pip": "seaborn"},
    {"pip": "matplotlib"},
    {"pip":'io'},
])
# -----------------------------------------------------------------------------

import io
import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt
from matplotlib.colors import ListedColormap
from matplotlib.patches import FancyArrowPatch
import streamlit as st

# ---------- Core math ----------

def annuity_factor(r, T):
    """(1 - (1+r)^(-T)) / r, with r->0 limit = T"""
    return T if np.isclose(r, 0.0) else (1.0 - (1.0 + r) ** (-T)) / r

def breakeven_portfolio_return_traced(cost_basis_pct, tax_rate, mm_yield, horizon_years,
                                      mortgage_rate, shield_sell, shield_use):
    """
    r* = ( ((1 - g*tax) * (1 + i*Δshield*AF))^(1/T) * (1 + r_mm) ) - 1
    g = 1 - cost_basis_pct ; Δshield = shield_sell - shield_use ; AF = annuity_factor(r_mm, T)
    """
    cb = np.asarray(cost_basis_pct, dtype=float)
    tr = np.asarray(tax_rate, dtype=float)
    g  = np.clip(1.0 - cb, 0.0, 1.0)
    one_minus_wedge = 1.0 - g * tr
    dshield = np.asarray(shield_sell, dtype=float) - np.asarray(shield_use, dtype=float)
    adj = 1.0 + mortgage_rate * dshield * annuity_factor(mm_yield, horizon_years)
    with np.errstate(invalid="ignore"):
        rp_star = np.where(
            (one_minus_wedge > 0.0) & (horizon_years > 0),
            ((one_minus_wedge * adj) ** (1.0 / horizon_years)) * (1.0 + mm_yield) - 1.0,
            np.nan
        )
    return rp_star

# ---------- Plotting ----------

def make_decision_heatmap(cb_vals, tax_vals, r_mm, T, r_mort, r_exp, shield_sell, shield_use,
                          title):
    """
    Returns (fig, df_rstar, df_decision)
    - two-color squares: light green = Use MM, light blue = Sell
    - per-cell label with r*
    - decision boundary with upward arrow
    """
    CB, TR = np.meshgrid(cb_vals, tax_vals)  # rows=tax, cols=cb
    Rstar = breakeven_portfolio_return_traced(CB, TR, r_mm, T, r_mort, shield_sell, shield_use)
    Delta = Rstar - r_exp
    decision_idx = (Delta >= 0).astype(int)  # 0=MM, 1=SELL

    color_mm, color_sell = "#CDECCF", "#ADD8E6"
    cmap = ListedColormap([color_mm, color_sell])

    fig, ax = plt.subplots(figsize=(12, 7))
    sns.heatmap(
        decision_idx,
        ax=ax, cmap=cmap, vmin=-0.5, vmax=1.5, cbar=False,
        linewidths=0.8, linecolor="white", square=True,
        xticklabels=[f"{x:.0%}" for x in cb_vals],
        yticklabels=[f"{y:.0%}" for y in tax_vals]
    )

    # Per-cell r* labels
    M, N = decision_idx.shape
    for i in range(M):
        for j in range(N):
            rs = Rstar[i, j]
            if np.isfinite(rs):
                ax.text(j + 0.5, i + 0.5, f"{rs*100:.1f}%", ha="center", va="center",
                        fontsize=9, color="#0f172a")

    # Decision boundary + upward arrow
    finite = np.isfinite(Delta)
    if finite.any() and (np.nanmin(Delta) <= 0.0 <= np.nanmax(Delta)):
        Xc = np.arange(N); Yc = np.arange(M)
        XX, YY = np.meshgrid(Xc, Yc)
        CS = ax.contour(XX + 0.5, YY + 0.5, Delta, levels=[0.0], colors="black", linewidths=2)

        try:
            path = max(CS.collections[0].get_paths(), key=lambda p: p.vertices.shape[0])
            verts = path.vertices
            mid = len(verts) // 2
            p0, p1 = verts[mid-1], verts[mid+1]
            if p1[1] < p0[1]:  # ensure arrow points upward (toward SELL region)
                p0, p1 = p1, p0
            arrow = FancyArrowPatch((p0[0], p0[1]), (p1[0], p1[1]),
                                    arrowstyle='->', mutation_scale=16, lw=2, color='black')
            ax.add_patch(arrow)
            #ax.text(p1[0] + 0.2, min(p1[1] + 0.3, M+0.3), "Sell portfolio",
                #    fontsize=11, weight="bold")
            #ax.text(max(p0[0] - 1.0, -0.1), max(p0[1] - 0.5, -0.3), "Use money market",
               #     fontsize=11, weight="bold")
        except Exception:
            pass
    else:
        ax.text(0.5, 1.02, "No decision boundary within shown range",
                transform=ax.transAxes, ha="center", va="bottom", fontsize=10, color="dimgray")

    # Legend chips
    mm_patch   = plt.Line2D([0],[0], marker='s', color='w', label='Use money market',
                             markerfacecolor=color_mm, markersize=14)
    sell_patch = plt.Line2D([0],[0], marker='s', color='w', label='Sell portfolio',
                             markerfacecolor=color_sell, markersize=14)
    ax.legend(handles=[mm_patch, sell_patch], loc="upper left")

    ax.set_xlabel("Cost basis (% of market value)")
    ax.set_ylabel("Capital gains tax rate")
    subtitle = (f"Horizon {T:.0f}y | MM {r_mm:.1%} | Mortgage {r_mort:.2%} | "
                f"Expected rₚ {r_exp:.1%} | Δshield {(shield_sell - shield_use):.1%}")
    ax.set_title(f"{title}\n{subtitle}", fontsize=12)
    fig.text(0.5, -0.02,
             "Decision boundary (black line): below the line → expected portfolio return rₚ is ABOVE breakeven r* → Use money-market proceeds; "
             "above the line → rₚ is BELOW r* → Sell portfolio.",
             ha='center', va='top', fontsize=10, color='dimgray')

    fig.tight_layout()
    return fig, pd.DataFrame(Rstar, index=[f"{y:.0%}" for y in tax_vals],
                             columns=[f"{x:.0%}" for x in cb_vals]), \
           pd.DataFrame(np.where(decision_idx==1, "SELL", "MM"),
                        index=[f"{y:.0%}" for y in tax_vals],
                        columns=[f"{x:.0%}" for x in cb_vals])

# ---------- Streamlit UI ----------

st.set_page_config(page_title="Liquidity Decision Map", layout="wide")

st.title("Liquidity Decision Map (Python)")
st.caption("Square-cell decision heatmap comparing **Sell portfolio** vs **Use money-market cash** with IRS tracing-aware deductibility.")

with st.sidebar:
    st.header("Assumptions")
    T = st.number_input("Horizon (years)", value=10, min_value=1, max_value=60, step=1)
    r_mm = st.number_input("Money market yield (decimal)", value=0.042, step=0.001, format="%.3f")
    r_mort = st.number_input("Mortgage rate (decimal)", value=0.06, step=0.001, format="%.3f")
    r_exp = st.number_input("Expected portfolio return (decimal)", value=0.05, step=0.001, format="%.3f")

    st.header("Scenario / Shields")
    scenario = st.radio(
        "Preset",
        ["Personal use (Use-MM loses deduction)",
         "Investment use (both retain deduction)",
         "Personal + NII cap (partial in SELL)"],
        index=0
    )
    if scenario == "Personal use (Use-MM loses deduction)":
        shield_sell, shield_use = 0.37, 0.00
    elif scenario == "Investment use (both retain deduction)":
        shield_sell, shield_use = 0.37, 0.37
    else:
        shield_sell, shield_use = 0.15, 0.00

    st.caption("Override shields (effective tax value of interest deductibility):")
    shield_sell = st.number_input("SELL path shield (decimal)", value=float(shield_sell), step=0.01, min_value=0.0, max_value=0.5)
    shield_use  = st.number_input("USE-MM path shield (decimal)", value=float(shield_use), step=0.01, min_value=0.0, max_value=0.5)

    st.header("Grid")
    cb_min  = st.number_input("Cost basis min (decimal)", value=0.30, step=0.05, min_value=0.0, max_value=1.0)
    cb_max  = st.number_input("Cost basis max (decimal)", value=0.90, step=0.05, min_value=0.0, max_value=1.0)
    cb_steps= st.number_input("# cost basis steps", value=13, step=1, min_value=3, max_value=51)
    tax_min = st.number_input("CGT min (decimal)", value=0.10, step=0.01, min_value=0.0, max_value=0.6)
    tax_max = st.number_input("CGT max (decimal)", value=0.35, step=0.01, min_value=0.0, max_value=0.6)
    tax_steps=st.number_input("# CGT steps", value=11, step=1, min_value=3, max_value=51)

# ---------- Step-by-step breakeven explainer ----------
def explain_breakeven(cb_pct, tax_rate, r_mm, T, r_mort, shield_sell, shield_use):
    """
    Returns a dict with all intermediate pieces for the traced breakeven formula:
    r* = ( ((1 - g*τ) * (1 + i*Δs*AF))^(1/T) * (1 + r_mm) ) - 1
    where g = 1 - cb,  Δs = shield_sell - shield_use,  AF = (1 - (1+r_mm)^(-T))/r_mm
    """
    g = 1.0 - cb_pct                                # embedded gain ratio
    one_minus_wedge = 1.0 - g * tax_rate            # net $ after CGT per $ sold
    dshield = shield_sell - shield_use              # difference in tax shields
    AF = annuity_factor(r_mm, T)                    # annuity factor
    adj = 1.0 + r_mort * dshield * AF               # deductibility adjustment
    r_star = ((one_minus_wedge * adj) ** (1.0 / T)) * (1.0 + r_mm) - 1.0

    return {
        "cb": cb_pct, "tax": tax_rate, "g": g,
        "one_minus_wedge": one_minus_wedge,
        "dshield": dshield, "AF": AF, "adj": adj,
        "r_mm": r_mm, "T": T, "r_mort": r_mort, "r_star": r_star
    }

def linspace(a, b, n):
    if n <= 1: return np.array([a])
    return np.linspace(a, b, int(n))

cb_vals = linspace(cb_min, cb_max, cb_steps)
tax_vals = linspace(tax_min, tax_max, tax_steps)

# Plot
title = ("Decision map — PERSONAL use of proceeds (tracing breaks if you use MM)"
         if not np.isclose(shield_sell, shield_use) else
         "Decision map — INVESTMENT use of proceeds (both retain deductibility)")
fig, df_rstar, df_decision = make_decision_heatmap(cb_vals, tax_vals, r_mm, T, r_mort, r_exp,
                                                   shield_sell, shield_use, title)
st.pyplot(fig, clear_figure=True)

# Explanation


# Download PNG
buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=150, bbox_inches="tight")
st.download_button("Download PNG", data=buf.getvalue(), file_name="liquidity_decision_map.png", mime="image/png")

with st.expander("Show data tables"):
    st.subheader("Breakeven r* (annualized)")
    st.dataframe(df_rstar.style.format("{:.2%}"))
    st.subheader("Decision")
    st.dataframe(df_decision)

st.markdown(
"""
**How to read:**  
- Each square shows the **breakeven return** \(r^*\).  
- **Light green = Use money market** (your expected return \(r_p\) is **above** \(r^*\)).  
- **Light blue = Sell portfolio** (your \(r_p\) is **below** \(r^*\)).  
- The **black curve** is the decision boundary; the arrow points toward the **Sell** region.
"""
)