Update app.py

app.py

@@ -0,0 +1,282 @@
+import os
+import io
+import ast
+import json
+import math
+import time
+import faiss
+import numpy as np
+import pandas as pd
+import matplotlib.pyplot as plt
+from typing import List, Tuple
+from sklearn.preprocessing import StandardScaler
+from sentence_transformers import SentenceTransformer
+import gradio as gr
+
+# --------------------
+# Finance parameters
+# --------------------
+TICKERS = ["VOO","QQQ","VNQ","IEF","HYG","GLD","EEM","XLP","XLK","XLE"]
+BETAS   = [1.00, 1.25, 0.60, 0.10, 0.40, 0.10, 1.10, 0.70, 1.20, 1.10]
+SIGMAS  = [0.16, 0.25, 0.18, 0.05, 0.10, 0.14, 0.22, 0.12, 0.20, 0.22]
+DEFAULT_RF = 0.03
+DEFAULT_MKT_PREM = 0.05
+DEFAULT_CORR = 0.2
+
+DATA_PATH = "data"
+CSV_PATH = os.path.join(DATA_PATH, "portfolios.csv")
+
+# --------------------
+# Helpers
+# --------------------
+def ensure_data_dir():
+    os.makedirs(DATA_PATH, exist_ok=True)
+
+def normalize_weights(w: np.ndarray) -> np.ndarray:
+    w = np.clip(np.array(w, dtype=float), 0.0, None)
+    s = w.sum()
+    if s <= 0:
+        return np.ones_like(w) / len(w)
+    return w / s
+
+def portfolio_sigma(weights: np.ndarray, sigmas: List[float], corr: float = DEFAULT_CORR) -> float:
+    sig = np.array(sigmas, dtype=float)
+    w = np.array(weights, dtype=float)
+    cov = np.outer(sig, sig) * corr
+    np.fill_diagonal(cov, sig**2)
+    var = float(w @ cov @ w)
+    return math.sqrt(var)
+
+def portfolio_beta(weights: np.ndarray, betas: List[float]) -> float:
+    return float(np.dot(weights, np.array(betas, dtype=float)))
+
+def capm_expected_return(beta: float, rf: float, mkt_prem: float) -> float:
+    return float(rf + beta * mkt_prem)
+
+def synth_profile(seed: int) -> str:
+    rng = np.random.default_rng(seed)
+    risk = rng.choice(["cautious", "balanced", "moderate", "growth", "aggressive"])
+    horizon = rng.choice(["three years", "five years", "seven years", "ten years", "fifteen years"])
+    goal = rng.choice([
+        "retirement savings",
+        "first home",
+        "education fund",
+        "wealth building",
+        "travel fund",
+        "emergency buffer"
+    ])
+    return f"{risk} investor, {horizon} horizon, goal is {goal}."
+
+def make_one_row(pid: int, rf: float, mkt_prem: float, corr: float) -> dict:
+    w = np.random.dirichlet(np.ones(len(TICKERS)))
+    b = portfolio_beta(w, BETAS)
+    er = capm_expected_return(b, rf, mkt_prem)
+    s = portfolio_sigma(w, SIGMAS, corr=corr)
+    return {
+        "id": pid,
+        "profile_text": synth_profile(1000 + pid),
+        "tickers": ",".join(TICKERS),
+        "weights": ",".join(f"{x:.4f}" for x in w),
+        "beta_p": b,
+        "er_p": er,
+        "sigma_p": s
+    }
+
+def generate_small_dataset(n: int = 300,
+                           rf: float = DEFAULT_RF,
+                           mkt_prem: float = DEFAULT_MKT_PREM,
+                           corr: float = DEFAULT_CORR) -> pd.DataFrame:
+    rows = [make_one_row(i, rf, mkt_prem, corr) for i in range(n)]
+    return pd.DataFrame(rows)
+
+def load_or_build_csv() -> pd.DataFrame:
+    ensure_data_dir()
+    if os.path.exists(CSV_PATH):
+        df = pd.read_csv(CSV_PATH)
+        # Backward compatibility if weights stored as list text
+        if isinstance(df.get("weights", pd.Series([None])).iloc[0], str) and "[" in str(df["weights"].iloc[0]):
+            df["weights"] = df["weights"].apply(lambda s: ",".join(f"{float(x):.4f}" for x in ast.literal_eval(s)))
+        return df
+    # Build a small dataset so the Space is usable without uploads
+    df = generate_small_dataset()
+    df.to_csv(CSV_PATH, index=False)
+    return df
+
+# --------------------
+# Embeddings and index
+# --------------------
+class Recommender:
+    def __init__(self, df: pd.DataFrame):
+        self.df = df.copy()
+        self.model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
+        self.scaler = StandardScaler()
+        self.index = None
+        self.feature_dim = None
+        self.vecs = None
+        self._build()
+
+    def _text_embed(self, texts: List[str]) -> np.ndarray:
+        return self.model.encode(texts, normalize_embeddings=True)
+
+    def _build(self):
+        texts = self.df["profile_text"].astype(str).tolist()
+        text_vecs = self._text_embed(texts)
+        nums = self.df[["er_p","sigma_p","beta_p"]].to_numpy(dtype=float)
+        nums = self.scaler.fit_transform(nums)
+        feats = np.hstack([text_vecs, nums])
+        faiss.normalize_L2(feats)
+        self.vecs = feats.astype("float32")
+        self.feature_dim = self.vecs.shape[1]
+        self.index = faiss.IndexFlatIP(self.feature_dim)
+        self.index.add(self.vecs)
+
+    def query(self, profile_text: str, er_p: float, sigma_p: float, beta_p: float, topk: int = 3):
+        text_vec = self._text_embed([profile_text])
+        nums = np.array([[er_p, sigma_p, beta_p]], dtype=float)
+        nums = self.scaler.transform(nums)
+        q = np.hstack([text_vec, nums]).astype("float32")
+        faiss.normalize_L2(q)
+        D, I = self.index.search(q, topk)
+        idxs = I[0].tolist()
+        scores = D[0].tolist()
+        out = self.df.iloc[idxs].copy()
+        out["score"] = scores
+        return out
+
+# --------------------
+# Plot CML
+# --------------------
+def plot_cml(rf: float, mkt_prem: float, market_sigma: float, port_sigma: float, port_er: float):
+    fig = plt.figure(figsize=(5, 4), dpi=120)
+    xs = np.linspace(0, max(market_sigma*1.4, port_sigma*1.2, 0.25), 50)
+    cml = rf + (mkt_prem / market_sigma) * xs
+    plt.plot(xs, cml, label="CML")
+    plt.scatter([0.0], [rf], label="Risk free")
+    plt.scatter([market_sigma], [rf + mkt_prem], label="Market")
+    plt.scatter([port_sigma], [port_er], label="Your portfolio")
+    plt.xlabel("Standard deviation")
+    plt.ylabel("Expected return")
+    plt.legend()
+    buf = io.BytesIO()
+    plt.tight_layout()
+    plt.savefig(buf, format="png")
+    plt.close(fig)
+    buf.seek(0)
+    return buf
+
+# --------------------
+# App state
+# --------------------
+DF = load_or_build_csv()
+RECO = Recommender(DF)
+
+# --------------------
+# Gradio logic
+# --------------------
+def sum_to_one(*w_list):
+    w = np.array([float(x) for x in w_list], dtype=float)
+    w = normalize_weights(w)
+    return [float(f"{x:.4f}") for x in w]
+
+def compute_and_recommend(goal_text: str,
+                          rf: float,
+                          mkt_prem: float,
+                          mkt_sigma: float,
+                          *weights) -> Tuple[str, pd.DataFrame, gr.Image, str]:
+    w = normalize_weights(np.array(weights, dtype=float))
+    b = portfolio_beta(w, BETAS)
+    er = capm_expected_return(b, rf, mkt_prem)
+    s = portfolio_sigma(w, SIGMAS, corr=DEFAULT_CORR)
+
+    # Query top 3
+    q_text = goal_text.strip() or "balanced investor, five years horizon, goal is retirement savings."
+    recs = RECO.query(q_text, er, s, b, topk=3).reset_index(drop=True)
+
+    # Prepare nice table
+    show = recs[["profile_text","er_p","sigma_p","beta_p","score"]].copy()
+    show.columns = ["profile", "er", "sigma", "beta", "similarity"]
+
+    # Plot
+    img_buf = plot_cml(rf, mkt_prem, mkt_sigma, s, er)
+
+    summary = (
+        f"Expected return {er:.2%}. "
+        f"Risk or sigma {s:.2%}. "
+        f"Beta {b:.2f}. "
+        f"Weights order {', '.join(TICKERS)}. "
+        f"Weights {', '.join(f'{x:.2%}' for x in w)}."
+    )
+
+    return summary, show, img_buf, q_text
+
+def upload_csv(file):
+    global DF, RECO
+    if file is None:
+        return "No file received."
+    try:
+        df = pd.read_csv(file.name)
+        required = {"profile_text","weights","er_p","sigma_p","beta_p"}
+        if not required.issubset(set(df.columns)):
+            return f"CSV must have columns {sorted(required)}"
+        DF = df.copy()
+        RECO = Recommender(DF)
+        return f"Loaded {len(DF)} rows and rebuilt index."
+    except Exception as e:
+        return f"Failed to load CSV. {e}"
+
+# --------------------
+# UI
+# --------------------
+with gr.Blocks(title="Personal Portfolio Risk Return Analyzer") as demo:
+    gr.Markdown(
+        "## Personal Portfolio Risk Return Analyzer\n"
+        "Enter a goal sentence and set weights. The app computes expected return, risk, and beta, "
+        "then shows three similar portfolios from the dataset."
+    )
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            goal = gr.Textbox(
+                label="Goal or profile sentence",
+                value="balanced investor, five years horizon, goal is retirement savings."
+            )
+            rf_in = gr.Number(label="Risk free rate", value=DEFAULT_RF, precision=4)
+            prem_in = gr.Number(label="Market premium", value=DEFAULT_MKT_PREM, precision=4)
+            mkt_sigma_in = gr.Number(label="Market sigma for CML plot", value=0.17, precision=4)
+
+            gr.Markdown("#### Weights, must sum to one")
+            sliders = []
+            for t in TICKERS:
+                sliders.append(gr.Slider(minimum=0.0, maximum=1.0, step=0.001, value=0.1, label=t))
+            sum_btn = gr.Button("Normalize weights to one")
+
+            upload = gr.File(label="Upload portfolios.csv to replace dataset", file_count="single")
+            status = gr.Markdown()
+
+        with gr.Column(scale=1):
+            summary = gr.Textbox(label="Your portfolio summary", interactive=False)
+            table = gr.Dataframe(headers=["profile","er","sigma","beta","similarity"], row_count=3)
+            plot = gr.Image(label="Capital Market Line", type="pil")
+            used_text = gr.Textbox(label="Query text used for retrieval", interactive=False)
+
+    sum_btn.click(fn=sum_to_one, inputs=sliders, outputs=sliders)
+    upload.upload(fn=upload_csv, inputs=upload, outputs=status)
+
+    compute_btn = gr.Button("Compute and recommend")
+    compute_btn.click(
+        fn=compute_and_recommend,
+        inputs=[goal, rf_in, prem_in, mkt_sigma_in] + sliders,
+        outputs=[summary, table, plot, used_text]
+    )
+
+    gr.Examples(
+        examples=[
+            ["cautious investor, ten years horizon, goal is education fund.", 0.03, 0.05, 0.17] + [0.1]*10,
+            ["aggressive investor, seven years horizon, goal is wealth building.", 0.03, 0.05, 0.17] + [0.05,0.15,0.05,0.05,0.05,0.05,0.2,0.1,0.2,0.1],
+            ["balanced investor, five years horizon, goal is first home.", 0.03, 0.05, 0.17] + [0.12,0.12,0.10,0.06,0.08,0.06,0.12,0.06,0.16,0.12],
+        ],
+        inputs=[goal, rf_in, prem_in, mkt_sigma_in] + sliders
+    )
+
+if __name__ == "__main__":
+    demo.launch()