手把手教你用Python+迅投miniqmt构建实时大盘涨跌幅图形变化监控

from xtquant import xtdata
import matplotlib.animation as animation
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
import warnings
# 忽略警告信息
warnings.filterwarnings('ignore')
# 设置绘图风格
sns.set(style="whitegrid")
# 设置字体 (兼容 Windows 和 Mac)
plt.rcParams['font.sans-serif'] = ['Arial Unicode MS', 'PingFang SC', 'SimHei']
plt.rcParams['axes.unicode_minus'] = False
def get_all_a_stock_codes():
    # 获取沪深A股板块
    stocks = xtdata.get_stock_list_in_sector('沪深京A股')
    return stocks
fig, ax = plt.subplots(figsize=(10, 6))
# 用于存储历史数据
time_history = []
up_history = []
down_history = []
flat_history = []
def update_market_charts(frame):
    """
    动态更新函数，会被matplotlib周期性调用
    """
    try:
        # 1. 获取全市场代码 (只在第一次运行时获取，避免重复请求)
        if not hasattr(update_market_charts, "stock_codes"):
            update_market_charts.stock_codes = get_all_a_stock_codes()
        stock_list = update_market_charts.stock_codes
        # 2. 获取实时行情快照
        # 返回结构: { stock_code: { 'lastPrice': ..., 'lastClose': ... }, ... }
        tick_data = xtdata.get_full_tick(stock_list)
        up_count = 0
        down_count = 0
        flat_count = 0
        # 3. 统计涨跌数
        for code, data in tick_data.items():
            last_price = data.get('lastPrice')  # 最新价
            last_close = data.get('lastClose')  # 昨收价
            # 过滤停牌或无数据的情况 (最新价为0通常表示无数据/停牌)
            if last_price is None or last_price == 0:
                continue
            if last_price > last_close:
                up_count += 1
            elif last_price < last_close:
                down_count += 1
            else:
                flat_count += 1
        current_time = datetime.now().strftime("%H:%M:%S")
        # 4. 更新历史数据
        time_history.append(current_time)
        up_history.append(up_count)
        down_history.append(down_count)
        flat_history.append(flat_count)
        # 保持数据列表长度，比如只显示最近 100 个时间点
        max_len = 100
        if len(time_history) > max_len:
            time_history.pop(0)
            up_history.pop(0)
            down_history.pop(0)
            flat_history.pop(0)
        # 5. 绘图
        ax.clear()  # 清空上一帧
        # 绘制三条线
        ax.plot(time_history, up_history, label='上涨家数', color='red', linewidth=2)
        ax.plot(time_history, down_history, label='下跌家数', color='green', linewidth=2)
        # ax.plot(time_history, flat_history, label='平盘家数', color='gray', linestyle='--') # 平盘通常波动小，可选
        # 设置图表格式
        ax.set_title(f"大盘实时涨跌家数监控 (更新于: {current_time})", fontsize=14)
        ax.set_xlabel("时间")
        ax.set_ylabel("家数")
        ax.legend(loc='upper left')
        # 优化X轴显示，避免时间标签重叠
        plt.xticks(rotation=45)
        # 在图表上显示当前数值
        info_text = f"当前涨: {up_count}\n当前跌: {down_count}\n当前平: {flat_count}"
        ax.text(0.02, 0.5, info_text, transform=ax.transAxes, fontsize=12,
                verticalalignment='center', bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.5))
    except Exception as e:
        print(f"Error: {e}")
# --- 启动实时监控 ---
print("开始启动实时行情监控...")
print("如果是在QMT脚本里运行，请确保已连接行情接口。")
# 使用FuncAnimation进行动画循环，interval单位是毫秒 (例如3000ms = 3秒刷新一次)

ani = animation.FuncAnimation(fig, update_market_charts, interval=3000)
plt.tight_layout()
plt.show()