技术指标

技术指标是量化交易策略的核心组成部分，宽图量化平台提供了丰富的技术指标库，帮助您构建强大的交易策略。

指标分类

我们的技术指标按功能分为以下几类：

• 趋势指标: 识别市场趋势方向
• 震荡指标: 判断超买超卖状态
• 成交量指标: 分析成交量变化
• 波动率指标: 衡量价格波动程度
• 支撑阻力指标: 识别关键价位

趋势指标

1. 移动平均线 (Moving Average)

移动平均线是最基础也是最重要的趋势指标。

简单移动平均线 (SMA)

# SMA计算公式
def sma(prices, period):
    return sum(prices[-period:]) / period

# 使用示例
sma_20 = self.add_indicator('SMA', period=20)
sma_50 = self.add_indicator('SMA', period=50)

# 金叉死叉信号
if sma_20.value > sma_50.value and sma_20.prev_value <= sma_50.prev_value:
    # 金叉买入信号
    self.buy_signal()
elif sma_20.value < sma_50.value and sma_20.prev_value >= sma_50.prev_value:
    # 死叉卖出信号
    self.sell_signal()

指数移动平均线 (EMA)

# EMA计算公式
def ema(prices, period, prev_ema=None):
    multiplier = 2 / (period + 1)
    if prev_ema is None:
        return prices[0]
    return (prices[-1] * multiplier) + (prev_ema * (1 - multiplier))

# 使用示例
ema_12 = self.add_indicator('EMA', period=12)
ema_26 = self.add_indicator('EMA', period=26)

参数建议

2. MACD指标

MACD是最受欢迎的趋势跟踪指标之一。

# MACD指标使用
macd = self.add_indicator('MACD', fast=12, slow=26, signal=9)

def analyze_macd(self, bar):
    macd_line = macd.macd
    signal_line = macd.signal
    histogram = macd.histogram
    
    # MACD金叉
    if (macd_line > signal_line and 
        macd.prev_macd <= macd.prev_signal):
        if macd_line > 0:  # 零轴上方金叉更可靠
            return 'STRONG_BUY'
        else:
            return 'BUY'
    
    # MACD死叉
    elif (macd_line < signal_line and 
          macd.prev_macd >= macd.prev_signal):
        if macd_line < 0:  # 零轴下方死叉更可靠
            return 'STRONG_SELL'
        else:
            return 'SELL'
    
    # 背离分析
    if self.detect_bullish_divergence(macd_line):
        return 'DIVERGENCE_BUY'
    elif self.detect_bearish_divergence(macd_line):
        return 'DIVERGENCE_SELL'
    
    return 'HOLD'

3. 布林带 (Bollinger Bands)

布林带结合了趋势和波动率信息。

# 布林带指标
bb = self.add_indicator('BollingerBands', period=20, std=2)

def analyze_bollinger_bands(self, bar):
    upper = bb.upper
    middle = bb.middle  # 20日SMA
    lower = bb.lower
    current_price = bar.close
    
    # 价格位置
    bb_position = (current_price - lower) / (upper - lower)
    
    # 布林带宽度（波动率指标）
    bb_width = (upper - lower) / middle
    
    # 交易信号
    if current_price <= lower and bb_width > 0.1:
        # 价格触及下轨且波动率足够
        return 'BUY'
    elif current_price >= upper and bb_width > 0.1:
        # 价格触及上轨且波动率足够
        return 'SELL'
    elif bb_position > 0.8:
        # 价格在上轨附近
        return 'OVERBOUGHT'
    elif bb_position < 0.2:
        # 价格在下轨附近
        return 'OVERSOLD'
    
    return 'NEUTRAL'

震荡指标

1. RSI指标

RSI是最常用的超买超卖指标。

# RSI指标使用
rsi = self.add_indicator('RSI', period=14)

def analyze_rsi(self, bar):
    rsi_value = rsi.value
    rsi_prev = rsi.prev_value
    
    # 基本超买超卖信号
    if rsi_value < 30:
        return 'OVERSOLD'
    elif rsi_value > 70:
        return 'OVERBOUGHT'
    
    # RSI背离
    if self.detect_rsi_bullish_divergence():
        return 'BULLISH_DIVERGENCE'
    elif self.detect_rsi_bearish_divergence():
        return 'BEARISH_DIVERGENCE'
    
    # RSI突破信号
    if rsi_value > 50 and rsi_prev <= 50:
        return 'RSI_BULLISH_BREAKOUT'
    elif rsi_value < 50 and rsi_prev >= 50:
        return 'RSI_BEARISH_BREAKOUT'
    
    return 'NEUTRAL'

def detect_rsi_bullish_divergence(self):
    # 价格创新低，RSI未创新低
    recent_prices = self.get_recent_lows(5)
    recent_rsi = self.get_recent_rsi_lows(5)
    
    if (recent_prices[-1] < recent_prices[-2] and 
        recent_rsi[-1] > recent_rsi[-2]):
        return True
    return False

2. 随机指标 (Stochastic)

随机指标衡量收盘价在一定周期内的相对位置。

# 随机指标
stoch = self.add_indicator('Stochastic', k_period=14, d_period=3)

def analyze_stochastic(self, bar):
    k_value = stoch.k
    d_value = stoch.d
    
    # 超买超卖区域
    if k_value < 20 and d_value < 20:
        if k_value > d_value:  # K线上穿D线
            return 'OVERSOLD_BULLISH'
    elif k_value > 80 and d_value > 80:
        if k_value < d_value:  # K线下穿D线
            return 'OVERBOUGHT_BEARISH'
    
    # 中线突破
    if k_value > 50 and d_value > 50:
        return 'BULLISH'
    elif k_value < 50 and d_value < 50:
        return 'BEARISH'
    
    return 'NEUTRAL'

3. 威廉指标 (%R)

威廉指标反映市场超买超卖程度。

# 威廉指标
williams_r = self.add_indicator('WilliamsR', period=14)

def analyze_williams_r(self, bar):
    wr_value = williams_r.value
    
    # 威廉指标的值在-100到0之间
    if wr_value < -80:
        return 'OVERSOLD'
    elif wr_value > -20:
        return 'OVERBOUGHT'
    elif wr_value > -50 and williams_r.prev_value <= -50:
        return 'BULLISH_SIGNAL'
    elif wr_value < -50 and williams_r.prev_value >= -50:
        return 'BEARISH_SIGNAL'
    
    return 'NEUTRAL'

成交量指标

1. 成交量移动平均线

# 成交量指标
volume_sma = self.add_indicator('VolumeSMA', period=20)
volume_ratio = self.add_indicator('VolumeRatio', period=10)

def analyze_volume(self, bar):
    current_volume = bar.volume
    avg_volume = volume_sma.value
    vol_ratio = volume_ratio.value
    
    # 放量突破
    if current_volume > avg_volume * 2:
        if bar.close > bar.open:
            return 'VOLUME_BREAKOUT_BULLISH'
        else:
            return 'VOLUME_BREAKOUT_BEARISH'
    
    # 缩量整理
    elif current_volume < avg_volume * 0.5:
        return 'LOW_VOLUME_CONSOLIDATION'
    
    # 价量配合
    price_change = (bar.close - bar.open) / bar.open
    if price_change > 0.02 and vol_ratio > 1.5:
        return 'PRICE_VOLUME_CONFIRMATION'
    
    return 'NORMAL_VOLUME'

2. OBV指标 (On Balance Volume)

# OBV指标
obv = self.add_indicator('OBV')

def analyze_obv(self, bar):
    obv_value = obv.value
    obv_sma = self.add_indicator('SMA', data=obv.values, period=10)
    
    # OBV趋势
    if obv_value > obv_sma.value:
        return 'OBV_BULLISH_TREND'
    elif obv_value < obv_sma.value:
        return 'OBV_BEARISH_TREND'
    
    # OBV背离
    if self.detect_obv_divergence():
        return 'OBV_DIVERGENCE'
    
    return 'OBV_NEUTRAL'

波动率指标

1. ATR指标 (Average True Range)

ATR衡量价格波动的幅度。

# ATR指标
atr = self.add_indicator('ATR', period=14)

def calculate_position_size_with_atr(self, account_balance, risk_per_trade=0.02):
    atr_value = atr.value
    current_price = self.get_current_price()
    
    # 基于ATR的止损距离
    stop_distance = atr_value * 2
    
    # 计算仓位大小
    risk_amount = account_balance * risk_per_trade
    position_size = risk_amount / stop_distance
    
    return min(position_size, account_balance * 0.1 / current_price)

def dynamic_stop_loss(self, entry_price, is_long=True):
    atr_value = atr.value
    
    if is_long:
        stop_loss = entry_price - (atr_value * 2)
    else:
        stop_loss = entry_price + (atr_value * 2)
    
    return stop_loss

2. 历史波动率

# 历史波动率计算
def calculate_historical_volatility(self, period=20):
    prices = self.get_recent_closes(period + 1)
    returns = []
    
    for i in range(1, len(prices)):
        daily_return = (prices[i] - prices[i-1]) / prices[i-1]
        returns.append(daily_return)
    
    # 计算标准差
    mean_return = sum(returns) / len(returns)
    variance = sum((r - mean_return) ** 2 for r in returns) / len(returns)
    volatility = (variance ** 0.5) * (365 ** 0.5)  # 年化波动率
    
    return volatility

def adjust_strategy_by_volatility(self, volatility):
    if volatility > 0.6:  # 高波动率
        self.reduce_position_size(0.7)
        self.tighten_stop_loss(0.8)
    elif volatility < 0.2:  # 低波动率
        self.increase_position_size(1.2)
        self.widen_stop_loss(1.2)

自定义指标

1. 创建自定义指标

from kuantu_sdk import Indicator

class CustomTrendStrength(Indicator):
    def __init__(self, period=20):
        super().__init__()
        self.period = period
        self.prices = []
        self.trend_strength = 0
    
    def update(self, price):
        self.prices.append(price)
        if len(self.prices) > self.period:
            self.prices.pop(0)
        
        if len(self.prices) == self.period:
            self.trend_strength = self.calculate_trend_strength()
            self.ready = True
    
    def calculate_trend_strength(self):
        # 计算趋势强度
        up_moves = 0
        down_moves = 0
        
        for i in range(1, len(self.prices)):
            if self.prices[i] > self.prices[i-1]:
                up_moves += 1
            elif self.prices[i] < self.prices[i-1]:
                down_moves += 1
        
        total_moves = up_moves + down_moves
        if total_moves == 0:
            return 0
        
        # 趋势强度 = (上涨次数 - 下跌次数) / 总变动次数
        return (up_moves - down_moves) / total_moves
    
    @property
    def value(self):
        return self.trend_strength

# 在策略中使用自定义指标
def setup_indicators(self):
    self.trend_strength = CustomTrendStrength(period=20)
    self.add_custom_indicator('trend_strength', self.trend_strength)

2. 组合指标

class CompositeSignal:
    def __init__(self, strategy):
        self.strategy = strategy
        self.rsi = strategy.add_indicator('RSI', period=14)
        self.macd = strategy.add_indicator('MACD', fast=12, slow=26, signal=9)
        self.bb = strategy.add_indicator('BollingerBands', period=20, std=2)
    
    def get_signal_strength(self, bar):
        signals = []
        
        # RSI信号
        if self.rsi.value < 30:
            signals.append(1)  # 买入信号
        elif self.rsi.value > 70:
            signals.append(-1)  # 卖出信号
        else:
            signals.append(0)
        
        # MACD信号
        if self.macd.macd > self.macd.signal and self.macd.macd > 0:
            signals.append(1)
        elif self.macd.macd < self.macd.signal and self.macd.macd < 0:
            signals.append(-1)
        else:
            signals.append(0)
        
        # 布林带信号
        bb_position = (bar.close - self.bb.lower) / (self.bb.upper - self.bb.lower)
        if bb_position < 0.2:
            signals.append(1)
        elif bb_position > 0.8:
            signals.append(-1)
        else:
            signals.append(0)
        
        # 计算综合信号强度
        signal_strength = sum(signals) / len(signals)
        return signal_strength

指标优化技巧

1. 参数优化

def optimize_indicator_parameters():
    # 定义参数范围
    rsi_periods = range(10, 21, 2)  # 10, 12, 14, 16, 18, 20
    ma_periods = range(15, 31, 5)   # 15, 20, 25, 30
    
    best_params = None
    best_performance = -float('inf')
    
    for rsi_period in rsi_periods:
        for ma_period in ma_periods:
            # 运行回测
            result = backtest_with_params(rsi_period, ma_period)
            
            # 评估性能（使用夏普比率）
            performance = result['sharpe_ratio']
            
            if performance > best_performance:
                best_performance = performance
                best_params = {
                    'rsi_period': rsi_period,
                    'ma_period': ma_period
                }
    
    return best_params

2. 指标过滤

def filter_signals(self, primary_signal):
    # 使用多个指标确认信号
    confirmations = 0
    
    # 趋势确认
    if self.sma_20.value > self.sma_50.value:
        if primary_signal == 'BUY':
            confirmations += 1
    elif self.sma_20.value < self.sma_50.value:
        if primary_signal == 'SELL':
            confirmations += 1
    
    # 成交量确认
    if self.current_volume > self.volume_sma.value * 1.2:
        confirmations += 1
    
    # 波动率确认
    if self.atr.value > self.atr_sma.value:
        confirmations += 1
    
    # 需要至少2个确认信号
    return confirmations >= 2

3. 动态参数调整

def adjust_parameters_by_market_condition(self):
    # 计算市场波动率
    volatility = self.calculate_market_volatility()
    
    # 根据波动率调整RSI参数
    if volatility > 0.5:  # 高波动率
        self.rsi.period = 10  # 缩短周期，更敏感
        self.rsi.overbought = 75  # 提高阈值
        self.rsi.oversold = 25
    elif volatility < 0.2:  # 低波动率
        self.rsi.period = 20  # 延长周期，更平滑
        self.rsi.overbought = 65  # 降低阈值
        self.rsi.oversold = 35
    else:  # 正常波动率
        self.rsi.period = 14
        self.rsi.overbought = 70
        self.rsi.oversold = 30

指标使用最佳实践

1. 指标组合原则

• 不要使用相似的指标: 避免信号重复
• 趋势+震荡组合: 趋势指标确定方向，震荡指标确定时机
• 价格+成交量组合: 价格指标配合成交量确认
• 多时间周期: 长周期确定趋势，短周期确定入场点

2. 常见错误

3. 性能监控

def monitor_indicator_performance(self):
    # 统计各指标的准确率
    indicator_stats = {
        'rsi': {'correct': 0, 'total': 0},
        'macd': {'correct': 0, 'total': 0},
        'bb': {'correct': 0, 'total': 0}
    }
    
    # 定期评估指标表现
    for trade in self.recent_trades:
        for indicator_name in indicator_stats:
            if trade.entry_reason.startswith(indicator_name):
                indicator_stats[indicator_name]['total'] += 1
                if trade.profit > 0:
                    indicator_stats[indicator_name]['correct'] += 1
    
    # 计算准确率
    for indicator_name, stats in indicator_stats.items():
        if stats['total'] > 0:
            accuracy = stats['correct'] / stats['total']
            self.log(f"{indicator_name} 准确率: {accuracy:.2%}")

下一步

掌握了技术指标的使用后，建议您：

1. 学习回测系统 - 验证指标效果
2. 了解风险管理 - 结合指标进行风险控制
3. 查看策略框架 - 将指标整合到完整策略中
4. 参与社区讨论 - 与其他用户交流指标使用经验

重要提示: 技术指标是工具，不是万能的。任何指标都有其局限性，关键是要理解每个指标的特点，合理组合使用，并结合良好的风险管理。