北向资金被称为A股的"聪明钱"，但大部分人对它的理解还停留在"今天净流入多少亿"的层面。本文用Python+TuShare从总量趋势、个股偏好、持股变化三个维度构建一个完整的北向资金追踪系统，帮你读懂聪明钱的真实意图。

一、为什么北向资金值得追踪？

北向资金（沪股通+深股通）由海外机构投资者主导，包括摩根士丹利、高盛、贝莱德等国际顶级资管。这些机构的投研能力远超普通散户，因此北向资金的流向往往被视为"聪明钱"的信号。

但单纯看净流入金额是个大坑。单日净流入100亿，可能是被动指数基金MSCI调仓，不代表主动看好；连续3天净流入，也未必是趋势性买入。真正的信号藏在结构里——聪明钱在买什么、卖什么、持有什么。

本文用TuShare的三组接口，从浅到深拆解北向资金的完整图景。

二、维度一：总量趋势分析——别被单日数据骗了

TuShare的moneyflow_hsgt接口可以获取每日北向资金流向数据。

import tushare as ts
import pandas as pd

ts.set_token('你的token')
pro = ts.pro_api()

# 获取近30天北向资金数据
df = pro.moneyflow_hsgt(start_date='20260710', end_date='20260810')
df = df.sort_values('trade_date').reset_index(drop=True)

# 关键：看连续趋势，不看单日
df['north_5d_avg'] = df['north_money'].rolling(5).mean()  # 5日均值
df['north_cum'] = df['north_money'].cumsum()  # 累计净流入

# 计算连续流入/流出天数
def calc_consecutive(series, positive=True):
    """计算连续流入或流出天数"""
    count = 0
    for val in series[::-1]:  # 从最近一天往前数
        if (positive and val > 0) or (not positive and val < 0):
            count += 1
        else:
            break
    return count

consecutive_inflow = calc_consecutive(df['north_money'], positive=True)
consecutive_outflow = calc_consecutive(df['north_money'], positive=False)

print(f"最近连续流入天数：{consecutive_inflow}")
print(f"最近连续流出天数：{consecutive_outflow}")
print(f"5日累计净流入：{df['north_money'].tail(5).sum():.0f} 百万元")

核心判断逻辑：

• 连续5天以上净流入才算趋势性买入，单日大幅流入大概率是调仓
• 5日累计净流入>30亿且沪股通+深股通同步流入=真看多
• 沪股通流入但深股通流出=分歧信号，可能只是买大盘蓝筹避险

一个容易忽略的细节：2024年8月20日起，港交所停止发布日度北向资金数据，改为季度披露。但TuShare通过其他渠道仍在更新这个数据，所以用TuShare获取北向数据是目前最便捷的方式之一。

三、维度二：十大成交股分析——聪明钱今天买了谁

hsgt_top10接口获取每日北向资金成交金额最大的10只股票，这才是真正有含金量的数据。

# 获取沪股通十大成交股
sh_top10 = pro.hsgt_top10(trade_date='20260808', market_type='1')
# 获取深股通十大成交股
sz_top10 = pro.hsgt_top10(trade_date='20260808', market_type='3')

# 合并分析
all_top = pd.concat([sh_top10, sz_top10], ignore_index=True)

# 计算净买入额（买入-卖出）
all_top['net_buy'] = all_top['buy'] - all_top['sell']

# 排序看谁被聪明钱最看好
top_buys = all_top.nlargest(10, 'net_buy')[['ts_code', 'name', 'net_buy', 'trade_date']]
top_sells = all_top.nsmallest(10, 'net_buy')[['ts_code', 'name', 'net_buy', 'trade_date']]

print("=== 聪明钱最看好的股票 ===")
print(top_buys.to_string(index=False))
print("\n=== 聪明钱最不看好的股票 ===")
print(top_sells.to_string(index=False))

进阶玩法：连续多日追踪

单日Top10噪音大，关键是看哪些股票反复出现在买入榜。

import datetime

def track_smart_money(days=10):
    """追踪近N天聪明钱持续买入的股票"""
    # 获取交易日列表
    cal = pro.trade_cal(exchange='SSE', start_date='20260701', end_date='20260810')
    trade_days = cal[cal['is_open'] == 1]['cal_date'].tolist()[-days:]
    
    all_net_buys = {}
    for day in trade_days:
        try:
            sh = pro.hsgt_top10(trade_date=day, market_type='1')
            sz = pro.hsgt_top10(trade_date=day, market_type='3')
            daily_top = pd.concat([sh, sz])
            daily_top['net_buy'] = daily_top['buy'] - daily_top['sell']
            
            for _, row in daily_top.iterrows():
                code = row['ts_code']
                if code not in all_net_buys:
                    all_net_buys[code] = {
                        'name': row['name'],
                        'total_net': 0,
                        'appear_days': 0,
                        'buy_days': 0
                    }
                all_net_buys[code]['total_net'] += row['net_buy']
                all_net_buys[code]['appear_days'] += 1
                if row['net_buy'] > 0:
                    all_net_buys[code]['buy_days'] += 1
        except Exception as e:
            print(f"{day}数据获取失败: {e}")
    
    # 筛选：出现>=3天且总净买入>0的股票
    results = []
    for code, info in all_net_buys.items():
        if info['appear_days'] >= 3 and info['total_net'] > 0:
            results.append({
                'ts_code': code,
                'name': info['name'],
                'total_net_million': info['total_net'],
                'appear_days': info['appear_days'],
                'buy_ratio': info['buy_days'] / info['appear_days']
            })
    
    result_df = pd.DataFrame(results)
    if not result_df.empty:
        result_df = result_df.sort_values('total_net_million', ascending=False)
    return result_df

smart_picks = track_smart_money(days=10)
print(smart_picks.to_string(index=False))

这段代码的输出告诉你：过去10个交易日里，哪些股票反复被北向资金净买入，且累计净买入额最大。这种"持续性买入"比单日大买更有参考价值。

四、维度三：个股持股变化——聪明钱到底囤了多少货

hk_hold接口获取港交所披露的北向资金持股明细，这是最底层的数据——告诉你每只股票被北向资金持有多少股、占流通盘比例多少。

# 获取某日北向资金全部持股
hold_all = pro.hk_hold(trade_date='20260808')

# 筛选沪股通持股
hold_sh = pro.hk_hold(trade_date='20260808', exchange='SH')

# 看持股占比最高的股票（聪明钱重仓股）
top_holds = hold_all.nlargest(20, 'ratio')[['ts_code', 'name', 'vol', 'ratio']]
print("=== 北向资金持股占比TOP20 ===")
print(top_holds.to_string(index=False))

最有价值的用法：追踪持股变化

def track_hold_change(ts_code, start_date, end_date):
    """追踪单只股票的北向资金持股变化"""
    df = pro.hk_hold(ts_code=ts_code, start_date=start_date, end_date=end_date)
    df = df.sort_values('trade_date').reset_index(drop=True)
    
    if df.empty:
        print(f"{ts_code} 无北向资金持仓数据")
        return None
    
    df['vol_change'] = df['vol'].diff()  # 持股数量变化
    df['ratio_change'] = df['ratio'].diff()  # 持股比例变化
    
    # 计算累计增减仓
    total_vol_change = df['vol'].iloc[-1] - df['vol'].iloc[0]
    total_ratio_change = df['ratio'].iloc[-1] - df['ratio'].iloc[0]
    
    print(f"股票代码：{ts_code}")
    print(f"期初持股：{df['vol'].iloc[0]:,.0f}股（占比{df['ratio'].iloc[0]:.2f}%）")
    print(f"期末持股：{df['vol'].iloc[-1]:,.0f}股（占比{df['ratio'].iloc[-1]:.2f}%）")
    print(f"累计增减：{total_vol_change:+,.0f}股（占比变化{total_ratio_change:+.2f}%）")
    
    # 判断趋势
    if total_ratio_change > 0.5:
        print(">>> 趋势：大幅加仓，聪明钱看好")
    elif total_ratio_change > 0.1:
        print(">>> 趋势：小幅加仓，持续配置")
    elif total_ratio_change < -0.5:
        print(">>> 趋势：大幅减仓，注意风险")
    else:
        print(">>> 趋势：小幅减仓或持平")
    
    return df

# 示例：追踪贵州茅台的北向资金持仓变化
track_hold_change('600519.SH', '20260701', '20260810')

这个功能的实战意义在于：当你发现一只股票北向资金持股比例从1%悄悄升到3%，同时股价还没明显启动，这可能就是聪明钱在"悄悄囤货"的信号。

五、综合信号：三维度联动判断

单看一个维度容易误判，三个维度联动能大幅提高判断准确性：

维度组合	信号含义	操作建议
总量连续流入 + 个股反复上买入榜 + 持股比例上升	强烈看多	可跟随布局
总量流入 + 十大成交股集中买入1-2只	定向看好，非普涨	只跟进目标股
总量流入 + 持股变化不大	短线交易，非趋势性	观望为主
总量流出 + 个股持股比例下降	聪明钱撤退	警惕风险
总量平稳 + 持股结构大幅调整	换仓不是跑路	关注新进入的股票

六、踩坑提醒

1. 积分门槛：moneyflow_hsgt需2000积分，hsgt_top10需600积分，hk_hold需2000积分。免费用户用不了，这是TuShare主要的付费墙之一。

2. 数据延迟：北向资金持股数据（hk_hold）有1-2天延迟，不是实时的。盘中看到的"实时北向资金"数据通常来自券商或东财，TuShare是盘后数据。

3. 港交所政策变化：2024年8月20日起港交所改为季度披露日度数据，但TuShare通过其他数据源仍在更新。如果接口返回空数据，先检查是否是数据源问题。

4. 别盲目跟聪明钱：北向资金也会犯错。2025年以来北向资金的多只重仓股跑输大盘。它只是参考维度之一，不能当成唯一买卖依据。

七、完整代码

"""
北向资金三维追踪系统
功能：总量趋势 + 十大成交股 + 持股变化
依赖：tushare, pandas
"""
import tushare as ts
import pandas as pd
import warnings
warnings.filterwarnings('ignore')

ts.set_token('你的token')
pro = ts.pro_api()

class NorthMoneyTracker:
    def __init__(self):
        self.pro = pro
    
    def trend_analysis(self, days=20):
        """维度一：总量趋势分析"""
        df = self.pro.moneyflow_hsgt()
        df = df.sort_values('trade_date').tail(days)
        
        result = {
            'period_days': days,
            'total_north': df['north_money'].sum(),
            'avg_daily': df['north_money'].mean(),
            'max_inflow': df['north_money'].max(),
            'max_outflow': df['north_money'].min(),
            'inflow_days': (df['north_money'] > 0).sum(),
            'outflow_days': (df['north_money'] < 0).sum(),
            'consecutive_inflow': self._calc_consecutive(df['north_money'], True),
            'consecutive_outflow': self._calc_consecutive(df['north_money'], False),
            '5d_avg': df['north_money'].tail(5).mean()
        }
        
        # 趋势判断
        if result['consecutive_inflow'] >= 5:
            result['signal'] = '趋势性流入'
        elif result['consecutive_outflow'] >= 5:
            result['signal'] = '趋势性流出'
        else:
            result['signal'] = '震荡无趋势'
        
        return result
    
    def top10_tracker(self, days=10):
        """维度二：十大成交股追踪"""
        cal = self.pro.trade_cal(exchange='SSE')
        cal = cal[cal['is_open'] == 1].sort_values('cal_date', ascending=False)
        trade_days = cal['cal_date'].head(days).tolist()
        
        stocks = {}
        for day in trade_days:
            for market in ['1', '3']:
                try:
                    top = self.pro.hsgt_top10(trade_date=day, market_type=market)
                    for _, row in top.iterrows():
                        code = row['ts_code']
                        net = row['buy'] - row['sell']
                        if code not in stocks:
                            stocks[code] = {'name': row['name'], 'total_net': 0, 
                                           'days': 0, 'buy_days': 0}
                        stocks[code]['total_net'] += net
                        stocks[code]['days'] += 1
                        if net > 0:
                            stocks[code]['buy_days'] += 1
                except:
                    pass
        
        results = []
        for code, info in stocks.items():
            if info['days'] >= 3:
                results.append({
                    'ts_code': code, 'name': info['name'],
                    'total_net': info['total_net'],
                    'appear_days': info['days'],
                    'buy_ratio': info['buy_days'] / info['days']
                })
        
        df = pd.DataFrame(results).sort_values('total_net', ascending=False)
        return df
    
    def hold_change(self, ts_code, days=30):
        """维度三：持股变化追踪"""
        df = self.pro.hk_hold(ts_code=ts_code)
        if df.empty:
            return None
        df = df.sort_values('trade_date').tail(days)
        
        return {
            'start_ratio': df['ratio'].iloc[0],
            'end_ratio': df['ratio'].iloc[-1],
            'change': df['ratio'].iloc[-1] - df['ratio'].iloc[0],
            'start_vol': df['vol'].iloc[0],
            'end_vol': df['vol'].iloc[-1],
            'signal': '加仓' if df['ratio'].iloc[-1] - df['ratio'].iloc[0] > 0.1 else 
                      '减仓' if df['ratio'].iloc[-1] - df['ratio'].iloc[0] < -0.1 else '持平'
        }
    
    def _calc_consecutive(self, series, positive=True):
        count = 0
        for val in series.values[::-1]:
            if (positive and val > 0) or (not positive and val < 0):
                count += 1
            else:
                break
        return count

# 使用示例
tracker = NorthMoneyTracker()

# 1. 总量趋势
trend = tracker.trend_analysis(days=20)
print(f"北向资金信号：{trend['signal']}")
print(f"20日累计净流入：{trend['total_north']:.0f}百万元")
print(f"连续流入天数：{trend['consecutive_inflow']}")

# 2. 聪明钱持续买入的股票
smart = tracker.top10_tracker(days=10)
print("\n聪明钱持续买入TOP5：")
print(smart.head(5)[['name', 'total_net', 'buy_ratio']].to_string(index=False))

# 3. 个股持股变化
hold = tracker.hold_change('600519.SH', days=30)
if hold:
    print(f"\n贵州茅台北向持股变化：{hold['signal']}")
    print(f"持股比例：{hold['start_ratio']:.2f}% → {hold['end_ratio']:.2f}%")

总结

追踪北向资金不是看个总数就完事。总量看趋势方向，十大成交股看聪明钱买了谁，持股变化看聪明钱囤了多少货。三个维度交叉验证，才能分辨出聪明钱的真实意图。

TuShare提供了一站式的数据接口，配合Python可以轻松构建自己的北向资金追踪系统。代码已经在上面，改个token就能跑。

作者h29hj，持续输出TuShare+Python量化实战文章，关注我获取更多A股量化分析技巧。