Update app.py

app.py

@@ -2,6 +2,7 @@ import os, io, math, warnings
 warnings.filterwarnings("ignore")
 
 from typing import List, Tuple, Dict, Optional
+from functools import partial
 
 import numpy as np
 import pandas as pd
@@ -23,8 +24,9 @@ MAX_TICKERS = 30
 DEFAULT_LOOKBACK_YEARS = 5
 MARKET_TICKER = "VOO"
 
-# column schema (weights shown in percent in UI tables)
+# column schemas (weights shown in percent in UI tables)
 POS_COLS = ["ticker", "amount_usd", "weight_%", "beta"]
+SUG_RISK_COLS = ["ticker", "suggested_weight_%"]
 
 FRED_MAP = [
     (1,  "DGS1"),
@@ -45,6 +47,9 @@ def ensure_data_dir():
 def empty_positions_df():
     return pd.DataFrame(columns=POS_COLS)
 
+def empty_risk_df():
+    return pd.DataFrame(columns=SUG_RISK_COLS)
+
 def fred_series_for_horizon(years: float) -> str:
     y = max(1.0, min(100.0, float(years)))
     for cutoff, code in FRED_MAP:
@@ -76,7 +81,6 @@ def fetch_prices_monthly(tickers: List[str], years: int) -> pd.DataFrame:
     )["Close"]
     if isinstance(df, pd.Series):
         df = df.to_frame()
-    # yfinance sometimes returns MultiIndex columns
     if isinstance(df.columns, pd.MultiIndex):
         df.columns = [c[-1] if isinstance(c, tuple) else str(c) for c in df.columns]
     else:
@@ -225,7 +229,7 @@ def plot_cml(
     plt.scatter([same_sigma_sigma], [same_sigma_mu], label="Efficient same sigma")
     plt.scatter([same_mu_sigma], [same_mu_mu], label="Efficient same return")
     if targ_sigma is not None and targ_mu is not None:
-        plt.scatter([targ_sigma], [targ_mu], label="Target suggestion")
+        plt.scatter([targ_sigma], [targ_mu], label="Dataset suggestion")
 
     # Guides + annotations (in percent)
     plt.plot([pt_sigma, same_sigma_sigma], [pt_mu, same_sigma_mu],
@@ -258,7 +262,7 @@ def plot_cml(
     buf.seek(0)
     return Image.open(buf)
 
-# -------------- synthetic dataset (for the optional predictor) --------------
+# -------------- synthetic dataset (for predictor + risk buttons) --------------
 def synth_profile(seed: int) -> str:
     rng = np.random.default_rng(seed)
     risk = rng.choice(["cautious", "balanced", "moderate", "growth", "aggressive"])
@@ -339,6 +343,73 @@ def predict_from_surrogate(amounts_map: Dict[str, float], universe: List[str],
     er_hat, sigma_hat, beta_hat = float(yhat[0]), float(yhat[1]), float(yhat[2])
     return er_hat, sigma_hat, beta_hat
 
+# ---- dataset risk buttons helpers (purely CSV-based) ----
+def pick_row_by_risk(df: pd.DataFrame, level: str) -> Optional[pd.Series]:
+    df = df.dropna(subset=["sigma_p"])
+    if df.empty:
+        return None
+    if level == "low":
+        return df.loc[df["sigma_p"].idxmin()]
+    if level == "high":
+        return df.loc[df["sigma_p"].idxmax()]
+    # medium: closest to median sigma
+    med = float(df["sigma_p"].median())
+    idx = (df["sigma_p"] - med).abs().idxmin()
+    return df.loc[idx]
+
+def row_to_suggestion(row: pd.Series, universe: List[str]) -> Optional[Dict]:
+    x = _row_to_exposures(row, universe)
+    if x is None:
+        return None
+    wmap = {universe[i]: float(x[i]) for i in range(len(universe)) if abs(float(x[i])) > 1e-4}
+    # sort top exposures
+    wmap = dict(sorted(wmap.items(), key=lambda kv: -abs(kv[1]))[:12])
+    return {
+        "weights": wmap,
+        "er": float(row["er_p"]),
+        "sigma": float(row["sigma_p"]),
+        "beta": float(row["beta_p"]),
+    }
+
+def suggest_by_risk(level: str, state: dict):
+    # State must come from a previous "Compute"
+    if not isinstance(state, dict) or not state.get("csv_path") or not os.path.exists(state["csv_path"]):
+        return gr.update(), empty_risk_df(), "Run analysis first to build the dataset."
+
+    try:
+        df = pd.read_csv(state["csv_path"])
+    except Exception:
+        return gr.update(), empty_risk_df(), "Could not read dataset."
+
+    row = pick_row_by_risk(df, {"low":"low","med":"med","high":"high"}[level])
+    if row is None:
+        return gr.update(), empty_risk_df(), "Dataset is empty."
+
+    cand = row_to_suggestion(row, UNIVERSE)
+    if cand is None:
+        return gr.update(), empty_risk_df(), "No suggestion available."
+
+    # Build table in percents
+    rows = [{"ticker": k, "suggested_weight_%": v * 100.0} for k, v in cand["weights"].items()]
+    risk_table = pd.DataFrame(rows, columns=SUG_RISK_COLS)
+
+    # Overlay the dataset suggestion on the existing CML
+    img = plot_cml(
+        state["rf_ann"], state["erp_ann"], state["sigma_mkt"],
+        state["pt_sigma"], state["pt_mu"],
+        state["same_sigma_sigma"], state["same_sigma_mu"],
+        state["same_mu_sigma"], state["same_mu_mu"],
+        targ_sigma=cand["sigma"], targ_mu=cand["er"]
+    )
+
+    msg = (
+        f"**Dataset suggestion ({'Lowest' if level=='low' else 'Medium' if level=='med' else 'Highest'} risk)**  \n"
+        f"- Predicted expected return: {fmt_pct(cand['er'])}  \n"
+        f"- Predicted sigma: {fmt_pct(cand['sigma'])}  \n"
+        f"- Predicted beta: {cand['beta']:.2f}"
+    )
+    return img, risk_table, msg
+
 # -------------- summary --------------
 def fmt_pct(x: float) -> str:
     return f"{x*100:.2f}%"
@@ -453,11 +524,11 @@ def compute(years_lookback: int, table: pd.DataFrame, use_synth: bool):
 
     symbols = [t for t in df["ticker"].tolist() if t]
     if len(symbols) == 0:
-        return None, "Add at least one ticker", "Universe empty", empty_positions_df(), None
+        return None, "Add at least one ticker", "Universe empty", empty_positions_df(), None, {}
 
     symbols = validate_tickers(symbols, years_lookback)
     if len(symbols) == 0:
-        return None, "Could not validate any tickers", "Universe invalid", empty_positions_df(), None
+        return None, "Could not validate any tickers", "Universe invalid", empty_positions_df(), None, {}
 
     global UNIVERSE
     UNIVERSE = list(sorted(set([s for s in symbols if s != MARKET_TICKER] + [MARKET_TICKER])))[:MAX_TICKERS]
@@ -471,7 +542,7 @@ def compute(years_lookback: int, table: pd.DataFrame, use_synth: bool):
 
     gross = sum(abs(v) for v in amounts.values())
     if gross == 0:
-        return None, "All amounts are zero", "Universe ok", empty_positions_df(), None
+        return None, "All amounts are zero", "Universe ok", empty_positions_df(), None, {}
     weights = {k: v / gross for k, v in amounts.items()}
 
     beta_p, er_p, sigma_p = portfolio_stats(weights, covA, betas, rf_ann, erp_ann)
@@ -479,7 +550,7 @@ def compute(years_lookback: int, table: pd.DataFrame, use_synth: bool):
     a_sigma, b_sigma, mu_eff_sigma = efficient_same_sigma(sigma_p, rf_ann, erp_ann, sigma_mkt)
     a_mu, b_mu, sigma_eff_mu = efficient_same_return(er_p, rf_ann, erp_ann, sigma_mkt)
 
-    # ensure synthetic dataset exists once (for predictor only)
+    # ensure synthetic dataset exists once (for predictor + risk buttons)
     if not os.path.exists(DATASET_PATH):
         synth_df = build_synthetic_dataset(
             universe=list(sorted(set(symbols + [MARKET_TICKER]))),
@@ -530,8 +601,22 @@ def compute(years_lookback: int, table: pd.DataFrame, use_synth: bool):
         })
     pos_table = pd.DataFrame(rows, columns=POS_COLS)
 
+    # Pack state for risk buttons
+    state = {
+        "csv_path": csv_path,
+        "rf_ann": rf_ann,
+        "erp_ann": erp_ann,
+        "sigma_mkt": sigma_mkt,
+        "pt_sigma": sigma_p,
+        "pt_mu": er_p,
+        "same_sigma_sigma": sigma_p,
+        "same_sigma_mu": mu_eff_sigma,
+        "same_mu_sigma": sigma_eff_mu,
+        "same_mu_mu": er_p,
+    }
+
     uni_msg = f"Universe set to {', '.join(UNIVERSE)}"
-    return img, info, uni_msg, pos_table, csv_path
+    return img, info, uni_msg, pos_table, csv_path, state
 
 # -------------- UI --------------
 ensure_data_dir()
@@ -540,9 +625,12 @@ with gr.Blocks(title="Efficient Portfolio Advisor") as demo:
     gr.Markdown(
         "## Efficient Portfolio Advisor\n"
         "Search symbols, enter dollar amounts, set your horizon. "
-        "Prices come from Yahoo Finance. Risk free comes from FRED."
+        "Prices come from Yahoo Finance. Risk free comes from FRED.\n\n"
+        "**New:** Dataset-only risk suggestions (Low / Medium / High) from the 1,000-row synthetic CSV."
     )
 
+    app_state = gr.State({})
+
     with gr.Row():
         with gr.Column(scale=1):
             q = gr.Textbox(label="Search symbol")
@@ -565,6 +653,12 @@ with gr.Blocks(title="Efficient Portfolio Advisor") as demo:
             use_synth = gr.Checkbox(label="Use synthetic predictor (fast check)", value=True)
 
             run_btn = gr.Button("Compute and suggest")
+
+            gr.Markdown("### Dataset-based risk suggestions")
+            with gr.Row():
+                btn_low = gr.Button("Lowest risk (dataset)")
+                btn_med = gr.Button("Medium risk (dataset)")
+                btn_high = gr.Button("Highest risk (dataset)")
         with gr.Column(scale=1):
             plot = gr.Image(label="Capital Market Line", type="pil")
             summary = gr.Markdown(label="Summary")
@@ -577,6 +671,15 @@ with gr.Blocks(title="Efficient Portfolio Advisor") as demo:
                 value=empty_positions_df(),
                 interactive=False
             )
+            risk_table = gr.Dataframe(
+                label="Suggested portfolio from dataset",
+                headers=SUG_RISK_COLS,
+                datatype=["str", "number"],
+                col_count=(len(SUG_RISK_COLS), "fixed"),
+                value=empty_risk_df(),
+                interactive=False
+            )
+            risk_msg = gr.Markdown(label="Suggestion metrics")
             dl = gr.File(label="Session CSV path (synthetic predictor data)", value=None, visible=True)
 
     def do_search(query):
@@ -591,8 +694,13 @@ with gr.Blocks(title="Efficient Portfolio Advisor") as demo:
     run_btn.click(
         fn=compute,
         inputs=[lookback, table, use_synth],
-        outputs=[plot, summary, universe_msg, positions, dl]
+        outputs=[plot, summary, universe_msg, positions, dl, app_state]
     )
 
+    # Risk buttons (purely dataset-driven)
+    btn_low.click(fn=partial(suggest_by_risk, "low"), inputs=[app_state], outputs=[plot, risk_table, risk_msg])
+    btn_med.click(fn=partial(suggest_by_risk, "med"), inputs=[app_state], outputs=[plot, risk_table, risk_msg])
+    btn_high.click(fn=partial(suggest_by_risk, "high"), inputs=[app_state], outputs=[plot, risk_table, risk_msg])
+
 if __name__ == "__main__":
     demo.launch()