28 KiB
28 KiB
In [ ]:
# Imports
import pandas as pd
import numpy as np
import MetaTrader5 as mt
import pandas_ta as ta
from scipy.signal import savgol_filter, find_peaks
from sklearn.linear_model import LinearRegression
from tabulate import tabulate
from datetime import datetime, timedelta
import json
import keyring as kr
print("✅ All imports successful")In [ ]:
# MT5 Login
mt.initialize()
login = 10800246
server = 'VantageInternational-Demo'
password = kr.get_password(server, str(login))
login_result = mt.login(login, password, server)
print(f"Login successful: {login_result}")
# Trading Parameter
symbol = "XAUUSD"
strategy_name = "TradingBot_V1.4_Complete"
print(f"Symbol: {symbol}")In [ ]:
# Helper Functions
def get_rates(timeframe="h4", count=200, symbol="XAUUSD"):
timeframes_dict = {
"m1": mt.TIMEFRAME_M1, "m5": mt.TIMEFRAME_M5, "m15": mt.TIMEFRAME_M15,
"m30": mt.TIMEFRAME_M30, "h1": mt.TIMEFRAME_H1, "h4": mt.TIMEFRAME_H4, "d1": mt.TIMEFRAME_D1
}
try:
rates = mt.copy_rates_from_pos(symbol, timeframes_dict[timeframe], 0, count)
if rates is None: return None
df = pd.DataFrame(rates)
df['time'] = pd.to_datetime(df['time'], unit='s')
df.set_index('time', inplace=True)
df['atr'] = ta.atr(df['high'], df['low'], df['close'], length=14)
return df
except Exception as e:
print(f"Error getting rates: {e}")
return None
def check_risk_limits(symbol, volume=None, order_type="buy", max_risk_per_trade=0.01):
try:
account_info = mt.account_info()
if not account_info: return False
balance, equity = account_info.balance, account_info.equity
if equity < balance * 0.8: return False
return True
except: return False
def market_order(symbol, volume, order_type, stoploss=None, take_profit=None, deviation=20):
try:
price_dict = {'buy': mt.symbol_info_tick(symbol).ask, 'sell': mt.symbol_info_tick(symbol).bid}
order_type_dict = {'buy': mt.ORDER_TYPE_BUY, 'sell': mt.ORDER_TYPE_SELL}
request = {
"action": mt.TRADE_ACTION_DEAL, "symbol": symbol, "volume": volume,
"type": order_type_dict[order_type], "price": price_dict[order_type],
"sl": stoploss, "tp": take_profit, "deviation": deviation,
"magic": 234000, "comment": strategy_name, "type_time": mt.ORDER_TIME_GTC,
"type_filling": mt.ORDER_FILLING_IOC
}
return mt.order_send(request)
except Exception as e:
print(f"Error in market order: {e}")
return None
print("✅ Helper functions defined")In [ ]:
# Market Regime Detection
def detect_market_regime(df, lookback=50):
try:
adx_data = ta.adx(df['high'], df['low'], df['close'], length=14)
adx = adx_data['ADX_14'].iloc[-1] if adx_data is not None and 'ADX_14' in adx_data.columns else 25.0
try:
bb = ta.bbands(df['close'], length=20)
if bb is not None and len(bb.columns) >= 3:
bb_cols = bb.columns.tolist()
bb_width = ((bb[bb_cols[0]] - bb[bb_cols[2]]) / bb[bb_cols[1]] * 100).iloc[-lookback:].mean()
else: bb_width = 4.0
except: bb_width = 4.0
price_range = df['high'].iloc[-lookback:].max() - df['low'].iloc[-lookback:].min()
atr_avg = df['atr'].iloc[-lookback:].mean()
range_ratio = price_range / (atr_avg * lookback) if atr_avg > 0 else 1.0
vol_cluster = df['atr'].iloc[-10:].std() / df['atr'].iloc[-50:].mean() if len(df) >= 50 else 1.0
if adx > 25 and range_ratio > 1.5:
regime, strength = 'trending', min(100, adx * 2)
elif vol_cluster > 1.5:
regime, strength = 'volatile', min(100, vol_cluster * 50)
else:
regime, strength = 'ranging', max(0, 100 - adx * 2)
return {'regime': regime, 'strength': strength, 'adx': adx, 'bb_width': bb_width, 'range_ratio': range_ratio, 'vol_cluster': vol_cluster}
except Exception as e:
return {'regime': 'ranging', 'strength': 50, 'adx': 20, 'bb_width': 4.0, 'range_ratio': 1.0, 'vol_cluster': 1.0}
print("✅ Market Regime Detection defined")In [ ]:
# Adaptive Confidence System
def calculate_adaptive_confidence_threshold(regime_info, base_confidence=70):
regime = regime_info['regime']
adx = regime_info['adx']
if regime == 'trending':
return max(60, base_confidence - 15) if adx > 30 else base_confidence - 10
elif regime == 'ranging':
return base_confidence + 15
elif regime == 'volatile':
return base_confidence + 20
return base_confidence
print("✅ Adaptive Confidence System defined")In [ ]:
# Enhanced Trend Analysis
def get_enhanced_trend(timeframe="H4", lookback=150, symbol="XAUUSD"):
tf_map = {"D1": "d1", "H4": "h4", "H1": "h1", "M30": "m30", "M15": "m15", "M5": "m5"}
tf = tf_map.get(timeframe, timeframe.lower())
try:
df = get_rates(tf, lookback, symbol)
if df is None or len(df) < 50: return None
df['close_smooth'] = savgol_filter(df['close'], min(15, len(df)//10), 3)
X = np.arange(len(df)).reshape(-1, 1)
y = df['close_smooth'].values
model = LinearRegression().fit(X, y)
slope = model.coef_[0]
regime_info = detect_market_regime(df.iloc[-50:])
base_threshold = df['atr'].iloc[-1] * 0.0001
if regime_info['regime'] == 'trending':
slope_threshold = base_threshold * 0.7
elif regime_info['regime'] == 'ranging':
slope_threshold = base_threshold * 1.5
else:
slope_threshold = base_threshold * 1.2
trend = "uptrend" if slope > slope_threshold else "downtrend" if slope < -slope_threshold else "sideways"
trend_strength = abs(slope) / slope_threshold if slope_threshold > 0 else 0
return {
"trend": trend, "slope": slope, "slope_threshold": slope_threshold,
"trend_strength": trend_strength, "atr": df['atr'].iloc[-1],
"price": df['close'].iloc[-1], "regime_info": regime_info
}
except Exception as e:
print(f"Error in get_enhanced_trend: {e}")
return None
print("✅ Enhanced Trend Analysis defined")In [ ]:
# Extended Top-Down Analysis V2
def extended_top_down_v2(symbol="XAUUSD", lookback=150):
timeframes = ["D1", "H4", "H1", "M30", "M15", "M5"]
trend_info = {}
for tf in timeframes:
trend_info[tf] = get_enhanced_trend(tf, lookback, symbol)
if trend_info[tf] is None:
print(f"⚠️ Keine Daten für {tf}")
return None
main_regime = trend_info["H4"]["regime_info"]
adaptive_confidence_threshold = calculate_adaptive_confidence_threshold(main_regime)
# Standard-Trend (D1 + H4)
d1_trend = trend_info["D1"]["trend"]
h4_trend = trend_info["H4"]["trend"]
d1_strength = trend_info["D1"]["trend_strength"]
h4_strength = trend_info["H4"]["trend_strength"]
if d1_trend == h4_trend and d1_trend != "sideways":
standard_trend = d1_trend
standard_strength = (d1_strength * 0.6 + h4_strength * 0.4)
elif d1_strength > h4_strength * 1.5:
standard_trend = d1_trend
standard_strength = d1_strength * 0.8
elif h4_strength > d1_strength * 1.5:
standard_trend = h4_trend
standard_strength = h4_strength * 0.8
else:
standard_trend = "sideways"
standard_strength = 0
# Fast-Trend
fast_timeframes = ["H1", "M30", "M15", "M5"]
fast_trends = [trend_info[tf]["trend"] for tf in fast_timeframes]
fast_strengths = [trend_info[tf]["trend_strength"] for tf in fast_timeframes]
required_alignment = 2 if main_regime['regime'] == 'trending' else 3
trend_counts = {'uptrend': 0, 'downtrend': 0, 'sideways': 0}
weighted_strengths = {'uptrend': 0, 'downtrend': 0}
weights = [1.0, 0.8, 0.6, 0.4]
for i, (trend, strength) in enumerate(zip(fast_trends, fast_strengths)):
trend_counts[trend] += 1
if trend != 'sideways':
weighted_strengths[trend] += strength * weights[i]
max_count = max(trend_counts['uptrend'], trend_counts['downtrend'])
if max_count >= required_alignment:
if trend_counts['uptrend'] > trend_counts['downtrend']:
fast_trend = "uptrend"
elif trend_counts['downtrend'] > trend_counts['uptrend']:
fast_trend = "downtrend"
else:
fast_trend = "uptrend" if weighted_strengths['uptrend'] > weighted_strengths['downtrend'] else "downtrend"
else:
fast_trend = "sideways"
# Top-Down-Trend
if standard_trend == fast_trend and standard_trend != "sideways":
top_down_trend = standard_trend
combined_strength = (standard_strength + weighted_strengths.get(fast_trend, 0)) / 2
else:
top_down_trend = "sideways"
combined_strength = 0
# Confidence Calculation
weights = {"D1": 2.5, "H4": 2.0, "H1": 1.5, "M30": 1.0, "M15": 0.8, "M5": 0.6}
weighted_matching = sum(
weights[tf] * trend_info[tf]["trend_strength"]
for tf in timeframes
if trend_info[tf]["trend"] == top_down_trend and trend_info[tf]["trend"] != "sideways"
)
weighted_total = sum(
weights[tf] * trend_info[tf]["trend_strength"]
for tf in timeframes
if trend_info[tf]["trend"] != "sideways"
)
confidence = round((weighted_matching / weighted_total) * 100, 2) if weighted_total > 0 else 0.0
# Risk-Adjusted Signal Strength
atr = trend_info["M5"]["atr"]
rrr = 2.5
risk_adjusted_strength = confidence * combined_strength * min(2.0, rrr)
# Entry Signal
entry_signal = 0
signal_quality = "none"
if (top_down_trend != "sideways" and
confidence >= adaptive_confidence_threshold and
risk_adjusted_strength >= 100):
entry_signal = 1 if top_down_trend == "uptrend" else -1
if confidence >= 85 and risk_adjusted_strength >= 150:
signal_quality = "excellent"
elif confidence >= 75 and risk_adjusted_strength >= 120:
signal_quality = "good"
else:
signal_quality = "fair"
# Debug Output
debug_data = []
for tf in timeframes:
info = trend_info[tf]
debug_data.append([tf, info["trend"], f"{info['trend_strength']:.2f}",
f"{info['atr']:.4f}", f"{info['slope']:.6f}", f"{info['price']:.2f}"])
print(f"📊 Enhanced Trend-Analyse für {symbol}")
print(f"🎯 Market Regime: {main_regime['regime'].upper()} (Strength: {main_regime['strength']:.0f}%)")
print(f"🎚️ Adaptive Confidence Threshold: {adaptive_confidence_threshold}%")
print(tabulate(debug_data, headers=["TF", "Trend", "Strength", "ATR", "Slope", "Price"], tablefmt="psql"))
print(f"➡️ Standard-Trend: {standard_trend} (Strength: {standard_strength:.2f})")
print(f"➡️ Fast-Trend: {fast_trend}")
print(f"➡️ Top-Down-Trend: {top_down_trend}")
print(f"➡️ Confidence: {confidence}% (Threshold: {adaptive_confidence_threshold}%)")
print(f"➡️ Risk-Adjusted Strength: {risk_adjusted_strength:.1f}")
print(f"➡️ Signal Quality: {signal_quality.upper()}")
return {
"symbol": symbol, "trend_info": trend_info, "market_regime": main_regime,
"standard_trend": standard_trend, "fast_trend": fast_trend, "top_down_trend": top_down_trend,
"confidence": confidence, "adaptive_threshold": adaptive_confidence_threshold,
"risk_adjusted_strength": risk_adjusted_strength, "entry_signal": entry_signal,
"signal_quality": signal_quality, "combined_strength": combined_strength
}
print("✅ Extended Top-Down V2 defined")In [ ]:
# Entry Timing Optimization
def check_pullback_entry(symbol, signal_info, timeframe="M5"):
if signal_info["entry_signal"] == 0:
return False, "No base signal"
try:
df = get_rates(timeframe.lower(), 50, symbol)
if df is None or len(df) < 20:
return False, "Insufficient data"
df['ema21'] = df['close'].ewm(span=21).mean()
df['ema50'] = df['close'].ewm(span=50).mean()
current_price = df['close'].iloc[-1]
ema21 = df['ema21'].iloc[-1]
ema50 = df['ema50'].iloc[-1]
signal_direction = signal_info["entry_signal"]
if signal_direction == 1: # Long
if current_price <= ema21 * 1.002 and ema21 > ema50:
return True, "Pullback to EMA21 for Long"
elif current_price <= ema21 * 0.998:
return True, "Below EMA21 - Good Long Entry"
elif signal_direction == -1: # Short
if current_price >= ema21 * 0.998 and ema21 < ema50:
return True, "Pullback to EMA21 for Short"
elif current_price >= ema21 * 1.002:
return True, "Above EMA21 - Good Short Entry"
return False, "Waiting for better entry timing"
except Exception as e:
return True, "Using immediate entry (fallback)"
print("✅ Entry Timing Optimization defined")In [ ]:
# Enhanced Execute Trade Function
def execute_trade_v2(
symbol="XAUUSD",
atr_mult=1.5,
base_confidence=70,
max_risk_per_trade=0.01,
risk_filter=True,
min_atr=0.0010,
use_pullback_entry=True,
debug=True
):
signal_info = extended_top_down_v2(symbol)
if signal_info is None:
print("❌ Signal-Analyse fehlgeschlagen")
return None
entry_signal = signal_info["entry_signal"]
confidence = signal_info["confidence"]
adaptive_threshold = signal_info["adaptive_threshold"]
signal_quality = signal_info["signal_quality"]
market_regime = signal_info["market_regime"]
m5_info = signal_info["trend_info"]["M5"]
price = m5_info["price"]
atr = m5_info["atr"]
reason = ""
if confidence < adaptive_threshold:
reason = f"Confidence {confidence}% < threshold {adaptive_threshold}%"
elif entry_signal == 0:
reason = f"No entry signal (Trend: {signal_info['top_down_trend']})"
elif price is None or atr is None:
reason = "Price/ATR not available"
elif risk_filter and atr < min_atr:
reason = f"ATR {atr:.5f} < min_atr {min_atr}"
elif signal_quality == "none":
reason = "Signal quality insufficient"
else:
if use_pullback_entry:
pullback_ok, pullback_reason = check_pullback_entry(symbol, signal_info)
if not pullback_ok:
reason = f"Entry timing: {pullback_reason}"
if not reason:
risk_ok = check_risk_limits(symbol, max_risk_per_trade=max_risk_per_trade)
if not risk_ok:
reason = "Risk limits exceeded"
if not reason:
regime_mult = 1.0
if market_regime['regime'] == 'volatile':
regime_mult = 1.3
elif market_regime['regime'] == 'ranging':
regime_mult = 0.8
adjusted_atr_mult = atr_mult * regime_mult
if entry_signal == 1:
stop_loss = price - adjusted_atr_mult * atr
take_profit = price + adjusted_atr_mult * atr * 2.5
else:
stop_loss = price + adjusted_atr_mult * atr
take_profit = price - adjusted_atr_mult * atr * 2.5
account_info = mt.account_info()
if account_info:
balance = account_info.balance
risk_amount = balance * max_risk_per_trade
if symbol == "XAUUSD":
volume = min(0.1, max(0.01, risk_amount / (adjusted_atr_mult * atr * 100)))
else:
volume = 0.01
else:
volume = 0.01
print(f"🚀 ENHANCED TRADE EXECUTION")
print(f"Symbol: {symbol}")
print(f"Direction: {'LONG' if entry_signal == 1 else 'SHORT'}")
print(f"Price: {price:.5f}")
print(f"Volume: {volume:.2f}")
print(f"Stop Loss: {stop_loss:.5f}")
print(f"Take Profit: {take_profit:.5f}")
print(f"Confidence: {confidence}% (Threshold: {adaptive_threshold}%)")
print(f"Signal Quality: {signal_quality.upper()}")
print(f"Market Regime: {market_regime['regime'].upper()}")
try:
order_result = market_order(
symbol=symbol,
volume=volume,
order_type="buy" if entry_signal == 1 else "sell",
stoploss=stop_loss,
take_profit=take_profit
)
return order_result
except Exception as e:
print(f"❌ Trade execution failed: {e}")
return None
else:
if debug:
print(f"⏸️ TRADE SKIPPED: {reason}")
print(f"Confidence: {confidence}% | Threshold: {adaptive_threshold}%")
print(f"Signal Quality: {signal_quality} | Regime: {market_regime['regime']}")
return None
print("✅ Enhanced Execute Trade defined")In [ ]:
# Test der optimierten Funktionen
print("🔍 Testing Market Regime Detection...")
df_test = get_rates("h4", 100, symbol)
if df_test is not None:
regime = detect_market_regime(df_test)
print(f"Regime: {regime['regime'].upper()}")
print(f"Strength: {regime['strength']:.1f}%")
print(f"ADX: {regime['adx']:.1f}")
adaptive_threshold = calculate_adaptive_confidence_threshold(regime)
print(f"Adaptive Threshold: {adaptive_threshold}% (vs 80% fixed)")
else:
print("❌ Could not get test data")In [ ]:
# Test Enhanced Top-Down Analysis
print("🔍 Testing Enhanced Top-Down Analysis...")
signal_result = extended_top_down_v2(symbol)
if signal_result:
print(f"🎯 SIGNAL SUMMARY:")
print(f"Entry Signal: {signal_result['entry_signal']}")
print(f"Confidence: {signal_result['confidence']}%")
print(f"Adaptive Threshold: {signal_result['adaptive_threshold']}%")
print(f"Signal Quality: {signal_result['signal_quality'].upper()}")
print(f"Market Regime: {signal_result['market_regime']['regime'].upper()}")
print(f"Risk-Adjusted Strength: {signal_result['risk_adjusted_strength']:.1f}")
if signal_result['entry_signal'] != 0:
direction = "LONG" if signal_result['entry_signal'] == 1 else "SHORT"
print(f"🚀 TRADING SIGNAL: {direction}")
else:
print(f"⏸️ NO TRADING SIGNAL")
else:
print("❌ Signal analysis failed")In [ ]:
# Trading Configuration
TRADING_CONFIG = {
'symbol': symbol,
'atr_mult': 1.5,
'base_confidence': 70,
'max_risk_per_trade': 0.01,
'risk_filter': True,
'min_atr': 0.0010,
'use_pullback_entry': True,
'debug': True
}
print("⚙️ Trading Configuration:")
for key, value in TRADING_CONFIG.items():
print(f" {key}: {value}")In [ ]:
# Test Trading Execution
def test_trading():
print("🚀 Testing Trade Execution...")
try:
result = execute_trade_v2(**TRADING_CONFIG)
if result:
print("✅ Trade executed successfully!")
return result
else:
print("⏸️ No trade executed")
return None
except Exception as e:
print(f"❌ Error: {e}")
return None
test_result = test_trading()In [ ]:
# Status Check
def check_bot_status():
print("🔍 Trading Bot Status:")
print(f" MT5 Connection: {'✅' if mt.terminal_info() else '❌'}")
try:
signal_info = extended_top_down_v2(symbol)
if signal_info:
print(f" Current Signal: {signal_info['entry_signal']}")
print(f" Confidence: {signal_info['confidence']}%")
print(f" Market Regime: {signal_info['market_regime']['regime'].upper()}")
print(f" Signal Quality: {signal_info['signal_quality'].upper()}")
except Exception as e:
print(f" Signal Check: ❌ Error: {e}")
check_bot_status()