DEV Community

Henry Lin
Henry Lin

Posted on

1.3 FreqTrade技术分析基础

1.3 技术分析基础

学习目标

  • 掌握技术分析的基本原理和假设
  • 理解趋势、支撑阻力等核心概念
  • 熟悉常用技术指标的计算和应用
  • 学会多时间框架分析方法

1.3.1 技术分析基本原理

技术分析的三大假设

1. 市场行为包含一切信息

def market_efficiency_principle():
    """
    市场行为包含一切信息的含义
    """
    principle = {
        '基本面因素': '财务数据、经济指标等已反映在价格中',
        '技术面因素': '供需关系、市场情绪等已反映在价格中', 
        '政策因素': '监管政策、政府态度等已反映在价格中',
        '心理因素': '市场参与者的情绪和预期已反映在价格中'
    }

    implication = """
    因此:只需要分析价格和成交量数据,
    就能做出有效的投资决策
    """

    return principle, implication
Enter fullscreen mode Exit fullscreen mode

2. 价格沿趋势运动

  • 趋势惯性: 一旦形成趋势,往往会持续一段时间
  • 趋势转换: 趋势改变需要明确的信号确认
  • 多级趋势: 同时存在主要趋势、次要趋势和短期波动

3. 历史会重演

def historical_pattern_repetition():
    """
    历史重演的逻辑基础
    """
    logic = {
        '人性不变': '贪婪、恐惧等情绪是人类共性',
        '集体行为': '群体心理具有相似的表现模式',
        '市场结构': '供需关系的基本模式相对稳定',
        '技术模式': '相似的市场条件产生相似的价格模式'
    }

    application = """
    应用:通过识别历史模式,
    预测未来可能的价格运动
    """

    return logic, application
Enter fullscreen mode Exit fullscreen mode

1.3.2 趋势分析

趋势的定义和分类

1. 按方向分类

import pandas as pd
import numpy as np

def identify_trend_direction(prices, window=20):
    """
    识别趋势方向
    """
    # 计算移动平均线
    ma = prices.rolling(window).mean()

    # 趋势判断逻辑
    current_price = prices.iloc[-1]
    ma_current = ma.iloc[-1]
    ma_previous = ma.iloc[-window]

    if current_price > ma_current and ma_current > ma_previous:
        return "上升趋势 (Uptrend)"
    elif current_price < ma_current and ma_current < ma_previous:
        return "下降趋势 (Downtrend)" 
    else:
        return "横盘整理 (Sideways)"

def trend_characteristics():
    """
    不同趋势的特征
    """
    trends = {
        '上升趋势': {
            'price_action': '高点和低点逐步抬高',
            'volume': '上涨时放量,下跌时缩量',
            'ma_alignment': '短期均线在长期均线上方',
            'psychology': '乐观情绪主导'
        },
        '下降趋势': {
            'price_action': '高点和低点逐步降低',
            'volume': '下跌时放量,反弹时缩量', 
            'ma_alignment': '短期均线在长期均线下方',
            'psychology': '悲观情绪主导'
        },
        '横盘趋势': {
            'price_action': '在一定区间内震荡',
            'volume': '成交量相对较低',
            'ma_alignment': '均线纠缠,方向不明',
            'psychology': '犹豫观望情绪'
        }
    }
    return trends
Enter fullscreen mode Exit fullscreen mode

2. 按时间周期分类

def trend_time_classification():
    """
    按时间周期划分趋势
    """
    trend_periods = {
        '主要趋势 (Primary Trend)': {
            'duration': '1年以上',
            'description': '长期牛市或熊市',
            'importance': '最重要,决定投资方向'
        },
        '次要趋势 (Secondary Trend)': {
            'duration': '3周-3个月',
            'description': '主趋势中的回调或反弹',
            'importance': '重要,提供交易机会'
        },
        '短期波动 (Minor Trend)': {
            'duration': '数天到3周',
            'description': '日常价格波动',
            'importance': '对短线交易有意义'
        }
    }
    return trend_periods
Enter fullscreen mode Exit fullscreen mode

趋势线绘制

1. 上升趋势线

def draw_uptrend_line(prices, dates):
    """
    绘制上升趋势线
    """
    # 寻找连续低点
    lows = []
    for i in range(1, len(prices)-1):
        if prices[i] < prices[i-1] and prices[i] < prices[i+1]:
            lows.append((dates[i], prices[i]))

    # 连接两个或多个低点形成上升趋势线
    if len(lows) >= 2:
        trend_line = {
            'points': lows,
            'slope': (lows[-1][1] - lows[0][1]) / (lows[-1][0] - lows[0][0]),
            'support_level': '趋势线作为动态支撑',
            'validity': '触及次数越多越有效'
        }
        return trend_line
Enter fullscreen mode Exit fullscreen mode

2. 下降趋势线

def draw_downtrend_line(prices, dates):
    """
    绘制下降趋势线
    """
    # 寻找连续高点
    highs = []
    for i in range(1, len(prices)-1):
        if prices[i] > prices[i-1] and prices[i] > prices[i+1]:
            highs.append((dates[i], prices[i]))

    # 连接两个或多个高点形成下降趋势线
    if len(highs) >= 2:
        trend_line = {
            'points': highs,
            'slope': (highs[-1][1] - highs[0][1]) / (highs[-1][0] - highs[0][0]),
            'resistance_level': '趋势线作为动态阻力',
            'break_signal': '有效突破预示趋势反转'
        }
        return trend_line
Enter fullscreen mode Exit fullscreen mode

1.3.3 支撑与阻力

支撑阻力的形成机制

def support_resistance_psychology():
    """
    支撑阻力的心理学基础
    """
    psychology = {
        '支撑位心理': {
            '买方集中': '历史低点附近买方密集',
            '止损聚集': '空头止损单集中在支撑位',
            '价值认同': '市场认为该价位具有投资价值',
            '技术信仰': '技术分析者在此设置买点'
        },
        '阻力位心理': {
            '卖方集中': '历史高点附近卖方密集',
            '获利了结': '多头在此价位获利出局',
            '套牢盘': '早期套牢者等待解套',
            '心理关口': '整数价位具有心理意义'
        }
    }
    return psychology
Enter fullscreen mode Exit fullscreen mode

支撑阻力的类型

1. 水平支撑阻力

def horizontal_support_resistance(prices, window=20, min_touches=2):
    """
    识别水平支撑阻力位
    """
    # 找出局部高点和低点
    highs = []
    lows = []

    for i in range(window, len(prices) - window):
        # 局部高点
        if all(prices[i] >= prices[i-j] for j in range(1, window+1)) and \
           all(prices[i] >= prices[i+j] for j in range(1, window+1)):
            highs.append(prices[i])

        # 局部低点  
        if all(prices[i] <= prices[i-j] for j in range(1, window+1)) and \
           all(prices[i] <= prices[i+j] for j in range(1, window+1)):
            lows.append(prices[i])

    # 聚类相近的价位
    def cluster_prices(price_list, tolerance=0.02):
        clusters = []
        for price in sorted(price_list):
            placed = False
            for cluster in clusters:
                if abs(price - cluster['center']) / cluster['center'] <= tolerance:
                    cluster['prices'].append(price)
                    cluster['center'] = np.mean(cluster['prices'])
                    cluster['touches'] += 1
                    placed = True
                    break
            if not placed:
                clusters.append({
                    'center': price,
                    'prices': [price], 
                    'touches': 1
                })

        # 过滤掉触及次数少的价位
        significant_levels = [cluster for cluster in clusters 
                            if cluster['touches'] >= min_touches]
        return significant_levels

    resistance_levels = cluster_prices(highs)
    support_levels = cluster_prices(lows)

    return {
        'resistance': resistance_levels,
        'support': support_levels
    }
Enter fullscreen mode Exit fullscreen mode

2. 动态支撑阻力

def dynamic_support_resistance(prices, ma_periods=[20, 50, 200]):
    """
    基于移动平均线的动态支撑阻力
    """
    dynamic_levels = {}

    for period in ma_periods:
        ma = prices.rolling(period).mean()

        # 判断MA的支撑阻力作用
        current_price = prices.iloc[-1]
        ma_current = ma.iloc[-1]

        if current_price > ma_current:
            role = f"MA{period}作为动态支撑"
        else:
            role = f"MA{period}作为动态阻力"

        dynamic_levels[f'MA{period}'] = {
            'value': ma_current,
            'role': role,
            'strength': '周期越长,支撑阻力越强'
        }

    return dynamic_levels
Enter fullscreen mode Exit fullscreen mode

支撑阻力的强度判断

def support_resistance_strength(level_data):
    """
    判断支撑阻力的强度
    """
    strength_factors = {
        '触及次数': {
            'weight': 0.3,
            'calculation': 'touches / max_touches',
            'interpretation': '触及次数越多越强'
        },
        '成交量': {
            'weight': 0.25, 
            'calculation': 'avg_volume_at_level / avg_volume',
            'interpretation': '该价位成交量越大越强'
        },
        '时间跨度': {
            'weight': 0.2,
            'calculation': 'time_span / total_period',
            'interpretation': '形成时间越长越强'
        },
        '距离现价': {
            'weight': 0.15,
            'calculation': '1 / abs(level - current_price)',
            'interpretation': '距离现价越近影响越大'
        },
        '突破历史': {
            'weight': 0.1,
            'calculation': 'false_breakouts / total_tests',
            'interpretation': '假突破越多越强'
        }
    }

    # 计算综合强度分数
    total_score = 0
    for factor, config in strength_factors.items():
        factor_score = config['calculation']  # 这里应该是实际计算
        weighted_score = factor_score * config['weight']
        total_score += weighted_score

    return {
        'strength_score': total_score,
        'level': 'Strong' if total_score > 0.7 else 'Medium' if total_score > 0.4 else 'Weak'
    }
Enter fullscreen mode Exit fullscreen mode

1.3.4 常用技术指标

趋势跟踪指标

1. 移动平均线 (Moving Average)

def moving_averages(prices, periods=[5, 10, 20, 50, 200]):
    """
    计算不同周期的移动平均线
    """
    mas = {}

    for period in periods:
        # 简单移动平均 (SMA)
        sma = prices.rolling(period).mean()

        # 指数移动平均 (EMA)
        ema = prices.ewm(span=period).mean()

        mas[f'SMA{period}'] = sma
        mas[f'EMA{period}'] = ema

    return mas

def ma_trading_signals(prices, short_period=20, long_period=50):
    """
    基于移动平均线的交易信号
    """
    short_ma = prices.rolling(short_period).mean()
    long_ma = prices.rolling(long_period).mean()

    # 金叉死叉信号
    golden_cross = (short_ma > long_ma) & (short_ma.shift(1) <= long_ma.shift(1))
    death_cross = (short_ma < long_ma) & (short_ma.shift(1) >= long_ma.shift(1))

    signals = pd.Series(0, index=prices.index)
    signals[golden_cross] = 1   # 买入信号
    signals[death_cross] = -1   # 卖出信号

    return {
        'signals': signals,
        'short_ma': short_ma,
        'long_ma': long_ma,
        'interpretation': {
            'golden_cross': '短期均线上穿长期均线,看涨信号',
            'death_cross': '短期均线下穿长期均线,看跌信号'
        }
    }
Enter fullscreen mode Exit fullscreen mode

2. MACD指标

def calculate_macd(prices, fast=12, slow=26, signal=9):
    """
    计算MACD指标
    """
    # 计算快速和慢速EMA
    ema_fast = prices.ewm(span=fast).mean()
    ema_slow = prices.ewm(span=slow).mean()

    # MACD线 = 快速EMA - 慢速EMA
    macd_line = ema_fast - ema_slow

    # 信号线 = MACD线的EMA
    signal_line = macd_line.ewm(span=signal).mean()

    # 柱状图 = MACD线 - 信号线
    histogram = macd_line - signal_line

    return {
        'macd': macd_line,
        'signal': signal_line, 
        'histogram': histogram,
        'interpretation': {
            'macd_above_signal': '看涨信号',
            'macd_below_signal': '看跌信号',
            'histogram_increasing': '动能增强',
            'histogram_decreasing': '动能减弱'
        }
    }

def macd_trading_signals(macd_data):
    """
    基于MACD的交易信号
    """
    macd = macd_data['macd']
    signal = macd_data['signal']
    histogram = macd_data['histogram']

    # MACD金叉死叉
    bullish_crossover = (macd > signal) & (macd.shift(1) <= signal.shift(1))
    bearish_crossover = (macd < signal) & (macd.shift(1) >= signal.shift(1))

    # 零轴交叉
    zero_line_cross_up = (macd > 0) & (macd.shift(1) <= 0)
    zero_line_cross_down = (macd < 0) & (macd.shift(1) >= 0)

    # 柱状图信号
    histogram_turning_positive = (histogram > histogram.shift(1)) & (histogram.shift(1) < histogram.shift(2))
    histogram_turning_negative = (histogram < histogram.shift(1)) & (histogram.shift(1) > histogram.shift(2))

    return {
        'bullish_crossover': bullish_crossover,
        'bearish_crossover': bearish_crossover,
        'zero_cross_up': zero_line_cross_up,
        'zero_cross_down': zero_line_cross_down,
        'momentum_up': histogram_turning_positive,
        'momentum_down': histogram_turning_negative
    }
Enter fullscreen mode Exit fullscreen mode

震荡指标

1. RSI相对强弱指标

def calculate_rsi(prices, period=14):
    """
    计算RSI指标
    """
    # 计算价格变化
    delta = prices.diff()

    # 分离上涨和下跌
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)

    # 计算平均涨跌幅
    avg_gain = gain.rolling(period).mean()
    avg_loss = loss.rolling(period).mean()

    # 计算相对强度
    rs = avg_gain / avg_loss

    # 计算RSI
    rsi = 100 - (100 / (1 + rs))

    return {
        'rsi': rsi,
        'interpretation': {
            'overbought': 'RSI > 70,超买区域',
            'oversold': 'RSI < 30,超卖区域',
            'neutral': '30 <= RSI <= 70,中性区域'
        }
    }

def rsi_trading_signals(rsi, overbought=70, oversold=30):
    """
    基于RSI的交易信号
    """
    # 超买超卖信号
    overbought_condition = rsi > overbought
    oversold_condition = rsi < oversold

    # 背离信号需要结合价格分析
    def detect_divergence(prices, rsi):
        # 这里简化处理,实际需要更复杂的算法
        price_highs = prices.rolling(20).max()
        rsi_highs = rsi.rolling(20).max()

        bearish_divergence = (prices == price_highs) & (rsi < rsi_highs)
        bullish_divergence = (prices == prices.rolling(20).min()) & (rsi > rsi.rolling(20).min())

        return bearish_divergence, bullish_divergence

    return {
        'overbought': overbought_condition,
        'oversold': oversold_condition,
        'signal_strength': 'RSI极值信号强度取决于市场环境'
    }
Enter fullscreen mode Exit fullscreen mode

2. 布林带指标

def calculate_bollinger_bands(prices, period=20, std_dev=2):
    """
    计算布林带指标
    """
    # 计算中轨(移动平均线)
    middle_band = prices.rolling(period).mean()

    # 计算标准差
    std = prices.rolling(period).std()

    # 计算上下轨
    upper_band = middle_band + (std * std_dev)
    lower_band = middle_band - (std * std_dev)

    # 计算带宽和位置
    band_width = (upper_band - lower_band) / middle_band
    position = (prices - lower_band) / (upper_band - lower_band)

    return {
        'upper': upper_band,
        'middle': middle_band,
        'lower': lower_band,
        'width': band_width,
        'position': position,
        'interpretation': {
            'squeeze': '带宽收窄预示波动率将增加',
            'expansion': '带宽扩张显示波动率增加',
            'upper_touch': '触及上轨可能回调',
            'lower_touch': '触及下轨可能反弹'
        }
    }

def bollinger_band_signals(bb_data, prices):
    """
    基于布林带的交易信号
    """
    upper = bb_data['upper']
    lower = bb_data['lower']
    middle = bb_data['middle']
    width = bb_data['width']

    # 布林带突破信号
    upper_breakout = prices > upper
    lower_breakout = prices < lower

    # 回归信号
    mean_reversion_up = (prices < lower) & (prices.shift(1) >= lower.shift(1))
    mean_reversion_down = (prices > upper) & (prices.shift(1) <= upper.shift(1))

    # 带宽收缩信号(波动率压缩)
    squeeze = width < width.rolling(20).quantile(0.1)

    return {
        'breakout_up': upper_breakout,
        'breakout_down': lower_breakout,
        'reversion_up': mean_reversion_up,
        'reversion_down': mean_reversion_down,
        'squeeze': squeeze
    }
Enter fullscreen mode Exit fullscreen mode

成交量指标

1. 成交量加权平均价 (VWAP)

def calculate_vwap(prices, volumes, period=20):
    """
    计算成交量加权平均价
    """
    # 计算典型价格
    typical_price = (prices['high'] + prices['low'] + prices['close']) / 3

    # 计算成交量加权价格
    vwap = (typical_price * volumes).rolling(period).sum() / volumes.rolling(period).sum()

    return {
        'vwap': vwap,
        'interpretation': {
            'above_vwap': '价格在VWAP上方,买方力量强',
            'below_vwap': '价格在VWAP下方,卖方力量强',
            'vwap_slope': 'VWAP斜率反映趋势强度'
        }
    }
Enter fullscreen mode Exit fullscreen mode

2. 能量潮指标 (OBV)

def calculate_obv(prices, volumes):
    """
    计算能量潮指标
    """
    # 价格变化方向
    price_change = prices.diff()

    # OBV计算逻辑
    obv = pd.Series(0.0, index=prices.index)

    for i in range(1, len(prices)):
        if price_change.iloc[i] > 0:
            obv.iloc[i] = obv.iloc[i-1] + volumes.iloc[i]
        elif price_change.iloc[i] < 0:
            obv.iloc[i] = obv.iloc[i-1] - volumes.iloc[i]
        else:
            obv.iloc[i] = obv.iloc[i-1]

    return {
        'obv': obv,
        'interpretation': {
            'rising_obv': 'OBV上升确认价格上涨趋势',
            'falling_obv': 'OBV下降确认价格下跌趋势',
            'divergence': 'OBV与价格背离预示趋势可能反转'
        }
    }
Enter fullscreen mode Exit fullscreen mode

1.3.5 多时间框架分析

多时间框架的重要性

def multi_timeframe_strategy():
    """
    多时间框架分析策略
    """
    framework = {
        '长期时间框架 (日线/周线)': {
            'purpose': '确定主要趋势方向',
            'indicators': ['MA200', '周线支撑阻力', '长期RSI'],
            'decision': '决定做多还是做空'
        },
        '中期时间框架 (4小时/1小时)': {
            'purpose': '寻找入场时机',
            'indicators': ['MACD', '中期支撑阻力', 'RSI背离'],
            'decision': '确定具体入场点位'
        },
        '短期时间框架 (15分钟/5分钟)': {
            'purpose': '精确执行入场',
            'indicators': ['短期MA', '布林带', '成交量'],
            'decision': '优化入场时机和价格'
        }
    }

    workflow = """
    分析流程:
    1. 长期看趋势:判断大方向
    2. 中期找机会:确定入场区域  
    3. 短期抓时机:精确执行交易
    4. 风险控制:各时间框架协调一致
    """

    return framework, workflow
Enter fullscreen mode Exit fullscreen mode

时间框架协调性分析

def timeframe_alignment_analysis(data_dict):
    """
    分析不同时间框架的协调性
    """
    # data_dict包含不同时间框架的数据
    # {'1d': daily_data, '4h': hourly_data, '15m': minute_data}

    alignment_score = {}

    for timeframe, data in data_dict.items():
        # 计算各时间框架的信号
        ma_signal = 1 if data['close'].iloc[-1] > data['ma20'].iloc[-1] else -1
        rsi_signal = 1 if data['rsi'].iloc[-1] < 70 else -1 if data['rsi'].iloc[-1] > 30 else 0
        macd_signal = 1 if data['macd'].iloc[-1] > data['signal'].iloc[-1] else -1

        alignment_score[timeframe] = {
            'ma_signal': ma_signal,
            'rsi_signal': rsi_signal, 
            'macd_signal': macd_signal,
            'total_score': ma_signal + rsi_signal + macd_signal
        }

    # 计算整体协调性
    total_alignment = sum(score['total_score'] for score in alignment_score.values())

    decision = {
        'strong_buy': total_alignment >= 6,
        'buy': 3 <= total_alignment < 6,
        'neutral': -3 < total_alignment < 3,
        'sell': -6 < total_alignment <= -3,
        'strong_sell': total_alignment <= -6
    }

    return alignment_score, decision
Enter fullscreen mode Exit fullscreen mode

1.3.6 技术形态分析

经典反转形态

1. 头肩顶/头肩底

def detect_head_shoulders(prices, window=20):
    """
    检测头肩形态
    """
    # 寻找三个连续的高点(头肩顶)或低点(头肩底)
    peaks = []
    troughs = []

    for i in range(window, len(prices) - window):
        # 检测峰值
        if all(prices[i] >= prices[i-j] for j in range(1, window+1)) and \
           all(prices[i] >= prices[i+j] for j in range(1, window+1)):
            peaks.append((i, prices[i]))

        # 检测谷值
        if all(prices[i] <= prices[i-j] for j in range(1, window+1)) and \
           all(prices[i] <= prices[i+j] for j in range(1, window+1)):
            troughs.append((i, prices[i]))

    # 头肩顶模式识别
    def identify_head_shoulders_top(peaks):
        if len(peaks) >= 3:
            for i in range(len(peaks) - 2):
                left_shoulder = peaks[i][1]
                head = peaks[i+1][1] 
                right_shoulder = peaks[i+2][1]

                # 头部高于两肩,两肩高度相近
                if (head > left_shoulder and head > right_shoulder and
                    abs(left_shoulder - right_shoulder) / left_shoulder < 0.05):
                    return {
                        'pattern': 'Head and Shoulders Top',
                        'left_shoulder': left_shoulder,
                        'head': head,
                        'right_shoulder': right_shoulder,
                        'signal': 'Bearish Reversal'
                    }
        return None

    # 头肩底模式识别  
    def identify_head_shoulders_bottom(troughs):
        if len(troughs) >= 3:
            for i in range(len(troughs) - 2):
                left_shoulder = troughs[i][1]
                head = troughs[i+1][1]
                right_shoulder = troughs[i+2][1]

                # 头部低于两肩,两肩高度相近
                if (head < left_shoulder and head < right_shoulder and
                    abs(left_shoulder - right_shoulder) / left_shoulder < 0.05):
                    return {
                        'pattern': 'Head and Shoulders Bottom',
                        'left_shoulder': left_shoulder,
                        'head': head,
                        'right_shoulder': right_shoulder,
                        'signal': 'Bullish Reversal'
                    }
        return None

    top_pattern = identify_head_shoulders_top(peaks)
    bottom_pattern = identify_head_shoulders_bottom(troughs)

    return {
        'head_shoulders_top': top_pattern,
        'head_shoulders_bottom': bottom_pattern
    }
Enter fullscreen mode Exit fullscreen mode

2. 双顶/双底

def detect_double_top_bottom(prices, window=15, tolerance=0.02):
    """
    检测双顶双底形态
    """
    peaks = []
    troughs = []

    # 寻找峰值和谷值
    for i in range(window, len(prices) - window):
        if all(prices[i] >= prices[i-j] for j in range(1, window+1)) and \
           all(prices[i] >= prices[i+j] for j in range(1, window+1)):
            peaks.append((i, prices[i]))

        if all(prices[i] <= prices[i-j] for j in range(1, window+1)) and \
           all(prices[i] <= prices[i+j] for j in range(1, window+1)):
            troughs.append((i, prices[i]))

    # 双顶识别
    def identify_double_top(peaks):
        for i in range(len(peaks) - 1):
            peak1 = peaks[i][1]
            peak2 = peaks[i+1][1]

            # 两个峰值高度相近
            if abs(peak1 - peak2) / peak1 <= tolerance:
                return {
                    'pattern': 'Double Top',
                    'peak1': peak1,
                    'peak2': peak2,
                    'signal': 'Bearish Reversal',
                    'target': min(peak1, peak2) - abs(peak1 - peak2)
                }
        return None

    # 双底识别
    def identify_double_bottom(troughs):
        for i in range(len(troughs) - 1):
            trough1 = troughs[i][1]
            trough2 = troughs[i+1][1]

            # 两个谷值高度相近
            if abs(trough1 - trough2) / trough1 <= tolerance:
                return {
                    'pattern': 'Double Bottom',
                    'trough1': trough1,
                    'trough2': trough2,
                    'signal': 'Bullish Reversal',
                    'target': max(trough1, trough2) + abs(trough1 - trough2)
                }
        return None

    double_top = identify_double_top(peaks)
    double_bottom = identify_double_bottom(troughs)

    return {
        'double_top': double_top,
        'double_bottom': double_bottom
    }
Enter fullscreen mode Exit fullscreen mode

持续形态

1. 三角形整理

def detect_triangle_patterns(prices, min_points=4):
    """
    检测三角形整理形态
    """
    highs = []
    lows = []

    # 收集高点和低点
    window = 10
    for i in range(window, len(prices) - window):
        if all(prices[i] >= prices[i-j] for j in range(1, window+1)) and \
           all(prices[i] >= prices[i+j] for j in range(1, window+1)):
            highs.append((i, prices[i]))

        if all(prices[i] <= prices[i-j] for j in range(1, window+1)) and \
           all(prices[i] <= prices[i+j] for j in range(1, window+1)):
            lows.append((i, prices[i]))

    if len(highs) >= min_points // 2 and len(lows) >= min_points // 2:
        # 计算高点和低点的趋势线斜率
        highs_slope = (highs[-1][1] - highs[0][1]) / (highs[-1][0] - highs[0][0])
        lows_slope = (lows[-1][1] - lows[0][1]) / (lows[-1][0] - lows[0][0])

        # 判断三角形类型
        if highs_slope < 0 and lows_slope > 0:
            pattern_type = "Symmetrical Triangle"
            signal = "Continuation pattern, breakout direction uncertain"
        elif highs_slope < 0 and abs(lows_slope) < 0.001:
            pattern_type = "Descending Triangle"  
            signal = "Bearish pattern, expect downward breakout"
        elif abs(highs_slope) < 0.001 and lows_slope > 0:
            pattern_type = "Ascending Triangle"
            signal = "Bullish pattern, expect upward breakout"
        else:
            pattern_type = "Unknown Triangle"
            signal = "Pattern unclear"

        return {
            'pattern': pattern_type,
            'signal': signal,
            'resistance_slope': highs_slope,
            'support_slope': lows_slope,
            'breakout_target': abs(highs[0][1] - lows[0][1])
        }

    return None
Enter fullscreen mode Exit fullscreen mode

本节总结

技术分析是量化交易策略开发的重要基础。通过系统学习趋势分析、支撑阻力、技术指标和形态识别,可以为策略开发提供丰富的信号来源。

关键要点:

  1. 技术分析基于市场行为包含一切信息的假设
  2. 趋势识别和支撑阻力分析是技术分析的核心
  3. 技术指标提供量化的买卖信号,但需要结合市场环境使用
  4. 多时间框架分析能够提高交易决策的准确性
  5. 技术形态为中长期价格预测提供参考

实践练习

练习1.3.1: 指标编程实现

# 实现以下技术指标
def practice_indicators(prices, volumes):
    """
    编程实现各种技术指标
    """
    # 1. 实现威廉指标(%R)
    # 2. 实现KDJ指标
    # 3. 实现ATR平均真实范围
    # 4. 实现CCI商品通道指标
    pass
Enter fullscreen mode Exit fullscreen mode

练习1.3.2: 多时间框架信号系统

def multi_timeframe_signals(daily_data, hourly_data, minute_data):
    """
    构建多时间框架信号系统
    """
    # 1. 分析各时间框架趋势
    # 2. 计算信号强度
    # 3. 给出综合交易建议
    pass
Enter fullscreen mode Exit fullscreen mode

练习1.3.3: 形态识别算法

def pattern_recognition_system(prices):
    """
    开发自动形态识别系统
    """
    # 1. 识别更多经典形态
    # 2. 计算形态可靠性
    # 3. 给出目标价位预测
    pass
Enter fullscreen mode Exit fullscreen mode

思考题

  1. 为什么技术指标在不同市场环境下的效果会有差异?
  2. 如何避免技术分析中的主观偏见?
  3. 多时间框架分析中,如何平衡不同时间框架信号的权重?
  4. 技术形态的有效性如何量化评估?

下一节:模块2 - Freqtrade平台入门

Top comments (0)