2、数据获取与清洗：北向资金数据源介绍

做北向资金分析，第一步就是搞定数据。我见过太多人，策略想得天花乱坠，结果卡在数据获取这一步。说白了，数据源选不对，后面全是白搭。

今天咱们就把这事聊透。我会把市面上主流的几个数据源挨个捋一遍，再手把手带你写代码把数据抓下来，最后做清洗。嗯，这套流程我在实盘项目里跑了两年多，踩过的坑不少，今天一并告诉你。

2.1 三大主流数据源对比

我个人习惯把数据源分成三类：官方源、商业源、开源源。各有各的脾气，咱们一个一个看。

数据源	类型	数据质量	获取难度	更新频率	费用
港交所	官方	★★★★★	高	日频	免费
Wind	商业	★★★★★	低	实时	高
Tushare	开源	★★★★	中	日频	低

2.1.1 港交所：最权威，但最折腾

港交所官网每天会发布北向资金的持股明细和资金流向。数据绝对准确，毕竟是官方一手货。但问题来了——它没有API接口。

你想想看，每次都得手动去下载Excel文件。一天两天还行，长期做量化分析，这谁受得了？我曾经试过写爬虫去抓，结果人家页面结构三天两头改，维护成本高得离谱。

适用场景：做历史数据回测，或者只需要偶尔拉一次数据做验证。

2.1.2 Wind：真香，但贵

Wind金融终端，业内老大哥。数据全、更新快、接口稳定。用Python的WindPy库，一行代码就能拉数据。

但价格嘛...一年几万块。个人投资者基本不用想。我在机构的时候用过两年，说实话，体验确实好。但如果你是自己做研究，我建议先看看别的选择。

适用场景：机构用户，或者预算充足的团队。

2.1.3 Tushare：个人玩家的首选

Tushare是我个人最推荐给个人投资者的。免费版就能拿到北向资金的日频数据，包括持股明细和资金流向。注册后获取token，用Python调接口就行。

数据质量嘛，我对比过港交所的官方数据，误差在0.5%以内。做策略分析完全够用。

适用场景：个人量化研究、中小团队。

2.2 Python数据获取实战

好，理论说完了，咱们直接上代码。我用Tushare演示，因为大家都能用。

2.2.1 安装与配置

# 安装Tushare
pip install tushare

# 导入并设置token
import tushare as ts
ts.set_token('你的token')
pro = ts.pro_api()

token怎么拿？去Tushare官网注册，在个人中心就能看到。嗯，这一步别省，没有token啥也干不了。

2.2.2 获取北向资金每日流向

# 获取2024年1月的北向资金数据
df = pro.moneyflow_hsgt(
    start_date='20240101',
    end_date='20240131'
)
print(df.head())

返回的数据长这样：

trade_date	gt_amt	sgt_amt	north_amt
20240102	12.5亿	8.3亿	20.8亿
20240103	-3.2亿	5.1亿	1.9亿

字段说明：gt_amt是沪股通，sgt_amt是深股通，north_amt是两者之和。

2.2.3 获取个股持股明细

# 获取北向资金持股明细
df_stock = pro.hk_hold(
    ts_code='600519.SH',
    start_date='20240101',
    end_date='20240131'
)
print(df_stock.head())

这里要注意，ts_code的格式是「股票代码.交易所」，上交所是.SH，深交所是.SZ。

2.3 数据清洗与预处理

数据拉下来，你以为就能直接用？太天真了。原始数据里全是坑，不洗一洗，策略跑出来全是噪音。

2.3.1 常见问题与处理

我总结了一下，北向资金数据常见的脏数据有这几类：

• 缺失值：节假日没有交易，数据是空的
• 异常值：某天资金突然暴增或暴减，可能是数据录入错误
• 重复值：同一日期的数据出现多次
• 格式问题：日期是字符串，金额是带「亿」的文本

咱们一个一个处理。

import pandas as pd
import numpy as np

# 1. 处理缺失值
df = df.dropna(subset=['gt_amt', 'sgt_amt'])  # 删除关键字段为空的行
df = df.fillna(0)  # 其他字段用0填充

# 2. 处理重复值
df = df.drop_duplicates(subset=['trade_date'])

# 3. 转换日期格式
df['trade_date'] = pd.to_datetime(df['trade_date'])

# 4. 清洗金额字段（去掉"亿"字，转为数值）
def clean_amount(x):
    if isinstance(x, str):
        return float(x.replace('亿', ''))
    return x

df['gt_amt'] = df['gt_amt'].apply(clean_amount)
df['sgt_amt'] = df['sgt_amt'].apply(clean_amount)

2.3.2 异常值检测

异常值怎么找？我一般用3σ原则。说白了，就是看数据偏离均值多少个标准差。

# 3σ异常值检测
mean = df['north_amt'].mean()
std = df['north_amt'].std()
threshold = 3

df['is_outlier'] = np.abs(df['north_amt'] - mean) > threshold * std

# 查看异常值
outliers = df[df['is_outlier']]
print(f"发现 {len(outliers)} 个异常值")

发现异常值后怎么办？别急着删。先看看是不是特殊事件导致的。比如某天突然有大量资金流入，可能是MSCI调整成分股。这种情况应该保留，而不是删除。

2.3.3 最终数据格式

清洗完之后，我习惯把数据存成这样的格式：

trade_date	gt_amt	sgt_amt	north_amt	is_outlier
2024-01-02	12.5	8.3	20.8	False
2024-01-03	-3.2	5.1	1.9	False

日期是datetime格式，金额是浮点数，异常值打了标记。这样后面做分析，直接拿来用就行。

2.4 本章知识体系

下面这张图，是我自己梳理的北向资金数据获取与清洗的完整流程。你照着这个走，基本不会出大问题。

这张图把整个流程串起来了。从左到右，从上到下，每一步都清楚。你写代码的时候，就照着这个流程走，不会漏掉任何一步。

---

公众号:蓝海资料掘金营，微信deep3321