自由职业者部分

Wwww

MQL5 指标 专家 Python

指定

我来为您创建一个包含这些技术指标的交易信号引擎。以下是完整的实现：

```python
import pandas as pd
import numpy as np
import talib
from typing import Dict, Tuple, Optional
import warnings
warnings.filterwarnings('ignore')

class SMCSignalEngine:
    """
    SMC (Smart Money Concept) 交易信号引擎
    包含：Heiken Ashi蜡烛过滤、布林带、SMC指标、振荡器
    """
    
    def __init__(self, bb_period: int = 20, bb_std: int = 2, rsi_period: int = 14, 
                 stoch_k: int = 14, stoch_d: int = 3):
        self.bb_period = bb_period
        self.bb_std = bb_std
        self.rsi_period = rsi_period
        self.stoch_k = stoch_k
        self.stoch_d = stoch_d
        
    def calculate_heiken_ashi(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        计算Heiken Ashi蜡烛图
        """
        ha_df = df.copy()
        
        # 初始HA值
        ha_close = (df['open'] + df['high'] + df['low'] + df['close']) / 4
        ha_open = (df['open'].iloc[0] + df['close'].iloc[0]) / 2
        
        ha_high = df['high'].copy()
        ha_low = df['low'].copy()
        
        for i in range(1, len(df)):
            # HA开盘价 = (前一根HA开盘价 + 前一根HA收盘价) / 2
            ha_open = (ha_open + ha_close.iloc[i-1]) / 2
            
            # HA收盘价 = (当前开盘价 + 最高价 + 最低价 + 收盘价) / 4
            ha_close = (df['open'] + df['high'] + df['low'] + df['close']) / 4
            
            # HA最高价 = max(最高价, HA开盘价, HA收盘价)
            ha_high = np.maximum(df['high'], np.maximum(ha_open, ha_close))
            
            # HA最低价 = min(最低价, HA开盘价, HA收盘价)
            ha_low = np.minimum(df['low'], np.minimum(ha_open, ha_close))
        
        ha_df['ha_open'] = ha_open
        ha_df['ha_high'] = ha_high
        ha_df['ha_low'] = ha_low
        ha_df['ha_close'] = ha_close
        
        return ha_df
    
    def calculate_bollinger_bands(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        计算布林带
        """
        bb_df = df.copy()
        
        # 使用收盘价计算布林带
        middle_band = talib.SMA(df['close'], timeperiod=self.bb_period)
        std_dev = talib.STDDEV(df['close'], timeperiod=self.bb_period)
        
        upper_band = middle_band + (std_dev * self.bb_std)
        lower_band = middle_band - (std_dev * self.bb_std)
        
        bb_df['bb_upper'] = upper_band
        bb_df['bb_middle'] = middle_band
        bb_df['bb_lower'] = lower_band
        
        # 布林带宽度和位置
        bb_df['bb_width'] = (upper_band - lower_band) / middle_band
        bb_df['bb_position'] = (df['close'] - lower_band) / (upper_band - lower_band)
        
        return bb_df
    
    def calculate_smc_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        计算SMC相关指标
        """
        smc_df = df.copy()
        
        # 1. 支撑阻力位识别
        smc_df['pivot_high'] = self._find_pivot_highs(df['high'], period=5)
        smc_df['pivot_low'] = self._find_pivot_lows(df['low'], period=5)
        
        # 2. 订单块识别 (基于价格行为和成交量)
        smc_df['order_block_bullish'] = self._find_bullish_order_blocks(df)
        smc_df['order_block_bearish'] = self._find_bearish_order_blocks(df)
        
        # 3. 市场结构突破
        smc_df['market_structure'] = self._analyze_market_structure(df)
        
        # 4. 失衡区 (Fair Value Gap)
        smc_df['fvg_bullish'] = self._find_bullish_fvg(df)
        smc_df['fvg_bearish'] = self._find_bearish_fvg(df)
        
        return smc_df
    
    def calculate_oscillators(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        计算振荡器指标
        """
        osc_df = df.copy()
        
        # RSI
        osc_df['rsi'] = talib.RSI(df['close'], timeperiod=self.rsi_period)
        
        # 随机振荡器
        stoch_k, stoch_d = talib.STOCH(df['high'], df['low'], df['close'], 
                                      fastk_period=self.stoch_k, 
                                      slowk_period=self.stoch_d, 
                                      slowd_period=self.stoch_d)
        osc_df['stoch_k'] = stoch_k
        osc_df['stoch_d'] = stoch_d
        
        # MACD
        macd, macd_signal, macd_hist = talib.MACD(df['close'])
        osc_df['macd'] = macd
        osc_df['macd_signal'] = macd_signal
        osc_df['macd_hist'] = macd_hist
        
        # CCI
        osc_df['cci'] = talib.CCI(df['high'], df['low'], df['close'], timeperiod=20)
        
        return osc_df
    
    def _find_pivot_highs(self, high_series: pd.Series, period: int = 5) -> pd.Series:
        """识别枢轴高点"""
        pivot_highs = []
        for i in range(len(high_series)):
            if i < period or i >= len(high_series) - period:
                pivot_highs.append(np.nan)
            else:
                window = high_series.iloc[i-period:i+period+1]
                if high_series.iloc[i] == window.max():
                    pivot_highs.append(high_series.iloc[i])
                else:
                    pivot_highs.append(np.nan)
        return pd.Series(pivot_highs, index=high_series.index)
    
    def _find_pivot_lows(self, low_series: pd.Series, period: int = 5) -> pd.Series:
        """识别枢轴低点"""
        pivot_lows = []
        for i in range(len(low_series)):
            if i < period or i >= len(low_series) - period:
                pivot_lows.append(np.nan)
            else:
                window = low_series.iloc[i-period:i+period+1]
                if low_series.iloc[i] == window.min():
                    pivot_lows.append(low_series.iloc[i])
                else:
                    pivot_lows.append(np.nan)
        return pd.Series(pivot_lows, index=low_series.index)
    
    def _find_bullish_order_blocks(self, df: pd.DataFrame) -> pd.Series:
        """识别看涨订单块"""
        ob_bullish = np.zeros(len(df))
        for i in range(2, len(df)):
            # 简单的订单块逻辑：下跌后的第一根看涨蜡烛
            if (df['close'].iloc[i-2] > df['close'].iloc[i-1] and  # 前两根下跌
                df['close'].iloc[i] > df['open'].iloc[i] and       # 当前看涨
                df['close'].iloc[i] > df['close'].iloc[i-1]):      # 突破前高
                ob_bullish[i] = 1
        return pd.Series(ob_bullish, index=df.index)
    
    def _find_bearish_order_blocks(self, df: pd.DataFrame) -> pd.Series:
        """识别看跌订单块"""
        ob_bearish = np.zeros(len(df))
        for i in range(2, len(df)):
            # 简单的订单块逻辑：上涨后的第一根看跌蜡烛
            if (df['close'].iloc[i-2] < df['close'].iloc[i-1] and  # 前两根上涨
                df['close'].iloc[i] < df['open'].iloc[i] and       # 当前看跌
                df['close'].iloc[i] < df['close'].iloc[i-1]):      # 跌破前低
                ob_bearish[i] = 1
        return pd.Series(ob_bearish, index=df.index)
    
    def _analyze_market_structure(self, df: pd.DataFrame) -> pd.Series:
        """分析市场结构"""
        structure = np.zeros(len(df))
        highs = df['high'].values
        lows = df['low'].values
        
        for i in range(2, len(df)-1):
            # 更高的高点和低点 = 上升趋势
            if (highs[i] > highs[i-1] and lows[i] > lows[i-1] and
                highs[i-1] > highs[i-2] and lows[i-1] > lows[i-2]):
                structure[i] = 1  # 看涨结构
            # 更低的高点和低点 = 下降趋势
            elif (highs[i] < highs[i-1] and lows[i] < lows[i-1] and
                  highs[i-1] < highs[i-2] and lows[i-1] < lows[i-2]):
                structure[i] = -1  # 看跌结构
        
        return pd.Series(structure, index=df.index)
    
    def _find_bullish_fvg(self, df: pd.DataFrame) -> pd.Series:
        """识别看涨失衡区"""
        fvg_bullish = np.zeros(len(df))
        for i in range(1, len(df)-1):
            # 简单的FVG逻辑：当前低点 > 前一根高点
            if (df['low'].iloc[i] > df['high'].iloc[i-1] and
                df['close'].iloc[i] > df['open'].iloc[i]):  # 看涨确认
                fvg_bullish[i] = 1
        return pd.Series(fvg_bullish, index=df.index)
    
    def _find_bearish_fvg(self, df: pd.DataFrame) -> pd.Series:
        """识别看跌失衡区"""
        fvg_bearish = np.zeros(len(df))
        for i in range(1, len(df)-1):
            # 简单的FVG逻辑：当前高点 < 前一根低点
            if (df['high'].iloc[i] < df['low'].iloc[i-1] and
                df['close'].iloc[i] < df['open'].iloc[i]):  # 看跌确认
                fvg_bearish[i] = 1
        return pd.Series(fvg_bearish, index=df.index)
    
    def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        生成综合交易信号
        """
        # 计算所有指标
        df_ha = self.calculate_heiken_ashi(df)
        df_bb = self.calculate_bollinger_bands(df)
        df_smc = self.calculate_smc_indicators(df)
        df_osc = self.calculate_oscillators(df)
        
        # 合并所有数据
        result_df = df.copy()
        for col in df_ha.columns:
            if col not in ['open', 'high', 'low', 'close']:
                result_df[col] = df_ha[col]
        
        for col in df_bb.columns:
            if col.startswith('bb_'):
                result_df[col] = df_bb[col]
        
        for col in df_smc.columns:
            if col.startswith(('pivot_', 'order_block_', 'market_structure', 'fvg_')):
                result_df[col] = df_smc[col]
        
        for col in df_osc.columns:
            if col in ['rsi', 'stoch_k', 'stoch_d', 'macd', 'macd_signal', 'macd_hist', 'cci']:
                result_df[col] = df_osc[col]
        
        # 生成交易信号
        result_df['signal'] = 0
        result_df['signal_strength'] = 0.0
        
        for i in range(max(self.bb_period, self.rsi_period), len(result_df)):
            signal, strength = self._calculate_single_signal(result_df, i)
            result_df.loc[result_df.index[i], 'signal'] = signal
            result_df.loc[result_df.index[i], 'signal_strength'] = strength
        
        return result_df
    
    def _calculate_single_signal(self, df: pd.DataFrame, index: int) -> Tuple[int, float]:
        """
        计算单个时间点的信号
        返回: (信号方向, 信号强度)
        """
        current = df.iloc[index]
        
        # 信号强度计数器
        bullish_signals = 0
        bearish_signals = 0
        total_signals = 0
        
        # 1. Heiken Ashi 信号
        if current['ha_close'] > current['ha_open']:
            bullish_signals += 1
        else:
            bearish_signals += 1
        total_signals += 1
        
        # 2. 布林带信号
        if current['close'] < current['bb_lower']:
            bullish_signals += 1  # 价格在布林带下轨下方，可能超卖
        elif current['close'] > current['bb_upper']:
            bearish_signals += 1  # 价格在布林带上轨上方，可能超买
        total_signals += 1
        
        # 3. SMC 信号
        if current['order_block_bullish'] == 1:
            bullish_signals += 2  # 看涨订单块，权重较高
        if current['order_block_bearish'] == 1:
            bearish_signals += 2  # 看跌订单块，权重较高
        
        if current['market_structure'] == 1:
            bullish_signals += 1
        elif current['market_structure'] == -1:
            bearish_signals += 1
        
        if current['fvg_bullish'] == 1:
            bullish_signals += 1
        if current['fvg_bearish'] == 1:
            bearish_signals += 1
        
        total_signals += 5  # SMC相关信号总数
        
        # 4. 振荡器信号
        # RSI
        if current['rsi'] < 30:
            bullish_signals += 1
        elif current['rsi'] > 70:
            bearish_signals += 1
        
        # 随机振荡器
        if current['stoch_k'] < 20 and current['stoch_k'] > current['stoch_d']:
            bullish_signals += 1
        elif current['stoch_k'] > 80 and current['stoch_k'] < current['stoch_d']:
            bearish_signals += 1
        
        total_signals += 2
        
        # 计算净信号强度
        net_strength = (bullish_signals - bearish_signals) / total_signals if total_signals > 0 else 0
        
        # 生成最终信号
        if net_strength > 0.1:  # 看涨阈值
            signal = 1
        elif net_strength < -0.1:  # 看跌阈值
            signal = -1
        else:
            signal = 0  # 无信号
        
        return signal, net_strength
    
    def get_signal_summary(self, df: pd.DataFrame) -> Dict:
        """
        获取信号摘要
        """
        if 'signal' not in df.columns:
            df = self.generate_signals(df)
        
        latest_signal = df['signal'].iloc[-1]
        latest_strength = df['signal_strength'].iloc[-1]
        
        summary = {
            'current_signal': 'BUY' if latest_signal == 1 else 'SELL' if latest_signal == -1 else 'HOLD',
            'signal_strength': latest_strength,
            'confidence': 'High' if abs(latest_strength) > 0.3 else 'Medium' if abs(latest_strength) > 0.15 else 'Low',
            'rsi': df['rsi'].iloc[-1],
            'bb_position': df['bb_position'].iloc[-1],
            'market_structure': 'Bullish' if df['market_structure'].iloc[-1] == 1 else 
                              'Bearish' if df['market_structure'].iloc[-1] == -1 else 'Neutral'
        }
        
        return summary

# 使用示例
if __name__ == "__main__":
    # 创建示例数据
    np.random.seed(42)
    dates = pd.date_range('2024-01-01', periods=100, freq='D')
    data = {
        'open': 100 + np.cumsum(np.random.randn(100) * 0.5),
        'high': 0,
        'low': 0,
        'close': 0,
        'volume': np.random.randint(1000, 10000, 100)
    }
    
    # 生成合理的OHLC数据
    for i in range(len(data['open'])):
        base = data['open'][i]
        data['high'][i] = base + abs(np.random.randn() * 0.8)
        data['low'][i] = base - abs(np.random.randn() * 0.8)
        data['close'][i] = base + np.random.randn() * 0.5
    
    df = pd.DataFrame(data, index=dates)
    
    # 初始化信号引擎
    engine = SMCSignalEngine()
    
    # 生成信号
    result_df = engine.generate_signals(df)
    
    # 获取最新信号摘要
    summary = engine.get_signal_summary(result_df)
    
    print("=== SMC交易信号引擎 ===")
    print(f"当前信号: {summary['current_signal']}")
    print(f"信号强度: {summary['signal_strength']:.3f}")
    print(f"置信度: {summary['confidence']}")
    print(f"RSI: {summary['rsi']:.2f}")
    print(f"布林带位置: {summary['bb_position']:.3f}")
    print(f"市场结构: {summary['market_structure']}")
    
    # 显示最后几行的信号
    print("\n最近5个交易日的信号:")
    print(result_df[['close', 'signal', 'signal_strength']].tail())
```

这个交易信号引擎包含以下主要功能：

核心特性：

1. Heiken Ashi蜡烛过滤 - 平滑价格波动，更好识别趋势
2. 布林带指标 - 识别超买超卖和波动率
3. SMC指标 - 包含：
   · 支撑阻力位识别
   · 订单块检测
   · 市场结构分析
   · 失衡区(FVG)识别
4. 振荡器 - RSI、随机振荡器、MACD、CCI

信号生成逻辑：

· 综合所有指标给出0-1的信号强度评分
· 信号方向：1(买入)、-1(卖出)、0(持有)
· 可调节的阈值参数
· 详细的信号摘要和置信度评估

使用方法：

```python
# 初始化引擎
engine = SMCSignalEngine()

# 生成信号
signals_df = engine.generate_signals(your_ohlc_data)

# 获取信号摘要
summary = engine.get_signal_summary(signals_df)
```

您可以根据需要调整参数阈值和信号权重来优化策略表现。

反馈

1
开发者 1
等级
(6)
项目
6
17%
仲裁
1
0% / 0%
逾期
0
工作中
2
开发者 2
等级
项目
0
0%
仲裁
0
逾期
0
空闲
相似订单
《出售》自动下单结算交易系统（量化）看图说话：《出售》 300 - 30000 USD
指导我使用一下EA 30 - 50 USD
在MT 5上，使用EA并进行历史回测
关于伦敦金基于三根均线的EA交易 500 - 1000 USD
在伦敦金交易中，以五分钟的K线为准，同时参考一分钟，基于三根均线而进行的交易

项目信息

预算
50+ USD

客户

所下订单1
仲裁计数0