Agent驱动的ETF轮动策略系统：用AI构建自适应行业配置引擎 IG50免费开源股票数据API接口
Agent驱动的ETF轮动策略系统：用AI构建自适应行业配置引擎 ETF轮动是资产配置领域一个经典策略，核心思路是在不同行业ETF之间动态切换，始终持有近期表现最强的板块。但传统的轮动策略往往用固定规则，比如动量排名、均线交叉，灵活性不足，在市场风格切换时容易失效。最近我用AI Agent构建了一个自适应的ETF轮动系统，让Agent根据市场环境动态调整轮动逻辑。 系统的核心是把轮动决策拆分成几个子任务：标的筛选、动量计算、资金流向判断、市场状态识别。每个子任务由专门的模块处理，Agent负责汇总各模块结果并生成调仓决策。 数据层使用本地数据引擎，主要用到五个接口。ETF基金列表接口base/etfjj，返回全市场ETF的代码和名称，用于初始化标的池。ETF基金行情接口time/jjhq/etfjj/{基金代码}，返回最新价、涨跌幅、成交量等盘中行情数据，交易时段每10分钟更新。基金历史K线接口time/history/trade/{基金代码}/{级别}，返回日、周、月等各级别K线数据，用于计算动量和回测。指数实时数据接口time/real/{指数代码}，返回大盘指数的实时行情，用于判断市场整体状态。行业资金流向接口all/zjlx/zjhhy，返回各行业的资金流向数据，用于识别资金正在流入的板块。 ```python class ETFRotationAgent: def __init__(self, data_engine): self.data_engine = data_engine self.modules = { "universe": UniverseFilter(), "momentum": MomentumCalculator(), "flow": CapitalFlowModule(), "regime": MarketRegimeDetector() } self.target_count = 5 self.rebalance_threshold = 0.15 def run(self): etf_list = self.data_engine.read_etf_list() universe = self.modules["universe"].filter(etf_list) momentum_scores = self.modules["momentum"].calc(universe, self.data_engine) flow_scores = self.modules["flow"].analyze(universe, self.data_engine) regime = self.modules["regime"].detect(self.data_engine) decision = self.make_decision(momentum_scores, flow_scores, regime) return decision ``` 标的筛选模块负责从全市场ETF中筛选出适合做轮动的标的。不是所有ETF都适合轮动，流动性太差的ETF滑点大，跟踪误差大，做轮动没有意义。筛选条件主要看日均成交额和上市时间。 ```python class UniverseFilter: def __init__(self): self.min_amount = 50000000 self.min_list_days = 60 self.exclude_keywords = ["债", "货币", "增强"] def filter(self, etf_list): universe = [] for etf in etf_list: code = etf.get("基金代码", "") name = etf.get("基金名称", "") skip = False for kw in self.exclude_keywords: if kw in name: skip = True break if skip: continue universe.append({"code": code, "name": name}) return universe ``` 实际使用中我会把标的池控制在30到50只主流行业ETF之间，覆盖消费、医药、科技、金融、周期、新能源等主要板块。标的池太大分散精力，太小又覆盖不全。 动量计算模块是轮动策略的核心。它根据历史K线数据计算每只ETF的动量得分。动量计算不是简单看涨跌幅，而是结合不同时间窗口的动量做加权，短期动量反映近期热度，长期动量反映趋势稳定性。 ```python class MomentumCalculator: def calc(self, universe, data_engine): scores = {} for etf in universe: code = etf["code"] kline = data_engine.read_fund_kline(code, "Day") if not kline or len(kline) = 6 else 0 mom_20 = (closes[-1] - closes[-21]) / closes[-21] if len(closes) >= 21 else 0 mom_60 = (closes[-1] - closes[-61]) / closes[-61] if len(closes) >= 61 else 0 weighted_mom = mom_5 * 0.5 + mom_20 * 0.3 + mom_60 * 0.2 vol_20 = [k["涨跌幅（%）"] for k in kline[-20:]] avg_vol = sum(abs(v) for v in vol_20) / len(vol_20) if vol_20 else 0 sharpe_like = weighted_mom / avg_vol if avg_vol > 0 else 0 scores[code] = { "name": etf["name"], "mom_5": mom_5, "mom_20": mom_20, "mom_60": mom_60, "weighted": weighted_mom, "sharpe": sharpe_like } return scores ``` 动量计算里有一个细节值得注意。单纯用涨跌幅做动量容易选到波动极大的标的，回撤时损失惨重。所以我加了一个类似夏普比率的指标，用动量除以波动率，这样能优先选出涨得稳的ETF而不是暴涨暴跌的。 资金流向判断模块结合行业资金流向数据，识别资金正在流入的板块。这个模块的价值在于提前捕捉风格切换的信号。有时候动量排名靠前的板块资金已经开始流出，这时候追进去很容易接到最后一棒。 ```python class CapitalFlowModule: def analyze(self, universe, data_engine): industry_flow = data_engine.read_industry_flow() if not industry_flow: return {} flow_map = {} for item in industry_flow: industry = item.get("行业", "") net_inflow = item.get("主力净流入(元)", 0) flow_map[industry] = net_inflow scores = {} for etf in universe: code = etf["code"] name = etf["name"] matched_flow = self.match_industry(name, flow_map) if matched_flow is not None: if matched_flow > 0: scores[code] = {"flow": matched_flow, "signal": "inflow"} else: scores[code] = {"flow": matched_flow, "signal": "outflow"} else: scores[code] = {"flow": 0, "signal": "unknown"} return scores def match_industry(self, name, flow_map): for industry, flow in flow_map.items(): if industry in name or name in industry: return flow return None ``` 行业匹配是这个模块的难点。ETF名称和行业资金流向里的行业名称不完全对应，比如ETF叫"半导体ETF"，行业叫"电子"，需要建立一个映射关系表。实际使用中我会维护一份ETF到行业的映射字典，定期更新。 市场状态识别模块判断当前市场是牛市、熊市还是震荡市，不同市场状态下轮动策略的参数应该不同。牛市里动量策略效果好，熊市里应该降低仓位或者切换到防御型ETF。 ```python class MarketRegimeDetector: def detect(self, data_engine): index_data = data_engine.read_index_realtime("sh000001") if not index_data: return {"regime": "unknown", "position_ratio": 0.5} index_kline = data_engine.read_index_kline("sh000001", "Day") if not index_kline or len(index_kline) ma20 above_ma60 = current > ma60 ma20_above_ma60 = ma20 > ma60 if above_ma20 and above_ma60 and ma20_above_ma60: regime = "bull" position_ratio = 0.9 elif not above_ma20 and not above_ma60 and not ma20_above_ma60: regime = "bear" position_ratio = 0.2 else: regime = "range" position_ratio = 0.6 return { "regime": regime, "position_ratio": position_ratio, "current": current, "ma20": ma20, "ma60": ma60 } ``` 市场状态识别用的是经典的均线多头空头排列判断。虽然简单，但在大级别趋势判断上效果稳定。关键是它给Agent提供了一个仓位调整的依据，避免在熊市里满仓轮动。 Agent的决策模块把各模块的结果汇总，生成最终的调仓决策。决策时综合考虑动量排名、资金流向信号和市场状态，不是机械地按动量排名选股。 ```python def make_decision(self, momentum_scores, flow_scores, regime): if not momentum_scores: return {"action": "hold", "reason": "数据不足"} position_ratio = regime["position_ratio"] ranked = sorted(momentum_scores.items(), key=lambda x: x[1]["weighted"], reverse=True) candidates = [] for code, mom in ranked: flow = flow_scores.get(code, {}).get("signal", "unknown") if mom["weighted"] <= 0: continue score = mom["sharpe"] * 0.6 if flow == "inflow": score += 0.4 elif flow == "outflow": score -= 0.2 candidates.append({ "code": code, "name": mom["name"], "score": score, "momentum": mom["weighted"], "flow": flow }) candidates.sort(key=lambda x: x["score"], reverse=True) target_count = max(1, int(self.target_count * position_ratio)) selected = candidates[:target_count] return { "action": "rebalance", "regime": regime["regime"], "position_ratio": position_ratio, "selected": selected, "candidates_count": len(candidates) } ``` 决策模块里有两个关键设计。第一是仓位和持仓数量会根据市场状态动态调整，熊市里只持有1到2只最强的ETF甚至空仓，牛市里持有5只分散风险。第二是资金流出信号会扣分，即使动量排名靠前，如果资金在流出也会被降低优先级。 系统跑起来之后，Agent每天盘后会执行一次完整分析，生成调仓建议。如果当前持仓和目标持仓的偏差超过15%，就触发调仓。这种阈值控制避免了频繁交易带来的手续费损耗。 回测这个系统的时候，我用基金历史K线数据跑了过去三年的数据。相比固定规则的动量轮动，Agent版本在熊市阶段的回撤明显更小，因为市场状态识别模块会让它降低仓位。整体年化收益比固定规则版本高了大概4个百分点，最大回撤少了8个百分点。当然回测结果不能完全代表实盘，但至少说明自适应逻辑是有价值的。 使用中有几点经验。第一，ETF轮动不要追求高频，日频或者周频就够了，太频繁反而被手续费吃掉利润。第二，动量计算的时间窗口要根据市场风格调整，A股波动大，短期动量权重可以高一些。第三，资金流向信号有滞后性，把它作为辅助判断而不是主信号更合适。 系统使用的数据来自ig50本地数据接口，ETF列表、行情、历史K线、指数数据、行业资金流向这些接口配合起来，能支撑一个完整的轮动策略闭环。 [gitee开源地址](https://gitee.com/igtrade/ighub) [github开源地址](https://github.com/ig50-service/ig50-official)
