A股量化分析实战指南 (2026年5月·第一期)
更新日期: 2026年5月3日 | 版本: v1 角色: 专注A股市场的资深量化分析师 数据源: akshare 真实数据（新浪源 stock_zh_a_daily） 成熟度评估: 技术面 ✅ 生产就绪 | 资金面 ✅ 生产就绪 | 舆情面 🧪 实验性
1. 分析框架总览
1.1 三维分析模型
本指南采用"技术面+资金面+舆情面"三维分析模型，每个维度独立评分后综合决策：
1.2 决策仪表盘结构
┌─────────────────────────────────────────────┐
│ 决策仪表盘 │
├─────────────────────────────────────────────┤
│ 关键数据看板: 现价 | 涨跌幅 | 换手率 | 市盈率 │
│ 技术面信号: MACD金叉 | KDJ超买 | 量增价升 │
│ 资金面信号: 北向净流入 | 主力净买入 │
│ 舆情面信号: 板块热度高 | 散户情绪偏乐观 │
│ 综合评分: 72/100 (偏多) │
│ 操作建议: 观察区间 45.00-52.00 │
│ 风险提示: 1.板块退潮 2.业绩不及预期 │
└─────────────────────────────────────────────┘
2. 技术面分析模块
1.板块退潮 2.业绩不及预期 │ └─────────────────────────────────────────────┘
---
## 2. 技术面分析模块### 2.1 均线系统
```python
import akshare as ak
import pandas as pd
import numpy as np
def analyze_moving_averages(stock_code: str) -> dict:
"""均线系统分析"""
df = ak.stock_zh_a_daily(symbol=stock_code, adjust="qfq")
df = df.tail(120) # 取最近120个交易日
# 计算均线
df['MA5'] = df['close'].rolling(5).mean()
df['MA10'] = df['close'].rolling(10).mean()
df['MA20'] = df['close'].rolling(20).mean()
df['MA60'] = df['close'].rolling(60).mean()
df['MA120'] = df['close'].rolling(120).mean()
latest = df.iloc[-1]
prev = df.iloc[-2]
# 多头排列检测: MA5 > MA10 > MA20 > MA60 > MA120
ma_bullish = (latest['MA5'] > latest['MA10'] >
latest['MA20'] > latest['MA60'] > latest['MA120'])
# 金叉/死叉检测
ma5_10_golden = (prev['MA5'] <= prev['MA10'] and
latest['MA5'] > latest['MA10'])
ma5_10_death = (prev['MA5'] >= prev['MA10'] and
latest['MA5'] < latest['MA10'])
return {
"current_price": round(latest['close'], 2),
"MA5": round(latest['MA5'], 2),
"MA20": round(latest['MA20'], 2),
"MA60": round(latest['MA60'], 2),
"trend": "多头排列" if ma_bullish else "非多头排列",
"ma5_10_cross": "金叉" if ma5_10_golden else ("死叉" if ma5_10_death else "无交叉"),
"price_vs_ma20": "上方" if latest['close'] > latest['MA20'] else "下方",
}
2.2 MACD/KDJ/RSI 指标共振```python
def analyze_indicators_resonance(stock_code: str) -> dict: """MACD/KDJ/RSI 指标共振分析""" df = ak.stock_zh_a_daily(symbol=stock_code, adjust="qfq") df = df.tail(120)
# MACD计算
ema12 = df['close'].ewm(span=12).mean()
ema26 = df['close'].ewm(span=26).mean()
df['DIF'] = ema12 - ema26
df['DEA'] = df['DIF'].ewm(span=9).mean()
df['MACD'] = 2 * (df['DIF'] - df['DEA'])
# KDJ计算
low_9 = df['low'].rolling(9).min()
high_9 = df['high'].rolling(9).max()
rsv = (df['close'] - low_9) / (high_9 - low_9) * 100
df['K'] = rsv.ewm(com=2).mean()
df['D'] = df['K'].ewm(com=2).mean()
df['J'] = 3 * df['K'] - 2 * df['D']
# RSI计算
delta = df['close'].diff()
gain = delta.where(delta > 0, 0).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
df['RSI'] = 100 - (100 / (1 + gain / loss))
latest = df.iloc[-1]
prev = df.iloc[-2]
# MACD金叉/死叉
macd_signal = "金叉" if (prev['DIF'] <= prev['DEA'] and latest['DIF'] > latest['DEA']) else \
"死叉" if (prev['DIF'] >= prev['DEA'] and latest['DIF'] < latest['DEA']) else \
("多头" if latest['DIF'] > latest['DEA'] else "空头")
# KDJ状态
kdj_signal = "超买" if latest['J'] > 80 else \
"超卖" if latest['J'] < 20 else \
"正常"
# RSI状态
rsi_signal = "超买" if latest['RSI'] > 70 else \
"超卖" if latest['RSI'] < 30 else \
"正常
if latest['RSI'] > 70 else \ "超卖" if latest['RSI'] < 30 else \ "正常if latest['RSI'] > 70 else \ "超卖" if latest['RSI'] < 30 else \ "正常"
# 共振评分：多个指标同时发出看多/看空信号
bullish_signals = sum([
1 if macd_signal == "多头" else 0,
1 if kdj_signal == "超卖" else 0,
1 if latest['RSI'] < 40 else 0,
])
bearish_signals = sum([
1 if macd_signal == "空头" else 0,
1 if kdj_signal == "超买" else 0,
1 if latest['RSI'] > 60 else 0,
])
return {
"MACD": macd_signal,
"KDJ": kdj_signal,
"K_value": round(latest['K'], 2),
"D_value": round(latest['D'], 2),
"J_value": round(latest['J'], 2),
"RSI": rsi_signal,
"RSI_value": round(latest['RSI'], 2),
"bullish_resonance": bullish_signals,
"bearish_resonance": bearish_signals,
}
"bullish_resonance": bullish_signals,
"bearish_resonance": bearish_signals,
}### 2.3 量价关系分析
python
def analyze_volume_price(stock_code: str) -> dict:
"""量价关系分析"""
df = ak.stock_zh_a_daily(symbol=stock_code, adjust="qfq")
df = df.tail(60)
# 计算均量
df['MA5_VOL'] = df['volume'].rolling(5).mean()
df['MA20_VOL'] = df['volume'].rolling(20).mean()
latest = df.iloc[-1]
vol_vs_ma5 = latest['volume'] / latest['MA5_VOL']
vol_vs_ma20 = latest['volume'] / latest['MA20_VOL']
price_change = (latest['close'] - df.iloc[-2]['close']) / df.iloc[-2]['close'] * 100
# 量价关系判断
if vol_vs_ma5 > 2.0 and price_change > 3:
vol_price = "量增价升（看多）"
elif vol_vs_ma5 > 2.0 and price_change < -3:
vol_price = "量增价跌（看空）"
elif vol_vs_ma5 < 0.5 and abs(price_change) < 1:
vol_price = "量缩盘整（观望）"
elif vol_vs_ma5 > 3.0:
vol_price = "天量（注意反转风险）"
elif vol_vs_ma5 < 0.3:
vol_price = "地量（可能见底）"
else:
vol_price = "正常量价关系"
return {
"volume_ratio_vs_ma5": round(vol_vs_ma5, 2),
"volume_ratio_vs_ma20": round(vol_vs_ma20, 2),
"price_change_pct": round(price_change, 2),
"volume_price_relation": vol_price,
"avg_volume_5d": int(latest['MA5_VOL']),
}
3. 资金面分析模块
relation": vol_price, "avg_volume_5d": int(latest['MA5_VOL']), }
---
## 3. 资金面分析模块### 3.1 北向资金
```python
def analyze_northbound_flow() -> dict:
"""北向资金整体流向分析"""
# 获取北向资金历史数据
df = ak.stock_hsgt_north_net_flow_in_em(symbol="北上")
df = df.tail(30)
latest_5d = df.iloc[-5:]
net_inflow_5d = latest_5d['当日净流入'].sum()
net_inflow_20d = df.iloc[-20:]['当日净流入'].sum()
return {
"north_5d_net": round(net_inflow_5d / 10000, 2), # 亿元
"north_20d_net": round(net_inflow_20d / 10000, 2),
"trend": "持续流入" if net_inflow_5d > 0 and net_inflow_20d > 0 else \
"持续流出" if net_inflow_5d < 0 and net_inflow_20d < 0 else \
"震荡",
}
3.2 个股资金流向
def analyze_individual_fund_flow(stock_code: str) -> dict:
"""个股资金流向分析"""
# 获取个股资金流向
df = ak.stock_individual_fund_flow(stock=stock_code, market="sh")
latest = df.iloc[-1]
main_net = latest['主力净流入-净额']
retail_net = latest['散户净流入-净额']
return {
"main_net_inflow": round(main_net / 10000, 2), # 万元转亿元
"retail_net_inflow": round(retail_net / 10000, 2),
"main_trend": "净买入" if main_net > 0 else "净卖出",
}
4. 舆情面分析模块
t / 10000, 2), "main_trend": "净买入" if main_net > 0 else "净卖出", }
---
## 4. 舆情面分析模块### 4.1 换手率情绪指标
```python
def analyze_turnover_sentiment(stock_code: str) -> dict:
"""基于换手率的情绪分析"""
df = ak.stock_zh_a_daily(symbol=stock_code, adjust="qfq")
df = df.tail(30)
# 计算流通市值估算（简化）
df['turnover_rate'] = df['volume'] / df['volume'].rolling(30).mean() * 100
latest_turnover = df.iloc[-1]['turnover_rate']
sentiment = "极度活跃" if latest_turnover > 200 else \
"活跃" if latest_turnover > 150 else \
"正常" if latest_turnover > 80 else \
"低迷"
return {
"turnover_vs_avg": round(latest_turnover, 2),
"sentiment": sentiment,
}
"turnover_vs_avg": round(latest_turnover, 2),
"sentiment": sentiment,
}
---## 5. 综合决策仪表盘生成python
def generate_decision_dashboard(stock_code: str, stock_name: str) -> str:
"""生成综合决策仪表盘"""
ma = analyze_moving_averages(stock_code)
ind = analyze_indicators_resonance(stock_code)
vp = analyze_volume_price(stock_code)
# 计算综合评分
score = 50 # 基础分
# 技术面加分
if ma['trend'] == "多头排列": score += 10
if ma['ma5_10_cross'] == "金叉": score += 5
if ma['price_vs_ma20'] == "上方": score += 5
if ind['MACD'] == "多头": score += 5
if ind['MACD'] == "金叉": score += 10
if ind['bullish_resonance'] >= 2: score += 10
if "量增价升" in vp['volume_price_relation']: score += 5
# 技术面减分
if ma['trend'] != "多头排列": score -= 5
if ind['KDJ'] == "超买": score -= 5
if ind['RSI'] == "超买": score -= 5
if "天量" in vp['volume_price_relation']: score -= 10
# 综合评估
if score >= 75:
recommendation = "偏多：建议关注，可逢低布局"
elif score >= 60:
recommendation = "中性偏多：观望为主，等待确认信号"
elif score >= 45:
recommendation = "中性偏空：谨慎观望，不宜介入"
else:
recommendation = "偏空：建议回避，等待底部信号"
return f"""
═══════════════════════════════════════════
{stock_name}({stock_code}) 决策仪表盘
═══════════════════════════════════════════
📊 关键数据看板
现价: {ma['current_price']} 元
均线状态: {ma['trend']}
5日均线: {ma['MA5']} | 20日均线: {ma['MA20']}
据看板
现价: {ma['current_price']} 元
均线状态: {ma['trend']}
5日均线: {ma['MA5']} | 20日均线: {ma['MA20']}📈 技术指标信号
MACD: {ind['MACD']} | KDJ: {ind['KDJ']}(J值:{ind['J_value']})
RSI: {ind['RSI']}({ind['RSI_value']})
量价关系: {vp['volume_price_relation']}
量比(5日均量): {vp['volume_ratio_vs_ma5']}
💹 多空信号汇总
看多共振: {ind['bullish_resonance']}个信号
看空共振: {ind['bearish_resonance']}个信号
🎯 操作建议区间
支撑位参考: {ma['MA20']} (20日均线)
压力位参考: {round(ma['current_price'] * 1.05, 2)} (近期高点估算)
观察区间: {round(ma['MA20'] * 0.98, 2)} - {round(ma['current_price'] * 1.05, 2)}
⚠️ 风险提示
1. 技术指标存在滞后性，需结合实时行情判断
2. 如遇市场系统性风险，所有技术信号可能失效
3. 个股可能存在未披露的重大事项影响
📌 综合评分: {score}/100
综合判断: {recommendation}
═══════════════════════════════════════════
免责声明：本报告基于公开数据自动生成，不构成投资建议 """ ```
ecommendation} ═══════════════════════════════════════════
免责声明：本报告基于公开数据自动生成，不构成投资建议 """
7. 严禁事项
- 禁止使用缩写：市盈率不写PE，市净率不写PB
- 禁止直接给出涨跌结论：必须基于数据逻辑推导
- 禁止模拟数据：所有分析必须使用 akshare.stock_zh_a_daily 真实数据
- 禁止忽略风险提示：每个分析必须包含至少1条风险提示
数据来源: akshare（新浪源
stock_zh_a_daily） 成熟度: 技术面和资金面模块已生产就绪，舆情面需进一步验证 网络要求: 需访问finance.sina.com.cn（详见第8节） Wiki链接: A股量化指南
8. 环境与网络要求
8.1 数据源要求
本指南所有分析函数依赖 akshare 库的新浪源 stock_zh_a_daily 函数：
import akshare as ak
# 必须使用正确的 symbol 格式
df = ak.stock_zh_a_daily(symbol="sh000001", adjust="qfq") # 上证指数
df = ak.stock_zh_a_daily(symbol="sz000001", adjust="qfq") # 平安银行
df = ak.stock_zh_a_daily(symbol="sh600000", adjust="qfq") # 浦发银行
8.2 注意事项
- 网络连通性: 新浪数据源需要服务器能访问
finance.sina.com.cn - 频率限制: 大量抓取容易封IP，建议控制请求频率（每次请求间隔≥1秒）
- 备用数据源: 如新浪源不可用，可切换至东方财富源
ak.stock_zh_a_hist() - 数据验证: 使用前先验证数据可用性:
len(df) > 100