量化入门-用Python筛选爆量上涨的股票
用Python筛选爆量上涨的股票最近在研究成交量异动的票，写了个小工具自动筛选爆量上涨的股票，分享一下框架。
思路
爆量上涨通常意味着资金关注度突然提升，可能是主力进场或者利好消息刺激。我的筛选逻辑：
选股条件：
[*]沪深主板，排除ST风险票
[*]流通市值200-2000亿（盘子适中，避免庄股，太小风险高）
[*]年利润>0.5亿（基本面过得去）
[*]近100天涨幅3%（有机构关注，一般不是垃圾票）
爆量定义：
[*]当日成交量 ≥ 近100天平均成交量的3倍
[*]当日收阳线（收盘价>开盘价）
环境
pip install jvQuanttoken去官网注册拿一下就行。
代码
# -*- 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 name = stock 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 = ) for k in klines] avg_volume = sum(volumes) / len(volumes) # 找爆量上涨 count = 0 for k in klines: date = k open_price = safe_float(k) close = safe_float(k) volume = safe_float(k) pct_chg = safe_float(k) turnover = safe_float(k) # 爆量且上涨 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"{'序号':
页:
[1]