Agent驱动的北向资金追踪系统：用AI捕捉聪明钱的行业轮动信号

北向资金被市场称为"聪明钱"，但大部分人对北向资金的分析还停留在"今天流入多少亿"的层面。聪明钱到底聪明在哪里？它在买入哪些行业、卖出哪些行业？它的动向对个股有什么指引？最近我用AI Agent构建了一个北向资金追踪系统，让Agent自动从北向资金数据中挖掘行业轮动信号。

系统的核心思路是把北向资金分析拆解成几个子任务：总量趋势分析、个股动向追踪、行业偏好识别、市场状态判断。每个子任务交给专门的模块处理，Agent负责汇总各模块结果生成综合判断。

数据层使用本地数据引擎，主要用到五个接口。沪股通历史数据all/nxbx/hgtlssj，返回每天沪股通的净买额、买入额、卖出额。深股通历史数据all/nxbx/sgtlssj，结构相同。沪股通十大成交股all/nxbx/hgtsdcj/{交易日}，返回北向资金当天买卖最多的十只沪市股票。深股通十大成交股all/nxbx/sgtsdcj/{交易日}，深市版本。沪股通成分股行情all/nxbx/hgthq和深股通成分股行情all/nxbx/sgthq，提供所有北向资金标的的行情数据。

```python
class NorthboundAgent:
    def __init__(self, data_engine):
        self.data_engine = data_engine
        self.modules = {
            "trend": TrendAnalyzer(),
            "stock": StockFlowTracker(),
            "sector": SectorPreferenceDetector(),
            "regime": MarketRegimeJudge()
        }
    
    def run(self, date_str):
        hgt_hist = self.data_engine.read_hgt_history()
        sgt_hist = self.data_engine.read_sgt_history()
        hgt_top10 = self.data_engine.read_hgt_top10(date_str)
        sgt_top10 = self.data_engine.read_sgt_top10(date_str)
        
        context = {
            "date": date_str,
            "hgt_hist": hgt_hist,
            "sgt_hist": sgt_hist,
            "hgt_top10": hgt_top10,
            "sgt_top10": sgt_top10
        }
        
        trend_result = self.modules["trend"].analyze(context)
        stock_result = self.modules["stock"].track(context)
        sector_result = self.modules["sector"].detect(context)
        regime_result = self.modules["regime"].judge(context)
        
        decision = self.synthesize(trend_result, stock_result, sector_result, regime_result)
        return decision
```

第一个模块是趋势分析器。它不简单看单日净流入流出，而是分析北向资金的连续性。连续5天以上净流入才算趋势性流入，单日大幅流入可能是调仓行为。

```python
class TrendAnalyzer:
    def analyze(self, context):
        hgt = context["hgt_hist"]
        sgt = context["sgt_hist"]
        if hgt is None or sgt is None:
            return None
        
        recent_20 = hgt.tail(20)["净买额"].values + sgt.tail(20)["净买额"].values
        
        inflow_days = sum(1 for x in recent_20 if x > 0)
        outflow_days = sum(1 for x in recent_20 if x < 0)
        
        consecutive_inflow = self.calc_consecutive(recent_20, positive=True)
        consecutive_outflow = self.calc_consecutive(recent_20, positive=False)
        
        total_net = sum(recent_20)
        avg_daily = total_net / len(recent_20)
        
        signal = {
            "module": "trend",
            "inflow_days": inflow_days,
            "outflow_days": outflow_days,
            "consecutive_inflow": consecutive_inflow,
            "consecutive_outflow": consecutive_outflow,
            "total_net": total_net,
            "avg_daily": avg_daily
        }
        
        if consecutive_inflow >= 5:
            signal["signal"] = "北向资金连续流入，趋势性看好"
            signal["score"] = 72
        elif consecutive_outflow >= 5:
            signal["signal"] = "北向资金连续流出，趋势性谨慎"
            signal["score"] = 35
        elif abs(total_net) < avg_daily * 2:
            signal["signal"] = "北向资金流入流出均衡"
            signal["score"] = 50
        else:
            signal["signal"] = "北向资金方向不明"
            signal["score"] = 50
        
        return signal
    
    def calc_consecutive(self, series, positive=True):
        count = 0
        for x in reversed(series):
            if positive and x > 0:
                count += 1
            elif not positive and x < 0:
                count += 1
            else:
                break
        return count
```

第二个模块是个股动向追踪器。它统计北向资金十大成交股的数据，找出北向资金连续买入的个股。连续买入的信号比单日买入更可靠。

```python
class StockFlowTracker:
    def track(self, context):
        date_str = context["date"]
        
        date_list = self.get_recent_dates(date_str, 10)
        
        buy_count = {}
        sell_count = {}
        
        for d in date_list:
            for channel in ["hgt", "sgt"]:
                top10 = self.data_engine.read_top10(channel, d)
                if top10 is None:
                    continue
                for _, row in top10.iterrows():
                    dm = row["dm"]
                    jme = row["jme"]
                    if jme > 0:
                        buy_count[dm] = buy_count.get(dm, 0) + 1
                    elif jme < 0:
                        sell_count[dm] = sell_count.get(dm, 0) + 1
        
        continuous_buy = {dm: c for dm, c in buy_count.items() if c >= 3}
        continuous_sell = {dm: c for dm, c in sell_count.items() if c >= 3}
        
        signal = {
            "module": "stock",
            "continuous_buy": continuous_buy,
            "continuous_sell": continuous_sell
        }
        
        if len(continuous_buy) >= 5:
            signal["signal"] = "北向资金集中买入多只股票"
            signal["score"] = 70
        elif len(continuous_sell) >= 5:
            signal["signal"] = "北向资金集中卖出多只股票"
            signal["score"] = 35
        else:
            signal["signal"] = "北向资金个股动向分散"
            signal["score"] = 50
        
        return signal
```

第三个模块是行业偏好识别器。它把北向资金买入的股票按行业分类，识别北向资金正在流入的行业。行业轮动信号比个股信号更有参考价值。

```python
class SectorPreferenceDetector:
    def detect(self, context):
        date_str = context["date"]
        date_list = self.get_recent_dates(date_str, 5)
        
        sector_inflow = {}
        
        for d in date_list:
            for channel in ["hgt", "sgt"]:
                top10 = self.data_engine.read_top10(channel, d)
                if top10 is None:
                    continue
                for _, row in top10.iterrows():
                    dm = row["dm"]
                    jme = row["jme"]
                    sector = self.get_sector(dm)
                    sector_inflow[sector] = sector_inflow.get(sector, 0) + jme
        
        sorted_sectors = sorted(sector_inflow.items(), key=lambda x: x[1], reverse=True)
        
        signal = {
            "module": "sector",
            "top_inflow": sorted_sectors[:3],
            "top_outflow": sorted_sectors[-3:]
        }
        
        top_sector = sorted_sectors[0] if sorted_sectors else None
        if top_sector and top_sector[1] > 0:
            signal["signal"] = "北向资金偏好{}行业".format(top_sector[0])
            signal["score"] = 68
        else:
            signal["signal"] = "北向资金行业偏好不明显"
            signal["score"] = 50
        
        return signal
```

第四个模块是市场状态判断器。它结合北向资金数据和大盘指数数据，判断当前市场环境。北向资金连续流入+大盘上升趋势是最佳组合。

```python
class MarketRegimeJudge:
    def judge(self, context):
        index_data = self.data_engine.read_index_kline("sh000001", "day")
        if index_data is None or len(index_data) < 60:
            return {"module": "regime", "score": 50, "signal": "数据不足"}
        
        closes = index_data["收盘价"].values
        ma20 = sum(closes[-20:]) / 20
        ma60 = sum(closes[-60:]) / 60
        current = closes[-1]
        
        trend = "bull" if current > ma20 > ma60 else "bear" if current < ma20 < ma60 else "range"
        
        trend_module = context.get("trend_result", {})
        nb_consecutive = trend_module.get("consecutive_inflow", 0)
        
        signal = {"module": "regime", "market_trend": trend}
        
        if trend == "bull" and nb_consecutive >= 3:
            signal["signal"] = "大盘上升+北向流入，最佳组合"
            signal["score"] = 78
        elif trend == "bear" and nb_consecutive >= 3:
            signal["signal"] = "大盘下跌但北向流入，可能见底"
            signal["score"] = 62
        elif trend == "bull" and trend_module.get("consecutive_outflow", 0) >= 3:
            signal["signal"] = "大盘上升但北向流出，警惕"
            signal["score"] = 45
        else:
            signal["signal"] = "市场状态中性"
            signal["score"] = 50
        
        return signal
```

Agent的决策模块把四个模块的结果汇总，生成综合判断。

```python
def synthesize(self, trend, stock, sector, regime):
    results = [r for r in [trend, stock, sector, regime] if r is not None]
    if not results:
        return {"score": 50, "signal": "数据不足"}
    
    weights = {"trend": 0.3, "stock": 0.25, "sector": 0.25, "regime": 0.2}
    
    total_score = 0
    total_weight = 0
    for r in results:
        w = weights.get(r.get("module", ""), 0.1)
        total_score += r.get("score", 50) * w
        total_weight += w
    
    final_score = total_score / total_weight if total_weight > 0 else 50
    
    if final_score >= 65:
        action = "北向资金信号偏多，可关注"
    elif final_score <= 40:
        action = "北向资金信号偏空，谨慎"
    else:
        action = "北向资金信号中性"
    
    return {
        "score": round(final_score, 1),
        "action": action,
        "details": [r.get("signal", "") for r in results]
    }
```

实际使用中，Agent每天盘后运行一次，生成北向资金的综合分析报告。如果综合得分超过65分，我会把北向资金连续买入的股票加入观察池，配合自己的选股策略做交叉验证。

几个使用经验。第一，北向资金的连续性比单日数据重要，连续5天流入比单日流入100亿信号更强。第二，北向资金大幅流出时不要恐慌，这往往是市场恐慌的标志，反而可能是反向指标。第三，北向资金的行业偏好变化比个股变化更有前瞻性，跟着行业走比跟着个股走胜率高。

系统使用的数据来自ig50本地数据接口，北向资金相关接口数据更新及时，字段完整，适合构建这种多维分析系统。

资料参考：ig50


北向资金被市场称为"聪明钱"，但大部分人对北向资金的分析还停留在"今天流入多少亿"的层面。聪明钱到底聪明在哪里？它在买入哪些行业、卖出哪些行业？它的动向对个股有什么指引？最近我用AI Agent构建了一个北向资金追踪系统，让Agent自动从北向资金数据中挖掘行业轮动信号。

Agent驱动的北向资金追踪系统：用AI捕捉聪明钱的行业轮动信号

人工智能

后端

AI驱动的北向资金追踪系统深度解析：通过Python构建智能分析模块，实时捕捉"聪明钱"行业轮动信号。系统采用多模块协同分析，包括趋势分析、个股追踪、行业偏好和市场状态判断，自动生成投资决策建议。连续5天流入、行业集中买入等关键指标揭示北向资金真实动向，为投资者提供精准市场信号。

数据接口

代码解释

Agent

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Agent驱动的北向资金追踪系统：用AI捕捉聪明钱的行业轮动信号-腾讯云开发者社区-腾讯云

Agent驱动的北向资金追踪系统：用AI捕捉聪明钱的行业轮动信号

Agent驱动的北向资金追踪系统：用AI捕捉聪明钱的行业轮动信号

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐