📈 Python实现代码
# AKShare 板块轮动分析
# pip install akshare pandas numpy
import akshare as ak
import pandas as pd
import numpy as np
def get_sector_momentum(lookback=20):
"""计算申万一级行业动量"""
try:
# 获取行业板块资金流向
df = ak.stock_sector_fund_flow_rank(indicator="今日")
df.columns = ['板块名称','涨跌幅','净流入','净流入占比','主力净流入','主力净流入占比','跟踪次数']
# 计算动量得分
df['动量得分'] = df['主力净流入'].rank(pct=True) * 0.6 + \
df['涨跌幅'].rank(pct=True) * 0.4
return df.sort_values('动量得分', ascending=False)
except Exception as e:
return f"数据获取失败: {e}"
def get_north_bound_flow():
"""北向资金（沪深港通）流向"""
try:
# 沪深港通资金流向
df = ak.stock_hsgt_north_net_flow_in_em()
today = df.iloc[-1]
net_flow = today['净流入'] if '净流入' in today.index else today.get('当日成交净买入', 0)
return float(net_flow) if net_flow else 0
except:
return None
def detect_rotation_signal(sector_df, north_flow):
"""生成轮动信号"""
if sector_df is None or len(sector_df) == 0:
return "无信号"
top_sectors = sector_df.head(5)['板块名称'].tolist()
if north_flow > 50:
signal = "北向大幅买入 → 关注消费/科技"
level = "buy"
elif north_flow < -30:
signal = "北向大幅卖出 → 防御为主"
level = "sell"
elif sector_df['涨跌幅'].std() > 3:
signal = f"板块分化加剧，强者: {top_sectors[0]}"
level = "warn"
else:
signal = "市场平稳，观望为主"
level = "neutral"
return signal, level, top_sectors
print("=== A股板块轮动分析 ===")
sector_df = get_sector_momentum(20)
print(sector_df.head(10))
north = get_north_bound_flow()
print(f"\n北向资金: {north}亿")
if north:
signal, level, tops = detect_rotation_signal(sector_df, north)
print(f"轮动信号: {signal}")