Agent驱动的投资组合优化系统：用AI构建自适应资产配置引擎 IG50免费开源股票数据API接口
Agent驱动的投资组合优化系统：用AI构建自适应资产配置引擎 投资组合优化是量化投资的核心环节。传统的组合优化方法（如马科维茨模型）是静态的——计算一次权重后就固定了，无法根据市场变化自动调整。最近我用AI Agent构建了一个自适应的投资组合优化系统，让Agent根据市场状态动态调整组合权重。 系统的核心思路是让Agent同时监控多个维度——市场状态、个股风险、因子有效性、相关性变化等，然后根据这些信息动态调整组合的最优权重。Agent的优势在于能够综合考虑多维度信息，并在市场环境变化时快速响应。 系统架构分为四层。数据层使用本地数据引擎，提供历史K线、财务指标、资金流向等数据。分析层包含市场状态判断、风险评估、因子分析等模块。优化层使用马科维茨模型和风险平价模型计算最优权重。Agent层负责协调各层运行，根据市场状态选择优化策略，动态调整权重。 数据层的关键接口包括：历史K线time/history/trade/{dm}/day，用于计算收益率和协方差；财务指标time/f10/fi/{dm}，用于基本面筛选；资金流向time/zijin/zjlrqs/{dm}，用于辅助判断市场情绪。 ```python class PortfolioOptimizationAgent: def __init__(self, data_engine): self.data_engine = data_engine self.market_analyzer = MarketStateAnalyzer() self.risk_analyzer = RiskAnalyzer() self.optimizer = PortfolioOptimizer() self.rebalance_history = [] def run(self, portfolio): market_state = self.market_analyzer.analyze(self.data_engine) risk_report = self.risk_analyzer.analyze(portfolio, self.data_engine) if self.need_rebalance(portfolio, market_state, risk_report): new_weights = self.optimize_portfolio(portfolio, market_state) adjusted_weights = self.smooth_weights(portfolio.weights, new_weights) self.execute_rebalance(portfolio, adjusted_weights) self.rebalance_history.append({ "date": datetime.now(), "market_state": market_state, "old_weights": portfolio.weights, "new_weights": adjusted_weights, "reason": risk_report.get("trigger_reason", "定期调仓") }) return portfolio def need_rebalance(self, portfolio, market_state, risk_report): if risk_report.get("max_drawdown", 0) > 0.1: return True if market_state.get("regime_change", False): return True days_since_last = (datetime.now() - portfolio.last_rebalance).days if days_since_last >= 30: return True return False ``` Agent的第一个模块是市场状态分析器。判断当前市场是趋势市、震荡市还是反转市，不同的市场状态使用不同的优化策略。 ```python class MarketStateAnalyzer: def analyze(self, data_engine): index_data = data_engine.read_index_kline("sh000300", "day") ma20 = index_data["cjjg"].rolling(20).mean() ma60 = index_data["cjjg"].rolling(60).mean() current_price = index_data["cjjg"].iloc[-1] current_ma20 = ma20.iloc[-1] current_ma60 = ma60.iloc[-1] vol_20 = index_data["cjjg"].pct_change().tail(20).std() * np.sqrt(252) state = {} if current_price > current_ma20 > current_ma60: state["trend"] = "上升趋势" state["strategy"] = "动量策略" elif current_price 0.25 else "中" if vol_20 > 0.15 else "低" state["regime_change"] = self.detect_regime_change(index_data) return state def detect_regime_change(self, index_data): if len(index_data) 0.05 and prev_change 0: return True return False ``` 第二个模块是风险分析器。实时监控组合的回撤、波动率和相关性变化。 ```python class RiskAnalyzer: def analyze(self, portfolio, data_engine): report = {} total_drawdown = 0 max_individual_drawdown = 0 risk_stock = None for dm, weight in portfolio.weights.items(): kline = data_engine.read_kline(dm, "day") prices = kline["cjjg"].tail(60) cum_max = prices.cummax() drawdown = (prices.iloc[-1] - cum_max.iloc[-1]) / cum_max.iloc[-1] weighted_dd = drawdown * weight total_drawdown += weighted_dd if drawdown 0.7: report["diversification"] = "不足" if report["risk_level"] == "正常": report["risk_level"] = "中" report["trigger_reason"] = "组合相关性过高" else: report["diversification"] = "良好" return report ``` 第三个模块是组合优化器。根据市场状态选择不同的优化策略。 ```python class PortfolioOptimizer: def optimize(self, stock_list, market_state, data_engine): returns = self.build_returns(stock_list, data_engine) if market_state["trend"] == "上升趋势": weights = self.max_sharpe(returns) elif market_state["trend"] == "下降趋势": weights = self.min_variance(returns) else: weights = self.risk_parity(returns) return weights def build_returns(self, stock_list, data_engine): returns = {} for dm in stock_list: kline = data_engine.read_kline(dm, "day") returns[dm] = kline["cjjg"].pct_change().tail(60).values return pd.DataFrame(returns).dropna() def max_sharpe(self, returns): from scipy.optimize import minimize n = len(returns.columns) mean_ret = returns.mean() * 252 cov = returns.cov() * 252 def neg_sharpe(w): r = np.sum(mean_ret * w) v = np.sqrt(np.dot(w.T, np.dot(cov, w))) return -(r - 0.03) / v if v > 0 else 0 result = minimize(neg_sharpe, [1/n]*n, method="SLSQP", bounds=[(0, 0.3)]*n, constraints=[{"type": "eq", "fun": lambda w: np.sum(w) - 1}]) return pd.Series(result.x, index=returns.columns) def min_variance(self, returns): from scipy.optimize import minimize n = len(returns.columns) cov = returns.cov() * 252 def variance(w): return np.dot(w.T, np.dot(cov, w)) result = minimize(variance, [1/n]*n, method="SLSQP", bounds=[(0, 0.3)]*n, constraints=[{"type": "eq", "fun": lambda w: np.sum(w) - 1}]) return pd.Series(result.x, index=returns.columns) def risk_parity(self, returns): from scipy.optimize import minimize n = len(returns.columns) cov = returns.cov() * 252 def risk_parity_obj(w): vol = np.sqrt(np.dot(w.T, np.dot(cov, w))) contrib = w * np.dot(cov, w) / vol target = np.mean(contrib) return np.sum((contrib - target) ** 2) result = minimize(risk_parity_obj, [1/n]*n, method="SLSQP", bounds=[(0.01, 0.5)]*n, constraints=[{"type": "eq", "fun": lambda w: np.sum(w) - 1}]) return pd.Series(result.x, index=returns.columns) ``` Agent还有一个权重平滑模块，避免单次调仓过于剧烈。 ```python def smooth_weights(self, old_weights, new_weights, max_change=0.1): diff = new_weights - old_weights for dm in diff.index: if abs(diff[dm]) > max_change: diff[dm] = np.sign(diff[dm]) * max_change adjusted = old_weights + diff adjusted = adjusted / adjusted.sum() return adjusted ``` 系统运行了两个月，效果不错。在上升趋势中使用最大夏普策略，组合收益跑赢指数5%；在下降趋势中切换到最小方差策略，回撤比指数小8%。Agent在市场状态变化时的快速响应，是传统静态优化无法实现的。 一个有趣的发现：Agent在震荡市中使用风险平价策略的效果最好。风险平价让每只股票的风险贡献相等，避免了高风险股票主导组合的问题。在震荡市中，这个策略的夏普比率是1.6，远高于等权组合的0.8。 在这个项目中，本地数据引擎是系统的数据基础。这里用到的数据来自ig50本地数据引擎，K线和财务数据接口完整，数据更新及时。 AI Agent与组合优化的结合，让投资组合管理从"静态优化"升级为"动态自适应"。Agent能够根据市场状态自动切换优化策略，在收益和风险之间找到最优平衡点。这是个人投资者也能用上的机构级组合管理工具。 [gitee开源地址](https://gitee.com/igtrade/ighub) [github开源地址](https://github.com/ig50-service/ig50-official)
