主题切换
19.4 北向资金与龙虎榜
互联互通机制下的北向资金（沪股通/深股通）和龙虎榜交易数据为 A 股量化策略提供了独特的增量信息。北向资金常被视为"聪明钱"（Smart Money），其对 A 股的配置行为具有一定领先性。龙虎榜则披露了当日异常交易股票的席位明细——机构席位与游资席位的行为差异为量化策略提供了丰富的信号来源。
北向资金的预测能力
什么是北向资金
北向资金是指通过沪港通（2014年开通）和深港通（2016年开通）机制，从香港流入内地 A 股市场的资金。由于香港是国际资本进入中国的主要门户，北向资金的动向在很大程度上反映了外资对中国资产的态度。
北向资金每日公布的数据包括：
- 当日成交净买入额（买入成交额 - 卖出成交额）
- 当日买入成交额和卖出成交额
- 每日持股数量变化（沪深交易所按个股披露）
- 十大成交活跃股（沪股通和深股通分别披露）
学术与实证证据
大量研究表明，北向资金流对 A 股收益具有显著的预测能力：
- 领先-滞后效应：北向资金净流入增加后的 1-5 个交易日，A 股往往有正收益。这一效应在中小盘股上更为显著。
- 信息优势假说：外资机构具有更成熟的研究框架和全球比较视野，能够更早识别 A 股的价值洼地或风险。
- 风格偏好稳定性：北向资金偏好高 ROE、稳定成长的品质股（如消费、医药白马股），这一风格偏好具有持续性，形成了"外资审美溢价"。
- 资金流持续性：北向资金的流入/流出方向具有一定的动量特征——连续净流入之后往往继续净流入，但持续时间有限（通常 1-2 周）。
预测能力的来源
北向资金预测能力并非来自某种神秘优势，而是以下因素的复合：
- 全球资产配置决策：外资在全球视野下进行国别配置，A 股获得超配时往往意味着在横向比较中具备性价比优势。
- 基本面研究深度：外资机构的卖方和买方研究覆盖深度通常高于国内同业。
- 资金体量带来的信息：大额配置行为本身会向市场释放信号。
- 互联互通机制的结构性特征：北向资金不能参与IPO打新、不能融资融券，其交易行为更加"纯粹"，反映了纯多头配置意图。
局限与注意事项
北向资金的预测能力不是绝对的。以下情况需要警惕：
- 假期效应：长假前后（特别是春节、国庆）北向资金可能出现异常的集中流入/流出，这往往反映的是流动性管理需求而非 Alpha 判断。
- 全球风险事件：全球性风险事件（如美联储加息、地缘政治冲突）可能导致北向资金的无差别撤出，不反映对个股的判断。
- MSCI 纳入效应：MSCI（明晟）等国际指数对 A 股的权重调整会引发被动资金在特定日期的集中买卖。
- "假外资"问题：部分北向资金实际为内地在香港设立的资金，其交易行为可能无法代表真正的"外资判断"。
北向资金因子开发
此处有展示代码展开 ▼
python
import numpy as np
import pandas as pd
from scipy import stats
class NorthboundFactorEngine:
"""北向资金因子引擎"""
def __init__(self, northbound_flow: pd.DataFrame, prices: pd.DataFrame):
"""
Parameters
----------
northbound_flow : pd.DataFrame
北向资金日度持股数据
columns: ['date', 'code', 'hold_shares', 'hold_pct',
'buy_amount', 'sell_amount', 'net_buy_amount']
prices : pd.DataFrame
股票价格数据（用于计算市值和收益）
"""
self.flow = northbound_flow
self.prices = prices
def compute_flow_factors(self) -> pd.DataFrame:
"""计算北向资金流因子"""
df = self.flow.copy()
# 按股票和日期排序
df = df.sort_values(['code', 'date'])
for code in df['code'].unique():
mask = df['code'] == code
code_data = df.loc[mask].copy()
# 因子1: 当日净买入占比（净买入额/当日成交额）
df.loc[mask, 'net_buy_ratio'] = (
code_data['net_buy_amount'] / code_data['total_amount'].replace(0, np.nan)
)
# 因子2: 持股比例变动（当日持股比例 - 前日持股比例）
df.loc[mask, 'hold_pct_change'] = code_data['hold_pct'].diff()
# 因子3: 5日累计净买入占比
df.loc[mask, 'net_buy_ratio_5d'] = (
code_data['net_buy_amount'].rolling(5).sum() /
code_data['total_amount'].rolling(5).sum().replace(0, np.nan)
)
# 因子4: 20日持股比例趋势（Z-score法）
hold_pct_20d = code_data['hold_pct'].rolling(20)
df.loc[mask, 'hold_pct_trend_20d'] = (
(code_data['hold_pct'] - hold_pct_20d.mean()) /
hold_pct_20d.std().replace(0, np.nan)
)
# 因子5: 连续净买入天数
net_buy_sign = (code_data['net_buy_amount'] > 0).astype(int)
df.loc[mask, 'consecutive_buy_days'] = (
net_buy_sign.groupby((net_buy_sign != net_buy_sign.shift()).cumsum()).cumcount() + 1
) * net_buy_sign
# 因子6: 北向资金加速度（持股变动的一阶差分）
df.loc[mask, 'flow_acceleration'] = code_data['hold_pct_change'].diff()
# 因子7: 净买入的波动率调整
net_buy_rolling = code_data['net_buy_ratio'].rolling(20)
df.loc[mask, 'net_buy_vol_adj'] = (
code_data['net_buy_ratio'] / net_buy_rolling.std().replace(0, np.nan)
)
return df
def compute_holding_concentration(self, sector_mapping: dict = None) -> pd.DataFrame:
"""
计算北向资金持仓集中度
高集中度增持的行业/个股可能具有更强的Alpha信号
"""
df = self.flow.copy()
# 按日期统计各股票的北向资金持有市值
if 'market_cap' not in df.columns:
# 合并价格和持股量计算持仓市值
df = df.merge(
self.prices[['date', 'code', 'close']],
on=['date', 'code'], how='left'
)
df['hold_market_value'] = df['hold_shares'] * df['close']
# 每日北向资金持仓总市值
daily_total = df.groupby('date')['hold_market_value'].sum().reset_index()
daily_total.columns = ['date', 'total_hold_mv']
df = df.merge(daily_total, on='date', how='left')
# 持仓权重
df['northbound_weight'] = df['hold_market_value'] / df['total_hold_mv']
# 权重变化：北向资金的边际配置
df['weight_change'] = df.groupby('code')['northbound_weight'].diff()
return df
def generate_composite_signal(self) -> pd.DataFrame:
"""
生成北向资金综合信号
将多个北向因子合成为一个综合评分
"""
flow_factors = self.compute_flow_factors()
holding_data = self.compute_holding_concentration()
merged = flow_factors.merge(
holding_data[['date', 'code', 'northbound_weight', 'weight_change']],
on=['date', 'code'], how='left'
)
# 因子列表和方向（1=因子值越大越好，-1=因子值越小越好）
factor_config = {
'net_buy_ratio': 1,
'hold_pct_change': 1,
'net_buy_ratio_5d': 1,
'hold_pct_trend_20d': 1,
'consecutive_buy_days': 1,
'flow_acceleration': 1,
'net_buy_vol_adj': 1,
'weight_change': 1
}
# 截面标准化
merged = merged.set_index(['date', 'code'])
composite = pd.Series(0.0, index=merged.index)
valid_count = pd.Series(0, index=merged.index)
for factor, direction in factor_config.items():
if factor in merged.columns:
# 截面Z-score
zscore = merged[factor].groupby('date').transform(
lambda x: (x - x.mean()) / x.std() if x.std() > 0 else 0
) * direction
composite += zscore.fillna(0)
valid_count += zscore.notna().astype(float)
# 取平均
composite = composite / valid_count.replace(0, np.nan)
composite = composite.reset_index()
composite.columns = ['date', 'code', 'northbound_signal']
return composite点击展开可浏览运行结果
📘 本段代码定义了 1 个函数/类：类 `NorthboundFactorEngine`（北向资金因子引擎）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
龙虎榜席位分析
龙虎榜制度概述
龙虎榜是沪深交易所每日公布的异常交易股票信息，主要包括以下阈值条件触发的股票：
- 日涨跌幅偏离值达到
的前五名证券 - 日换手率达到
的前五名证券 - 日价格振幅达到
的前五名证券 - 连续三个交易日内涨跌幅偏离值累计达到
龙虎榜披露的具体信息包括：买入/卖出金额最大的前五名营业部名称及其买入/卖出金额。
机构席位 vs 游资席位
龙虎榜数据分析的核心是区分两类不同性质的交易者：
| 特征 | 机构专用席位 | 游资营业部 |
|---|---|---|
| 交易主体 | 公募基金、券商自营、保险资管等 | 个人大户、私募基金、涨停板敢死队 |
| 交易逻辑 | 基本面驱动，中长期持有 | 技术面驱动，短线博弈 |
| 买入后表现 | 买入后次日上涨概率较高，延续性强 | 买入后次日波动大，可能出现"一日游" |
| 卖出后表现 | 卖出信号较强，预示基本面转差 | 卖出后股价可能反弹（游资获利了结后新资金接盘） |
| 席位数稳定性 | 相对固定 | 变化频繁 |
席位信号构建
此处有展示代码展开 ▼
python
class DragonTigerAnalyzer:
"""龙虎榜数据量化分析器"""
# 常见机构席位标识（示例）
INSTITUTION_KEYWORDS = ['机构专用', '深股通专用', '沪股通专用',
'QFII', '社保基金']
def __init__(self, dragon_tiger_data: pd.DataFrame):
"""
Parameters
----------
dragon_tiger_data : pd.DataFrame
columns: ['date', 'code', 'name', 'reason',
'buy_seat_1', 'buy_amount_1', ..., 'buy_seat_5', 'buy_amount_5',
'sell_seat_1', 'sell_amount_1', ..., 'sell_seat_5', 'sell_amount_5',
'total_buy', 'total_sell']
"""
self.data = dragon_tiger_data
def classify_seats(self) -> pd.DataFrame:
"""对龙虎榜席位进行分类"""
df = self.data.copy()
def is_institution(seat_name: str) -> bool:
if pd.isna(seat_name):
return False
return any(kw in str(seat_name) for kw in self.INSTITUTION_KEYWORDS)
# 统计机构买入/卖出金额
inst_buy_cols = []
inst_sell_cols = []
for i in range(1, 6):
buy_seat = f'buy_seat_{i}'
buy_amount = f'buy_amount_{i}'
sell_seat = f'sell_seat_{i}'
sell_amount = f'sell_amount_{i}'
if buy_seat in df.columns:
df[f'inst_buy_{i}'] = df[buy_seat].apply(
lambda x: df.loc[df[buy_seat] == x, buy_amount].iloc[0]
if is_institution(x) else 0
) if False else np.where(
df[buy_seat].apply(is_institution), df[buy_amount], 0
)
if sell_seat in df.columns:
df[f'inst_sell_{i}'] = np.where(
df[sell_seat].apply(is_institution), df[sell_amount], 0
)
# 汇总机构净买入
if 'total_buy' in df.columns:
df['inst_total_buy'] = df[[f'inst_buy_{i}' for i in range(1, 6)
if f'inst_buy_{i}' in df.columns]].sum(axis=1)
df['inst_total_sell'] = df[[f'inst_sell_{i}' for i in range(1, 6)
if f'inst_sell_{i}' in df.columns]].sum(axis=1)
df['inst_net_buy'] = df['inst_total_buy'] - df['inst_total_sell']
df['inst_buy_ratio'] = df['inst_total_buy'] / df['total_buy'].replace(0, np.nan)
df['inst_sell_ratio'] = df['inst_total_sell'] / df['total_sell'].replace(0, np.nan)
return df
def seat_following_signal(self) -> pd.DataFrame:
"""
席位跟踪信号
识别具有持续盈利能力的"实力席位"，跟踪其操作方向
"""
classified = self.classify_seats()
# 计算历史上这五个席位的后续收益表现
seat_performance = {}
for i in range(1, 6):
buy_seat_col = f'buy_seat_{i}'
if buy_seat_col not in classified.columns:
continue
for seat in classified[buy_seat_col].dropna().unique():
trades = classified[classified[buy_seat_col] == seat]
if len(trades) < 10:
continue
# 假设持有1天的收益
avg_1d_return = trades.get('next_day_return', pd.Series(0)).mean()
win_rate = (trades.get('next_day_return', pd.Series(0)) > 0).mean()
seat_performance[seat] = {
'n_trades': len(trades),
'avg_1d_return': avg_1d_return,
'win_rate': win_rate,
'score': avg_1d_return * win_rate # 综合评分
}
# 筛选实力席位（前20%）
if seat_performance:
perf_df = pd.DataFrame(seat_performance).T
elite_seats = perf_df[perf_df['score'] > perf_df['score'].quantile(0.8)].index
# 生成跟随信号：实力席位净买入 > 0
classified['elite_seat_signal'] = classified.apply(
lambda row: self._check_elite_seats(row, elite_seats), axis=1
)
return classified
def _check_elite_seats(self, row, elite_seats) -> int:
"""检查某日的龙虎榜是否有实力席位参与"""
for i in range(1, 6):
if row.get(f'buy_seat_{i}') in elite_seats:
if row.get(f'buy_amount_{i}', 0) > row.get(f'sell_amount_{i}', 0):
return 1
return 0
def list_reason_analysis(self) -> pd.DataFrame:
"""按上榜原因分析后续表现"""
df = self.data.copy()
if 'reason' not in df.columns or 'next_day_return' not in df.columns:
return pd.DataFrame()
reason_stats = df.groupby('reason').agg(
n_events=('next_day_return', 'count'),
avg_next_day_return=('next_day_return', 'mean'),
win_rate=('next_day_return', lambda x: (x > 0).mean()),
avg_inst_net_buy=('inst_net_buy', 'mean')
).sort_values('avg_next_day_return', ascending=False)
return reason_stats点击展开可浏览运行结果
📘 本段代码定义了 1 个函数/类：类 `DragonTigerAnalyzer`（龙虎榜数据量化分析器）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
综合策略：北向 + 龙虎榜
将北向资金流和龙虎榜信号结合可以构建互补性更强的策略：
此处有展示代码展开 ▼
python
def composite_northbound_dragontiger(
northbound_signal: pd.DataFrame,
dragontiger_signal: pd.DataFrame,
price_data: pd.DataFrame
) -> pd.DataFrame:
"""
北向资金 + 龙虎榜综合策略
核心逻辑：
- 北向资金提供中长期方向信号（2-4周维度）
- 龙虎榜提供短期催化剂信号（1-5天维度）
- 两者共振时信号置信度最高
"""
# 合并信号
merged = price_data[['date', 'code', 'close']].merge(
northbound_signal[['date', 'code', 'northbound_signal']],
on=['date', 'code'], how='left'
).merge(
dragontiger_signal[['date', 'code', 'inst_net_buy', 'inst_buy_ratio']],
on=['date', 'code'], how='left'
)
# 北向信号标准化（截面排序）
merged['nb_signal_percentile'] = merged.groupby('date')['northbound_signal'].transform(
lambda x: x.rank(pct=True)
)
# 龙虎榜信号（只有上榜的股票有值，填充0）
merged['dt_signal'] = merged['inst_net_buy'].fillna(0) / merged['close'] * 10000
merged['dt_signal_percentile'] = merged.groupby('date')['dt_signal'].transform(
lambda x: np.where(x.sum() > 0, x.rank(pct=True), 0)
)
# 综合信号
merged['composite_signal'] = (
merged['nb_signal_percentile'].fillna(0.5) * 0.6 + # 北向权重60%
merged['dt_signal_percentile'].fillna(0.5) * 0.4 # 龙虎榜权重40%
)
# 共振增强：北向和龙虎榜同时强信号
nb_strong = merged['nb_signal_percentile'] > 0.8
dt_strong = merged['dt_signal_percentile'] > 0.8
merged['resonance'] = nb_strong & dt_strong
# 共振信号加10%的额外权重
merged.loc[merged['resonance'], 'composite_signal'] *= 1.10
merged['composite_signal'] = merged['composite_signal'].clip(upper=1.0)
return merged点击展开可浏览运行结果
📘 本段代码定义了 1 个函数/类：函数 `composite_northbound_dragontiger`（北向资金 + 龙虎榜综合策略）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
北向资金和龙虎榜数据为 A 股量化提供了独特的观察窗口。北向资金的趋势性信号适合构建中等频率的因子策略，龙虎榜的事件性信号适合构建短期的催化剂策略。两者的结合能够覆盖不同时间维度的 Alpha 来源，构建更全面的信号体系。