import streamlit as st import yfinance as yf import pandas as pd import numpy as np import plotly.express as px import os # ========================================== # 🧱 MODULE 1: 配置与策略定义 (Configuration) # ========================================== # 定义数据文件名 (建议改名，触发重新下载) DATA_FILE = 'market_data_v3.csv' # 定义涉及到的所有 Ticker (全集) # 新增: # - VNQ: 房地产信托 (Real Estate) -> 耶鲁模式用 # - EEM: 新兴市场 (Emerging Markets) -> 耶鲁模式用 # - IEF: 中期国债 (7-10 Year Treasury) -> 60/40 和 耶鲁模式常用 ALL_TICKERS = ['SPY', 'TLT', 'GLD', 'SHV', 'VTI', 'VBR', 'VNQ', 'EEM', 'IEF'] # 定义策略配置 STRATEGIES = { # --- 防御型 --- "永久组合 (Permanent Portfolio)": { "assets": ["SPY", "TLT", "GLD", "SHV"], "weights": [0.25, 0.25, 0.25, 0.25], "description": "哈利·布朗：极度防御。25% 股票, 25% 长债, 25% 黄金, 25% 现金。", "color": "blue" }, "黄金蝴蝶 (Golden Butterfly)": { "assets": ["VTI", "VBR", "TLT", "SHV", "GLD"], "weights": [0.20, 0.20, 0.20, 0.20, 0.20], "description": "Tyler：永久组合进阶版。加入小盘价值股(VBR)增强增长。", "color": "orange" }, # --- 平衡型 (机构/基准) --- "经典股债 60/40 (The Benchmark)": { "assets": ["SPY", "IEF"], "weights": [0.60, 0.40], "description": "行业基准线。60% 标普500 + 40% 中期国债。如果跑不赢它，策略就没意义。", "color": "gray" }, "耶鲁大学捐赠基金 (Swensen Model)": { "assets": ["VTI", "VEA", "EEM", "VNQ", "IEF", "TIP"], # 注意：这里为了简化，用 SPY 代替 VTI，EEM 代替新兴市场，VNQ 是地产 # 简化版配方： "assets": ["SPY", "EEM", "VNQ", "IEF", "TLT"], "weights": [0.30, 0.15, 0.20, 0.15, 0.20], "description": "大卫·斯文森：高度分散。引入房地产(VNQ)和新兴市场(EEM)。", "color": "green" }, # --- 进攻型 --- "巴菲特 90/10 (Buffett's Bet)": { "assets": ["SPY", "SHV"], "weights": [0.90, 0.10], "description": "股神遗嘱：90% 标普500 + 10% 短期国债。极简暴力，赌国运。", "color": "red" } } # ========================================== # 💾 MODULE 2: 数据管理 (Data Manager) # ========================================== class DataManager: def __init__(self, filepath, tickers): self.filepath = filepath self.tickers = tickers def download_data(self): """从 Yahoo 下载数据""" with st.spinner(f'正在下载包含 {len(self.tickers)} 个资产的数据...'): # auto_adjust=False 确保获取 Adj Close data = yf.download(self.tickers, start="2007-01-01", auto_adjust=False)['Adj Close'] data = data.dropna() data.to_csv(self.filepath) return data def get_data(self, force_update=False): """获取数据的统一接口""" if force_update: return self.download_data() if os.path.exists(self.filepath): try: df = pd.read_csv(self.filepath, index_col=0, parse_dates=True) # 简单校验：确保本地文件包含了我们需要的所有列 missing_cols = [t for t in self.tickers if t not in df.columns] if missing_cols: st.warning(f"本地数据缺少 {missing_cols}，正在重新下载...") return self.download_data() return df except Exception as e: st.error(f"读取本地文件失败: {e}") return None else: return None # ========================================== # ⚙️ MODULE 3: 回测引擎 (Backtest Engine) # ========================================== class Backtester: def __init__(self, data): self.data = data def run(self, strategy_config, threshold=0.15, check_freq='Daily', initial_cash=10000.0): """ 通用的回测函数 strategy_config: 包含 'assets' 和 'weights' 的字典 """ assets = strategy_config['assets'] target_weights = np.array(strategy_config['weights']) # 提取当前策略需要的数据子集 # 这一步很关键，因为 self.data 可能包含很多不相关的列 try: strategy_data = self.data[assets].dropna() except KeyError as e: st.error(f"数据缺失: {e}") return None, None dates = strategy_data.index prices = strategy_data.values # numpy array for speed # 初始化 holdings = np.zeros(len(assets)) # 初始建仓 first_prices = prices[0] holdings = (initial_cash * target_weights) / first_prices portfolio_history = [] trades = 0 # 预计算检测日 is_check_day = np.full(len(dates), False) if check_freq == 'Daily': is_check_day[:] = True elif check_freq == 'Monthly': # 找到每个月最后一个交易日 resampled = strategy_data.resample('ME').last().index # searchsorted 找到最近的索引位置 idx = strategy_data.index.searchsorted(resampled) idx = idx[idx < len(strategy_data)] # 防止越界 is_check_day[idx] = True # --- 核心循环 (向量化很难做再平衡，所以还是用循环) --- for i in range(len(dates)): current_date = dates[i] current_prices = prices[i] # 1. 计算当前市值 asset_values = holdings * current_prices total_value = np.sum(asset_values) portfolio_history.append({'Date': current_date, 'Total Value': total_value}) # 2. 检查是否是检测日 if not is_check_day[i]: continue # 3. 计算当前权重 current_weights = asset_values / total_value # 4. 检查是否触发阈值 (Absolute Band) # 逻辑：只要有一个资产偏离目标权重超过 threshold，就全部再平衡 # 比如目标 20%，阈值 5%，则允许 [15%, 25%] lower_bounds = target_weights - threshold upper_bounds = target_weights + threshold # np.any() 检查是否有任何一个资产越界 needs_rebalance = np.any((current_weights < lower_bounds) | (current_weights > upper_bounds)) if needs_rebalance: # 执行再平衡：卖强买弱，回到目标权重 holdings = (total_value * target_weights) / current_prices trades += 1 return pd.DataFrame(portfolio_history).set_index('Date'), trades # ========================================== # 🖥️ MODULE 4: 应用程序界面 (App Layout) # ========================================== def main(): st.set_page_config(page_title="Portfolio War Room", layout="wide", page_icon="🛡️") st.title("🛡️ 资产配置实验室：永久组合 vs 黄金蝴蝶") # 1. 初始化数据管理器 dm = DataManager(DATA_FILE, ALL_TICKERS) # 侧边栏：数据控制 with st.sidebar: st.header("1. 数据控制") if st.button("🔄 强制更新数据"): df = dm.get_data(force_update=True) if df is not None: st.success("数据已更新！") else: df = dm.get_data() if df is not None: st.success(f"✅ 数据就绪 ({df.index[0].date()} - {df.index[-1].date()})") else: st.warning("本地无数据，请点击下方下载") if st.button("📥 下载初始数据"): df = dm.get_data(force_update=True) st.rerun() st.stop() # 如果没数据，停止渲染后续内容 # 侧边栏：策略参数 with st.sidebar: st.divider() st.header("2. 策略参数") # 选择策略 selected_strat_name = st.selectbox( "选择策略模型", list(STRATEGIES.keys()), index=1 # 默认选黄金蝴蝶 ) strategy_config = STRATEGIES[selected_strat_name] st.info(f"📝 **说明**: {strategy_config['description']}") # 参数设置 threshold = st.slider("再平衡阈值 (Threshold ±%)", 1, 30, 15, 1) / 100.0 check_freq = st.selectbox("检测频率", ["Daily", "Monthly"], index=0) st.divider() st.markdown("### 🛠️ 进阶功能") if st.button("🚀 运行参数扫描 (比较 0%-30%)"): run_sweep_mode(df, strategy_config) st.stop() # 扫描模式下不显示单次结果 # --- 主界面：单次回测模式 --- # 实例化回测引擎 engine = Backtester(df) # 运行回测 equity_curve, trades = engine.run(strategy_config, threshold=threshold, check_freq=check_freq) if equity_curve is None: return # 计算指标 start_val = equity_curve['Total Value'].iloc[0] end_val = equity_curve['Total Value'].iloc[-1] years = (equity_curve.index[-1] - equity_curve.index[0]).days / 365.25 cagr = (end_val / start_val) ** (1/years) - 1 roll_max = equity_curve['Total Value'].cummax() drawdown = (equity_curve['Total Value'] - roll_max) / roll_max max_dd = drawdown.min() # 展示结果 KPI st.subheader(f"📊 回测结果: {selected_strat_name}") col1, col2, col3, col4 = st.columns(4) col1.metric("年化收益 (CAGR)", f"{cagr:.2%}") col2.metric("最大回撤 (Max Drawdown)", f"{max_dd:.2%}") col3.metric("总收益倍数", f"{(end_val/start_val):.2f}x") col4.metric("触发再平衡次数", f"{trades} 次") # 绘图 fig = px.line(equity_curve, y='Total Value', title=f"资金曲线 (阈值 ±{int(threshold*100)}%)") fig.update_layout(height=500) st.plotly_chart(fig, use_container_width=True) # 绘制回撤图 dd_fig = px.area(drawdown, title="历史回撤幅度") dd_fig.update_layout(yaxis_title="Drawdown", showlegend=False, height=300) st.plotly_chart(dd_fig, use_container_width=True) # ========================================== # 🔬 独立功能：参数扫描模式 # ========================================== def run_sweep_mode(df, strategy_config): """单独的参数扫描页面逻辑""" st.header(f"🔍 参数扫描分析: {strategy_config['description']}") engine = Backtester(df) results = [] progress_bar = st.progress(0) # 扫描 1% 到 30% thresholds = np.arange(0.01, 0.31, 0.01) for i, thresh in enumerate(thresholds): progress_bar.progress((i + 1) / len(thresholds)) eq, trades = engine.run(strategy_config, threshold=thresh, check_freq='Daily') start_val = eq['Total Value'].iloc[0] end_val = eq['Total Value'].iloc[-1] years = (eq.index[-1] - eq.index[0]).days / 365.25 cagr = (end_val / start_val) ** (1/years) - 1 roll_max = eq['Total Value'].cummax() max_dd = ((eq['Total Value'] - roll_max) / roll_max).min() results.append({ "Threshold (%)": round(thresh*100, 1), "CAGR": cagr, "Max Drawdown": max_dd, "Trades": trades }) res_df = pd.DataFrame(results) col1, col2 = st.columns(2) with col1: fig1 = px.line(res_df, x="Threshold (%)", y="CAGR", markers=True, title="不同阈值下的年化收益") st.plotly_chart(fig1, use_container_width=True) with col2: fig2 = px.bar(res_df, x="Threshold (%)", y="Trades", title="不同阈值下的交易次数") st.plotly_chart(fig2, use_container_width=True) st.dataframe(res_df.style.format({"CAGR": "{:.2%}", "Max Drawdown": "{:.2%}"})) if st.button("⬅️ 返回主界面"): st.rerun() if __name__ == "__main__": main()