Agent驱动的集合竞价分析系统：用AI挖掘开盘前的市场信号

集合竞价是每个交易日开盘前的15分钟，大部分散户觉得这段时间没什么信息量。但实际上集合竞价阶段包含了大量有价值的信号，特别是对日内短线交易来说。最近我用AI Agent构建了一个集合竞价分析系统，让Agent自动从集合竞价数据中识别开盘信号。

系统的核心思路是把集合竞价分析拆解成几个子任务：放量检测、价格走势分析、未匹配量分析、高开幅度评估。每个子任务交给专门的模块处理，Agent负责汇总生成综合判断。

数据层使用本地数据引擎的集合竞价接口time/real/trace/jhjj/{dm}，这个接口在集合竞价时间段每1分钟更新一次，返回申报时间、匹配量、未匹配量、标志（1买单-1卖单）、成交价。同时配合实时行情接口time/real/{dm}获取前日收盘价，历史K线接口time/history/trade/{dm}/day获取历史数据用于计算平均集合竞价量。

```python
class AuctionAgent:
    def __init__(self, data_engine):
        self.data_engine = data_engine
        self.modules = {
            "volume": VolumeDetector(),
            "price": PriceTrendAnalyzer(),
            "unmatched": UnmatchedAnalyzer(),
            "premium": OpenPremiumEvaluator()
        }
    
    def analyze(self, dm):
        jhjj_data = self.data_engine.read_jhjj(dm)
        realtime = self.data_engine.read_realtime(dm)
        kline = self.data_engine.read_kline(dm, "day")
        
        if not jhjj_data or not realtime or not kline:
            return {"dm": dm, "score": 50, "signal": "数据不足"}
        
        prev_close = realtime.get("昨日收盘价", 0)
        history_vols = self.calc_history_auction_vols(kline)
        
        context = {
            "dm": dm,
            "jhjj": jhjj_data,
            "prev_close": prev_close,
            "history_vols": history_vols,
            "realtime": realtime
        }
        
        results = []
        for name, module in self.modules.items():
            result = module.analyze(context)
            if result:
                results.append(result)
        
        decision = self.synthesize(results, dm)
        return decision
    
    def calc_history_auction_vols(self, kline):
        return [k.get("集合竞价量", 0) for k in kline[-10:]]
```

第一个模块是放量检测器。集合竞价放量说明市场关注度突然提高，是重要的异动信号。

```python
class VolumeDetector:
    def analyze(self, context):
        jhjj = context["jhjj"]
        history_vols = context["history_vols"]
        
        today_vol = sum(item["匹配量"] for item in jhjj)
        avg_vol = sum(history_vols) / len(history_vols) if history_vols else 0
        
        ratio = today_vol / avg_vol if avg_vol > 0 else 0
        
        signal = {
            "module": "volume",
            "today_vol": today_vol,
            "avg_vol": avg_vol,
            "ratio": ratio
        }
        
        if ratio >= 3:
            signal["signal"] = "集合竞价显著放量"
            signal["score"] = 72
        elif ratio >= 2:
            signal["signal"] = "集合竞价放量"
            signal["score"] = 65
        elif ratio <= 0.5:
            signal["signal"] = "集合竞价缩量"
            signal["score"] = 40
        else:
            signal["signal"] = "集合竞价量正常"
            signal["score"] = 50
        
        return signal
```

第二个模块是价格走势分析器。9:20到9:25之间的价格变化最有参考价值，因为这段时间不能撤单，申报都是真实的。

```python
class PriceTrendAnalyzer:
    def analyze(self, context):
        jhjj = context["jhjj"]
        
        late_items = [item for item in jhjj if item["申报时间"] >= "09:20:00"]
        if len(late_items) < 2:
            return None
        
        prices = [item["成交价"] for item in late_items]
        first_price = prices[0]
        last_price = prices[-1]
        
        if first_price <= 0:
            return None
        
        change = (last_price - first_price) / first_price
        
        signal = {
            "module": "price",
            "change": change,
            "first_price": first_price,
            "last_price": last_price
        }
        
        if change > 0.02:
            signal["signal"] = "集合竞价尾段强势拉升"
            signal["score"] = 70
        elif change > 0.005:
            signal["signal"] = "集合竞价尾段小幅上行"
            signal["score"] = 60
        elif change < -0.02:
            signal["signal"] = "集合竞价尾段下挫"
            signal["score"] = 35
        else:
            signal["signal"] = "集合竞价尾段平稳"
            signal["score"] = 50
        
        return signal
```

第三个模块是未匹配量分析器。9:25时的未匹配量反映买卖力量失衡。

```python
class UnmatchedAnalyzer:
    def analyze(self, context):
        jhjj = context["jhjj"]
        
        if not jhjj:
            return None
        
        last_item = jhjj[-1]
        wppl = last_item.get("未匹配量", 0)
        ppl = last_item.get("匹配量", 0)
        bz = last_item.get("标志", 0)
        
        if ppl <= 0:
            return None
        
        ratio = abs(wppl) / ppl
        
        signal = {
            "module": "unmatched",
            "unmatched": wppl,
            "matched": ppl,
            "ratio": ratio,
            "direction": "buy" if bz == 1 else "sell"
        }
        
        if bz == 1 and ratio > 3:
            signal["signal"] = "未匹配买单大量堆积"
            signal["score"] = 68
        elif bz == -1 and ratio > 3:
            signal["signal"] = "未匹配卖单大量堆积"
            signal["score"] = 32
        else:
            signal["signal"] = "买卖力量基本均衡"
            signal["score"] = 50
        
        return signal
```

第四个模块是高开幅度评估器。高开幅度要结合成交量和未匹配量综合判断。

```python
class OpenPremiumEvaluator:
    def analyze(self, context):
        jhjj = context["jhjj"]
        prev_close = context["prev_close"]
        
        if prev_close <= 0 or not jhjj:
            return None
        
        open_price = jhjj[-1].get("成交价", 0)
        if open_price <= 0:
            return None
        
        premium = (open_price - prev_close) / prev_close
        
        signal = {
            "module": "premium",
            "open_price": open_price,
            "prev_close": prev_close,
            "premium": premium
        }
        
        if premium > 0.05:
            signal["signal"] = "大幅高开"
            signal["score"] = 60
        elif premium > 0.02:
            signal["signal"] = "温和高开"
            signal["score"] = 65
        elif premium < -0.05:
            signal["signal"] = "大幅低开"
            signal["score"] = 35
        elif premium < -0.02:
            signal["signal"] = "温和低开"
            signal["score"] = 42
        else:
            signal["signal"] = "平开"
            signal["score"] = 50
        
        return signal
```

Agent的决策模块把四个模块的结果汇总。

```python
def synthesize(self, results, dm):
    if not results:
        return {"dm": dm, "score": 50, "signal": "无明显信号"}
    
    weights = {"volume": 0.25, "price": 0.3, "unmatched": 0.25, "premium": 0.2}
    
    total_score = 0
    total_weight = 0
    signals = []
    
    for r in results:
        w = weights.get(r.get("module", ""), 0.1)
        total_score += r.get("score", 50) * w
        total_weight += w
        signals.append(r.get("signal", ""))
    
    final_score = total_score / total_weight if total_weight > 0 else 50
    
    if final_score >= 65:
        action = "集合竞价信号偏多，可关注"
    elif final_score <= 40:
        action = "集合竞价信号偏空，谨慎"
    else:
        action = "集合竞价信号中性"
    
    return {
        "dm": dm,
        "score": round(final_score, 1),
        "action": action,
        "signals": signals
    }
```

系统每天9:25运行，遍历全市场股票，筛选综合得分超过65分的股票作为当日观察池。实际使用下来，集合竞价信号对次新股和小盘股效果更好，大盘股的信号不太准。

几个经验分享。第一，9:20到9:25的信号比9:15到9:20可靠，因为9:20之后不能撤单。第二，集合竞价放量+高开+未匹配买单堆积是最佳组合，当天上涨概率较高。第三，集合竞价信号要配合前一日收盘状态看，前一天涨停且今天集合竞价继续放量的股票信号更强。

系统使用的数据来自ig50本地数据接口，集合竞价接口每分钟更新，配合实时行情和历史K线接口，能构建一个完整的开盘信号识别系统。

资料参考：ig50


集合竞价是每个交易日开盘前的15分钟，大部分散户觉得这段时间没什么信息量。但实际上集合竞价阶段包含了大量有价值的信号，特别是对日内短线交易来说。最近我用AI Agent构建了一个集合竞价分析系统，让Agent自动从集合竞价数据中识别开盘信号。

Agent驱动的集合竞价分析系统：用AI挖掘开盘前的市场信号

人工智能

编程语言

AI驱动的集合竞价分析系统精准捕捉开盘信号，通过放量检测、价格走势、未匹配量、高开幅度四大模块智能评估。系统特别适合次新股和小盘股，9:20-9:25数据最可靠，结合前日收盘状态提升预测准确率。

数据接口

代码解释

Agent

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Agent驱动的集合竞价分析系统：用AI挖掘开盘前的市场信号-腾讯云开发者社区-腾讯云

Agent驱动的集合竞价分析系统：用AI挖掘开盘前的市场信号

Agent驱动的集合竞价分析系统：用AI挖掘开盘前的市场信号

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐