# 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. """ )