Hummingbot logo Hummingbot
登录
6. V2脚本开发基础

6.V2脚本开发基础

从这一章开始，我们真正动手写代码。前面几章把环境搭好、交易所连上、V2框架的原理讲透，现在该把这些知识转化成能跑的脚本了。V2脚本开发并不复杂，核心是把交易逻辑塞进几个固定的方法里，再配上一套参数系统，就能让策略在Hummingbot里跑起来。

V2脚本架构与生命周期

V2脚本的骨架继承自 StrategyV2Base 类，这个基类在 hummingbot/strategy/strategy_v2_base.py 里定义。它本身又是 ScriptStrategyBase 的子类，所以V2脚本同时具备了V1脚本的简洁性和V2框架的模块化能力。

继承关系

整个继承链条是这样的：StrategyBase → ScriptStrategyBase → StrategyV2Base。最顶层的 StrategyBase 是Cython写的，处理底层事件循环和订单管理；ScriptStrategyBase 用Python封装了一层，提供了 buy()、sell() 这些直接操作市场的方法；StrategyV2Base 再往上加了两样东西：一是统一的市场数据接口，二是基于Executor的订单管理机制。

写V2脚本时，我们很少需要直接调用 buy() 或 sell()。取而代之的是创建Executor，让Executor去处理订单的生命周期。这种设计把"决策"和"执行"分开了，脚本专注生成交易信号，Executor专注把订单管好。

生命周期方法

一个V2脚本实例化后会经历几个阶段：

  1. __init__：初始化连接器、市场数据提供者、配置参数
  2. on_start：策略启动时调用一次，适合在这里初始化Candles数据源
  3. on_tick：每隔 tick_size（默认1秒）调用一次，是策略的核心循环
  4. format_status：用户执行 status 命令时调用，返回要展示的文本
  5. on_stop：策略停止时调用，做清理工作

on_tick 是最重要的方法。它里面通常会有一个主循环，检查市场数据、计算指标、生成信号，然后创建或更新Executor。这个方法必须轻量，不能阻塞，否则会影响整个事件循环。

配置参数定义与解析

V2框架最大的改进之一是把参数管理标准化了。以前写脚本得硬编码参数，现在可以定义一个配置类，让Hummingbot自动生成配置文件，还能在运行时动态更新。

配置类定义

配置类继承自 BaseClientModel，用Pydantic的 Field 定义每个参数。每个字段可以带 client_data，里面指定提示语、默认值、是否在新创建时提示用户等。

class SimpleMACDConfig(StrategyV2ConfigBase):
    script_file_name: str = Field(default_factory=lambda: os.path.basename(__file__))
    
    # 交易配置
    exchange: str = Field(
        "binance_perpetual",
        client_data=ClientFieldData(
            prompt_on_new=True,
            prompt=lambda mi: "交易所名称 (如 binance_perpetual):"
        )
    )
    trading_pair: str = Field(
        "BTC-USDT",
        client_data=ClientFieldData(
            prompt_on_new=True,
            prompt=lambda mi: "交易对 (如 BTC-USDT):"
        )
    )
    order_amount: Decimal = Field(
        Decimal("100"),
        client_data=ClientFieldData(
            prompt_on_new=True,
            prompt=lambda mi: "订单金额 (USD):"
        )
    )
    
    # 指标参数
    fast_period: int = Field(12, client_data=ClientFieldData(prompt_on_new=False))
    slow_period: int = Field(26, client_data=ClientFieldData(prompt_on_new=False))

prompt_on_new=True 的参数会在运行 create --script-config 时提示用户输入。prompt_on_new=False 的参数不会提示，但可以在配置文件里手动修改。default_factory 用来动态生成默认值，比如当前文件名。

配置文件创建

写好配置类后，在Hummingbot客户端执行：

create --script-config simple_macd.py

客户端会扫描 /scripts 目录，列出所有带配置类的脚本。选中我们的脚本后，会逐个提示 prompt_on_new=True 的参数。输入完毕后，配置文件保存在 conf/scripts/simple_macd.yml。

配置文件是YAML格式，可以手动编辑。修改后保存，脚本会在下一个 config_update_interval（默认60秒）自动加载新参数，无需重启。这种热更新能力在生产环境特别有用。

on_tick方法实现逻辑

on_tick 是策略的心跳，每秒跳动一次。它的实现模式通常分三步：获取数据、计算信号、执行动作。

数据获取

通过 market_data_provider 获取市场数据，这是V2框架的统一入口：

def on_tick(self):
    # 获取最新价格
    price = self.market_data_provider.get_price_by_type(
        self.config.exchange,
        self.config.trading_pair,
        PriceType.MidPrice
    )
    
    # 获取K线数据
    candles_df = self.market_data_provider.get_candles_df(
        self.config.exchange,
        self.config.trading_pair,
        "1m",
        100
    )
    
    # 获取订单簿
    bids_df, asks_df = self.market_data_provider.get_order_book_snapshot(
        self.config.exchange,
        self.config.trading_pair
    )

所有数据获取都是异步的，但 market_data_provider 帮我们封装好了，调用时数据已经准备好。candles_df 是Pandas DataFrame，可以直接用 pandas_ta 计算指标。

信号计算

拿到数据后，用技术指标生成交易信号。这里以MACD为例：

import pandas_ta as ta

def on_tick(self):
    # 获取数据
    candles_df = self.market_data_provider.get_candles_df(...)
    
    # 计算MACD
    macd_df = ta.macd(
        candles_df["close"],
        fast=self.config.fast_period,
        slow=self.config.slow_period
    )
    
    # 获取最新信号
    current_macd = macd_df.iloc[-1]["MACD_12_26_9"]
    current_signal = macd_df.iloc[-1]["MACDs_12_26_9"]
    
    # 生成信号
    if current_macd > current_signal:
        signal = 1  # 做多
    elif current_macd < current_signal:
        signal = -1  # 做空
    else:
        signal = 0  # 观望

信号可以是简单的多空判断，也可以是复杂的打分系统。关键是把逻辑封装成纯函数，方便测试。

执行动作

根据信号创建或更新Executor。V2框架提供了几种内置Executor：

  • PositionExecutor：管理单向仓位，带止损止盈
  • GridExecutor：网格交易
  • TWAPExecutor：时间加权平均价格执行
from hummingbot.smart_components.executors.position_executor import PositionExecutor

def on_tick(self):
    # ... 获取信号
    
    if signal == 1 and not self.long_executor:
        # 创建多头Executor
        self.long_executor = PositionExecutor(
            strategy=self,
            connector_name=self.config.exchange,
            trading_pair=self.config.trading_pair,
            side=TradeType.BUY,
            amount=self.config.order_amount,
            stop_loss=0.01,
            take_profit=0.02
        )
        self.long_executor.start()
    
    elif signal == -1 and self.long_executor:
        # 关闭多头
        self.long_executor.stop()
        self.long_executor = None

Executor一旦启动，会自动处理订单创建、跟踪、止损止盈。脚本只需要在合适的时机启动或停止它。

format_status状态展示

status 命令是调试策略的利器。format_status 方法返回的字符串会原样显示在客户端里。默认实现只展示余额和活动订单，我们需要重写它来展示策略核心状态。

基础实现

def format_status(self) -> str:
    if not self.ready_to_trade:
        return "市场连接器未就绪"
    
    lines = []
    lines.append(f"策略: {self.config.strategy_name}")
    lines.append(f"交易对: {self.config.trading_pair}")
    lines.append(f"最新价格: {self.market_data_provider.get_price_by_type(...)}")
    
    # 展示指标状态
    if self.all_candles_ready:
        candles_df = self.market_data_provider.get_candles_df(...)
        lines.append(f"MACD: {candles_df.iloc[-1]['MACD_12_26_9']:.4f}")
        lines.append(f"Signal: {candles_df.iloc[-1]['MACDs_12_26_9']:.4f}")
    
    # 展示Executor状态
    if self.long_executor:
        lines.append(f"多头仓位: {self.long_executor.status}")
        lines.append(f"  盈亏: {self.long_executor.pnl:.2f} USD")
    
    return "\n".join(lines)

状态信息分三类：市场数据、指标数值、Executor状态。展示太多会刷屏，只放关键信息。

高级技巧

可以用 tabulate 库格式化表格，或者用ANSI颜色高亮重要信息。Hummingbot客户端支持基本的颜色代码：

def format_status(self) -> str:
    lines = []
    
    # 红色表示亏损
    pnl = self.long_executor.pnl if self.long_executor else 0
    color = "\033[91m" if pnl < 0 else "\033[92m"  # 红/绿
    reset = "\033[0m"
    
    lines.append(f"盈亏: {color}{pnl:.2f} USD{reset}")
    return "\n".join(lines)

不过颜色要慎用，太多会眼花缭乱。

市场订单创建与取消

虽然V2推荐用Executor，但有时候需要直接操作订单。ScriptStrategyBase 提供了底层方法。

创建订单

# 限价买单
order_id = self.buy(
    connector_name="binance_perpetual",
    trading_pair="BTC-USDT",
    amount=Decimal("0.01"),
    order_type=OrderType.LIMIT,
    price=Decimal("65000")
)

# 市价卖单
order_id = self.sell(
    connector_name="binance_perpetual",
    trading_pair="BTC-USDT",
    amount=Decimal("0.01"),
    order_type=OrderType.MARKET
)

buy() 和 sell() 返回 order_id，可以用来跟踪订单状态。amount 是交易货币的数量，不是USD。OrderType 枚举包括 LIMIT、MARKET、LIMIT_MAKER 等。

取消订单

# 取消单个订单
self.cancel("binance_perpetual", "BTC-USDT", order_id)

# 取消所有订单
self.cancel_all_orders("binance_perpetual", "BTC-USDT")

取消是异步操作，命令发出后订单不会立刻消失，需要等待交易所确认。可以在 did_cancel_order 事件处理器里做后续处理。

事件处理

订单状态变化会触发事件，重写这些方法可以自定义处理逻辑：

def did_fill_order(self, event: OrderFilledEvent):
    self.logger().info(f"订单成交: {event.order_id} {event.trade_type} {event.amount}")
    
def did_cancel_order(self, event: OrderCancelledEvent):
    self.logger().info(f"订单取消: {event.order_id}")
    
def did_fail_order(self, event: MarketOrderFailureEvent):
    self.logger().error(f"订单失败: {event.order_id} {event.error}")

事件系统是基于回调的，不需要轮询订单状态。这在高频场景下效率更高。

账户数据访问接口

策略需要实时知道账户里有多少钱、多少仓位。V2框架提供了几个便捷方法。

余额查询

# 获取全部余额DataFrame
balance_df = self.get_balance_df()
print(balance_df)

# 获取单个资产余额
connector = self.connectors["binance_perpetual"]
btc_balance = connector.get_balance("BTC")
usdt_balance = connector.get_balance("USDT")

get_balance_df() 返回的DataFrame包含四列：Exchange、Asset、Total Balance、Available Balance。get_balance() 返回的是可用余额，不包括挂单冻结的部分。

活动订单

# 获取所有活动订单DataFrame
orders_df = self.active_orders_df()
print(orders_df)

# 获取单个订单
order = self.connectors["binance_perpetual"].get_order(order_id)

active_orders_df() 返回的DataFrame包含Exchange、Market、Side、Price、Amount、Age等列。Age是订单存活时间，用来判断是否需要刷新。

仓位信息

对于永续合约，还需要查询仓位：

connector = self.connectors["binance_perpetual"]
position = connector.get_position("BTC-USDT")

if position:
    print(f"仓位大小: {position.amount}")
    print(f"开仓价格: {position.entry_price}")
    print(f"未实现盈亏: {position.unrealized_pnl}")

get_position() 返回 Position 对象，包含仓位方向、数量、杠杆等信息。现货连接器调用这个方法会返回 None。

基础脚本示例解析

理论讲得差不多，来看一个完整的可运行示例。这个脚本实现了一个简单的RSI均值回归策略：RSI低于30时买入，高于70时卖出，用 PositionExecutor 管理仓位。

完整代码

import os
from decimal import Decimal
from pydantic import Field
import pandas_ta as ta

from hummingbot.strategy.strategy_v2_base import StrategyV2Base, StrategyV2ConfigBase
from hummingbot.core.data_type.common import OrderType, PriceType, TradeType
from hummingbot.smart_components.executors.position_executor import PositionExecutor
from hummingbot.client.config.config_data_types import ClientFieldData
from hummingbot.core.data_type.trade_fee import TokenAmount

class SimpleRSIConfig(StrategyV2ConfigBase):
    script_file_name: str = Field(default_factory=lambda: os.path.basename(__file__))
    
    exchange: str = Field(
        "binance_perpetual",
        client_data=ClientFieldData(
            prompt_on_new=True,
            prompt=lambda mi: "交易所名称:"
        )
    )
    trading_pair: str = Field(
        "BTC-USDT",
        client_data=ClientFieldData(
            prompt_on_new=True,
            prompt=lambda mi: "交易对:"
        )
    )
    order_amount: Decimal = Field(
        Decimal("50"),
        client_data=ClientFieldData(
            prompt_on_new=True,
            prompt=lambda mi: "订单金额 (USD):"
        )
    )
    rsi_length: int = Field(14, client_data=ClientFieldData(prompt_on_new=False))
    rsi_low: int = Field(30, client_data=ClientFieldData(prompt_on_new=False))
    rsi_high: int = Field(70, client_data=ClientFieldData(prompt_on_new=False))

class SimpleRSIStrategy(StrategyV2Base):
    def __init__(self, connectors):
        super().__init__(connectors)
        self.config = None
        self.position_executor = None

    def on_start(self):
        # 初始化配置
        self.config = SimpleRSIConfig.load_from_file(self.config_file_path)
        
        # 初始化K线数据源
        self.market_data_provider.initialize_candles_feed(
            connector=self.config.exchange,
            trading_pair=self.config.trading_pair,
            interval="1m",
            max_records=500
        )

    def on_tick(self):
        if not self.market_data_provider.ready:
            return
        
        # 获取RSI
        candles_df = self.market_data_provider.get_candles_df(
            self.config.exchange,
            self.config.trading_pair,
            "1m",
            self.config.rsi_length + 10
        )
        
        rsi_series = ta.rsi(candles_df["close"], length=self.config.rsi_length)
        current_rsi = rsi_series.iloc[-1]
        
        # 检查是否需要开平仓
        if current_rsi < self.config.rsi_low and not self.position_executor:
            # RSI过低，开多
            self.position_executor = PositionExecutor(
                strategy=self,
                connector_name=self.config.exchange,
                trading_pair=self.config.trading_pair,
                side=TradeType.BUY,
                amount=self.config.order_amount,
                stop_loss=0.02,
                take_profit=0.03
            )
            self.position_executor.start()
            self.notify_hb_app(f"RSI {current_rsi:.1f} < {self.config.rsi_low}，开多")
            
        elif current_rsi > self.config.rsi_high and self.position_executor:
            # RSI过高，平多
            self.position_executor.stop()
            self.position_executor = None
            self.notify_hb_app(f"RSI {current_rsi:.1f} > {self.config.rsi_high}，平多")

    def format_status(self) -> str:
        if not self.ready_to_trade:
            return "市场连接器未就绪"
        
        lines = []
        lines.append(f"策略: SimpleRSI")
        lines.append(f"交易对: {self.config.trading_pair}")
        
        # 显示RSI
        if self.market_data_provider.ready:
            candles_df = self.market_data_provider.get_candles_df(...)
            rsi = ta.rsi(candles_df["close"], length=self.config.rsi_length).iloc[-1]
            lines.append(f"RSI: {rsi:.2f}")
        
        # 显示Executor状态
        if self.position_executor:
            lines.append(f"仓位状态: {self.position_executor.status}")
            lines.append(f"盈亏: {self.position_executor.pnl:.2f} USD")
        
        return "\n".join(lines)

代码解析

这个脚本展示了V2开发的完整流程：

  1. 配置定义：SimpleRSIConfig 类定义了所有参数，带提示语，支持配置文件
  2. 初始化：on_start 里加载配置、初始化K线数据源
  3. 主逻辑：on_tick 里计算RSI，根据阈值创建或关闭Executor
  4. 状态展示：format_status 展示RSI值和Executor状态
  5. 通知系统：notify_hb_app 在关键操作时发送通知

运行步骤

把代码保存为 /scripts/simple_rsi.py，然后在Hummingbot客户端：

# 创建配置文件
create --script-config simple_rsi.py

# 按提示输入参数，保存为 conf_simple_rsi.yml

# 启动策略
start --script simple_rsi.py --conf conf_simple_rsi.yml

# 查看状态
status

# 实时状态
status --live

策略会每秒检查一次RSI，满足条件就自动下单。PositionExecutor 会管理止损止盈，不需要我们操心。

总结

V2脚本开发的核心是理解三个东西：配置系统、on_tick 循环、Executor机制。配置让参数管理变得规范；on_tick 是策略大脑，每秒做一次决策；Executor是执行层，负责把订单管好。把这三件事搞明白，就能写出健壮的交易策略。

这一章只讲了基础，下一章会深入控制器设计。控制器能把策略逻辑进一步抽象，让同一个脚本跑多个配置，甚至多个策略混跑。这是V2框架真正强大的地方。

上一章

V2策略框架核心原理

 下一章

V2控制器设计与实现