Update app.py

app.py

@@ -1,178 +1,134 @@
-import os, io, math, warnings
-warnings.filterwarnings("ignore")
+# app.py - Part 1
 
-from typing import List, Tuple, Dict
-import numpy as np
 import pandas as pd
-import matplotlib.pyplot as plt
-import gradio as gr
-from PIL import Image
-import requests
+import numpy as np
 import yfinance as yf
-
-# ---------------- config ----------------
-DATA_DIR = "data"
-MAX_TICKERS = 30
-DEFAULT_LOOKBACK_YEARS = 5
-MARKET_TICKER = "VOO"
-POS_COLS = ["ticker", "weight_exposure", "beta", "er_p", "sigma_p"]
-
-FRED_MAP = [
-    (1, "DGS1"), (2, "DGS2"), (3, "DGS3"),
-    (5, "DGS5"), (7, "DGS7"), (10, "DGS10"),
-    (20, "DGS20"), (30, "DGS30"), (100, "DGS30")
-]
-
-# ---------------- helpers ----------------
-def ensure_data_dir():
-    os.makedirs(DATA_DIR, exist_ok=True)
-
-def fred_series_for_horizon(years: float) -> str:
-    y = max(1.0, min(100.0, float(years)))
-    for cutoff, code in FRED_MAP:
-        if y <= cutoff:
-            return code
-    return "DGS30"
-
-def fetch_fred_yield_annual(code: str) -> float:
-    url = f"https://fred.stlouisfed.org/graph/fredgraph.csv?id={code}"
-    try:
-        r = requests.get(url, timeout=10)
-        r.raise_for_status()
-        df = pd.read_csv(io.StringIO(r.text))
-        s = pd.to_numeric(df.iloc[:, 1], errors="coerce").dropna()
-        return float(s.iloc[-1] / 100.0) if len(s) else 0.03
-    except Exception:
-        return 0.03
-
-def fetch_prices_monthly(tickers: List[str], years: int) -> pd.DataFrame:
-    start = pd.Timestamp.today(tz="UTC") - pd.DateOffset(years=years, days=7)
-    end = pd.Timestamp.today(tz="UTC")
-    frames = []
-    for t in tickers:
-        try:
-            s = yf.download(
-                t, start=start.date(), end=end.date(),
-                interval="1mo", auto_adjust=True, progress=False
-            )["Close"]
-            if isinstance(s, pd.Series) and s.dropna().size > 0:
-                frames.append(s.rename(t))
-        except Exception:
-            pass
-    if frames:
-        return pd.concat(frames, axis=1).dropna(how="any").fillna(method="ffill")
-    return pd.DataFrame()
-
-def monthly_returns(prices: pd.DataFrame) -> pd.DataFrame:
-    return prices.pct_change().dropna()
-
-def annualize_mean(m):
-    return np.asarray(m, dtype=float) * 12.0
-
-def annualize_sigma(s):
-    return np.asarray(s, dtype=float) * math.sqrt(12.0)
-
-def estimate_all_moments_aligned(symbols: List[str], years: int, rf_ann: float):
-    px = fetch_prices_monthly(symbols, years)
-    rets = monthly_returns(px)
-    rf_m = rf_ann / 12.0
-    mu = rets.mean()
-    sigma = rets.std(ddof=1)
-    betas = {}
-    mkt = rets[MARKET_TICKER]
-    var_m = np.var(mkt - rf_m, ddof=1)
-    for s in symbols:
-        if s == MARKET_TICKER: 
-            betas[s] = 1.0
-        else:
-            ex_s = rets[s] - rf_m
-            betas[s] = np.cov(ex_s, mkt - rf_m, ddof=1)[0,1] / var_m
-    erp = annualize_mean(mu[MARKET_TICKER]) - rf_ann
-    sigma_mkt = annualize_sigma(sigma[MARKET_TICKER])
-    covA = pd.DataFrame(np.cov(rets.T) * 12.0, index=symbols, columns=symbols)
-    return betas, covA, erp, sigma_mkt
-
-def capm_er(beta: float, rf_ann: float, erp_ann: float) -> float:
-    return float(rf_ann + beta * erp_ann)
-
-def portfolio_stats(weights: Dict[str, float], cov_ann: pd.DataFrame,
-                    betas: Dict[str, float], rf_ann: float, erp_ann: float):
-    tickers = list(weights.keys())
-    w = np.array(list(weights.values()))
-    w_expo = w / sum(abs(w))
-    beta_p = np.dot([betas[t] for t in tickers], w_expo)
-    er_p = capm_er(beta_p, rf_ann, erp_ann)
-    cov = cov_ann.loc[tickers, tickers].to_numpy()
-    sigma_p = math.sqrt(max(w_expo @ cov @ w_expo, 0.0))
-    return beta_p, er_p, sigma_p
-
-def efficient_same_sigma(sigma_target, rf_ann, erp_ann, sigma_mkt):
-    a = sigma_target / sigma_mkt
-    return a, 1 - a, rf_ann + a * erp_ann
-
-def efficient_same_return(mu_target, rf_ann, erp_ann, sigma_mkt):
-    a = (mu_target - rf_ann) / erp_ann
-    return a, 1 - a, abs(a) * sigma_mkt
-
-def build_synthetic_dataset(symbols: List[str], years: int, rf_ann: float, erp_ann: float):
-    betas, covA, _, _ = estimate_all_moments_aligned(symbols, years, rf_ann)
-    rng = np.random.default_rng(42)
-    rows = []
-    for _ in range(1000):
-        k = rng.integers(2, len(symbols)+1)
-        picks = list(rng.choice(symbols, size=k, replace=False))
-        raw = rng.dirichlet(np.ones(k))
-        gross = 1.0 + rng.gamma(2.0, 0.5)
-        w = gross * raw
-        stats = portfolio_stats({picks[i]: w[i] for i in range(k)}, covA, betas, rf_ann, erp_ann)
-        rows.append({
-            "tickers": ",".join(picks),
-            "weights": ",".join(f"{x:.4f}" for x in w),
-            "beta_p": stats[0], "er_p": stats[1], "sigma_p": stats[2]
+import gradio as gr
+from itertools import combinations_with_replacement
+
+# -------------------
+# Helper functions
+# -------------------
+
+def fetch_live_data(tickers, period="1y"):
+    """Fetch historical adjusted close prices for given tickers."""
+    data = yf.download(tickers, period=period)["Adj Close"]
+    return data.dropna()
+
+def calculate_portfolio_metrics(weights, mean_returns, cov_matrix, risk_free_rate=0.045):
+    """Return expected portfolio return, volatility, and beta."""
+    weights = np.array(weights)
+    portfolio_return = np.sum(mean_returns * weights)
+    portfolio_volatility = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
+    beta = np.sum(weights)  # Placeholder if no real beta calc
+    return portfolio_return, portfolio_volatility, beta
+
+def generate_synthetic_portfolios(tickers, num_portfolios=1000):
+    """Generate synthetic portfolios from live data for given tickers."""
+    df_prices = fetch_live_data(tickers)
+    returns = df_prices.pct_change().dropna()
+    mean_returns = returns.mean()
+    cov_matrix = returns.cov()
+
+    synthetic_data = []
+    for _ in range(num_portfolios):
+        weights = np.random.random(len(tickers))
+        weights /= np.sum(weights)
+        er, sigma, beta = calculate_portfolio_metrics(weights, mean_returns, cov_matrix)
+        synthetic_data.append({
+            "weights": weights,
+            "er_p": er,
+            "sigma_p": sigma,
+            "beta_p": beta
         })
-    return pd.DataFrame(rows)
 
-def select_risk_profiles(df):
-    high = df.sort_values("er_p", ascending=False).head(1)
-    low = df.sort_values("sigma_p", ascending=True).head(1)
-    med_idx = ((df["er_p"] - df["er_p"].median())**2 + (df["sigma_p"] - df["sigma_p"].median())**2).idxmin()
-    medium = df.loc[[med_idx]]
+    return pd.DataFrame(synthetic_data)
+
+def select_risk_profiles(synth_df):
+    """Select high/high, medium/medium, low/low risk profiles from synthetic dataset."""
+    high = synth_df.sort_values("er_p", ascending=False).iloc[0]
+    low = synth_df.sort_values("sigma_p", ascending=True).iloc[0]
+
+    median_idx = ((synth_df["sigma_p"] - synth_df["sigma_p"].median()).abs() +
+                  (synth_df["er_p"] - synth_df["er_p"].median()).abs()).idxmin()
+    medium = synth_df.loc[median_idx]
+
     return high, medium, low
 
-# ---------------- main compute ----------------
-def compute(years_lookback, tickers_df):
-    tickers_df["ticker"] = tickers_df["ticker"].str.upper().str.strip()
-    tickers = tickers_df["ticker"].tolist()
-    amounts = tickers_df["amount_usd"].tolist()
-    rf_ann = fetch_fred_yield_annual(fred_series_for_horizon(5))
-    betas, covA, erp_ann, sigma_mkt = estimate_all_moments_aligned(tickers + [MARKET_TICKER], years_lookback, rf_ann)
-    weights = {t: a for t, a in zip(tickers, amounts)}
-    beta_p, er_p, sigma_p = portfolio_stats(weights, covA, betas, rf_ann, erp_ann)
-    eff_sigma = efficient_same_sigma(sigma_p, rf_ann, erp_ann, sigma_mkt)
-    eff_return = efficient_same_return(er_p, rf_ann, erp_ann, sigma_mkt)
-    synth_df = build_synthetic_dataset(tickers + [MARKET_TICKER], years_lookback, rf_ann, erp_ann)
+def find_efficient_same_sigma(user_er, user_sigma, synth_df):
+    """Find portfolio with same sigma but highest return."""
+    close_sigma = synth_df[np.isclose(synth_df["sigma_p"], user_sigma, atol=0.002)]
+    if close_sigma.empty:
+        return synth_df.iloc[0]
+    return close_sigma.sort_values("er_p", ascending=False).iloc[0]
+
+def find_efficient_same_return(user_er, user_sigma, synth_df):
+    """Find portfolio with same return but lowest sigma."""
+    close_return = synth_df[np.isclose(synth_df["er_p"], user_er, atol=0.002)]
+    if close_return.empty:
+        return synth_df.iloc[0]
+    return close_return.sort_values("sigma_p", ascending=True).iloc[0]
+# -------------------
+# Main compute function
+# -------------------
+
+def compute(user_tickers):
+    # Convert comma-separated string into ticker list
+    tickers = [t.strip().upper() for t in user_tickers.split(",") if t.strip()]
+    if len(tickers) < 2:
+        return "Please enter at least two tickers.", None
+
+    # Fetch live data & compute user portfolio metrics (equal weights for now)
+    df_prices = fetch_live_data(tickers)
+    if df_prices.empty:
+        return "Could not fetch data. Check tickers.", None
+
+    returns = df_prices.pct_change().dropna()
+    mean_returns = returns.mean()
+    cov_matrix = returns.cov()
+    user_weights = np.ones(len(tickers)) / len(tickers)
+    user_er, user_sigma, user_beta = calculate_portfolio_metrics(user_weights, mean_returns, cov_matrix)
+
+    # Generate synthetic dataset
+    synth_df = generate_synthetic_portfolios(tickers, num_portfolios=1000)
+
+    # Select profiles
+    eff_sigma = find_efficient_same_sigma(user_er, user_sigma, synth_df)
+    eff_return = find_efficient_same_return(user_er, user_sigma, synth_df)
     high, medium, low = select_risk_profiles(synth_df)
-    return {
-        "user": (beta_p, er_p, sigma_p, weights),
-        "eff_sigma": eff_sigma,
-        "eff_return": eff_return,
-        "high": high,
-        "medium": medium,
-        "low": low
+
+    # Prepare results DataFrame
+    portfolios = {
+        "User Portfolio": [user_er, user_sigma, user_beta, user_weights],
+        "Efficient (Same Sigma)": [eff_sigma.er_p, eff_sigma.sigma_p, eff_sigma.beta_p, eff_sigma.weights],
+        "Efficient (Same Return)": [eff_return.er_p, eff_return.sigma_p, eff_return.beta_p, eff_return.weights],
+        "High Risk / High Return": [high.er_p, high.sigma_p, high.beta_p, high.weights],
+        "Medium Risk / Medium Return": [medium.er_p, medium.sigma_p, medium.beta_p, medium.weights],
+        "Low Risk / Low Return": [low.er_p, low.sigma_p, low.beta_p, low.weights],
     }
 
-# ---------------- UI ----------------
+    df_out = pd.DataFrame(portfolios, index=["Expected Return", "Sigma", "Beta", "Weights"])
+
+    return df_out.to_markdown(), df_out
+
+# -------------------
+# Gradio Interface
+# -------------------
+
 with gr.Blocks() as demo:
-    gr.Markdown("## Efficient Portfolio Advisor with Synthetic Risk Profiles")
-    table = gr.Dataframe(headers=["ticker", "amount_usd"], datatype=["str", "number"], row_count=3)
-    lookback = gr.Slider(1, 10, value=DEFAULT_LOOKBACK_YEARS, step=1, label="Lookback years")
-    run_btn = gr.Button("Compute")
-    output = gr.Textbox(label="Results")
-    def run_app(lookback, table):
-        res = compute(lookback, table)
-        return str(res)
-    run_btn.click(fn=run_app, inputs=[lookback, table], outputs=[output])
+    gr.Markdown("## Portfolio Optimizer and Risk Profiles")
+    tickers_input = gr.Textbox(label="Enter tickers (comma separated)", placeholder="AAPL, MSFT, GOOG")
+    output_md = gr.Markdown()
+    output_df = gr.Dataframe(headers=["Portfolio", "Value"], interactive=False)
+
+    def run_and_display(tickers):
+        md, df = compute(tickers)
+        if df is None:
+            return md, None
+        return md, df
+
+    run_btn = gr.Button("Run Analysis")
+    run_btn.click(fn=run_and_display, inputs=tickers_input, outputs=[output_md, output_df])
 
 if __name__ == "__main__":
-    ensure_data_dir()
     demo.launch()