第3章单智能体基础

3. 单智能体基础：设计一个简单的趋势跟踪Agent

好，咱们今天来聊点实在的。

前面两章我们把多智能体系统的概念框架搭起来了，也聊了聊市场环境怎么建模。但说实话，那些都是「虚」的。真正要上手干活，得从最基础的单元开始——单智能体。

这一章，我带你手写一个最简单的趋势跟踪Agent。别小看它，麻雀虽小五脏俱全。你把它搞明白了，后面几十个Agent协同作战的逻辑，无非就是把这个闭环复制粘贴、再组合。

3.1 智能体的核心：感知-决策-执行闭环

先问一个问题：一个交易智能体，本质上是什么？

说白了，就是一个循环。它不停地做三件事：

感知：看市场现在什么样（价格、成交量、指标）
决策：根据看到的信号，判断该买、该卖、还是该躺平
执行：把决策变成真实的订单，扔到交易所

嗯，就这么简单。但很多新手容易犯一个错——他们只关注「决策」那一步，觉得策略牛逼就完事了。我刚开始做量化的时候也这样，花三个月调参数，结果实盘一跑，数据源断了都不知道。为什么？因为感知层没做好。

  核心观点： 感知是决策的基础。数据不准，再好的策略也是垃圾。

所以这一章，我会把这三个环节拆开，每个环节都给你讲透。

3.2 感知层：Agent如何「看」市场

感知层要解决一个问题：Agent能看到什么？

我个人习惯，把感知数据分成三类：

数据类型	例子	更新频率
原始行情	开盘价、收盘价、最高价、最低价、成交量	实时（tick/1分钟）
技术指标	MA、MACD、RSI、布林带	每根K线更新
市场状态	趋势/震荡/高波动/低波动	按需计算

对于趋势跟踪Agent，最核心的感知信号就是移动平均线（MA）。为什么？因为它平滑了噪音，让你看到方向。

我曾经踩过一个坑：用5日均线和10日均线做金叉死叉，回测曲线漂亮得不行。结果实盘一周，连续被假突破打脸。后来才发现，我忘了加一个「过滤条件」——均线斜率。你想想看，两根均线交叉的时候，如果它们都是水平的，那这个交叉有意义吗？没意义。只有均线本身有斜率，交叉才代表趋势启动。

避坑指南： 感知层不要只取原始数据。加一层「特征工程」，比如计算均线斜率、波动率变化，能大幅减少噪音。

下面是一个简单的感知模块代码示例：

class TrendPerception:
    def __init__(self, short_window=5, long_window=20):
        self.short_window = short_window
        self.long_window = long_window
    
    def perceive(self, price_data):
        """感知当前市场状态"""
        # 计算两条均线
        short_ma = price_data['close'].rolling(self.short_window).mean()
        long_ma = price_data['close'].rolling(self.long_window).mean()
        
        # 计算均线斜率（用最近5个点的线性回归斜率）
        short_slope = self._calc_slope(short_ma)
        long_slope = self._calc_slope(long_ma)
        
        # 判断是否形成金叉/死叉
        cross_signal = 'none'
        if short_ma.iloc[-2] < long_ma.iloc[-2] and short_ma.iloc[-1] > long_ma.iloc[-1]:
            cross_signal = 'golden_cross'
        elif short_ma.iloc[-2] > long_ma.iloc[-2] and short_ma.iloc[-1] < long_ma.iloc[-1]:
            cross_signal = 'death_cross'
        
        return {
            'short_ma': short_ma.iloc[-1],
            'long_ma': long_ma.iloc[-1],
            'short_slope': short_slope,
            'long_slope': long_slope,
            'cross_signal': cross_signal
        }
    
    def _calc_slope(self, series):
        """计算最近5个点的斜率"""
        if len(series) < 5:
            return 0
        y = series.iloc[-5:].values
        x = np.array([0, 1, 2, 3, 4])
        # 最小二乘法求斜率
        slope = np.polyfit(x, y, 1)[0]
        return slope

3.3 决策层：从信号到动作

感知完了，接下来就是决策。决策层要回答一个问题：我现在该干什么？

对于趋势跟踪，决策逻辑其实很朴素：

如果金叉出现 + 均线斜率向上 → 开多仓
如果死叉出现 + 均线斜率向下 → 开空仓
其他情况 → 持仓不动或空仓

但这里有一个细节，很多人会忽略：仓位管理。你决策了「开多仓」，但开多少？全仓干？那万一回调呢？

我建议的做法是：把决策拆成两个子步骤——方向决策和仓位决策。

  方向决策： 用技术信号判断做多/做空/观望。

  仓位决策： 用风险模型（如ATR、波动率）计算开仓比例。

举个例子：

class TrendDecision:
    def __init__(self, base_position=0.1, max_position=0.5):
        self.base_position = base_position
        self.max_position = max_position
    
    def decide(self, perception_result, account_info):
        """根据感知结果做决策"""
        signal = perception_result['cross_signal']
        short_slope = perception_result['short_slope']
        long_slope = perception_result['long_slope']
        
        # 方向决策
        direction = 0  # 0: 观望, 1: 做多, -1: 做空
        if signal == 'golden_cross' and short_slope > 0 and long_slope > 0:
            direction = 1
        elif signal == 'death_cross' and short_slope < 0 and long_slope < 0:
            direction = -1
        
        # 仓位决策（基于当前波动率调整）
        if direction != 0:
            volatility = account_info.get('current_volatility', 0.01)
            # 波动率越大，仓位越小
            position = self.base_position * (0.02 / max(volatility, 0.005))
            position = min(position, self.max_position)
        else:
            position = 0
        
        return {
            'direction': direction,
            'position': position,
            'reason': f"信号: {signal}, 斜率: {short_slope:.4f}"
        }

你可能会问：为什么要把仓位和方向分开？

嗯，这里有个实战经验。我早期做策略的时候，把仓位写死在决策逻辑里。结果遇到一次黑天鹅，波动率瞬间翻倍，我的仓位没变，直接爆仓。从那以后，我所有的Agent都是「方向」和「仓位」分离的。方向由策略决定，仓位由风控决定。各司其职，互不干扰。

3.4 执行层：把决策变成订单

执行层是最后一步，也是最容易被低估的一步。很多人觉得「决策做好了，执行不就是发个单吗？」

大错特错。

执行层要考虑的事情多了去了：

订单类型：市价单还是限价单？市价单成交快但滑点大，限价单滑点小但可能成交不了。
订单拆分：大单要不要拆成小单？避免冲击成本。
重试机制：如果订单没成交，要不要重试？重试几次？
错误处理：交易所返回错误码怎么办？

我见过最离谱的事：一个团队的回测年化50%，实盘一跑直接亏20%。为什么？因为回测里假设「市价单永远以当前价成交」，但实盘里滑点能把利润全吃掉。

警告： 执行层一定要做滑点模拟。回测时至少加0.1%的滑点成本，否则你的回测曲线就是「照骗」。

下面是一个简单的执行模块：

class TrendExecutor:
    def __init__(self, exchange_api, max_slippage=0.001):
        self.api = exchange_api
        self.max_slippage = max_slippage
    
    def execute(self, decision, current_price):
        """执行交易决策"""
        if decision['direction'] == 0:
            return {'status': 'no_action', 'message': '无操作'}
        
        # 计算目标价格（考虑滑点）
        side = 'buy' if decision['direction'] == 1 else 'sell'
        if side == 'buy':
            target_price = current_price * (1 + self.max_slippage)
        else:
            target_price = current_price * (1 - self.max_slippage)
        
        # 发送订单
        order = {
            'symbol': 'BTCUSDT',
            'side': side,
            'type': 'limit',
            'price': target_price,
            'quantity': decision['position']
        }
        
        try:
            result = self.api.place_order(order)
            return {'status': 'success', 'order_id': result['order_id']}
        except Exception as e:
            # 重试一次
            try:
                result = self.api.place_order(order)
                return {'status': 'success_retry', 'order_id': result['order_id']}
            except:
                return {'status': 'failed', 'error': str(e)}

3.5 闭环整合：让Agent跑起来

三个模块都写好了，怎么把它们串起来？

很简单，一个主循环：

class TrendAgent:
    def __init__(self, perception, decision, executor):
        self.perception = perception
        self.decision = decision
        self.executor = executor
        self.position = 0  # 当前持仓
    
    def run(self, market_data, account_info):
        """一次完整的感知-决策-执行循环"""
        # 1. 感知
        perception_result = self.perception.perceive(market_data)
        
        # 2. 决策
        decision = self.decision.decide(perception_result, account_info)
        
        # 3. 执行
        current_price = market_data['close'].iloc[-1]
        execution_result = self.executor.execute(decision, current_price)
        
        # 4. 更新持仓状态
        if execution_result['status'] in ['success', 'success_retry']:
            self.position = decision['position'] * decision['direction']
        
        return {
            'perception': perception_result,
            'decision': decision,
            'execution': execution_result,
            'position': self.position
        }

你看，整个闭环就这么清晰。每次调用 run()，Agent就完成一次「看市场→想对策→下订单」的循环。

3.6 知识体系总览

为了让你更直观地理解这个闭环，我画了一张图：

3.7 小结

这一章我们做了三件事：

拆解了智能体的感知-决策-执行闭环
用代码实现了每个模块
把它们整合成一个可运行的Agent

你可能会觉得，这个趋势跟踪Agent太简单了，连机器学习都没用。没错，它确实简单。但你要明白，复杂系统都是从简单单元开始的。后面我们讲多智能体协作，每个Agent的底层逻辑都是这个闭环。

把基础打牢，后面才能飞得起来。

课后练习： 试着给这个Agent加一个「止损」功能。在决策层里判断：如果当前价格跌破最近10根K线的最低点，就平仓。看看回测效果有什么变化。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

3. 单智能体基础：设计一个简单的趋势跟踪Agent

3.1 智能体的核心：感知-决策-执行闭环

3.2 感知层：Agent如何「看」市场

3.3 决策层：从信号到动作

3.4 执行层：把决策变成订单

3.5 闭环整合：让Agent跑起来

3.6 知识体系总览

3.7 小结