11. K线数据与技术指标集成

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 线的策略是否真正有效。