量化投资实战:用 Python 复刻「主力资金监控」指标,捕捉个股起爆点!

import pandas as pdimport numpy as npimport matplotlib.pyplot as pltimport matplotlib.dates as mdatesfrom get_stock_data import df# 设置中文字体plt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = Falsedef EMA(series, period):    """自定义指数移动平均函数"""    return series.ewm(span=period, adjust=False).mean()def calculate_indicators(df):    """    计算所有技术指标（不依赖 talib）    """    # M5: 收盘价    M5 = df['close']    # M6: 数据总数（从首日到当前的计数）    M6 = np.arange(len(df)) + 1    # M7: 成交量加权平均价 (VWAP)    M7 = (df['amount'] * df['close']).cumsum() / df['amount'].cumsum()    # M8: 全周期均线（动态窗口均值）    M8 = df['close'].expanding().mean()    # M12: 单日涨幅超过2%    M12 = df['close'] / df['close'].shift(1) > 1.02    # M13: 单日跌幅超过2%    M13 = df['close'] / df['close'].shift(1) < 0.98    # M14: 近期波动幅度小于5%（横盘特征）    HHV_M5 = df['close'].expanding().max()    LLV_M5 = df['close'].expanding().min()    M14 = (HHV_M5 - LLV_M5) / df['close'] < 0.05    # M15: 成交额的30日指数移动平均    M15 = EMA(df['amount'], 30)    # M16: 成交量相对强度大于95%（放量特征）    LLV_M15 = M15.expanding().min()    HHV_M15 = M15.expanding().max()    M16 = (M15 - LLV_M15) / (HHV_M15 - LLV_M15) > 0.95    # M17: 当日大涨或大跌    M17 = M12 | M13    # M18: 主力关注条件——横盘+放量+VWAP上穿全周期均线    M18 = (~M17) & M14 & M16 & (M7 > M8) & (M7.shift(1) <= M8.shift(1))    # M19: 强势起爆条件——收盘价同时上穿VWAP和全周期均线    cross_close_M7 = (df['close'] > M7) & (df['close'].shift(1) <= M7.shift(1))    cross_close_M8 = (df['close'] > M8) & (df['close'].shift(1) <= M8.shift(1))    M19 = (~M17) & M14 & M16 & cross_close_M7 & cross_close_M8    # AL指标：快慢均线差值动量    EMA8 = EMA(df['close'], 8)    EMA55 = EMA(df['close'], 55)    AL = (EMA8 - EMA55) * 10    # TRK指标：双平滑均线    EMA13_1 = EMA(df['close'], 13)    TRK = EMA(EMA13_1, 13)    # MTQ指标：双均线变化率    MTQ = (TRK - TRK.shift(1)) / TRK.shift(1) * 1000    # 大资金介入判定条件组    VX1 = (df['close'] - M7) / M7 < -0.005      # 价格低于VWAP    VX2 = (df['close'] - M7) / M7 < -0.03       # 明显偏离VWAP    VX3 = (MTQ > MTQ.shift(1)) & (MTQ < -0.2)   # MTQ反弹但仍处低位    EMA_AL_3 = EMA(AL, 3)    VX4 = (AL > EMA_AL_3) & (AL < -0.3)         # AL指标超卖反弹    # 综合判定：大资金介入信号    VX5 = VX1 & VX3 & VX4    VX6 = VX2 & VX3 & VX4    大资金介入 = VX5 & VX6    return {        'M5': M5, 'M6': M6, 'M7': M7, 'M8': M8,        'M12': M12, 'M13': M13, 'M14': M14,        'M15': M15, 'M16': M16, 'M17': M17,        'M18': M18, 'M19': M19, 'AL': AL,        'TRK': TRK, 'MTQ': MTQ,         'VX1': VX1, 'VX2': VX2, 'VX3': VX3, 'VX4': VX4,        '大资金介入': 大资金介入    }

def generate_signals(df):    """    整合指标并生成交易信号表    """    df_copy = df.copy()    # 智能日期处理：兼容索引/列/序号三种模式    if hasattr(df_copy.index, 'strftime'):        dates = df_copy.index    elif 'date' in df_copy.columns:        try:            dates = pd.to_datetime(df_copy['date'])        except:            dates = pd.Series(range(len(df_copy)))    else:        dates = pd.Series(range(len(df_copy)))    indicators = calculate_indicators(df_copy)    results = pd.DataFrame({        'date': dates,        'close': df_copy['close'],        'open': df_copy['open'],        'high': df_copy['high'],        'low': df_copy['low'],        'amount': df_copy['amount']    })    # 批量添加指标列    for key, value in indicators.items():        results[key] = value    # 涨停板标记（对应通达信 STICKLINE）    results['涨停'] = ((df_copy['close'] - df_copy['close'].shift(1)) /                       df_copy['close'].shift(1) * 100 > 9.5)    return results

def visualize_results(signals, stock_code="sz002075"):    """    三图合一可视化：价格走势 + 成交量 + 指标曲线    """    fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(12, 10), sharex=True)    dates = signals['date']    # --- 图1：价格与均线体系 ---    ax1.plot(dates, signals['close'], label='收盘价', linewidth=1, color='black')    ax1.plot(dates, signals['M7'], label='VWAP均价', linewidth=1, color='blue', linestyle='--')    ax1.plot(dates, signals['M8'], label='全周期均线', linewidth=1, color='red', linestyle='--')    # 动态标记各类信号点    if signals['M18'].any():        m18_mask = signals['M18'].fillna(False)        ax1.scatter(dates[m18_mask], signals.loc[m18_mask, 'close'],                    color='yellow', s=50, label='主力关注', marker='^', zorder=5)    if signals['M19'].any():        m19_mask = signals['M19'].fillna(False)        ax1.scatter(dates[m19_mask], signals.loc[m19_mask, 'close'] * 0.99,                    color='red', s=80, label='强势起爆', marker='*', zorder=5)    if signals['大资金介入'].any():        big_money_mask = signals['大资金介入'].fillna(False)        ax1.scatter(dates[big_money_mask], signals.loc[big_money_mask, 'low'] * 0.985,                    color='purple', s=60, label='大资金介入', marker='v', zorder=5)    ax1.set_title(f'{stock_code} - 主力资金监控分析', fontsize=14)    ax1.set_ylabel('价格')    ax1.legend(loc='upper left')    ax1.grid(True, alpha=0.3)    # --- 图2：成交量分析 ---    ax2.bar(dates, signals['amount'], label='成交额', alpha=0.7, color='gray')    ax2.plot(dates, signals['M15'], label='成交额EMA', color='orange', linewidth=2)    ax2.set_ylabel('成交额')    ax2.legend(loc='upper left')    ax2.grid(True, alpha=0.3)    # --- 图3：趋势指标 ---    ax3.plot(dates, signals['AL'], label='AL(均线差值动量)', color='green', linewidth=1)    ax3.plot(dates, signals['MTQ'], label='MTQ(双均线变化率)', color='blue', linewidth=1)    ax3.axhline(-0.3, color='red', linestyle='--', alpha=0.5, label='超卖阈值')    ax3.set_ylabel('指标数值')    ax3.set_xlabel('交易日期')    ax3.legend(loc='upper left')    ax3.grid(True, alpha=0.3)    # 日期刻度优化    if pd.api.types.is_datetime64_any_dtype(dates):        ax1.xaxis.set_major_formatter(mdates.DateFormatter('%m-%d'))        ax1.xaxis.set_major_locator(mdates.DayLocator(interval=5))    plt.xticks(rotation=45)    plt.tight_layout()    return fig

def print_detailed_results(signals):    """    结构化输出信号明细与统计    """    print("="*60)    print("🔍 主力资金信号分析报告")    print("="*60)    signal_cols = ['M18', 'M19', '大资金介入']    has_signals = signals[signal_cols].any(axis=1)    if has_signals.any():        signal_days = signals[has_signals].copy()        # 智能日期格式化        if pd.api.types.is_datetime64_any_dtype(signal_days['date']):            signal_days['date_str'] = signal_days['date'].dt.strftime('%Y-%m-%d')        else:            signal_days['date_str'] = signal_days['date'].astype(str)        print(f"\n📅 发现信号天数: {len(signal_days)}")        print("-"*45)        for _, row in signal_days.iterrows():            date_str = row['date_str']            tags = []            if row['M18']: tags.append('🟡主力关注')            if row['M19']: tags.append('🔴强势起爆')            if row['大资金介入']: tags.append('💜大资金介入')            print(f"{date_str}: {' | '.join(tags)} —— 收盘价:{row['close']:.2f}")    else:        print("\n❌ 本期未检测到有效信号")    print(f"\n📊 信号汇总统计:")    print(f"  • 主力关注信号: {signals['M18'].sum():>2d} 次")    print(f"  • 强势起爆信号: {signals['M19'].sum():>2d} 次")    print(f"  • 大资金介入信号: {signals['大资金介入'].sum():>2d} 次")

if __name__ == "__main__":    # 🛠️ 执行指标计算    signals = generate_signals(df)    print("✅ 指标计算完成！")    print(f"📁 分析周期: {len(signals)} 个交易日")    # 📊 输出详细报告    print_detailed_results(signals)    # 🖼️ 可视化展示    try:        fig = visualize_results(signals, "sz002075")        plt.savefig('./主力监控分析.png', dpi=300, bbox_inches='tight')        plt.show()    except Exception as e:        print(f"⚠️ 绘图异常: {e}")        print("📄 原始数据预览:")        cols_to_show = ['date', 'close', 'M18', 'M19', '大资金介入']        print(signals[cols_to_show].tail(10))