从封板瞬间到主力意图识别：本地逐笔数据+Python实战 IG50免费开源股票数据API接口

disalong

398人浏览 · 2026-08-08 17:44:46

disalong · 2026-08-08 17:44:46 发布

从封板瞬间到主力意图识别：本地逐笔数据 + Python 实战

做股票数据本地化之后，我做了一件有意思的事——从"封板瞬间的逐笔数据"反推"主力意图"。今天把整个识别流程的 Python 实现完整记录下来。

一、为什么能从封板瞬间反推主力意图

封板瞬间（封板时间点前后 1 分钟）的逐笔数据包含了主力的"操作指纹"。这些指纹包括：

大单笔数：主力拉升时会用"大单密集扫货"
小单笔数：散户跟风时会出现"小单密集"
主力席位关联：同一席位连续买卖可能是"对倒"
价格弹性：少量成交拉动价格的程度反映"主力控盘度"

把这 4 个指纹组合起来，能在封板瞬间判断主力是在"拉升"还是"出货"。

二、数据准备

我用的是本地数据引擎的 3 个接口：

# 涨停板列表
def load_daily_boards(date):
    """拉取某日所有涨停板"""
    return ig50_get(f'time/real/data/zhangting/{date}')

# 逐笔成交
def load_tick_data(dm, date):
    """拉取某只股票的逐笔成交"""
    return ig50_get(f'time/real/trace/onebyone/{dm}/{date}')

# 资金流向
def load_money_flow(dm, date):
    """拉取某只股票的资金流向"""
    return ig50_get(f'time/zijin/zlzjzs/{dm}/{date}')

三、特征提取

从封板瞬间的逐笔数据里，我提取 4 个核心特征。

特征 1：主力买入笔数占比

def main_buy_ratio(tick_data, board_time, window=60):
    """封板瞬间主力买入笔数占比"""
    start = board_time - window
    end = board_time + window
    main_buy = 0
    total_buy = 0
    for t in tick_data:
        if t['cjsj'] < start or t['cjsj'] > end:
            continue
        if t['jyzd'] != 1:  # 只看买方向
            continue
        amount = t['cjjg'] * t['cjl'] * 100
        if amount >= 1000000:  # 单笔成交 >= 100 万算主力
            main_buy += 1
        total_buy += 1
    return main_buy / total_buy if total_buy > 0 else 0

特征 2：散户跟风笔数占比

def retail_follow_ratio(tick_data, board_time, window=60):
    """封板瞬间散户跟风笔数占比"""
    start = board_time - window
    end = board_time + window
    retail_buy = 0
    total_buy = 0
    for t in tick_data:
        if t['cjsj'] < start or t['cjsj'] > end:
            continue
        if t['jyzd'] != 1:
            continue
        amount = t['cjjg'] * t['cjl'] * 100
        if amount < 100000:  # 单笔成交 < 10 万算散户
            retail_buy += 1
        total_buy += 1
    return retail_buy / total_buy if total_buy > 0 else 0

特征 3：主力对倒嫌疑

def main_washsale_score(tick_data, board_time, window=60):
    """封板瞬间主力对倒嫌疑分数"""
    start = board_time - window
    end = board_time + window
    seat_trades = {}  # {seat: [trades]}
    for t in tick_data:
        if t['cjsj'] < start or t['cjsj'] > end:
            continue
        seat = t.get('seat', 'unknown')
        if seat not in seat_trades:
            seat_trades[seat] = []
        seat_trades[seat].append(t)
    
    # 对倒检测：同一席位 1 分钟内同时有买和卖
    washsale_score = 0
    for seat, trades in seat_trades.items():
        if len(trades) < 3:
            continue
        # 简化：检查是否有"先买后卖"模式
        buys = sum(1 for t in trades if t['jyzd'] == 1)
        sells = sum(1 for t in trades if t['jyzd'] == 2)
        if buys > 0 and sells > 0:
            washsale_score += min(buys, sells)
    return washsale_score

特征 4：价格弹性

def price_elasticity(tick_data, board_time, window=300):
    """封板瞬间价格弹性"""
    start = board_time - window
    end = board_time + window
    prices = [t['cjjg'] for t in tick_data if start <= t['cjsj'] <= end]
    volumes = sum(t['cjl'] for t in tick_data if start <= t['cjsj'] <= end)
    if len(prices) < 2 or volumes == 0:
        return 0
    price_range = max(prices) - min(prices)
    # 价格弹性 = 振幅 / 成交量
    return price_range / volumes

四、主力意图识别

把 4 个特征组合起来判断主力意图：

def identify_main_intent(stock):
    """识别涨停板主力意图"""
    dm = stock['dm']
    tick_data = load_tick_data(dm, stock['date'])
    board_time = stock['board_time']
    
    f1 = main_buy_ratio(tick_data, board_time)      # 主力买入笔数占比
    f2 = retail_follow_ratio(tick_data, board_time) # 散户跟风笔数占比
    f3 = main_washsale_score(tick_data, board_time) # 主力对倒嫌疑
    f4 = price_elasticity(tick_data, board_time)    # 价格弹性
    
    # 主力拉升特征
    if f1 > 0.3 and f2 < 0.3 and f3 < 5 and f4 < 0.5:
        return 'MAIN_PULLING'  # 主力拉升
    # 主力出货特征
    elif f1 < 0.1 and f2 > 0.5 and f3 > 10:
        return 'MAIN_DUMPING'  # 主力出货
    # 散户跟风
    elif f2 > 0.6:
        return 'RETAIL_FOLLOWING'  # 散户跟风
    else:
        return 'UNCLEAR'  # 不明