QMT实时Tick监控示例代码

#coding:gbk
"""
QMT实时Tick监控 - 只监控指定列表，不依赖主图
"""

import datetime

# 要监控的股票列表（只监控这两只）
watch_list = ['510300.SH', '600000.SH']

def get_current_time_str():
    return datetime.datetime.now().strftime('%Y%m%d %H:%M:%S')

def format_value(value):
    if isinstance(value, (int, float)):
        return round(float(value), 3)
    elif isinstance(value, list):
        return [round(float(x), 3) if isinstance(x, (int, float)) else x for x in value]
    return value

def determine_direction(last_price, bid1, ask1):
    if bid1 is None or ask1 is None or last_price is None:
        return 0
    if last_price >= ask1:
        return 1
    elif last_price <= bid1:
        return -1
    else:
        return 0

def init(ContextInfo):
    # 只监控用户指定的股票列表
    ContextInfo.set_universe(watch_list)
    print(f"[INFO] 策略初始化完成，监控股票: {watch_list}")
    print("[INFO] 等待tick数据...")

def handlebar(ContextInfo):
    # 获取所有监控股票的tick数据（一次性获取，更高效）
    all_ticks = ContextInfo.get_full_tick(watch_list)
    if all_ticks is None:
        return

    # 遍历监控列表中的每只股票
    for stock_code in watch_list:
        tick_data = all_ticks.get(stock_code)
        if tick_data is None:
            continue

        # 提取字段
        last_price = tick_data.get('lastPrice')
        bid_prices = tick_data.get('bidPrice', [])
        ask_prices = tick_data.get('askPrice', [])
        bid_vols = tick_data.get('bidVol', [])
        ask_vols = tick_data.get('askVol', [])

        if last_price is None or last_price == 0:
            continue

        bid1 = bid_prices[0] if bid_prices else None
        ask1 = ask_prices[0] if ask_prices else None

        direction = determine_direction(last_price, bid1, ask1)
        direction_desc = {
            1: "主动买盘(吃卖单)",
            -1: "主动卖盘(吃买单)",
            0: "中性成交(盘口内部)"
        }.get(direction, "未知方向")

        output = {
            '证券代码': stock_code,
            '时间': get_current_time_str(),
            '最新价': format_value(last_price),
            '买一价': format_value(bid1),
            '卖一价': format_value(ask1),
            '五档买价': format_value(bid_prices),
            '五档卖价': format_value(ask_prices),
            '五档买量': format_value(bid_vols),
            '五档卖量': format_value(ask_vols),
            '方向': direction_desc
        }

        for k, v in output.items():
            print(f"{k}: {v}")
        print("-" * 40)

def stop(ContextInfo):
    print("[INFO] 策略停止")