聊一聊N字涨停战法，附代码

import pandas as pd
import numpy as np
from sqlalchemy import create_engine
import sys
from pathlib import Path
from datetime import datetime, timedelta
engine = create_engine("mysql://root:12345678@localhost/db_stock?charset=utf8")

def get_n_shape_potential_stocks(lookback_days=30, bias_period=6):
    """
    筛选具备N字板潜力的股票。
    策略逻辑:
    1.  近5个交易日内，有且仅有1个涨停板。
    2.  当前股价处于回调状态，且乖离率(BIAS)小于-5，表示超卖。
    3.  当前成交量小于5日平均成交量，表示下跌缩量，抛压减弱。
    参数:
        lookback_days: 查询的历史数据天数，用于计算技术指标。
        bias_period: 计算BIAS指标所用的均线周期。
    返回:
        一个包含符合条件的股票信息的DataFrame。
    """
    # 1. 获取数据
    # 确定查询的日期范围
    end_date_query = "SELECT MAX(trade_date) as max_date FROM stock_daily"
    try:
        end_date = pd.read_sql(end_date_query, engine).iloc[0]['max_date']
    except IndexError:
        print("错误: 无法从数据库获取最新交易日期，请检查数据库连接和表名。")
        return pd.DataFrame()
    start_date = (pd.to_datetime(end_date) - timedelta(days=lookback_days)).strftime('%Y-%m-%d')
    print(f"正在分析数据，时间范围: {start_date} 至 {end_date}")
    data_query = f"""
    SELECT
        sd.trade_date,
        sd.ts_code,
        sb.name,
        sd.open,
        sd.high,
        sd.low,
        sd.close,
        sd.pre_close,
        sd.vol
    FROM stock_daily sd
    LEFT JOIN stock_basic sb ON sd.ts_code = sb.ts_code
    WHERE sd.trade_date >= '{start_date}' AND sd.trade_date <= '{end_date}'
    ORDER BY sd.ts_code, sd.trade_date
    """
    df = pd.read_sql(data_query, engine)
    if df.empty:
        print("未获取到数据，请检查数据库表 'stock_daily' 和 'stock_basic' 是否存在且有数据。")
        return pd.DataFrame()
    # 2. 数据处理与指标计算
    # 按股票分组，然后对每只股票进行计算
    def process_stock(group):
        # 按日期排序，确保计算顺序正确
        group = group.sort_values('trade_date').reset_index(drop=True)
        # 计算涨跌幅
        group['pct_chg'] = (group['close'] / group['pre_close'] - 1) * 100
        # 计算BIAS乖离率
        group['ma'] = group['close'].rolling(bias_period).mean()
        group['bias'] = (group['close'] / group['ma'] - 1) * 100
        # 计算5日成交量均线
        group['vol_ma5'] = group['vol'].rolling(5).mean()
        # 获取最近5个交易日的数据用于筛选
        recent_data = group.tail(5)
        # --- 筛选条件 ---
        # 条件1: 近5个交易日有且仅有1个涨停 (涨幅 >= 9.8%)
        # 使用9.8%作为阈值，避免因四舍五入导致10%的涨停无法被识别
        limit_up_days = recent_data[recent_data['pct_chg'] >= 9.8]
        if len(limit_up_days) != 1:
            return None
        # 条件2: 最新BIAS < -5 (股价超卖)
        latest_bias = group['bias'].iloc[-1]
        if pd.isna(latest_bias) or latest_bias >= -5:
            return None
        # 条件3: 最新成交量 < 5日平均成交量 (下跌缩量)
        latest_vol = group['vol'].iloc[-1]
        latest_vol_ma5 = group['vol_ma5'].iloc[-1]
        if pd.isna(latest_vol_ma5) or latest_vol >= latest_vol_ma5:
            return None
        # 如果所有条件都满足，返回最新一行的关键信息
        latest_row = group.iloc[-1]
        return pd.Series({
            'ts_code': latest_row['ts_code'],
            'name': latest_row['name'],
            'trade_date': latest_row['trade_date'],
            'close_price': round(latest_row['close'], 2),
            'pct_chg': round(latest_row['pct_chg'], 2),
            'bias': round(latest_bias, 2),
            'volume': int(latest_vol),
            'vol_ma5': int(latest_vol_ma5),
            'limit_up_date': limit_up_days['trade_date'].iloc[0]  # 记录涨停日期
        })
    # 应用筛选函数
    results = df.groupby('ts_code').apply(process_stock)
    # 过滤掉None值，并重置索引
    results_df = results.dropna().reset_index(drop=True)
    return results_df
if __name__ == "__main__":
    print("开始筛选具备N字板潜力的股票...")
    # 调用函数进行筛选
    # lookback_days=30 表示用最近30天的数据来计算均线等指标
    potential_stocks = get_n_shape_potential_stocks(lookback_days=30, bias_period=6)
    if not potential_stocks.empty:
        print(f"\n🎯 找到 {len(potential_stocks)} 只具备N字板潜力的股票:")
        print("=" * 90)
        # 格式化输出，让信息更易读
        for _, stock in potential_stocks.iterrows():
            print(f"股票代码: {stock['ts_code']}\t股票名称: {stock['name']}")
            print(f"分析日期: {stock['trade_date']}\t收盘价: {stock['close_price']}\t当日涨跌: {stock['pct_chg']}%")
            print(f"前次涨停日期: {stock['limit_up_date']}")
            print(f"乖离率(BIAS): {stock['bias']}\t成交量: {stock['volume']}\t5日均量: {stock['vol_ma5']}")
            print("-" * 90)
        # 保存结果到CSV文件
        output_filename = 'n_shape_potential_stocks.csv'
        potential_stocks.to_csv(output_filename, index=False, encoding='utf-8-sig')
        print(f"\n✅ 结果已成功保存至文件: {output_filename}")
    else:
        print("\n❌ 今日未找到符合条件的N字板潜力股票。")
        print("提示: 市场可能处于普涨或普跌状态，不符合回调缩量的条件。")