Agent驱动的十大股东追踪系统：用AI挖掘机构持仓变动的领先信号 IG50免费开源股票数据API接口
Agent驱动的十大股东追踪系统：用AI挖掘机构持仓变动的领先信号 十大股东数据是追踪机构动向的重要窗口。国家队、社保基金、QFII出现在十大股东里，往往意味着机构看好这只股票。但十大股东数据维度多、更新频率低，人工分析效率低。最近我用AI Agent构建了一个十大股东追踪系统，让Agent自动从股东数据中挖掘机构持仓变动的信号。 系统把十大股东分析拆解成几个子任务：机构身份识别、持仓变动追踪、股东人数分析、持股集中度评估。每个子任务交给专门的模块处理，Agent负责汇总各模块结果生成综合判断。 数据层使用本地数据引擎，主要用到四个接口。十大股东接口time/f10/zygd/{股票代码}，返回截止日期、公告日期、股东总数、平均持股、十大股东列表。十大流通股东接口time/f10/zygdlt/{股票代码}，结构类似但只看流通股东。股东变化趋势接口time/f10/gdbh/{股票代码}，返回各季度的股东户数变化。基金持股接口time/f10/jjcg/{股票代码}，返回基金持仓数据。 ```python class ShareholderAgent: def __init__(self, data_engine): self.data_engine = data_engine self.modules = { "institution": InstitutionDetector(), "change": HoldingChangeTracker(), "count": ShareholderCountAnalyzer(), "concentration": ConcentrationEvaluator() } self.institution_keywords = { "national": ["汇金", "证金", "国家队"], "social": ["社保"], "qfii": ["QFII", "瑞士联合", "摩根", "高盛", "富达"], "insurance": ["保险", "人寿", "平安"], "fund": ["基金", "资管", "信托"] } def analyze(self, dm): zygd = self.data_engine.read_zygd(dm) zygdlt = self.data_engine.read_zygdlt(dm) gdbh = self.data_engine.read_gdbh(dm) jjcg = self.data_engine.read_jjcg(dm) if not zygd: return {"dm": dm, "score": 50, "signal": "无股东数据"} context = { "dm": dm, "zygd": zygd, "zygdlt": zygdlt, "gdbh": gdbh, "jjcg": jjcg } results = [] for module in self.modules.values(): result = module.analyze(context) if result: results.append(result) decision = self.synthesize(results, dm) return decision ``` 第一个模块是机构身份识别器。它分析十大股东中有没有国家队、社保、QFII等重要机构。 ```python class InstitutionDetector: def analyze(self, context): zygd = context["zygd"] if not zygd: return None latest = zygd[0] holders = latest.get("ZygdSdgd", []) agent = ShareholderAgent.__new__(ShareholderAgent) keywords = { "national": ["汇金", "证金", "国家队"], "social": ["社保"], "qfii": ["QFII", "瑞士联合", "摩根", "高盛", "富达"], "insurance": ["保险", "人寿", "平安"], "fund": ["基金", "资管", "信托"] } institutions_found = {} for holder in holders: name = holder.get("股东名称", "") for inst_type, kws in keywords.items(): for kw in kws: if kw in name: if inst_type not in institutions_found: institutions_found[inst_type] = [] institutions_found[inst_type].append(name) break signal = { "module": "institution", "institutions": institutions_found } if "national" in institutions_found: signal["signal"] = "国家队进驻，中长期看好" signal["score"] = 72 elif "social" in institutions_found: signal["signal"] = "社保基金进驻" signal["score"] = 70 elif "qfii" in institutions_found: signal["signal"] = "QFII进驻" signal["score"] = 68 elif "fund" in institutions_found or "insurance" in institutions_found: signal["signal"] = "机构资金进驻" signal["score"] = 62 else: signal["signal"] = "十大股东以个人或普通企业为主" signal["score"] = 48 return signal ``` 第二个模块是持仓变动追踪器。它对比相邻季度的十大股东名单，找出新进和退出的股东。 ```python class HoldingChangeTracker: def analyze(self, context): zygd = context["zygd"] if not zygd or len(zygd) 0.15: signal["signal"] = "股东人数大幅增加，筹码分散" signal["score"] = 38 else: signal["signal"] = "股东人数变化不大" signal["score"] = 50 return signal ``` 第四个模块是持股集中度评估器。它通过平均持股和十大股东持股比例评估筹码集中度。 ```python class ConcentrationEvaluator: def analyze(self, context): zygd = context["zygd"] if not zygd: return None latest = zygd[0] avg_holding = latest.get("平均持股", 0) total_count = latest.get("股东总数", 0) holders = latest.get("ZygdSdgd", []) top10_ratio = sum(h.get("持股比例", 0) for h in holders) signal = { "module": "concentration", "avg_holding": avg_holding, "top10_ratio": top10_ratio } if top10_ratio > 70: signal["signal"] = "十大股东持股比例高，筹码高度集中" signal["score"] = 65 elif top10_ratio 0 else 50 if final_score >= 65: action = "股东结构信号偏多，机构看好" elif final_score <= 40: action = "股东结构信号偏空，机构谨慎" else: action = "股东结构信号中性" return { "dm": dm, "score": round(final_score, 1), "action": action, "signals": signals } ``` 几个使用经验。第一，十大股东数据是季度披露的，有滞后性，但机构动作往往领先股价。第二，机构新进比机构增持信号更强，因为新进意味着从无到有。第三，股东人数减少要配合业绩看，业绩增长的筹码集中才有意义。第四，十大股东的季度变动比静态名单更有价值。 系统使用的数据来自ig50本地数据接口，十大股东、十大流通股东、股东变化趋势、基金持股这些接口配合使用，能构建完整的股东结构分析系统。 [gitee开源地址](https://gitee.com/igtrade/ighub) [github开源地址](https://github.com/ig50-service/ig50-official)
