11.K线数据与技术指标集成
交易的世界里,价格不会说话,但K线会。每一根K线都是市场在某个时间段内的情绪快照,开盘价、最高价、最低价、收盘价,还有成交量,这五个数字组合起来,能讲出无数故事。Hummingbot 的 V2 框架把 K 线数据和技术指标集成做得相当优雅,不需要自己写爬虫,也不用操心数据清洗,几行代码就能拿到干净的 OHLCV 数据,还能顺手把 RSI、MACD、布林带这些经典指标算好。
K 线数据源配置与初始化
在 V2 策略里获取 K 线数据,核心是靠 CandlesFactory 这个工厂类。设计模式里的工厂模式在这里用得恰到好处,不需要关心底层是哪个交易所、数据格式什么样,统一接口,拿来就用。
基础配置方法
最简洁的初始化方式是这样的:
from hummingbot.data_feed.candles_feed.candles_factory import CandlesFactory
candles = CandlesFactory.get_candle(
connector="binance",
trading_pair="BTC-USDT",
interval="5m",
max_records=100
)
代码里,connector 指定数据源,可以是现货交易所,也可以是永续合约。trading_pair 是交易对,格式要跟交易所保持一致,一般是 BASE-QUOTE 这种形式。interval 是时间周期,支持 1m、3m、5m、15m、30m、1h、2h、4h、6h、8h、12h、1d、1w、1M 这些常见周期。max_records 是保留的 K 线数量,相当于一个滑动窗口的大小。
这段代码执行后,candles 对象会自己在后台启动一个异步任务,持续从交易所拉取数据,维护一个长度为 100 的 K 线窗口。数据不够 100 根的时候,is_ready 属性会是 False,等数据攒够了自动变成 True。
使用配置对象
如果策略需要管理多个 K 线源,用配置对象会更清晰:
from hummingbot.data_feed.candles_feed.candles_factory import CandlesFactory, CandlesConfig
config = CandlesConfig(
connector="binance_perpetual",
trading_pair="ETH-USDT",
interval="15m",
max_records=200
)
candles = CandlesFactory.get_candle(config)
CandlesConfig 把参数封装起来,方便复用和传递。特别是在控制器(Controller)模式里,配置对象可以直接序列化成 JSON,存到配置文件里,策略启动时动态加载。
数据源的选择策略
一个常见的误区是:策略在哪个交易所交易,就必须用哪个交易所的 K 线。其实不然。Hummingbot 允许跨交易所使用数据,比如在流动性较差的小交易所做市,但用 Binance 的 K 线数据计算指标。这种做法很务实,大交易所的数据更稳定,深度更好,算出来的指标更可靠。
# 在 Hyperliquid 永续合约上交易,但用 Binance 的 K 线数据
trading_candles = CandlesFactory.get_candle(
connector="binance",
trading_pair="BTC-USDT",
interval="1h",
max_records=50
)
需要注意的是,不同交易所的 K 线时间戳对齐方式可能略有差异,大部分主流交易所都使用 UTC 时间,实际使用中差异不大。但如果策略对时间精度极度敏感,最好做一下对齐校验。
多时间框架数据同步
专业交易员都明白,单一时间框架的指标容易产生误判。1 小时级别的上涨趋势,可能在 5 分钟级别已经处于超买状态。多时间框架分析是标配,Hummingbot 里实现起来也不复杂。
同时初始化多个 K 线源
在策略类里定义多个 K 线实例:
from hummingbot.strategy.script_strategy_base import ScriptStrategyBase
from hummingbot.data_feed.candles_feed.candles_factory import CandlesFactory, CandlesConfig
class MultiTimeframeStrategy(ScriptStrategyBase):
# 定义三个不同周期的 K 线配置
candles_1m_config = CandlesConfig(
connector="binance",
trading_pair="BTC-USDT",
interval="1m",
max_records=60
)
candles_1h_config = CandlesConfig(
connector="binance",
trading_pair="BTC-USDT",
interval="1h",
max_records=24
)
candles_1d_config = CandlesConfig(
connector="binance",
trading_pair="BTC-USDT",
interval="1d",
max_records=30
)
def __init__(self, connectors):
super().__init__(connectors)
# 初始化三个 K 线源
self.candles_1m = CandlesFactory.get_candle(self.candles_1m_config)
self.candles_1h = CandlesFactory.get_candle(self.candles_1h_config)
self.candles_1d = CandlesFactory.get_candle(self.candles_1d_config)
这段代码在策略初始化时创建了三个不同周期的 K 线源。1m 的存 60 根,刚好覆盖最近一小时;1h 的存 24 根,覆盖最近一天;1d 的存 30 根,覆盖最近一个月。这种配置适合日内交易策略,既能捕捉短期波动,又能把握中期趋势。
检查数据就绪状态
在 on_tick 里使用多框架数据前,务必检查所有数据源是否就绪:
def on_tick(self):
# 检查所有 K 线源是否准备就绪
if not all([self.candles_1m.is_ready, self.candles_1h.is_ready, self.candles_1d.is_ready]):
self.logger().info("等待 K 线数据就绪...")
return
# 数据就绪后开始逻辑处理
df_1m = self.candles_1m.candles_df
df_1h = self.candles_1h.candles_df
df_1d = self.candles_1d.candles_df
# 在这里进行多时间框架分析
# 例如:日线趋势向上,小时线回调,分钟线出现买入信号
is_ready 属性是线程安全的,可以放心在 on_tick 里调用。如果某个数据源还没准备好,策略直接返回,不执行后续逻辑。这种防御性编程能避免很多空指针异常和索引错误。
数据同步的底层机制
Hummingbot 的 K 线模块内部用异步 IO 实现,每个 K 线源独立运行,互不阻塞。数据到达后会触发回调,更新内部的 deque 队列。candles_df 属性每次访问时都会把队列转换成 pandas.DataFrame,这个转换是轻量级的,对性能影响不大。
多框架同步的关键在于,所有数据源都使用相同的 timestamp 对齐方式。Hummingbot 内部统一用毫秒级 Unix 时间戳,转换 DataFrame 时再格式化成可读时间。这样即使不同周期的 K 线更新频率不同,也能保证在任意时刻获取到的数据都是对齐的。
技术指标添加与计算
拿到 K 线数据只是第一步,更重要的是从中提取交易信号。Hummingbot 没有重复造轮子,而是直接集成了 pandas_ta 库,这个库包含了 130 多个技术指标,从简单的均线到复杂的斐波那契回调,应有尽有。
在 K 线上添加指标
最基础的用法是在 format_status 里动态计算指标:
import pandas_ta as ta
def format_status(self) -> str:
if not self.ready_to_trade:
return "市场连接器未就绪"
lines = []
if self.candles_1h.is_ready:
# 获取 K 线 DataFrame
df = self.candles_1h.candles_df
# 添加 RSI 指标
df.ta.rsi(length=14, append=True)
# 添加布林带指标
df.ta.bbands(length=20, std=2.0, append=True)
# 添加 MACD 指标
df.ta.macd(fast=12, slow=26, signal=9, append=True)
# 显示最近 5 根 K 线
display_cols = ["timestamp", "open", "high", "low", "close", "volume",
"RSI_14", "BBL_20_2.0", "BBM_20_2.0", "BBU_20_2.0"]
lines.append(f"BTC-USDT 1小时K线及指标:")
lines.append(df[display_cols].tail().to_string(index=False))
return "\n".join(lines)
pandas_ta 的 append=True 参数非常实用,它会把计算结果直接添加到原 DataFrame 里,列名会自动生成,比如 RSI 的列名是 RSI_14,布林带的下轨、中轨、上轨分别是 BBL_20_2.0、BBM_20_2.0、BBU_20_2.0。
常用指标参数配置
不同策略对指标参数敏感度不同,最好把参数提取成策略配置:
class TechnicalStrategy(ScriptStrategyBase):
# 策略配置
rsi_length = 14
bb_length = 20
bb_std = 2.0
macd_fast = 12
macd_slow = 26
macd_signal = 9
def add_indicators(self, df):
"""批量添加技术指标"""
# 趋势指标
df.ta.ema(length=20, append=True, suffix="EMA20")
df.ta.ema(length=50, append=True, suffix="EMA50")
df.ta.sma(length=200, append=True, suffix="SMA200")
# 动量指标
df.ta.rsi(length=self.rsi_length, append=True)
df.ta.macd(fast=self.macd_fast, slow=self.macd_slow,
signal=self.macd_signal, append=True)
# 波动率指标
df.ta.bbands(length=self.bb_length, std=self.bb_std, append=True)
df.ta.atr(length=14, append=True)
# 成交量指标
df.ta.obv(append=True)
return df
把指标计算封装成独立方法,既保持了 on_tick 的简洁,又方便单元测试。suffix 参数可以给指标列名加后缀,避免不同周期的同名指标冲突。
避免重复计算
指标计算虽然快,但没必要每秒钟都算一次。可以在 on_tick 里加一个简单的计数器:
def __init__(self, connectors):
super().__init__(connectors)
self.tick_count = 0
self.last_calc_time = 0
def on_tick(self):
current_time = self.current_timestamp
# 每 60 秒计算一次指标
if current_time - self.last_calc_time < 60:
return
self.last_calc_time = current_time
if self.candles_1h.is_ready:
df = self.candles_1h.candles_df.copy()
df = self.add_indicators(df)
# 基于最新指标值生成交易信号
latest_rsi = df[f"RSI_{self.rsi_length}"].iloc[-1]
latest_bb_upper = df[f"BBU_{self.bb_length}_{self.bb_std}"].iloc[-1]
latest_price = df["close"].iloc[-1]
# 交易逻辑...
这样能把 CPU 占用降低一个数量级,特别是在需要计算大量复杂指标的时候。
K 线数据在状态栏展示
策略运行起来后,实时看到 K 线和指标的变化很重要。Hummingbot 客户端的 status 命令会调用策略的 format_status 方法,在这里把 K 线数据格式化输出。
基础展示格式
def format_status(self) -> str:
if not self.ready_to_trade:
return "市场连接器未就绪"
lines = ["\n" + "="*50 + " K线数据 " + "="*50 + "\n"]
# 检查数据是否就绪
if not self.candles_1h.is_ready:
lines.append("1小时K线数据加载中...")
return "\n".join(lines)
# 获取数据并格式化时间戳
df = self.candles_1h.candles_df.copy()
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms").dt.strftime('%Y-%m-%d %H:%M:%S')
# 选择要显示的列
display_columns = ["timestamp", "open", "high", "low", "close", "volume"]
# 显示最近 5 根 K 线
lines.append("BTC-USDT 1小时K线 (最近5根):")
lines.append(df[display_columns].tail().to_string(index=False))
return "\n".join(lines)
pd.to_datetime(df["timestamp"], unit="ms") 把毫秒时间戳转成可读格式,dt.strftime 再格式化成字符串。tail() 取最后几行,默认是 5 行。
带指标的进阶展示
把指标值也一起展示出来:
def format_status(self) -> str:
if not self.ready_to_trade:
return "市场连接器未就绪"
lines = ["\n" + "="*50 + " 市场分析 " + "="*50 + "\n"]
if self.candles_1h.is_ready:
df = self.candles_1h.candles_df.copy()
# 计算指标
df.ta.rsi(length=14, append=True)
df.ta.bbands(length=20, std=2.0, append=True)
# 格式化时间
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms").dt.strftime('%m-%d %H:%M')
# 显示关键信息
latest = df.iloc[-1]
lines.append(f"当前价格: {latest['close']:.2f}")
lines.append(f"RSI(14): {latest['RSI_14']:.2f}")
lines.append(f"布林带: 下轨={latest['BBL_20_2.0']:.2f}, 中轨={latest['BBM_20_2.0']:.2f}, 上轨={latest['BBU_20_2.0']:.2f}")
lines.append(f"带宽: {((latest['BBU_20_2.0'] - latest['BBL_20_2.0']) / latest['BBM_20_2.0'] * 100):.2f}%")
lines.append("\n最近3根K线:")
display_cols = ["timestamp", "open", "high", "low", "close", "RSI_14"]
lines.append(df[display_cols].tail(3).to_string(index=False))
return "\n".join(lines)
这种展示方式把关键指标提炼出来,一眼就能看出当前市场状态。带宽(Bollinger Band Width)的计算能反映波动率大小,值越大说明波动越剧烈。
多框架综合展示
同时展示多个时间框架的数据:
def format_status(self) -> str:
if not self.ready_to_trade:
return "市场连接器未就绪"
lines = ["\n" + "="*60 + " 多时间框架分析 " + "="*60 + "\n"]
# 15分钟框架
if self.candles_15m.is_ready:
df_15m = self.candles_15m.candles_df.copy()
df_15m.ta.ema(length=20, append=True, suffix="15m")
latest_15m = df_15m.iloc[-1]
lines.append(f"【15分钟】 EMA20: {latest_15m['EMA20_15m']:.2f}, 收盘价: {latest_15m['close']:.2f}")
# 1小时框架
if self.candles_1h.is_ready:
df_1h = self.candles_1h.candles_df.copy()
df_1h.ta.ema(length=20, append=True, suffix="1h")
latest_1h = df_1h.iloc[-1]
lines.append(f"【1小时】 EMA20: {latest_1h['EMA20_1h']:.2f}, 收盘价: {latest_1h['close']:.2f}")
# 1天框架
if self.candles_1d.is_ready:
df_1d = self.candles_1d.candles_df.copy()
df_1d.ta.ema(length=20, append=True, suffix="1d")
latest_1d = df_1d.iloc[-1]
lines.append(f"【1天】 EMA20: {latest_1d['EMA20_1d']:.2f}, 收盘价: {latest_1d['close']:.2f}")
# 判断趋势一致性
if all([self.candles_15m.is_ready, self.candles_1h.is_ready, self.candles_1d.is_ready]):
price_15m = latest_15m['close']
price_1h = latest_1h['close']
price_1d = latest_1d['close']
ema_15m = latest_15m['EMA20_15m']
ema_1h = latest_1h['EMA20_1h']
ema_1d = latest_1d['EMA20_1d']
# 简单趋势判断:价格在EMA上方为多头趋势
trend_15m = "多头" if price_15m > ema_15m else "空头"
trend_1h = "多头" if price_1h > ema_1h else "空头"
trend_1d = "多头" if price_1d > ema_1d else "空头"
lines.append(f"\n趋势一致性: {trend_15m} | {trend_1h} | {trend_1d}")
# 如果三个周期都是多头,信号强度更高
if trend_15m == trend_1h == trend_1d == "多头":
lines.append("信号强度: 🔥🔥🔥 (强多头)")
elif trend_15m == trend_1h == "多头":
lines.append("信号强度: 🔥🔥 (中等多头)")
else:
lines.append("信号强度: 🔥 (弱多头或震荡)")
return "\n".join(lines)
这种多框架展示能直观看出不同周期的趋势是否一致。当短、中、长期趋势同向时,交易信号的可靠性会大幅提升。
K 线日志记录与分析
status 命令展示的是实时状态,有时候需要把历史数据记录下来,事后分析策略表现。Hummingbot 的日志系统支持把 DataFrame 直接输出到日志文件。
周期性记录 K 线
def on_tick(self):
# 每 300 秒记录一次完整 K 线数据
if self.current_timestamp - self.last_log_time > 300:
self.last_log_time = self.current_timestamp
if self.candles_1h.is_ready:
df = self.candles_1h.candles_df.copy()
# 添加指标
df.ta.rsi(length=14, append=True)
df.ta.bbands(length=20, std=2.0, append=True)
# 记录到日志
self.logger().info(f"\n{df.tail(10).to_string()}")
日志文件默认在 logs/ 目录下,按天分割。记录下来的 DataFrame 可以用 Python 的 pandas.read_csv 读取,做离线分析。
记录关键信号
比起记录完整 K 线,更有效的方法是只记录关键信号:
def on_tick(self):
if not self.candles_1h.is_ready:
return
df = self.candles_1h.candles_df.copy()
df.ta.rsi(length=14, append=True)
latest_rsi = df["RSI_14"].iloc[-1]
latest_price = df["close"].iloc[-1]
# 超买超卖信号
if latest_rsi > 70:
self.logger().info(f"超买信号 - 价格: {latest_price:.2f}, RSI: {latest_rsi:.2f}")
elif latest_rsi < 30:
self.logger().info(f"超卖信号 - 价格: {latest_price:.2f}, RSI: {latest_rsi:.2f}")
# 背离检测(简化版)
if len(df) > 14:
price_trend = df["close"].iloc[-1] > df["close"].iloc[-5]
rsi_trend = df["RSI_14"].iloc[-1] > df["RSI_14"].iloc[-5]
if price_trend and not rsi_trend:
self.logger().warning(f"潜在顶背离 - 价格创新高,RSI 未创新高")
elif not price_trend and rsi_trend:
self.logger().warning(f"潜在底背离 - 价格创新低,RSI 未创新低")
这种日志很精简,但能捕捉到策略决策的关键依据。事后复盘时,能快速定位问题。
日志分析脚本
记录下来的日志可以用简单脚本分析:
import pandas as pd
import re
def parse_strategy_log(log_file):
"""解析策略日志,提取信号"""
signals = []
with open(log_file, 'r') as f:
for line in f:
# 匹配超买超卖信号
if "超买信号" in line or "超卖信号" in line:
# 用正则提取时间和数值
match = re.search(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*价格: ([\d.]+), RSI: ([\d.]+)', line)
if match:
signals.append({
'timestamp': match.group(1),
'price': float(match.group(2)),
'rsi': float(match.group(3)),
'signal': 'overbought' if '超买' in line else 'oversold'
})
return pd.DataFrame(signals)
# 使用示例
df_signals = parse_strategy_log('logs/strategy.log')
print(df_signals.head())
这个脚本把非结构化的日志转成 DataFrame,方便做统计分析,比如计算信号准确率、平均盈亏等。
自定义指标扩展方法
pandas_ta 虽然强大,但总有覆盖不到的场景。比如一个基于订单簿深度的自定义波动率指标,或者结合资金费率的特殊均线。这时候就需要扩展自定义指标。
使用 pandas_ta 的自定义函数接口
pandas_ta 支持把自定义函数注册成指标:
import pandas_ta as ta
from pandas import DataFrame
def custom_pressure_indicator(close, volume, length=20, offset=0, **kwargs):
"""
自定义压力指标:结合价格和成交量的简单指标
计算方式:(收盘价 - 最低价) / (最高价 - 最低价) * 成交量
"""
# 计算价格位置
low = close.rolling(window=length).min()
high = close.rolling(window=length).max()
price_position = (close - low) / (high - low)
# 结合成交量
pressure = price_position * volume
# 平滑处理
pressure_smooth = pressure.rolling(window=length).mean()
# 处理偏移
if offset != 0:
pressure_smooth = pressure_smooth.shift(offset)
return pressure_smooth
# 注册自定义指标
ta.CustomIndicator = custom_pressure_indicator
# 在策略中使用
def add_custom_indicators(self, df):
"""添加自定义指标"""
# 调用自定义指标
df.ta.custom_pressure_indicator(close=df["close"],
volume=df["volume"],
length=20,
append=True,
suffix="CP")
# 自定义指标列名会是 CustomIndicator_CP_20
return df
这种方式的好处是复用了 pandas_ta 的参数处理机制,append、suffix 这些参数都能用。
直接操作 DataFrame
更直接的方式是不依赖 pandas_ta,直接在 DataFrame 上计算:
def calculate_custom_metrics(self, df):
"""计算完全自定义的指标"""
# 复制 DataFrame 避免修改原始数据
result_df = df.copy()
# 计算 VWAP (Volume Weighted Average Price)
result_df["vwap"] = (result_df["close"] * result_df["volume"]).cumsum() / result_df["volume"].cumsum()
# 计算价格偏离 VWAP 的程度
result_df["vwap_deviation"] = (result_df["close"] - result_df["vwap"]) / result_df["vwap"] * 100
# 计算 K 线实体大小
result_df["body_size"] = abs(result_df["close"] - result_df["open"])
# 计算上影线长度
result_df["upper_wick"] = result_df["high"] - result_df[["open", "close"]].max(axis=1)
# 计算下影线长度
result_df["lower_wick"] = result_df[["open", "close"]].min(axis=1) - result_df["low"]
# 计算影线比例
result_df["wick_ratio"] = result_df["upper_wick"] / (result_df["lower_wick"] + 1e-8)
# 识别锤子线 (简化版)
result_df["is_hammer"] = (
(result_df["lower_wick"] > result_df["body_size"] * 2) &
(result_df["upper_wick"] < result_df["body_size"] * 0.5)
)
return result_df
这种方法更灵活,可以实现任意复杂的逻辑。比如上面的 is_hammer 识别锤子线,这是 pandas_ta 里没有的。
性能优化技巧
自定义指标计算可能会比较耗时,特别是用到滚动窗口的时候。几个优化技巧:
def calculate_efficient_indicators(self, df):
"""高效计算指标"""
# 1. 避免链式赋值,用 loc
df.loc[:, "price_change"] = df["close"].diff()
# 2. 滚动窗口计算一次,多次使用
rolling_20 = df["close"].rolling(window=20)
df.loc[:, "sma_20"] = rolling_20.mean()
df.loc[:, "std_20"] = rolling_20.std()
# 3. 向量化运算,避免循环
df.loc[:, "z_score"] = (df["close"] - df["sma_20"]) / df["std_20"]
# 4. 用 numpy 加速复杂计算
import numpy as np
def custom_complex_calculation(close, volume):
# 模拟复杂计算
log_returns = np.log(close / close.shift(1))
vol_adj_returns = log_returns * np.sqrt(volume / volume.mean())
return vol_adj_returns.rolling(20).sum()
df.loc[:, "custom_signal"] = custom_complex_calculation(df["close"], df["volume"])
return df
向量化运算是 pandas 的性能关键,能用矩阵运算就别用循环。滚动窗口计算结果可以赋值给变量,避免重复计算。
指标持久化
有些指标计算代价很高,可以缓存起来,下次直接用:
import hashlib
import pickle
import os
class CachedIndicatorStrategy(ScriptStrategyBase):
def __init__(self, connectors):
super().__init__(connectors)
self.indicator_cache = {}
self.cache_dir = "indicator_cache"
os.makedirs(self.cache_dir, exist_ok=True)
def get_cached_indicator(self, df, indicator_name, calc_func, **params):
"""带缓存的指标计算"""
# 用 DataFrame 的 hash 作为缓存键
df_hash = hashlib.md5(df.to_json().encode()).hexdigest()[:8]
param_str = "_".join([f"{k}_{v}" for k, v in params.items()])
cache_key = f"{indicator_name}_{df_hash}_{param_str}"
cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl")
# 检查缓存
if os.path.exists(cache_file):
with open(cache_file, 'rb') as f:
return pickle.load(f)
# 计算并缓存
result = calc_func(df, **params)
with open(cache_file, 'wb') as f:
pickle.dump(result, f)
return result
这种缓存机制在回测时特别有用,同样的数据不需要重复计算。实盘时要注意缓存文件的大小,定期清理。
总结
K 线数据和技术指标是量化策略的基石。Hummingbot 的 V2 框架把这块做得足够简单,又保留了足够的灵活性。从数据源的初始化、多框架同步,到指标计算、状态展示,再到日志记录和自定义扩展,整个流程一气呵成。
核心要点回顾:用 CandlesFactory 初始化数据源,用 is_ready 检查数据完整性,用 pandas_ta 快速添加指标,用 format_status 实时展示,用 logger 记录关键信号,用自定义函数扩展特殊指标。
掌握了这些,就能在策略里自如地运用各种技术分析工具。下一章会深入回测系统,看看如何用历史数据验证这些基于 K 线的策略是否真正有效。