Python量化实战:用AKShare计算MACD金叉死叉,精准捕捉买卖点!

💡 导语：MACD是技术分析中最经典的指标之一，但手动计算太麻烦？今天教你用Python + AKShare自动计算MACD，识别金叉死叉，让量化交易变得简单！

一、MACD指标原理详解

MACD（Moving Average Convergence Divergence，指数平滑异同移动平均线）由Gerald Appel于1970年代提出，是判断趋势强度和买卖时机的重要工具。

1.1 MACD三大组成部分

1.2 金叉与死叉

📈 金叉（买入信号）：DIF 上穿 DEA
   - 发生在零轴上方：强势金叉，买入信号更强
   - 发生在零轴下方：弱势金叉，可能是反弹

📉 死叉（卖出信号）：DIF 下穿 DEA
   - 发生在零轴上方：高位死叉，警惕回调
   - 发生在零轴下方：弱势死叉，下跌趋势延续

二、环境准备与数据获取

2.1 安装依赖

pip install akshare pandas numpy matplotlib

2.2 使用AKShare获取股票数据

AKShare是开源的Python财经数据接口库，完全免费，无需注册API Key。

import akshare as ak
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta

# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False

# 获取股票数据（以贵州茅台为例）
stock_code = "600519"  # 贵州茅台
start_date = (datetime.now() - timedelta(days=365)).strftime("%Y%m%d")
end_date = datetime.now().strftime("%Y%m%d")

# 使用AKShare获取日线数据
df = ak.stock_zh_a_hist(
    symbol=stock_code,
    period="daily",
    start_date=start_date,
    end_date=end_date,
    adjust="qfq"  # 前复权
)

df['日期'] = pd.to_datetime(df['日期'])
df.set_index('日期', inplace=True)
df.rename(columns={
    '开盘': 'open',
    '收盘': 'close',
    '最高': 'high',
    '最低': 'low',
    '成交量': 'volume'
}, inplace=True)

print(f"获取到 {len(df)} 条数据")
print(df.head())

三、MACD计算实战

3.1 手动计算MACD

def calculate_macd(df, fast=12, slow=26, signal=9):
    """
    计算MACD指标
    
    参数:
        df: DataFrame，包含收盘价
        fast: 快线周期，默认12
        slow: 慢线周期，默认26
        signal: 信号线周期，默认9
    """
    df = df.copy()
    
    # 计算EMA
    df['ema_fast'] = df['close'].ewm(span=fast, adjust=False).mean()
    df['ema_slow'] = df['close'].ewm(span=slow, adjust=False).mean()
    
    # 计算DIF（快线）
    df['DIF'] = df['ema_fast'] - df['ema_slow']
    
    # 计算DEA（慢线/信号线）
    df['DEA'] = df['DIF'].ewm(span=signal, adjust=False).mean()
    
    # 计算MACD柱状图
    df['MACD'] = 2 * (df['DIF'] - df['DEA'])
    
    return df

# 计算MACD
df = calculate_macd(df)
print(df[['close', 'DIF', 'DEA', 'MACD']].tail(10))

3.2 使用TA-Lib快速计算（可选）

如果你安装了TA-Lib库，可以更简洁地计算：

import talib

# 使用TA-Lib计算MACD
dif, dea, macd = talib.MACD(
    df['close'].values,
    fastperiod=12,
    slowperiod=26,
    signalperiod=9
)

df['DIF_talib'] = dif
df['DEA_talib'] = dea
df['MACD_talib'] = macd * 2  # TA-Lib的MACD是(DIF-DEA)，需要乘以2

四、识别金叉死叉信号

4.1 信号检测逻辑

def find_signals(df):
    """
    识别金叉和死叉信号
    """
    df = df.copy()
    
    # 金叉：DIF上穿DEA（今日DIF>DEA，昨日DIF<=DEA）
    df['golden_cross'] = (
        (df['DIF'] > df['DEA']) & 
        (df['DIF'].shift(1) <= df['DEA'].shift(1))
    )
    
    # 死叉：DIF下穿DEA（今日DIF<DEA，昨日DIF>=DEA）
    df['death_cross'] = (
        (df['DIF'] < df['DEA']) & 
        (df['DIF'].shift(1) >= df['DEA'].shift(1))
    )
    
    # 判断金叉位置（零轴上方还是下方）
    df['gc_above_zero'] = df['golden_cross'] & (df['DIF'] > 0)
    df['gc_below_zero'] = df['golden_cross'] & (df['DIF'] <= 0)
    
    # 判断死叉位置
    df['dc_above_zero'] = df['death_cross'] & (df['DIF'] > 0)
    df['dc_below_zero'] = df['death_cross'] & (df['DIF'] <= 0)
    
    return df

# 识别信号
df = find_signals(df)

# 查看最近的信号
recent_signals = df[df['golden_cross'] | df['death_cross']].tail(10)
print("\n最近的交易信号：")
print(recent_signals[['close', 'DIF', 'DEA', 'golden_cross', 'death_cross']])

4.2 统计信号效果

def analyze_signals(df, hold_days=5):
    """
    分析金叉死叉信号的后续收益
    """
    results = []
    
    # 分析金叉
    for idx in df[df['golden_cross']].index:
        entry_price = df.loc[idx, 'close']
        future_idx = df.index.get_loc(idx) + hold_days
        
        if future_idx < len(df):
            exit_price = df.iloc[future_idx]['close']
            return_pct = (exit_price - entry_price) / entry_price * 100
            results.append({
                'date': idx,
                'signal': '金叉',
                'entry_price': entry_price,
                'exit_price': exit_price,
                'return_pct': return_pct
            })
    
    # 分析死叉
    for idx in df[df['death_cross']].index:
        entry_price = df.loc[idx, 'close']
        future_idx = df.index.get_loc(idx) + hold_days
        
        if future_idx < len(df):
            exit_price = df.iloc[future_idx]['close']
            return_pct = (exit_price - entry_price) / entry_price * 100
            results.append({
                'date': idx,
                'signal': '死叉',
                'entry_price': entry_price,
                'exit_price': exit_price,
                'return_pct': return_pct
            })
    
    results_df = pd.DataFrame(results)
    
    if len(results_df) > 0:
        print(f"\n信号统计（持有{hold_days}天）：")
        print(f"金叉平均收益: {results_df[results_df['signal']=='金叉']['return_pct'].mean():.2f}%")
        print(f"死叉平均收益: {results_df[results_df['signal']=='死叉']['return_pct'].mean():.2f}%")
        print(f"金叉胜率: {(results_df[results_df['signal']=='金叉']['return_pct'] > 0).mean()*100:.1f}%")
        print(f"死叉胜率: {(results_df[results_df['signal']=='死叉']['return_pct'] > 0).mean()*100:.1f}%")
    
    return results_df

# 分析信号效果
signal_analysis = analyze_signals(df, hold_days=5)

五、可视化：绘制MACD图表

5.1 完整可视化代码

def plot_macd(df, stock_name="股票", save_path=None):
    """
    绘制股票价格和MACD指标
    """
    fig, axes = plt.subplots(3, 1, figsize=(14, 12), 
                             gridspec_kw={'height_ratios': [3, 1, 1]})
    
    # 获取最近120天的数据用于展示
    plot_df = df.tail(120).copy()
    
    # ========== 子图1：K线 + 买卖点 ==========
    ax1 = axes[0]
    
    # 绘制收盘价
    ax1.plot(plot_df.index, plot_df['close'], 
             label='收盘价', color='black', linewidth=1.5)
    
    # 标注金叉（买入点）
    golden_crosses = plot_df[plot_df['golden_cross']]
    if len(golden_crosses) > 0:
        ax1.scatter(golden_crosses.index, golden_crosses['close'], 
                   color='red', marker='^', s=150, label='金叉买入', zorder=5)
        # 添加价格标签
        for idx, row in golden_crosses.iterrows():
            ax1.annotate(f'买入\n{row["close"]:.1f}', 
                        xy=(idx, row['close']),
                        xytext=(0, 20), textcoords='offset points',
                        ha='center', fontsize=8, color='red',
                        arrowprops=dict(arrowstyle='->', color='red', lw=1))
    
    # 标注死叉（卖出点）
    death_crosses = plot_df[plot_df['death_cross']]
    if len(death_crosses) > 0:
        ax1.scatter(death_crosses.index, death_crosses['close'], 
                   color='green', marker='v', s=150, label='死叉卖出', zorder=5)
        # 添加价格标签
        for idx, row in death_crosses.iterrows():
            ax1.annotate(f'卖出\n{row["close"]:.1f}', 
                        xy=(idx, row['close']),
                        xytext=(0, -35), textcoords='offset points',
                        ha='center', fontsize=8, color='green',
                        arrowprops=dict(arrowstyle='->', color='green', lw=1))
    
    ax1.set_title(f'{stock_name} - MACD金叉死叉买卖点分析', fontsize=14, fontweight='bold')
    ax1.set_ylabel('价格（元）', fontsize=11)
    ax1.legend(loc='upper left')
    ax1.grid(True, alpha=0.3)
    
    # ========== 子图2：DIF和DEA线 ==========
    ax2 = axes[1]
    ax2.plot(plot_df.index, plot_df['DIF'], 
             label='DIF（快线）', color='blue', linewidth=1.5)
    ax2.plot(plot_df.index, plot_df['DEA'], 
             label='DEA（慢线）', color='orange', linewidth=1.5)
    ax2.axhline(y=0, color='gray', linestyle='--', alpha=0.5)
    
    # 标注金叉死叉位置
    for idx in plot_df[plot_df['golden_cross']].index:
        ax2.axvline(x=idx, color='red', linestyle=':', alpha=0.5)
    for idx in plot_df[plot_df['death_cross']].index:
        ax2.axvline(x=idx, color='green', linestyle=':', alpha=0.5)
    
    ax2.set_ylabel('DIF/DEA', fontsize=11)
    ax2.legend(loc='upper left')
    ax2.grid(True, alpha=0.3)
    
    # ========== 子图3：MACD柱状图 ==========
    ax3 = axes[2]
    
    # 绘制MACD柱状图（红色表示正值，绿色表示负值）
    positive = plot_df['MACD'] >= 0
    negative = plot_df['MACD'] < 0
    
    ax3.bar(plot_df[positive].index, plot_df[positive]['MACD'], 
            color='red', alpha=0.7, label='MACD>0')
    ax3.bar(plot_df[negative].index, plot_df[negative]['MACD'], 
            color='green', alpha=0.7, label='MACD<0')
    ax3.axhline(y=0, color='black', linewidth=0.5)
    
    # 标注金叉死叉位置
    for idx in plot_df[plot_df['golden_cross']].index:
        ax3.axvline(x=idx, color='red', linestyle=':', alpha=0.5)
    for idx in plot_df[plot_df['death_cross']].index:
        ax3.axvline(x=idx, color='green', linestyle=':', alpha=0.5)
    
    ax3.set_ylabel('MACD柱状图', fontsize=11)
    ax3.set_xlabel('日期', fontsize=11)
    ax3.legend(loc='upper left')
    ax3.grid(True, alpha=0.3)
    
    plt.tight_layout()
    
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches='tight', facecolor='white')
        print(f"图表已保存至: {save_path}")
    
    plt.show()
    return fig

# 绘制图表
plot_macd(df, stock_name="贵州茅台(600519)", 
          save_path="macd_analysis.png")

六、完整可运行代码

以下是整合所有功能的完整代码，复制即可运行：

"""
MACD金叉死叉实战 - 完整代码
作者: AI成长
功能: 使用AKShare获取股票数据，计算MACD，识别买卖信号
"""

import akshare as ak
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta

# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False


def get_stock_data(stock_code, days=365):
    """获取股票历史数据"""
    end_date = datetime.now().strftime("%Y%m%d")
    start_date = (datetime.now() - timedelta(days=days)).strftime("%Y%m%d")
    
    df = ak.stock_zh_a_hist(
        symbol=stock_code,
        period="daily",
        start_date=start_date,
        end_date=end_date,
        adjust="qfq"
    )
    
    df['日期'] = pd.to_datetime(df['日期'])
    df.set_index('日期', inplace=True)
    df.rename(columns={
        '开盘': 'open',
        '收盘': 'close',
        '最高': 'high',
        '最低': 'low',
        '成交量': 'volume'
    }, inplace=True)
    
    return df


def calculate_macd(df, fast=12, slow=26, signal=9):
    """计算MACD指标"""
    df = df.copy()
    df['ema_fast'] = df['close'].ewm(span=fast, adjust=False).mean()
    df['ema_slow'] = df['close'].ewm(span=slow, adjust=False).mean()
    df['DIF'] = df['ema_fast'] - df['ema_slow']
    df['DEA'] = df['DIF'].ewm(span=signal, adjust=False).mean()
    df['MACD'] = 2 * (df['DIF'] - df['DEA'])
    return df


def find_signals(df):
    """识别金叉和死叉信号"""
    df = df.copy()
    df['golden_cross'] = (df['DIF'] > df['DEA']) & (df['DIF'].shift(1) <= df['DEA'].shift(1))
    df['death_cross'] = (df['DIF'] < df['DEA']) & (df['DIF'].shift(1) >= df['DEA'].shift(1))
    return df


def plot_macd(df, stock_name="股票", save_path=None):
    """绘制MACD分析图表"""
    fig, axes = plt.subplots(3, 1, figsize=(14, 12), 
                             gridspec_kw={'height_ratios': [3, 1, 1]})
    
    plot_df = df.tail(120).copy()
    
    # 价格图 + 买卖点
    ax1 = axes[0]
    ax1.plot(plot_df.index, plot_df['close'], label='收盘价', color='black', linewidth=1.5)
    
    golden_crosses = plot_df[plot_df['golden_cross']]
    if len(golden_crosses) > 0:
        ax1.scatter(golden_crosses.index, golden_crosses['close'], 
                   color='red', marker='^', s=150, label='金叉买入', zorder=5)
    
    death_crosses = plot_df[plot_df['death_cross']]
    if len(death_crosses) > 0:
        ax1.scatter(death_crosses.index, death_crosses['close'], 
                   color='green', marker='v', s=150, label='死叉卖出', zorder=5)
    
    ax1.set_title(f'{stock_name} - MACD金叉死叉分析', fontsize=14, fontweight='bold')
    ax1.set_ylabel('价格（元）')
    ax1.legend(loc='upper left')
    ax1.grid(True, alpha=0.3)
    
    # DIF/DEA图
    ax2 = axes[1]
    ax2.plot(plot_df.index, plot_df['DIF'], label='DIF（快线）', color='blue', linewidth=1.5)
    ax2.plot(plot_df.index, plot_df['DEA'], label='DEA（慢线）', color='orange', linewidth=1.5)
    ax2.axhline(y=0, color='gray', linestyle='--', alpha=0.5)
    ax2.set_ylabel('DIF/DEA')
    ax2.legend(loc='upper left')
    ax2.grid(True, alpha=0.3)
    
    # MACD柱状图
    ax3 = axes[2]
    positive = plot_df['MACD'] >= 0
    negative = plot_df['MACD'] < 0
    ax3.bar(plot_df[positive].index, plot_df[positive]['MACD'], color='red', alpha=0.7)
    ax3.bar(plot_df[negative].index, plot_df[negative]['MACD'], color='green', alpha=0.7)
    ax3.axhline(y=0, color='black', linewidth=0.5)
    ax3.set_ylabel('MACD')
    ax3.set_xlabel('日期')
    ax3.grid(True, alpha=0.3)
    
    plt.tight_layout()
    
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches='tight', facecolor='white')
        print(f"图表已保存至: {save_path}")
    
    plt.show()
    return fig


# ========== 主程序 ==========
if __name__ == "__main__":
    # 设置股票代码
    stock_code = "600519"  # 贵州茅台，可修改为你关注的股票
    stock_name = "贵州茅台"
    
    print(f"正在获取 {stock_name}({stock_code}) 的数据...")
    
    # 1. 获取数据
    df = get_stock_data(stock_code, days=365)
    print(f"✓ 获取到 {len(df)} 条数据")
    
    # 2. 计算MACD
    df = calculate_macd(df)
    print("✓ MACD计算完成")
    
    # 3. 识别信号
    df = find_signals(df)
    
    # 4. 输出最近信号
    recent_gc = df[df['golden_cross']].tail(3)
    recent_dc = df[df['death_cross']].tail(3)
    
    print(f"\n📈 最近3次金叉:")
    for idx, row in recent_gc.iterrows():
        print(f"   {idx.strftime('%Y-%m-%d')}: 价格 {row['close']:.2f}, DIF={row['DIF']:.3f}")
    
    print(f"\n📉 最近3次死叉:")
    for idx, row in recent_dc.iterrows():
        print(f"   {idx.strftime('%Y-%m-%d')}: 价格 {row['close']:.2f}, DIF={row['DIF']:.3f}")
    
    # 5. 绘制图表
    print("\n正在生成图表...")
    plot_macd(df, stock_name=f"{stock_name}({stock_code})", save_path="macd_analysis.png")
    
    print("\n✅ 分析完成！")

七、实战技巧与注意事项

7.1 使用技巧

1. 1. 多周期验证：结合日线、周线MACD信号，提高准确性
2. 2. 背离判断：价格创新高但MACD未创新高，可能是顶背离
3. 3. 零轴意义：零轴上方为多头市场，下方为空头市场
4. 4. 结合成交量：金叉时放量更可靠

7.2 常见误区

⚠️ 不要盲目跟单：MACD是滞后指标，单独使用胜率有限
⚠️ 震荡市失效：横盘震荡时会产生大量假信号
⚠️ 参数优化：不同股票可能需要调整12/26/9的参数

7.3 进阶应用

# 多周期MACD策略
def multi_timeframe_macd(df_daily, df_weekly):
    """
    多周期验证：周线MACD向上时，只做多日线金叉
    """
    # 实现你的多周期策略...
    pass

# MACD背离检测
def detect_divergence(df, window=20):
    """
    检测价格与MACD的背离
    """
    # 实现背离检测逻辑...
    pass

八、总结

今天我们一起学习了：

✅ MACD指标原理：DIF、DEA、MACD柱的计算方法
✅ 金叉死叉识别：用Python自动检测买卖信号
✅ AKShare数据获取：免费获取A股历史数据
✅ 可视化分析：绘制专业级MACD分析图表
✅ 完整可运行代码：复制即可使用

💪 下一步：尝试修改代码分析你关注的股票，或者将MACD与其他指标（如RSI、KDJ）结合使用！

参考资料

• • AKShare官方文档^[1]
• • Gerald Appel. Technical Analysis: Power Tools for Active Investors

本文代码已在Python 3.9+、AKShare 1.10+环境下测试通过。如有问题欢迎留言交流！

关注「AI成长」，每天一个Python量化技巧！ 🚀

引用链接

[1] AKShare官方文档: https://www.akshare.xyz/