一、完整的Python脚本

import pandas as pd
import numpy as np
import talib
import akshare as ak
import time
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')

class MACDAnalyzer:
    def __init__(self):
        self.results = []
    
    def calculate_macd(self, df, fastperiod=12, slowperiod=26, signalperiod=9):
        """计算MACD指标"""
        if len(df) < max(fastperiod, slowperiod, signalperiod):
            return None
        
        # 计算MACD
        macd, signal, hist = talib.MACD(
            df['close'].values,
            fastperiod=fastperiod,
            slowperiod=slowperiod,
            signalperiod=signalperiod
        )
        
        df['macd'] = macd
        df['signal'] = signal
        df['hist'] = hist
        
        return df
    
    def detect_golden_cross(self, df, days=2):
        """检测最近N天内的MACD金叉"""
        if len(df) < days + 1:
            return False, []
        
        golden_cross_days = []
        
        # 检查最近days天是否有金叉
        for i in range(-days, 0):
            if i == -1:  # 最后一天（今天）
                current_macd = df['macd'].iloc[i]
                current_signal = df['signal'].iloc[i]
                prev_macd = df['macd'].iloc[i-1]
                prev_signal = df['signal'].iloc[i-1]
                
                # 金叉条件：MACD从下向上穿过Signal线
                if (prev_macd <= prev_signal) and (current_macd > current_signal):
                    golden_cross_days.append(len(df) + i)  # 记录日期索引
            
            elif i < -1:  # 倒数第2天到倒数第days天
                current_macd = df['macd'].iloc[i]
                current_signal = df['signal'].iloc[i]
                prev_macd = df['macd'].iloc[i-1]
                prev_signal = df['signal'].iloc[i-1]
                
                if (prev_macd <= prev_signal) and (current_macd > current_signal):
                    golden_cross_days.append(len(df) + i)
        
        return len(golden_cross_days) > 0, golden_cross_days
    
    def analyze_volume_trend(self, df, cross_day_index):
        """分析金叉前后的成交量趋势"""
        if cross_day_index < 3 or cross_day_index >= len(df) - 1:
            return {
                'volume_trend': 'insufficient_data',
                'volume_change_pct': 0,
                'avg_volume_before': 0,
                'avg_volume_after': 0
            }
        
        # 金叉前3天的平均成交量
        before_start = max(0, cross_day_index - 3)
        before_end = cross_day_index
        avg_volume_before = df['volume'].iloc[before_start:before_end].mean()
        
        # 金叉后2天的平均成交量（包括金叉当天）
        after_start = cross_day_index
        after_end = min(len(df), cross_day_index + 3)
        avg_volume_after = df['volume'].iloc[after_start:after_end].mean()
        
        # 计算成交量变化百分比
        if avg_volume_before > 0:
            volume_change_pct = ((avg_volume_after - avg_volume_before) / avg_volume_before) * 100
        else:
            volume_change_pct = 0
        
        # 判断趋势
        if volume_change_pct > 20:
            trend = '显著放大'
        elif volume_change_pct > 5:
            trend = '温和放大'
        elif volume_change_pct > -5:
            trend = '基本持平'
        elif volume_change_pct > -20:
            trend = '温和萎缩'
        else:
            trend = '显著萎缩'
        
        return {
            'volume_trend': trend,
            'volume_change_pct': round(volume_change_pct, 2),
            'avg_volume_before': round(avg_volume_before, 2),
            'avg_volume_after': round(avg_volume_after, 2)
        }
    
    def get_stock_data(self, symbol, period=60):
        """获取股票历史数据"""
        try:
            # 获取最近60个交易日的数据
            end_date = datetime.now().strftime('%Y%m%d')
            start_date = (datetime.now() - timedelta(days=period)).strftime('%Y%m%d')
            
            # 使用akshare获取A股数据
            if symbol.startswith(('00', '30', '60')):
                df = ak.stock_zh_a_hist(
                    symbol=symbol,
                    period="daily",
                    start_date=start_date,
                    end_date=end_date,
                    adjust="qfq"
                )
                
                if df is not None and not df.empty:
                    # 重命名列以匹配预期格式
                    df = df.rename(columns={
                        '日期': 'date',
                        '开盘': 'open',
                        '收盘': 'close',
                        '最高': 'high',
                        '最低': 'low',
                        '成交量': 'volume'
                    })
                    df['date'] = pd.to_datetime(df['date'])
                    df = df.sort_values('date').reset_index(drop=True)
                    return df
        except Exception as e:
            print(f"获取股票 {symbol} 数据失败: {e}")
            return None
        
        return None
    
    def analyze_single_stock(self, symbol, stock_name):
        """分析单个股票"""
        try:
            # 获取数据
            df = self.get_stock_data(symbol)
            if df is None or len(df) < 30:
                return None
            
            # 计算MACD
            df = self.calculate_macd(df)
            if df is None:
                return None
            
            # 检测金叉
            has_cross, cross_days = self.detect_golden_cross(df, days=2)
            if not has_cross:
                return None
            
            # 分析最近一次金叉
            latest_cross_day = cross_days[-1]
            cross_date = df['date'].iloc[latest_cross_day]
            
            # 分析成交量趋势
            volume_analysis = self.analyze_volume_trend(df, latest_cross_day)
            
            # 获取价格信息
            current_price = df['close'].iloc[-1]
            price_change_pct = ((current_price - df['close'].iloc[-2]) / df['close'].iloc[-2]) * 100
            
            result = {
                'symbol': symbol,
                'name': stock_name,
                'cross_date': cross_date.strftime('%Y-%m-%d'),
                'current_price': round(current_price, 2),
                'price_change_pct': round(price_change_pct, 2),
                'volume_trend': volume_analysis['volume_trend'],
                'volume_change_pct': volume_analysis['volume_change_pct'],
                'macd_value': round(df['macd'].iloc[-1], 4),
                'signal_value': round(df['signal'].iloc[-1], 4),
                'hist_value': round(df['hist'].iloc[-1], 4)
            }
            
            return result
            
        except Exception as e:
            print(f"分析股票 {symbol} 失败: {e}")
            return None
    
    def get_all_stocks(self):
        """获取所有A股股票列表"""
        try:
            # 获取沪深A股列表
            stock_info = ak.stock_info_a_code_name()
            return stock_info
        except Exception as e:
            print(f"获取股票列表失败: {e}")
            return None
    
    def run_analysis(self, limit=50):
        """运行完整分析"""
        print("开始获取股票列表...")
        stocks = self.get_all_stocks()
        if stocks is None:
            print("无法获取股票列表")
            return []
        
        print(f"找到 {len(stocks)} 只股票，开始分析...")
        
        results = []
        processed = 0
        
        # 限制分析数量以避免API限制
        stocks_to_analyze = stocks.head(limit)
        
        for idx, row in stocks_to_analyze.iterrows():
            symbol = row['code']
            name = row['name']
            
            # 跳过ST股票和退市股票
            if 'ST' in name or '退' in name:
                continue
            
            result = self.analyze_single_stock(symbol, name)
            if result:
                results.append(result)
                print(f"发现金叉股票: {symbol} {name} - 金叉日期: {result['cross_date']}")
            
            processed += 1
            if processed % 10 == 0:
                print(f"已处理 {processed}/{len(stocks_to_analyze)} 只股票")
            
            # 添加延迟避免API限制
            time.sleep(0.5)
        
        return results
    
    def save_results(self, results, filename=None):
        """保存结果到CSV文件"""
        if not filename:
            filename = f"macd_golden_cross_{datetime.now().strftime('%Y%m%d')}.csv"
        
        if results:
            df = pd.DataFrame(results)
            df = df.sort_values('cross_date', ascending=False)
            df.to_csv(filename, index=False, encoding='utf-8-sig')
            print(f"结果已保存到: {filename}")
        else:
            print("没有找到符合条件的股票")

def main():
    """主函数"""
    analyzer = MACDAnalyzer()
    
    print("=" * 60)
    print("MACD金叉股票分析工具")
    print("=" * 60)
    
    # 运行分析（限制分析50只股票以演示）
    results = analyzer.run_analysis(limit=50)
    
    if results:
        print(f"\n找到 {len(results)} 只最近2天出现MACD金叉的股票:")
        print("-" * 80)
        
        # 显示结果
        for result in results:
            print(f"{result['symbol']} {result['name']}")
            print(f"  金叉日期: {result['cross_date']}")
            print(f"  当前价格: {result['current_price']} ({result['price_change_pct']:+.2f}%)")
            print(f"  成交量趋势: {result['volume_trend']} ({result['volume_change_pct']:+.2f}%)")
            print(f"  MACD: {result['macd_value']:.4f}, Signal: {result['signal_value']:.4f}")
            print("-" * 80)
        
        # 保存结果
        analyzer.save_results(results)
        
    else:
        print("最近2天没有找到MACD金叉的股票")

if __name__ == "__main__":
    main()

使用说明

 talib 库下载地址https://github.com/cgohlke/talib-build/releases

1. 基础使用

python

# 运行基础分析
python macd_analyzer.py

2. 自定义参数

python

# 修改分析参数
analyzer = MACDAnalyzer()
results = analyzer.run_analysis(limit=100)  # 分析100只股票

3. 输出示例

Plain Text

发现金叉股票: 600519 贵州茅台 - 金叉日期: 2024-01-15
  当前价格: 1850.25 (+2.35%)
  成交量趋势: 温和放大 (+15.67%)
  MACD: 12.4567, Signal: 10.2345

注意事项

1. API限制

• akshare有请求频率限制

• 建议添加适当的延迟（time.sleep）

• 可以分批处理大量股票

2. 数据质量

• 确保有足够的历史数据（至少30天）

• 处理停牌、退市等特殊情况

• 验证数据完整性

3. 技术指标参数

• MACD默认参数：(12, 26, 9)

• 可根据需要调整参数

• 不同市场可能需要不同参数

4. 风险提示

• MACD金叉只是技术信号之一

• 需要结合其他指标和基本面分析

• 过去表现不代表未来收益

这个脚本可以自动化的获取、分析和保存MACD金叉股票数据，帮助您快速识别潜在的投资机会。