第二章 数据获取与清洗：从公开API获取北向资金日频数据
做量化分析的人都知道一句话：垃圾进，垃圾出。北向资金分析也不例外。你拿到的数据如果是脏的，后面再漂亮的模型也是白搭。这一章，我就带你手把手搞定数据获取和清洗这两个最基础、也最关键的环节。
核心要点：北向资金日频数据的获取并不复杂，但清洗对齐才是真正的技术活。节假日处理、缺失值填充、数据对齐，这三件事做不好，你的分析结果可能全是错的。
2.1 数据源选择：从哪里拿数据？
我个人习惯用三个数据源：东方财富、新浪财经、以及港交所官方。为什么是三个？因为单一数据源偶尔会抽风，多一个备份心里踏实。
东方财富的接口最稳定，我用了三年基本没出过问题。新浪财经的响应速度最快，适合做实时监控。港交所官方数据最权威，但接口调用限制比较多。
这里我直接给出常用的API地址：
# 东方财富北向资金日频数据接口
url = "https://push2.eastmoney.com/api/qt/kamt.kline/get"
params = {
"fields1": "f1,f2,f3,f4",
"fields2": "f51,f52,f53,f54,f55",
"klt": "101", # 日频
"lmt": "1000", # 最近1000个交易日
"secid": "1.000001" # 沪股通
}
# 深股通类似，secid改为"0.159900"小技巧：我建议你同时拉取沪股通和深股通的数据，然后合并计算。因为有些时候沪股通和深股通的资金流向方向是相反的，单独看一个会失真。
2.2 数据获取实战：写一个稳定的爬虫
直接上代码。这是我用了很久的一个模板，稳定可靠：
import requests
import pandas as pd
import time
from datetime import datetime, timedelta
def fetch_north_flow(start_date, end_date):
"""
获取北向资金日频数据
start_date: '2024-01-01'
end_date: '2024-12-31'
"""
# 沪股通
sh_url = "https://push2.eastmoney.com/api/qt/kamt.kline/get"
sh_params = {
"fields1": "f1,f2,f3,f4",
"fields2": "f51,f52,f53,f54,f55",
"klt": "101",
"lmt": "1000",
"secid": "1.000001"
}
# 深股通
sz_url = "https://push2.eastmoney.com/api/qt/kamt.kline/get"
sz_params = {
"fields1": "f1,f2,f3,f4",
"fields2": "f51,f52,f53,f54,f55",
"klt": "101",
"lmt": "1000",
"secid": "0.159900"
}
try:
sh_resp = requests.get(sh_url, params=sh_params, timeout=10)
sz_resp = requests.get(sz_url, params=sz_params, timeout=10)
# 解析数据
sh_data = sh_resp.json()['data']['klines']
sz_data = sz_resp.json()['data']['klines']
# 合并处理
df_sh = parse_kline(sh_data, '沪股通')
df_sz = parse_kline(sz_data, '深股通')
# 合并
df = pd.merge(df_sh, df_sz, on='date', how='outer')
df['total_net'] = df['sh_net'] + df['sz_net']
return df
except Exception as e:
print(f"数据获取失败: {e}")
return None
def parse_kline(kline_data, name):
"""解析K线数据"""
records = []
for item in kline_data:
parts = item.split(',')
records.append({
'date': parts[0],
f'{name}_net': float(parts[1]), # 净买入额
f'{name}_buy': float(parts[2]), # 买入额
f'{name}_sell': float(parts[3]) # 卖出额
})
return pd.DataFrame(records)注意：我曾经遇到过一个问题——API返回的数据顺序偶尔会乱。所以拿到数据后，一定要按日期排序。另外，建议每次请求之间加0.5秒的延迟，别把人家服务器搞崩了。
2.3 数据清洗：剔除节假日
A股市场有它自己的交易日历。北向资金只在A股交易日才有数据。你想想看，如果直接把所有日期都拿来分析，周末和节假日的数据全是空的，那分析结果能对吗？
我的做法是维护一个交易日历表：
# A股交易日历（示例）
trading_days = [
'2024-01-02', '2024-01-03', '2024-01-04', # 元旦后
'2024-02-19', '2024-02-20', '2024-02-21', # 春节后
# ... 实际需要完整列表
]
def filter_trading_days(df, trading_days):
"""只保留交易日数据"""
df['date'] = pd.to_datetime(df['date'])
df = df[df['date'].isin(trading_days)]
return df.sort_values('date')避坑指南：我曾经直接用pd.bdate_range()生成交易日，结果发现它把调休上班的周末也算进去了，而真正的交易日反而漏了。后来我改用上交所和深交所官方发布的交易日历，这才彻底解决。
2.4 缺失值处理：别让NaN毁了你的模型
数据拿到手，你会发现有些日期是缺失的。原因很多：API超时、网络波动、或者那天确实没有交易数据。怎么处理？
我总结了三种常用方法：
| 方法 | 适用场景 | 我的建议 |
|---|---|---|
| 前向填充 | 短期缺失（1-2天） | 用前一天的数据填充，简单有效 |
| 线性插值 | 中期缺失（3-5天） | 用前后数据做线性拟合，更平滑 |
| 直接删除 | 长期缺失（超过5天） | 别犹豫，删掉最安全 |
具体实现：
def handle_missing_values(df, method='ffill'):
"""
处理缺失值
method: 'ffill' 前向填充, 'linear' 线性插值, 'drop' 删除
"""
if method == 'ffill':
df = df.fillna(method='ffill')
elif method == 'linear':
df = df.interpolate(method='linear')
elif method == 'drop':
df = df.dropna()
return df我的经验：对于北向资金数据，我一般先用前向填充。因为资金流向有惯性，今天和昨天的变化不会太大。但如果连续缺失超过3天，我会改用线性插值。记住，没有一种方法能通吃所有情况，你得根据数据特点灵活选择。
2.5 数据对齐：让北向资金和A股指数同步
这是最容易出错的一步。北向资金数据和A股指数数据，它们的日期可能不完全对齐。比如北向资金在某个交易日没有交易，但A股指数有数据。或者反过来。
我的对齐逻辑是这样的：
def align_data(north_flow, index_data):
"""
对齐北向资金和A股指数数据
以交易日历为准，两边都对齐
"""
# 合并
merged = pd.merge(north_flow, index_data, on='date', how='inner')
# 检查对齐后的数据量
print(f"对齐前北向数据: {len(north_flow)} 天")
print(f"对齐前指数数据: {len(index_data)} 天")
print(f"对齐后数据: {len(merged)} 天")
return merged重要提醒：对齐时一定要用inner join，别用outer。为什么？因为outer join会产生很多NaN，你后面还得花时间处理。我刚开始做的时候用了outer，结果数据量翻了一倍，但大部分都是无效数据，白白浪费时间。
2.6 本章知识体系
下面这张图是我自己画的，把整个数据获取和清洗的流程串起来了：
2.7 完整代码示例
最后，我把整个流程封装成一个完整的函数。你直接复制就能用：
def get_clean_north_flow(start_date, end_date):
"""
一站式获取并清洗北向资金数据
"""
# 1. 获取数据
df = fetch_north_flow(start_date, end_date)
if df is None:
return None
# 2. 加载交易日历
trading_days = load_trading_calendar()
# 3. 过滤交易日
df = filter_trading_days(df, trading_days)
# 4. 处理缺失值
df = handle_missing_values(df, method='ffill')
# 5. 对齐指数数据
index_data = fetch_index_data(start_date, end_date)
df = align_data(df, index_data)
return df
# 使用示例
data = get_clean_north_flow('2024-01-01', '2024-12-31')
print(data.head())最后说一句：数据清洗这件事，看起来枯燥，但它是整个量化分析的地基。我见过太多人急着跑模型，结果数据都没对齐，分析出来的结论全是错的。嗯，别走那条弯路。