[Python技术] miniqmt获取 通达信自选股成交量变化并告警

import osimport timeimport pandas as pdfrom xtquant import xtdatafrom xtquant.xttrader import XtQuantTrader#功能描述，读取通达信自选股，监听miniqmt，如果今天的成交量>昨日的成交量 考虑买入# ===== 1. 股票代码转换函数 =====defconvert_blk_to_qmt(blk_codes):    valid_codes = []    for code in blk_codes:        code = code.strip()        iflen(code) 7:  # 至少7位（首位+6位代码）            continue        prefix_char = code[0]        stock_num = code[1:].lstrip('0').zfill(6)  # 关键：去前导零后补足6位        if prefix_char == '0':            valid_codes.append(f"{stock_num}.SZ")        elif prefix_char == '1':            valid_codes.append(f"{stock_num}.SH")        elif prefix_char == '9':            valid_codes.append(f"{stock_num}.BJ")    return valid_codes# ===== 2. 加载自选股文件 =====defload_block_stocks(block_path):    """读取.blk文件并转换为标准代码"""    try:        withopen(block_path, 'r', encoding='utf-8') as f:            raw_codes = [line.strip() for line in f if line.strip()]        return convert_blk_to_qmt(raw_codes)    except Exception as e:        print(f"读取自选股文件失败: {e}")        return []# ===== 3. 全局变量初始化 =====block_path = r"D:\new_tdx\T0002\blocknew\zxg.blk"#换成自己的路径last_volume = {}  # 存储昨日成交量：{股票代码: 昨日成交量}monitored_stocks = set()  # 当前监听的股票集合trader = XtQuantTrader("", 123456)  # 初始化交易对象subscribe_seq = None  # 存储订阅序列号# ===== 4. 行情回调函数 =====defon_data(datas):    global last_volume, monitored_stocks    for stock_code, data in datas.items():        if stock_code notin monitored_stocks:            continue        # 确保获取的是标量值        realtime_vol = data.get('volume', 0)        ifisinstance(realtime_vol, pd.Series):  # 如果是Series，取最后一个值            realtime_vol = realtime_vol.iloc[-1].item()        prev_vol = last_volume.get(stock_code, 0)        # 确保prev_vol是数字类型        ifisinstance(prev_vol, pd.Series):            prev_vol = prev_vol.iloc[-1].item()        if realtime_vol > prev_vol and prev_vol > 0:            print(f"[信号] {stock_code} 实时成交量({realtime_vol}) > 昨日成交量({prev_vol})")            #买入逻辑，检测持仓是否已经有或已提交订单，如无则考虑买入# ===== 5. 主监控循环 =====defmain():    global last_volume, monitored_stocks, subscribe_seq    last_mtime = 0  # 文件最后修改时间    # 首次加载自选股    stocks = load_block_stocks(block_path)    monitored_stocks = set(stocks)


    
    print(f"初始监控列表: {monitored_stocks}")    # 获取昨日成交量    for stock in monitored_stocks:        df = xtdata.get_market_data(stock_list=[stock], period='1d', count=1)        if df isnotNone  and'volume'in df:            # 处理多维数组场景            volume_data = df['volume'].iloc[-1]            ifhasattr(volume_data, '__len__') andlen(volume_data) > 1:                last_volume[stock] = volume_data[0]  # 取首个元素            else:                last_volume[stock] = volume_data            print(f"初始化 {stock} 昨日成交量: {last_volume[stock]}")        else:            print(f"警告：无法获取 {stock} 的昨日成交量")    # 订阅实时行情（全推模式）    subscribe_seq = xtdata.subscribe_whole_quote(list(monitored_stocks), callback=on_data)    print(f"订阅序列号: {subscribe_seq}")    # 持续监听.blk文件变化    whileTrue:        try:            current_mtime = os.path.getmtime(block_path)            if current_mtime != last_mtime:                print("检测到自选股文件变化，重新加载...")                new_stocks = set(load_block_stocks(block_path))                # 处理新增股票                added = new_stocks - monitored_stocks                for stock in added:                    df = xtdata.get_market_data(stock_list=[stock], period='1d', count=1)                    if df isnotNoneandnot df.empty and'volume'in df:                        last_volume[stock] = df['volume'].iloc[-1].item()                        print(f"新增 {stock} 昨日成交量: {last_volume[stock]}")                    else:                        print(f"警告：无法获取 {stock} 的昨日成交量")                # 更新订阅（先取消旧订阅，再订阅新列表）                if subscribe_seq isnotNone:                    xtdata.unsubscribe_quote(subscribe_seq)                subscribe_seq = xtdata.subscribe_whole_quote(list(new_stocks), callback=on_data)                monitored_stocks = new_stocks                last_mtime = current_mtime                print(f"更新后监控列表: {monitored_stocks}")            time.sleep(5)  # 每5秒检查一次文件        except Exception as e:            print(f"监控循环异常: {e}")            time.sleep(10)if __name__ == "__main__":    main()