[Python技术]怎么用miniqmt筛选涨停放量下跌缩量选股

import time
import datetime
import pandas as pd
import numpy as np
from threading import Thread
from xtquant import xtdata
class StockSelector:
    def __init__(self):
        self.selected_stocks = {}  # 存储选股结果 {code: {'limit_up_vol': volume, 'last_vol': volume}}
        # 移除历史数据存储目录相关代码
    def get_history_data(self, code, count=20):
        """
        直接使用xtdata获取股票历史数据
        返回格式: [{'date': date, 'open', 'high', 'low', 'close', 'volume'}, ...]
        """
        # 生成xtquant标准代码格式
        xt_code = f"{code}.SZ" if code.startswith(('0', '3')) else f"{code}.SH"
        # 计算日期范围（覆盖最近count*2个交易日）
        end_date = datetime.datetime.now().strftime('%Y%m%d')
        start_date = (datetime.datetime.now() - datetime.timedelta(days=count * 2)).strftime('%Y%m%d')
        # 下载数据到miniQMT本地缓存
        xtdata.download_history_data(
            stock_code=xt_code,
            period="1d",  # 日线数据
            start_time=start_date,
            end_time=end_date
        )
        # 从本地缓存获取数据（前复权）
        data = xtdata.get_market_data_ex(
            stock_list=[xt_code],
            period="1d",
            start_time=start_date,
            end_time=end_date,
            dividend_type="front",  # 前复权
            fill_data=True
        )
        #print(data)
        if xt_code not in data:
            print(f"未获取到数据: {code}")
            return []
        # 转换为标准格式
        df = data[xt_code].reset_index()
        df = df.rename(columns={'index': 'date'})
        # 优化日期格式转换
        try:
            # 直接使用时间戳转换为日期字符串
            df['date'] = pd.to_datetime(df['date']).dt.strftime('%Y-%m-%d')
        except Exception as e:
            print(f"日期转换异常: {e}")
            df['date'] = df['date'].astype(str)
        return df.to_dict('records')
    def get_stock_list(self):
        """获取沪深A股所有股票代码（去掉后缀）"""
        all_stocks = xtdata.get_stock_list_in_sector("沪深A股")
        return [stock.split('.')[0] for stock in all_stocks]  # 去除交易所后缀
    def get_recent_limit_up_info(self, code, days=20):
        """
        获取最近20个交易日的涨停信息
        返回: [{'date': date, 'volume': volume}, ...]
        """
        try:
            # 获取历史数据
            data = self.get_history_data(code, days)
            if data is None or len(data) < 2:
                return []
            limit_up_list = []
            # 检查每天是否涨停（从最早到最新）
            for i in range(len(data) - 1):  # 不包括今天
                current = data[i]
                next_day = data[i + 1] if i + 1 < len(data) else None
                if next_day is None:
                    continue
                # 计算涨跌幅
                if current['close'] != 0:
                    change_pct = (next_day['close'] - current['close']) / current['close'] * 100
                    # 判断是否涨停（考虑不同板块的涨跌幅限制）
                    is_limit_up = False
                    if code.startswith('68'):  # 科创板
                        is_limit_up = change_pct >= 19.8
                    elif code.startswith('3') or code.startswith('0'):  # 创业板/中小板
                        is_limit_up = change_pct >= 19.8
                    else:  # 主板
                        is_limit_up = change_pct >= 9.8
                    if is_limit_up:
                        limit_up_list.append({
                            'date': next_day['date'],
                            'volume': next_day['volume']
                        })
            return limit_up_list
        except Exception as e:
            print(f"获取{code}涨停信息出错: {e}")
            return []
    def select_stocks(self):
        """
        选股逻辑：
        1. 前20个交易日有过涨停并记录下当时的成交量
        2. 上一个交易日缩量小于那个涨停成交量的4分之一
        3. 前20个交易日只有1个涨停
        """
        selected = {}
        # 获取所有股票代码
        stock_list = self.get_stock_list()
        total_stocks = len(stock_list)
        print(f"开始选股，共{total_stocks}只股票需要筛选...")
        for idx, code in enumerate(stock_list):
            if idx % 500 == 0:
                print(f"选股进度: {idx}/{total_stocks} ({idx / total_stocks * 100:.1f}%)")
            try:
                # 获取最近20个交易日的涨停信息
                limit_up_info = self.get_recent_limit_up_info(code, 20)
                # 条件1：前20个交易日只有1个涨停
                if len(limit_up_info) != 1:
                    continue
                # 获取最近2个交易日的数据
                recent_data = self.get_history_data(code, 2)
                if recent_data is None or len(recent_data) < 2:
                    continue
                last_limit_up = limit_up_info[0]
                last_day_data = recent_data[-1]  # 上一个交易日

                # 条件2：上一个交易日缩量小于涨停成交量的4分之一, 具体参数根据自己需求调整
                if last_day_data['volume'] >= last_limit_up['volume'] / 4:
                    continue
                selected[code] = {
                    'limit_up_vol': last_limit_up['volume'],
                    'last_vol': last_day_data['volume'],
                    'limit_up_date': last_limit_up['date']
                }
                print(
                    f"选中股票: {code}, 涨停日期: {last_limit_up['date']}, 涨停成交量: {last_limit_up['volume']}, 昨日成交量: {last_day_data['volume']}")
            except Exception as e:
                # 忽略异常继续下一只股票
                print(f"处理股票{code}时出错: {str(e)[:100]}...")  # 截断错误信息
                continue
        self.selected_stocks = selected
        return selected
def main():
    # 初始化选股器
    selector = StockSelector()
    print("开始选股...")
    selected = selector.select_stocks()
    print(f"选股完成，共选出 {len(selected)} 只股票")
    if len(selected) == 0:
        print("未选出符合条件的股票")
        return
    #监控
if __name__ == "__main__":
    main()