QMT里的“基础行情数据”和“快照指标”是什么？怎么使用？

QUANT 2025-12-02 11:10:54 80 举报

两者的基础区别

基础行情数据：由交易所直接给的原始数据（比如 :1分钟k线等）

快照指标：由交易所给的原始数据，通过迅投加工而成（比如 “量比”、“涨速”、“换手率”等）

获取方式的根本区别：period 参数在使用 get_market_data_ex 函数时

获取基础行情，period 参数应设为标准的时间周期：

# 获取1分钟K线数据
basic_data = C.get_market_data_ex([], ["000001.SZ"], period="1m", count=10)

获取快照指标，period 参数必须设为特定的字符串 'snapshotindex'：

# 获取快照指标数据
snapshot_data = C.get_market_data_ex([], ["000001.SZ"], period="snapshotindex", count=10)

数据获取流程对比:

获取基础行情数据 (1分钟线)案例：

# coding:gbk
 
def init(C):
    C.stock = "000001.SZ"
    C.has_executed = False
 
def handlebar(C):
    if C.has_executed:
        return
 
    print("--- 1. 获取基础行情数据 (1分钟) ---")
    basic_data = C.get_market_data_ex(
        [], 
        [C.stock], 
        period="1m", # 关键参数：标准时间周期
        count=5
    )
 
    if C.stock in basic_data:
        df = basic_data[C.stock]
        print("可用字段:", df.columns.tolist())
        print("最新数据:")
        print(df.iloc[-1])
 
    C.has_executed = True

获取结果：

获取快照指标数据：

# coding:gbk
"""
快照指标批量计算版 - 同时分析多只股票
"""

def init(C):
    # 设置要分析的股票列表
    C.stock_list = [
        "000001.SZ",  # 平安银行
        "000002.SZ",  # 万科A
        "000592.SZ",  # 平潭发展
        "600000.SH",  # 浦发银行
        "600036.SH"   # 招商银行
    ]
    C.has_executed = False

def handlebar(C):
    if C.has_executed:
        return
    
    print("\n" + "="*80)
    print("【快照指标批量分析】")
    print("="*80)
    
    # 存储结果
    results = []
    
    for stock in C.stock_list:
        indicators = calculate_indicators(C, stock)
        if indicators:
            results.append(indicators)
    
    # 按量比排序，找出异动股票
    if results:
        print("\n【量比排行榜 - 前5名】")
        print("-" * 80)
        sorted_by_volume_ratio = sorted(results, key=lambda x: x['volume_ratio'], reverse=True)
        for i, item in enumerate(sorted_by_volume_ratio[:5], 1):
            print(f"{i}. {item['stock']} - 量比: {item['volume_ratio']:.2f}, "
                  f"1分钟涨速: {item['speed_1m']:.2f}%, "
                  f"5日涨幅: {item['change_5d']:.2f}%")
        
        # 按1分钟涨速排序
        print("\n【1分钟涨速排行榜 - 前5名】")
        print("-" * 80)
        sorted_by_speed = sorted(results, key=lambda x: x['speed_1m'], reverse=True)
        for i, item in enumerate(sorted_by_speed[:5], 1):
            print(f"{i}. {item['stock']} - 1分钟涨速: {item['speed_1m']:.2f}%, "
                  f"量比: {item['volume_ratio']:.2f}")
    
    print("="*80)
    C.has_executed = True


def calculate_indicators(C, stock):
    """计算单只股票的快照指标"""
    try:
        # 获取数据
        minute_data = C.get_market_data_ex([], [stock], period="1m", count=240)
        daily_data = C.get_market_data_ex([], [stock], period="1d", count=11)
        
        if stock not in minute_data or stock not in daily_data:
            return None
        
        df_minute = minute_data[stock]
        df_daily = daily_data[stock]
        
        if len(df_minute) < 2 or len(df_daily) < 6:
            return None
        
        current_price = df_minute.iloc[-1]['close']
        
        # 计算各项指标
        indicators = {
            'stock': stock,
            'current_price': current_price
        }
        
        # 量比
        today_volume = df_minute['volume'].sum()
        avg_volume_5d = df_daily.iloc[-6:-1]['volume'].mean()
        indicators['volume_ratio'] = today_volume / avg_volume_5d if avg_volume_5d > 0 else 0
        
        # 1分钟涨速
        price_1m_ago = df_minute.iloc[-2]['close']
        indicators['speed_1m'] = (current_price - price_1m_ago) / price_1m_ago * 100
        
        # 5分钟涨速
        if len(df_minute) >= 6:
            price_5m_ago = df_minute.iloc[-6]['close']
            indicators['speed_5m'] = (current_price - price_5m_ago) / price_5m_ago * 100
        else:
            indicators['speed_5m'] = 0
        
        # 3日涨幅
        if len(df_daily) >= 4:
            price_3d_ago = df_daily.iloc[-4]['close']
            indicators['change_3d'] = (current_price - price_3d_ago) / price_3d_ago * 100
        else:
            indicators['change_3d'] = 0
        
        # 5日涨幅
        if len(df_daily) >= 6:
            price_5d_ago = df_daily.iloc[-6]['close']
            indicators['change_5d'] = (current_price - price_5d_ago) / price_5d_ago * 100
        else:
            indicators['change_5d'] = 0
        
        # 10日涨幅
        if len(df_daily) >= 11:
            price_10d_ago = df_daily.iloc[-11]['close']
            indicators['change_10d'] = (current_price - price_10d_ago) / price_10d_ago * 100
        else:
            indicators['change_10d'] = 0
        
        return indicators
        
    except Exception as e:
        print(f"计算 {stock} 时出错: {str(e)}")
        return None

打印结果：