量化入门-用Python筛选爆量上涨的股票

用Python筛选爆量上涨的股票

最近在研究成交量异动的票，写了个小工具自动筛选爆量上涨的股票，分享一下框架。
思路

爆量上涨通常意味着资金关注度突然提升，可能是主力进场或者利好消息刺激。我的筛选逻辑：
选股条件：

• 沪深主板，排除ST风险票
  
• 流通市值200-2000亿（盘子适中，避免庄股，太小风险高）
  
• 年利润>0.5亿（基本面过得去）
  
• 近100天涨幅3%（有机构关注，一般不是垃圾票）
  

爆量定义：

• 当日成交量 ≥ 近100天平均成交量的3倍
  
• 当日收阳线（收盘价>开盘价）
  

环境

1. pip install jvQuant

复制代码

token去官网注册拿一下就行。
代码

[code]# -*- coding: utf-8 -*-"""量化筛选-爆量上涨筛选工具功能说明：    1. 筛选符合条件的股票    2. 获取每只股票近100天前复权日K线数据    3. 分析并筛选出爆量上涨的日期（成交量>=n倍均值且当日上涨）    4. 按爆量倍数排序输出结果作者：Juice日期：2023-02-27"""import jvQuantimport time# ==================== 配置参数 ====================TOKEN = "你的token"  # 替换成你自己的# 筛选条件QUERY_CONDITION = "近100天涨幅小于50%，基金持股比例大于3%,沪深主板,非ST,流通市值200-2000亿,年利润大于0.5亿"# 爆量阈值VOLUME_RATIO_THRESHOLD = 3.0# K线参数KLINE_LIMIT = 100KLINE_TYPE = "day"KLINE_FQ = "前复权"def safe_float(val, default=0.0):    """安全转浮点数"""    try:        return float(val)    except (ValueError, TypeError):        return defaultdef main():    db = jvQuant.sql_client.Construct(TOKEN)        # 第一步：筛选股票    print("=" * 60)    print("第一步：筛选符合条件的股票，筛选条件:", QUERY_CONDITION)    print("=" * 60)        query_result = db.query(        QUERY_CONDITION,        page=1,        sort_type=1,        sort_key="TOTAL_MV"    )        if query_result['code'] != 0:        print(f"查询失败: {query_result['message']}")        return        data = query_result['data']    stock_list = data['list']        print(f"筛选条件: {data['query']}")    print(f"符合条件的股票数量: {data['count']}")    print()        # 第二步：获取K线分析爆量    print("=" * 60)    print("第二步：获取K线并分析爆量上涨")    print("=" * 60)        results = []        for i, stock in enumerate(stock_list):        code = stock[0]        name = stock[1]                print(f"[{i+1}/{len(stock_list)}] {code} {name}", end=" ")                try:            kline_result = db.kline(                code=code,                cate='stock',                fq=KLINE_FQ,                type=KLINE_TYPE,                limit=KLINE_LIMIT            )                        if kline_result['code'] != 0:                print("失败")                continue                        klines = kline_result['data']['list']                        if len(klines) < 20:                print("数据不足")                continue                        # 算平均成交量            volumes = [safe_float(k[5]) for k in klines]            avg_volume = sum(volumes) / len(volumes)                        # 找爆量上涨            count = 0            for k in klines:                date = k[0]                open_price = safe_float(k[1])                close = safe_float(k[2])                volume = safe_float(k[5])                pct_chg = safe_float(k[8])                turnover = safe_float(k[10])                                # 爆量且上涨                if close > open_price and volume >= VOLUME_RATIO_THRESHOLD * avg_volume:                    volume_ratio = volume / avg_volume                    results.append({                        'code': code,                        'name': name,                        'date': date,                        'close': close,                        'volume_ratio': volume_ratio,                        'turnover': turnover,                        'pct_chg': pct_chg                    })                    count += 1                        print(f"完成(爆量{count}次)")                    except Exception as e:            print(f"错误: {e}")                time.sleep(0.1)  # 别请求太快        # 第三步：输出结果    print()    print("=" * 60)    print("第三步：输出爆量上涨结果")    print("=" * 60)        if not results:        print("没有找到符合条件的爆量上涨记录")        return        results.sort(key=lambda x: x['volume_ratio'], reverse=True)        output_lines = []    output_lines.append(f"共发现 {len(results)} 条爆量上涨记录\n")    output_lines.append(f"{'序号':