# 从 0 搭建一个主力资金追踪 Agent：让 LLM 帮你识别真实建仓信号

做量化这几年，我一直在思考一个问题——**LLM 能不能直接帮我做主力资金策略的"信号识别"工作**？

主力资金数据本身是结构化的（净流入金额、连续天数、流通市值比值等），但"信号识别"是个开放问题——**哪些信号是真吸筹、哪些是诱多、哪些是市值管理动作**，需要大量经验判断。

我尝试搭了一个主力资金追踪 Agent，让 LLM 自动识别"真实建仓信号"。

今天把这套架构完整分享出来，重点讲工程实现和踩过的坑。

## 一、为什么主力资金策略需要 LLM

传统的主力资金策略是"硬编码规则"——比如"连续 5 天净流入就买入"。但这种硬编码规则的 alpha 早就被吃完了。

真实的主力资金策略需要做大量"软判断"：

- 这次净流入是"真吸筹"还是"市值管理"？
- 这只股票的"流通市值比值"是高还是低？
- 这家公司最近有没有"信息泄露窗口"（财报披露前 10 天）？

这些判断如果人工做，每天要看 5000 只股票，根本不可能。**用 LLM 做"软判断"+ 硬编码做"硬规则"，是最优组合**。

## 二、整体架构设计

主力资金追踪 Agent 的整体架构分 5 层：

```text
数据层（Data Layer）
  ↓
特征工程层（Feature Layer）
  ↓
LLM 信号层（LLM Signal Layer）
  ↓
规则引擎层（Rule Engine Layer）
  ↓
执行层（Execution Layer）
```

每一层的职责非常清晰：

**数据层**：本地股票数据（行情、资金流、财报、公告）
**特征工程层**：计算所有可能用到的硬指标
**LLM 信号层**：让 LLM 给出"软判断"信号
**规则引擎层**：硬编码的交易规则
**执行层**：下单、风控、日志

## 三、数据层实现

数据层是整套系统的基础。我用的是本地股票数据引擎，所有数据本地化存储、3 秒落盘。

```python
from typing import Dict, List
import pandas as pd

class LocalDataClient:
    """本地数据客户端"""
    
    def __init__(self):
        self.cache = {}
    
    def get_money_flow(self, dm: str, days: int = 20) -> pd.DataFrame:
        """获取主力资金流数据
        
        接口路径：time/zijin/zlzjzs/{dm}
        字段：cjsj, zlJlr, zlJlb, shJlb, ddx, ddy, ddz, ddf, f5MinZlJe
        """
        cache_key = f"flow_{dm}_{days}"
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        df = self._query(f"/time/zijin/zlzjzs/{dm}", {"days": days})
        df.columns = ['cjsj', 'zlJlr', 'zlJlb', 'shJlb', 
                      'ddx', 'ddy', 'ddz', 'ddf', 'f5MinZlJe']
        df['cjsj'] = pd.to_datetime(df['cjsj'])
        
        self.cache[cache_key] = df
        return df
    
    def get_kline(self, dm: str, days: int = 60) -> pd.DataFrame:
        """获取行情数据
        
        接口路径：time/history/trade/{dm}/1d
        字段：cjsj, open, high, low, close, cjl, cje
        """
        df = self._query(f"/time/history/trade/{dm}/1d", {"days": days})
        df.columns = ['cjsj', 'open', 'high', 'low', 'close', 'cjl', 'cje']
        df['cjsj'] = pd.to_datetime(df['cjsj'])
        return df
    
    def get_basic_info(self, dm: str) -> Dict:
        """获取股票基础信息
        
        接口路径：base/gplist + time/f10/company/{dm}
        """
        gplist = self._query("/base/gplist")
        company = self._query(f"/time/f10/company/{dm}")
        return {
            'dm': dm,
            'mc': gplist[gplist['dm'] == dm]['mc'].iloc[0],
            'industry': company['industry'].iloc[0],
            'list_date': company['list_date'].iloc[0]
        }
    
    def get_recent_announcements(self, dm: str, days: int = 30) -> pd.DataFrame:
        """获取最近公告
        
        接口路径：time/data/announcements/{dm}
        """
        df = self._query(f"/time/data/announcements/{dm}", {"days": days})
        df.columns = ['ann_date', 'ann_type', 'ann_title']
        return df
    
    def _query(self, path: str, params: Dict = None) -> pd.DataFrame:
        """本地数据查询（实际项目里换成 HTTP 调用）"""
        # 这里简化为占位实现
        # 实际项目里用 HTTP / IPC / 文件读取
        pass
```

## 四、特征工程层

特征工程层负责把所有可能用到的硬指标计算出来，给 LLM 当"上下文"。

```python
class FeatureBuilder:
    """特征工程层"""
    
    def __init__(self, data_client: LocalDataClient):
        self.data = data_client
    
    def build_features(self, dm: str) -> Dict:
        """为单只股票构造特征字典
        
        返回的特征包括：
        - 资金流特征
        - 行情特征
        - 基本面特征
        - 公告特征
        """
        flow = self.data.get_money_flow(dm, days=20)
        kline = self.data.get_kline(dm, days=60)
        basic = self.data.get_basic_info(dm)
        ann = self.data.get_recent_announcements(dm, days=30)
        
        features = {
            'dm': dm,
            'mc': basic['mc'],
            'industry': basic['industry'],
            
            # 资金流特征
            'continuous_inflow_days': self._calc_continuous_inflow(flow),
            'total_inflow_5d': flow.tail(5)['zlJlr'].sum(),
            'inflow_volatility': flow['zlJlr'].std(),
            'avg_inflow_ratio': self._calc_inflow_ratio(flow, kline),
            'recent_5d_return': self._calc_return(kline, 5),
            'recent_20d_return': self._calc_return(kline, 20),
            
            # 公告特征
            'has_recent_announcement': len(ann) > 0,
            'announcement_types': ann['ann_type'].tolist(),
            
            # L2 指标
            'avg_ddx': flow['ddx'].tail(5).mean(),
            'avg_ddy': flow['ddy'].tail(5).mean()
        }
        
        return features
    
    def _calc_continuous_inflow(self, flow: pd.DataFrame) -> int:
        """计算连续净流入天数"""
        is_inflow = (flow['zlJlr'] > 0).astype(int).values
        continuous = 0
        for v in reversed(is_inflow):
            if v == 1:
                continuous += 1
            else:
                break
        return continuous
    
    def _calc_inflow_ratio(self, flow: pd.DataFrame, kline: pd.DataFrame) -> float:
        """计算净流入 / 流通市值"""
        # 这里需要流通市值数据
        # 简化处理：用收盘价 × 流通股本近似
        total_inflow = flow.tail(5)['zlJlr'].sum()
        # 假设平均流通市值约等于 close × 流通股本
        # 实际项目里直接用 ig50 的 f10 接口取流通市值
        return total_inflow / 1e10  # 简化
    
    def _calc_return(self, kline: pd.DataFrame, days: int) -> float:
        """计算过去 N 天累计涨跌幅"""
        if len(kline) < days:
            return 0.0
        return (kline['close'].iloc[-1] / kline['close'].iloc[-days]) - 1
```

## 五、LLM 信号层（核心）

LLM 信号层是整个 Agent 的核心。我用 LLM 给每只股票打 4 个维度的"软判断"分数。

```python
import json
from openai import OpenAI

class LLMSignalLayer:
    """LLM 信号识别层"""
    
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.system_prompt = """
你是一个 A 股主力资金策略的资深研究员。
你的任务是：根据提供的股票特征，判断这次主力资金净流入是"真吸筹"、"诱多"还是"市值管理动作"。

请从以下 4 个维度分别打分（每个维度 0-10 分）：

1. **吸筹强度**：连续净流入天数越多，得分越高
2. **位置合理性**：股价处于阶段低位 + 净流入 = 逆向吸筹（高分）；股价高位 + 净流入 = 诱多（低分）
3. **市值管理嫌疑**：最近有减持公告、业绩预减、股权激励解锁等，得分降低
4. **行业协同度**：整个行业同步净流入，得分升高；只有个股净流入，得分中性

输出 JSON 格式：
{
    "absorption": 8,
    "position": 6,
    "manipulation_suspicion": 2,
    "industry_synergy": 7,
    "overall": "TRUE_BUILDING" | "BULL_TRAP" | "MARKET_MANAGEMENT" | "NEUTRAL",
    "reason": "..."
}
"""
    
    def score(self, features: Dict) -> Dict:
        """对单只股票打分
        
        参数：
        - features: FeatureBuilder.build_features 返回的特征字典
        
        返回：
        - LLM 的 4 个维度打分 + 综合判断 + 理由
        """
        user_prompt = f"""
请分析以下股票的主力资金信号：

股票代码：{features['dm']}
股票名称：{features['mc']}
行业：{features['industry']}

资金流特征：
- 连续净流入天数：{features['continuous_inflow_days']}
- 最近 5 天主力净流入：{features['total_inflow_5d']:.2f} 亿
- 净流入波动率：{features['inflow_volatility']:.2f}
- 净流入 / 流通市值比值：{features['avg_inflow_ratio']*100:.2f}%

行情特征：
- 最近 5 天累计涨跌幅：{features['recent_5d_return']*100:.2f}%
- 最近 20 天累计涨跌幅：{features['recent_20d_return']*100:.2f}%

公告特征：
- 最近 30 天有公告：{features['has_recent_announcement']}
- 公告类型：{features['announcement_types']}

L2 指标：
- 最近 5 天平均 ddx：{features['avg_ddx']:.2f}
- 最近 5 天平均 ddy：{features['avg_ddy']:.2f}
"""
        
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            response_format={"type": "json_object"},
            temperature=0.0
        )
        
        result = json.loads(response.choices[0].message.content)
        
        # 加入特征信息
        result['features'] = features
        result['total_score'] = (
            result['absorption'] + 
            result['position'] - 
            result['manipulation_suspicion'] + 
            result['industry_synergy']
        )
        
        return result
```

## 六、规则引擎层

规则引擎层负责把 LLM 的"软判断"和硬编码的"硬规则"组合起来。

```python
class RuleEngine:
    """规则引擎层"""
    
    def __init__(self, llm_layer: LLMSignalLayer, feature_builder: FeatureBuilder):
        self.llm = llm_layer
        self.features = feature_builder
    
    def decide(self, dm: str) -> Dict:
        """对单只股票做交易决策
        
        返回：
        - action: BUY / SELL / HOLD
        - reason: 决策理由
        - position_size: 建议仓位
        """
        # 1. 计算硬指标特征
        features = self.features.build_features(dm)
        
        # 2. 硬规则过滤
        if features['continuous_inflow_days'] < 5:
            return {
                'action': 'HOLD',
                'reason': '连续净流入天数 < 5，不符合买入条件',
                'position_size': 0
            }
        
        if features['avg_inflow_ratio'] > 0.01:
            return {
                'action': 'HOLD',
                'reason': '净流入 / 流通市值 > 1%，极端信号，反向',
                'position_size': 0
            }
        
        # 3. LLM 软判断
        llm_result = self.llm.score(features)
        
        # 4. 综合决策
        if (llm_result['overall'] == 'TRUE_BUILDING' and 
            llm_result['total_score'] >= 25):
            return {
                'action': 'BUY',
                'reason': f"LLM 判断：{llm_result['reason']}",
                'position_size': 0.05,  # 单只股票 5% 仓位
                'llm_signals': llm_result
            }
        
        elif llm_result['overall'] == 'BULL_TRAP':
            return {
                'action': 'SELL',
                'reason': f"LLM 判断为诱多：{llm_result['reason']}",
                'position_size': 0,
                'llm_signals': llm_result
            }
        
        else:
            return {
                'action': 'HOLD',
                'reason': f"LLM 判断：{llm_result['overall']}",
                'position_size': 0,
                'llm_signals': llm_result
            }
```

## 七、执行层

```python
class ExecutionLayer:
    """执行层"""
    
    def __init__(self, broker_api):
        self.broker = broker_api
        self.positions = {}
    
    def execute(self, decision: Dict, dm: str):
        """执行单只股票的决策"""
        action = decision['action']
        size = decision['position_size']
        
        if action == 'BUY':
            # 计算买入数量
            available_cash = self.broker.get_available_cash()
            amount = available_cash * size
            price = self.broker.get_current_price(dm)
            quantity = int(amount / price / 100) * 100  # A 股 100 股一手
            
            if quantity > 0:
                order_id = self.broker.buy(dm, price, quantity)
                self.positions[dm] = {
                    'order_id': order_id,
                    'entry_price': price,
                    'quantity': quantity,
                    'entry_date': pd.Timestamp.now()
                }
                
                # 记录日志
                self._log_trade(dm, 'BUY', price, quantity, decision)
        
        elif action == 'SELL':
            if dm in self.positions:
                pos = self.positions[dm]
                price = self.broker.get_current_price(dm)
                self.broker.sell(dm, price, pos['quantity'])
                del self.positions[dm]
                
                self._log_trade(dm, 'SELL', price, pos['quantity'], decision)
    
    def _log_trade(self, dm, action, price, quantity, decision):
        """记录交易日志"""
        log = {
            'timestamp': pd.Timestamp.now(),
            'dm': dm,
            'action': action,
            'price': price,
            'quantity': quantity,
            'reason': decision['reason']
        }
        # 实际项目里写入数据库
        pass
```

## 八、主循环

把整个 Agent 串起来：

```python
class MoneyFlowAgent:
    """主力资金追踪 Agent"""
    
    def __init__(self, data_client, broker_api, llm_api_key):
        self.data = data_client
        self.features = FeatureBuilder(data_client)
        self.llm = LLMSignalLayer(llm_api_key)
        self.rules = RuleEngine(self.llm, self.features)
        self.execution = ExecutionLayer(broker_api)
    
    def run_daily(self):
        """每日运行"""
        # 1. 获取全市场股票列表
        gplist = self.data._query("/base/gplist")
        gplist.columns = ['dm', 'mc']
        
        decisions = []
        
        # 2. 逐只股票做决策
        for dm in gplist['dm']:
            decision = self.rules.decide(dm)
            decisions.append({
                'dm': dm,
                'decision': decision
            })
            
            # 3. 执行决策
            if decision['action'] in ['BUY', 'SELL']:
                self.execution.execute(decision, dm)
        
        return decisions
```

## 九、踩过的 4 个坑

最后说几个工程实现中踩过的坑：

**坑 1：LLM 调用成本**

LLM 调用一次大概 $0.01-0.03，5000 只股票全跑一遍要 $50-150。**我后来优化成"先用硬规则过滤到 100 只，再让 LLM 判断"**，成本降到 $1-3。

**坑 2：LLM 判断的稳定性**

同一个特征，跑 5 次 LLM，可能有 3 种不同的 overall 判断。**解决方法是 temperature=0.0 + 多模型投票**。

**坑 3：特征工程必须做扎实**

LLM 不是万能的——**如果特征给的不全，LLM 也会瞎判断**。我花了一周时间专门做特征工程，确保所有可能用到的硬指标都给到 LLM。

**坑 4：本地数据是基础**

整套 Agent 跑得动的前提是**本地数据稳定、低延迟**。我用的是本地股票数据引擎，3 秒落盘、毫秒级查询。**没有本地数据，Agent 跑不动**。

## 十、写在最后

做这个主力资金追踪 Agent 最大的体会是——**LLM 不是来替代硬规则的，而是来做"软判断"的**。

硬指标（连续天数、净流入金额、流通市值比值）用代码算。
软判断（吸筹强度、位置合理性、市值管理嫌疑）让 LLM 来做。

**两者结合，alpha 才能稳定**。

整套 Agent 的核心思想就是——**用工程化的方式让 LLM 参与量化决策**，而不是让 LLM 替代量化决策。

[gitee开源地址](https://gitee.com/igtrade/ighub) 
[github开源地址](https://github.com/ig50-service/ig50-official)

做量化这几年，我一直在思考一个问题——LLM 能不能直接帮我做主力资金策略的"信号识别"工作？

从 0 搭建一个主力资金追踪 Agent：让 LLM 帮你识别真实建仓信号 IG50免费开源股票数据API接口

人工智能

算法

后端

从0搭建主力资金追踪Agent，通过LLM软判断+硬编码规则识别真实建仓信号。架构分5层：数据、特征工程、LLM信号、规则引擎、执行。核心是让LLM从吸筹强度、位置合理性、市值管理嫌疑、行业协同度4维度打分，结合连续净流入天数等硬指标过滤，实现高效信号识别。

代码解释

数据库

流数据

Python

Agent

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

从 0 搭建一个主力资金追踪 Agent：让 LLM 帮你识别真实建仓信号 IG50免费开源股票数据API接口-腾讯云开发者社区-腾讯云

从 0 搭建一个主力资金追踪 Agent：让 LLM 帮你识别真实建仓信号 IG50免费开源股票数据API接口

从 0 搭建一个主力资金追踪 Agent：让 LLM 帮你识别真实建仓信号 IG50免费开源股票数据API接口

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐