65 KiB
65 KiB
In [ ]:
#!pip install ta_lib-0.6.5-cp311-cp311-win_amd64.whlIn [ ]:
#%pip install talibIn [8]:
#!pip install ta
from ta.trend import ADXIndicator, EMAIndicator
from ta.momentum import RSIIndicator
from talib import CDLHAMMER, CDLSHOOTINGSTARIn [9]:
import pandas as pd
import matplotlib.pyplot as plt
import mplfinance as mpf
import keyring as kr
import MetaTrader5 as mt
import requests
import re
from time import sleep
import sqlite3 as db
#import matplotlib.pyplot as plt
import pandas_ta as ta
import numpy as np
from sklearn.linear_model import LinearRegression
from scipy.signal import savgol_filter
from scipy.signal import find_peaks
In [10]:
# login to your Trading Account - sign up in the description
mt.initialize()
login = 10800246
server = 'VantageInternational-Demo'
password = kr.get_password(server, str(login))
mt.login(login, password, server)Out [10]:
True
In [11]:
project = "trading-" + server[-4::1]
project = project.lower()
projectOut [11]:
'trading-demo'
In [12]:
pause_trading = 0
In [13]:
symbols = ['XAUUSD']
#symbols = ['BTCUSD']
#'BTCUSD', 'ETHUSD',
#'XRPUSD', , 'EURNZD', 'EURUSD'In [14]:
volume_dict = {
'BTCUSD' : 0.1,
'BTCUSD_short' : 0.1,
'ETHUSD' : 1.0,
'ETHUSD_short' : 1.0,
'XRPUSD': 0.1,
'XRPUSD_short': 0.1,
'XAUUSD': 0.1,
'XAUUSD_short': 0.1,
'EURUSD': 0.1,
'EURUSD_short': 0.1,
'EURNZD': 0.1,
'EURNZD_short': 0.1,
}In [15]:
periods_dict = {
'BTCUSD' : ['h1', 'm30', 'm15', 'm5', 'm1'],
'ETHUSD' : ['h1', 'm30', 'm15', 'm5', 'm1'],
'XRPUSD': ['h1', 'm30', 'm15', 'm5', 'm1'],
'XAUUSD': [ 'm5'],
'EURUSD': ['h1', 'm30', 'm15', 'm5', 'm1'],
'EURNZD': ['m5', 'm2', 'm1'],
}In [16]:
def get_symbol():
global symbols, pause_trading, periods_dict
pricemovement = {}
for s in symbols:
items = mt.symbol_info(s)
pricemovement[s] = round(items.price_change,2)
#print(s, round(items.price_change,2))
#percentage = pricemovement[max(pricemovement, key=pricemovement.get)]
#symbol = max(pricemovement, key=pricemovement.get)
#volume = volume_dict[symbol]
sorted_pricemovement = sorted(pricemovement.items(), key=lambda x:x[1], reverse=True)
converted_dict = dict(sorted_pricemovement)
print(converted_dict)
percentage = converted_dict[max(converted_dict, key=converted_dict.get)]
symbol = max(converted_dict, key=converted_dict.get)
volume = volume_dict[symbol]
print(symbol, percentage, volume)
# if percentage > 0: # and percentage < 0.8:
# periods_dict[symbol] = ['m15', 'm5', 'm2', 'm1']
# elif percentage > 0.8:
# periods_dict[symbol] = ['h1', 'm30', 'm15', 'm5', 'm1']
#
#print(periods_dict)
# if percentage > 0 and mt.positions_total() == 0:
# pause_trading = 0 #0 no puase
# return symbol, percentage, volume, periods_dict
# elif percentage < 0.1 and mt.positions_total() == 0:
# print("aktuell kein neues Symbol, pause Trading")
# pause_trading = 1 #1 pause
# return None
return symbol, percentage, volume, periods_dictIn [17]:
get_symbol()Out [17]:
{'XAUUSD': 0.93}
XAUUSD 0.93 0.1
('XAUUSD',
0.93,
0.1,
{'BTCUSD': ['h1', 'm30', 'm15', 'm5', 'm1'],
'ETHUSD': ['h1', 'm30', 'm15', 'm5', 'm1'],
'XRPUSD': ['h1', 'm30', 'm15', 'm5', 'm1'],
'XAUUSD': ['m5'],
'EURUSD': ['h1', 'm30', 'm15', 'm5', 'm1'],
'EURNZD': ['m5', 'm2', 'm1']})In [18]:
def set_symbol():
global symbol, volume, volume_dict
symb = get_symbol()
if symb != None:
symbol = symb[0]
volume = volume_dict[symb[0]]
In [19]:
get_symbol()Out [19]:
{'XAUUSD': 0.93}
XAUUSD 0.93 0.1
('XAUUSD',
0.93,
0.1,
{'BTCUSD': ['h1', 'm30', 'm15', 'm5', 'm1'],
'ETHUSD': ['h1', 'm30', 'm15', 'm5', 'm1'],
'XRPUSD': ['h1', 'm30', 'm15', 'm5', 'm1'],
'XAUUSD': ['m5'],
'EURUSD': ['h1', 'm30', 'm15', 'm5', 'm1'],
'EURNZD': ['m5', 'm2', 'm1']})In [20]:
set_symbol()
{'XAUUSD': 0.93}
XAUUSD 0.93 0.1
In [21]:
symbol, volume, pause_tradingOut [21]:
('XAUUSD', 0.1, 0)In [22]:
pos = mt.positions_get()
for i in pos:
#if i.comment == 'Retracement Bot':
# print(i.ticket)
if bool(re.search('^BuyStop[0-9]{2}', i.comment)):
print(i.ticket)
mt.positions_total()Out [22]:
0
In [23]:
strategy_name = 'Retracement Bot'
pos = mt.positions_get()
for p in pos:
if p.comment == strategy_name:
print(p.comment)
print(pos)()
In [25]:
def market_order(symbol, volume, order_type, deviation=20, magic=30, stoploss=0.0, take_profit=0.0,
strategy_name='Retracement Bot'):
global project, pause_trading
project_id_dict = {
'trading-demo': 'a3f3ae',
'trading-live': '747543'
}
order_type_dict = {
'buy': mt.ORDER_TYPE_BUY,
'sell': mt.ORDER_TYPE_SELL
}
price_dict = {
'buy': mt.symbol_info_tick(symbol).ask,
'sell': mt.symbol_info_tick(symbol).bid
}
buypos = []
pos = mt.positions_get()
for p in pos:
if p.comment == strategy_name:
buypos.append('true')
activepos = buypos.count('true')
if order_type == 'buy' and activepos == 0 and pause_trading == 0: # and mt.positions_total() == 0:
request = {
"action": mt.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": volume, # FLOAT
"type": order_type_dict[order_type],
"price": price_dict[order_type],
"sl": stoploss, # FLOAT
"tp": take_profit, # FLOAT
"deviation": deviation, # INTERGER
"magic": magic, # INTERGER
"comment": strategy_name,
"type_time": mt.ORDER_TIME_GTC,
"type_filling": mt.ORDER_FILLING_IOC, # mt.ORDER_FILLING_FOK if IOC does not work
}
requests.post('https://api.mynotifier.app', {
"apiKey": 'beafb52e-3cb6-477a-92ef-2f10bff50e20',
"message": "Es wrude ein Handel eröffnet!",
"description": "Bitte kontrolliere die Position",
"type": "info",#"info", # info, error, warning or success
"project": project_id_dict[project]
})
order_result = mt.order_send(request)
#return (order_result)
elif order_type == 'sell' and mt.positions_total() > 0:
pos = mt.positions_get()
for p in pos:
if p.comment == strategy_name:
# while schleife ?
positions = mt.positions_get()
ticket = p.ticket
request = {
"action": mt.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": volume, # FLOAT
"type": order_type_dict[order_type],
"price": price_dict[order_type],
"position": ticket,
"sl": stoploss, # FLOAT
"tp": take_profit, # FLOAT
"deviation": deviation, # INTERGER
"magic": magic, # INTERGER
"comment": strategy_name,
"type_time": mt.ORDER_TIME_GTC,
"type_filling": mt.ORDER_FILLING_IOC, # mt.ORDER_FILLING_FOK if IOC does not work
}
requests.post('https://api.mynotifier.app', {
"apiKey": 'beafb52e-3cb6-477a-92ef-2f10bff50e20',
"message": "Es wrude ein Handel geschlossen!",
"description": "Bitte prüfe die Position",
"type": "info",#"info", # info, error, warning or success
"project": project_id_dict[project]
})
order_result = mt.order_send(request)
#return (order_result)
In [26]:
debug = FalseIn [27]:
trend_dict = {
'm5': '',
#'m10': '',
#'m15': '',
#'m30': '',
#'h4': '',
}In [28]:
timeframes_dict = {
'm1': mt.TIMEFRAME_M1,
'm2': mt.TIMEFRAME_M2,
'm3': mt.TIMEFRAME_M3,
'm5': mt.TIMEFRAME_M5,
'm15': mt.TIMEFRAME_M15,
'm20': mt.TIMEFRAME_M20,
'm30': mt.TIMEFRAME_M30,
'h1': mt.TIMEFRAME_H1,
'h4': mt.TIMEFRAME_H4,
}
# timeframes = {
# 'm1': mt.TIMEFRAME_M1,
# 'm2': mt.TIMEFRAME_M2,
# 'm3': mt.TIMEFRAME_M3,
# 'm5': mt.TIMEFRAME_M5,
# 'm15': mt.TIMEFRAME_M15,
# 'm20': mt.TIMEFRAME_M20,
# 'm30': mt.TIMEFRAME_M30,
# 'h1': mt.TIMEFRAME_H1,
# }In [29]:
print(trend_dict){'m5': ''}
In [30]:
def get_rates(periode):
global symbol
ohlc = mt.copy_rates_from_pos(symbol, timeframes_dict[periode], 0, 300)
df = pd.DataFrame(ohlc)
df['time']=pd.to_datetime(df['time'], unit='s')
#df = df[["time","open","high","low","close"]]
df["open"] = df.open.astype(float)
df["high"] = df.high.astype(float)
df["low"] = df.low.astype(float)
df["close"] = df.close.astype(float)
## Take the rolling atr so the yaxis doesn't shake too much
df["atr"] = ta.atr(high=df.high, low=df.low, close=df.close)
df["atr"] = df.atr.rolling(window=30).mean()
df.set_index("time", inplace = True)
return dfIn [31]:
get_rates('m5')Out [31]:
| open | high | low | close | tick_volume | spread | real_volume | atr | |
|---|---|---|---|---|---|---|---|---|
| time | ||||||||
| 2025-08-28 22:00:00 | 3420.04 | 3420.35 | 3419.77 | 3419.85 | 1045 | 20 | 0 | NaN |
| 2025-08-28 22:05:00 | 3419.82 | 3420.01 | 3418.74 | 3418.90 | 1057 | 20 | 0 | NaN |
| 2025-08-28 22:10:00 | 3418.93 | 3419.20 | 3418.27 | 3418.62 | 1003 | 20 | 0 | NaN |
| 2025-08-28 22:15:00 | 3418.62 | 3419.81 | 3418.55 | 3419.69 | 1218 | 20 | 0 | NaN |
| 2025-08-28 22:20:00 | 3419.67 | 3420.01 | 3419.15 | 3419.17 | 956 | 20 | 0 | NaN |
| ... | ... | ... | ... | ... | ... | ... | ... | ... |
| 2025-08-29 23:35:00 | 3447.78 | 3448.55 | 3447.18 | 3448.21 | 546 | 20 | 0 | 1.968904 |
| 2025-08-29 23:40:00 | 3448.21 | 3449.35 | 3447.78 | 3449.17 | 421 | 20 | 0 | 1.961315 |
| 2025-08-29 23:45:00 | 3449.15 | 3449.18 | 3447.64 | 3447.90 | 578 | 20 | 0 | 1.955341 |
| 2025-08-29 23:50:00 | 3447.86 | 3448.36 | 3447.68 | 3448.22 | 467 | 20 | 0 | 1.947173 |
| 2025-08-29 23:55:00 | 3448.20 | 3448.91 | 3447.97 | 3448.87 | 228 | 20 | 0 | 1.938756 |
300 rows × 8 columns
In [32]:
def get_trend(period):
global trend_dict, periods_dict, symbol, debug
#current_periods = periods_dict[symbol][0]
df2 = get_rates(period).iloc[-200:]
df2["close_smooth"] = savgol_filter(df2.close, 25, 5)
fig, ax = plt.subplots()
plt.xticks(rotation=-30)
price, = ax.plot(df2.index, df2.close, c='grey', lw=2, alpha=0.5, zorder=5)
price_smooth, = ax.plot(df2.index, df2.close_smooth, c='b', lw=2, zorder=5)
atr = df2.atr.iloc[-1] # all the first atrs are NaN
peaks_idx, _ = find_peaks(df2.close_smooth, distance = 1,
width = 2, prominence=atr)
troughs_idx, _ = find_peaks(-1*df2.close_smooth, distance = 1,
width = 2, prominence=atr)
peaks, = ax.plot(df2.index[peaks_idx], df2.close_smooth.iloc[peaks_idx], \
c="r", linestyle='None', markersize = 10.0, marker = "o", zorder=10)
troughs, = ax.plot(df2.index[troughs_idx], df2.close_smooth.iloc[troughs_idx], \
c="g", linestyle='None', markersize = 10.0, marker = "o", zorder=10)
plt.show()
#print(peaks_idx[-1], troughs_idx[-1])
if peaks_idx[-1] > troughs_idx[-1]:
print("downtrend")
trend_dict[period] = 'downtrend'
else:
print("uptrend")
trend_dict[period] = 'uptrend'
In [33]:
def get_trend_fast(period):
global trend_dict, periods_dict, symbol
df2 = get_rates(period).iloc[-200:]
df2["close_smooth"] = savgol_filter(df2.close, 15, 5) # kleinere Glättung
atr = df2.atr.iloc[-1]
# Weniger strenge Peak-Erkennung
peaks_idx, _ = find_peaks(df2.close_smooth, distance=1, width=2, prominence=atr*0.5) #evtl kleiner 0.5
troughs_idx, _ = find_peaks(-1*df2.close_smooth, distance=1, width=2, prominence=atr*0.5)
# Trend über Peaks/Troughs
if len(peaks_idx) > 0 and len(troughs_idx) > 0:
if peaks_idx[-1] > troughs_idx[-1]:
trend = "downtrend"
else:
trend = "uptrend"
else:
# Falls keine klaren Peaks gefunden wurden → Slope nutzen
slope = df2.close_smooth.diff().iloc[-5:].mean()
trend = "downtrend" if slope < 0 else "uptrend"
trend_dict[period] = trend
return trend
In [57]:
get_trend_fast('m5')Out [57]:
'downtrend'
In [35]:
def set_trend():
global pause_trading, periods_dict, trend_dict
for k,v in trend_dict.items():
get_trend_fast(k)
print(k)
for k,v in reversed(trend_dict.items()):
if v == 'uptrend':
print(k,v)
periods_dict[symbol] = [k]
pause_trading = 0
print(f"Pause Trading: {pause_trading}")
break
elif v == 'downtrend':
periods_dict[symbol] = [k] #['m15']
pause_trading = 1
print(f"Pause Trading: {pause_trading}")In [56]:
set_trend()m5 Pause Trading: 1
In [ ]:
pause_trading, trend_dict, periods_dict, symbols[0]In [36]:
set_trend()m5 m5 uptrend Pause Trading: 0
In [37]:
trend_dict[periods_dict[symbol][0]]Out [37]:
'uptrend'
In [38]:
periods_dict[symbol], pause_tradingOut [38]:
(['m5'], 0)
In [ ]:
from apscheduler.schedulers.background import BackgroundScheduler
import time
scheduler = BackgroundScheduler()
#scheduler.add_job(main, 'date', run_date='2025-03-07 14:29:50')
#scheduler.add_job(decide_order, 'interval', minutes=1) #intervall
#scheduler.add_job(set_symbol, 'interval', minutes=30)
#scheduler.add_job(set_trend, 'interval', minutes=1)
#scheduler.add_job(decide_order, 'interval', minutes=1)
#scheduler.add_job(decide_order, 'cron', year="*", month="*", day_of_week="mon, tue, wed, thu, fri", hour='0-23', minute='*') #cron
#scheduler.add_job(get_buy_sell_signal, 'cron', year="*", month="*", day_of_week="mon, tue, wed, thu, fri", hour='0-23', minute='*/5') #cron
scheduler.add_job(get_m5_trade_signals, 'cron', year="*", month="*", day_of_week="mon, tue, wed, thu, fri", hour='0-23', minute='*/5') #cron
#scheduler.add_job(export_marketview, 'cron', year="*", month='*', day_of_week='mon, tue, wed; thu, fri', hour='8-22', minute=00)
#scheduler.add_job(pause_trading, 'cron', year="*", month="*", day_of_week="mon, tue, wed, thu, fri", hour=22, minute=00) #cron
scheduler.start()
In [ ]:
scheduler.get_jobs()In [ ]:
scheduler.remove_all_jobs()In [ ]:
scheduler.shutdown()In [48]:
def get_mt5_data(symbol=symbol, timeframe=mt.TIMEFRAME_M5, n_bars=500):
rates = mt.copy_rates_from_pos(symbol, timeframe, 0, n_bars)
df = pd.DataFrame(rates)
df["time"] = pd.to_datetime(df["time"], unit="s")
return dfIn [39]:
def get_m5_trade_signal(symbol, atr_mult=1.5):
global trend_dict, periods_dict
# Hole die letzten M5-Daten
df = get_rates('m5').iloc[-200:]
# Smoothen
df["close_smooth_std"] = savgol_filter(df.close, 25, 5)
df["close_smooth_fast"] = savgol_filter(df.close, 15, 5)
atr = df.atr.iloc[-1]
# --- Standard-Trend ---
peaks_std, _ = find_peaks(df.close_smooth_std, distance=1, width=2, prominence=atr)
troughs_std, _ = find_peaks(-df.close_smooth_std, distance=1, width=2, prominence=atr)
if len(peaks_std) > 0 and len(troughs_std) > 0:
if peaks_std[-1] > troughs_std[-1]:
trend_standard = "downtrend"
else:
trend_standard = "uptrend"
else:
trend_standard = "neutral"
# --- Fast-Trend ---
peaks_fast, _ = find_peaks(df.close_smooth_fast, distance=1, width=2, prominence=atr*0.5)
troughs_fast, _ = find_peaks(-df.close_smooth_fast, distance=1, width=2, prominence=atr*0.5)
if len(peaks_fast) > 0 and len(troughs_fast) > 0:
if peaks_fast[-1] > troughs_fast[-1]:
trend_fast = "downtrend"
else:
trend_fast = "uptrend"
else:
slope = df.close_smooth_fast.diff().iloc[-5:].mean()
trend_fast = "downtrend" if slope < 0 else "uptrend"
# --- Kombiniertes Signal ---
signal = 0 # 0 = neutral, 1 = long, -1 = short
stop_loss = None
if trend_standard == "uptrend" and trend_fast == "uptrend":
signal = 1
stop_loss = df.close.iloc[-1] - atr_mult * atr
elif trend_standard == "downtrend" and trend_fast == "downtrend":
signal = -1
stop_loss = df.close.iloc[-1] + atr_mult * atr
# Speichern
trend_dict['m5'] = {"standard": trend_standard, "fast": trend_fast, "signal": signal}
return {
"signal": signal,
"price": df.close.iloc[-1],
"stop_loss": stop_loss,
"trends": trend_dict['m5']
}
In [49]:
def generate_signal(df = get_mt5_data(symbol=symbol, timeframe=timeframes_dict['m5']), confirm_window=3):
"""
Berechnet drei Signalarten:
- fast_signal: schnelle, aggressive Variante
- standard_signal: konservative Basisstrategie
- optimized_signal: zusätzliche Filter (ATR, ADX, strengerer RSI)
"""
# Indikatoren
df["ema21"] = df["close"].ewm(span=3).mean()
df["ema50"] = df["close"].ewm(span=9).mean()
df["rsi9"] = ta.rsi(df["close"], length=9)
df["rsi14"] = ta.rsi(df["close"], length=14)
df["trend"] = savgol_filter(df["close"], 25, 3)
# Zusätzliche Filterindikatoren
df["atr"] = ta.atr(df["high"], df["low"], df["close"], length=14)
df["adx"] = ta.adx(df["high"], df["low"], df["close"], length=14)["ADX_14"]
# Spalten für Signale
df["fast_signal"] = 0
df["standard_signal"] = 0
df["optimized_signal"] = 0
for i in range(1, len(df)):
# -----------------------
# FAST SIGNAL (früh/aggressiv)
# -----------------------
if (
df["ema21"].iloc[i] > df["ema50"].iloc[i]
and df["rsi9"].iloc[i] > 30
):
df.at[i, "fast_signal"] = 1
elif (
df["ema21"].iloc[i] < df["ema50"].iloc[i]
and df["rsi9"].iloc[i] < 60
):
df.at[i, "fast_signal"] = -1
# -----------------------
# STANDARD SIGNAL (konservativ, ursprüngliche Logik)
# -----------------------
if (
df["ema21"].iloc[i] > df["ema50"].iloc[i]
and df["ema21"].iloc[i - 1] <= df["ema50"].iloc[i - 1]
and df["rsi14"].iloc[i] < 70
and df["rsi9"].iloc[i] > 40
and df["trend"].iloc[i] > df["trend"].iloc[i - 1]
):
df.at[i, "standard_signal"] = 1
elif (
df["ema21"].iloc[i] < df["ema50"].iloc[i]
and df["ema21"].iloc[i - 1] >= df["ema50"].iloc[i - 1]
and df["rsi14"].iloc[i] > 40
and df["rsi9"].iloc[i] < 70
and df["trend"].iloc[i] < df["trend"].iloc[i - 1]
):
df.at[i, "standard_signal"] = -1
# -----------------------
# OPTIMIZED SIGNAL (mit ADX + ATR + strengerem RSI)
# -----------------------
if (
df["ema21"].iloc[i] > df["ema50"].iloc[i]
and df["ema21"].iloc[i - 1] <= df["ema50"].iloc[i - 1]
and df["rsi14"].iloc[i] < 65 # strengerer Filter
and df["rsi9"].iloc[i] > 45 # Momentum klarer
and df["adx"].iloc[i] > 20 # Trendstärke vorhanden
and df["atr"].iloc[i] > df["atr"].rolling(50).mean().iloc[i] # Volatilität über Durchschnitt
):
df.at[i, "optimized_signal"] = 1
elif (
df["ema21"].iloc[i] < df["ema50"].iloc[i]
and df["ema21"].iloc[i - 1] >= df["ema50"].iloc[i - 1]
and df["rsi14"].iloc[i] > 35 # strengerer Filter unten
and df["rsi9"].iloc[i] < 55
and df["adx"].iloc[i] > 20
and df["atr"].iloc[i] > df["atr"].rolling(50).mean().iloc[i]
):
df.at[i, "optimized_signal"] = -1
return df
In [58]:
generate_signal()Out [58]:
| time | open | high | low | close | tick_volume | spread | real_volume | ema21 | ema50 | rsi9 | rsi14 | trend | atr | adx | fast_signal | standard_signal | optimized_signal | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 2025-08-28 05:25:00 | 3386.52 | 3386.74 | 3385.71 | 3386.64 | 714 | 20 | 0 | 3386.640000 | 3386.640000 | NaN | NaN | 3385.220875 | NaN | NaN | 0 | 0 | 0 |
| 1 | 2025-08-28 05:30:00 | 3386.75 | 3386.98 | 3385.81 | 3385.85 | 985 | 20 | 0 | 3386.113333 | 3386.201111 | NaN | NaN | 3386.775285 | NaN | NaN | 0 | 0 | 0 |
| 2 | 2025-08-28 05:35:00 | 3385.86 | 3387.38 | 3384.59 | 3387.17 | 1151 | 20 | 0 | 3386.717143 | 3386.598197 | NaN | NaN | 3388.065491 | NaN | NaN | 0 | 0 | 0 |
| 3 | 2025-08-28 05:40:00 | 3387.12 | 3388.36 | 3386.82 | 3388.18 | 798 | 20 | 0 | 3387.497333 | 3387.134038 | NaN | NaN | 3389.112151 | NaN | NaN | 0 | 0 | 0 |
| 4 | 2025-08-28 05:45:00 | 3388.16 | 3390.56 | 3387.94 | 3390.46 | 947 | 20 | 0 | 3389.026452 | 3388.123436 | NaN | NaN | 3389.935928 | NaN | NaN | 0 | 0 | 0 |
| ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... | ... |
| 495 | 2025-08-29 23:40:00 | 3448.21 | 3449.35 | 3447.78 | 3449.17 | 421 | 20 | 0 | 3448.743230 | 3448.456881 | 56.591716 | 56.452748 | 3448.267585 | 2.002968 | 15.874639 | 1 | 0 | 0 |
| 496 | 2025-08-29 23:45:00 | 3449.15 | 3449.18 | 3447.64 | 3447.90 | 578 | 20 | 0 | 3448.321615 | 3448.345505 | 50.150670 | 52.209309 | 3447.941698 | 1.969899 | 15.547755 | -1 | -1 | 0 |
| 497 | 2025-08-29 23:50:00 | 3447.86 | 3448.36 | 3447.68 | 3448.22 | 467 | 20 | 0 | 3448.270808 | 3448.320404 | 51.708684 | 53.164605 | 3447.570685 | 1.877764 | 15.244220 | -1 | 0 | 0 |
| 498 | 2025-08-29 23:55:00 | 3448.20 | 3448.91 | 3447.97 | 3448.87 | 228 | 20 | 0 | 3448.570404 | 3448.430323 | 54.927801 | 55.126747 | 3447.156651 | 1.810780 | 15.232676 | 1 | 0 | 0 |
| 499 | 2025-09-01 01:00:00 | 3444.73 | 3449.27 | 3444.67 | 3445.14 | 245 | 20 | 0 | 3446.855202 | 3447.772258 | 38.401814 | 43.789528 | 3446.701703 | 2.010010 | 14.843679 | -1 | -1 | 0 |
500 rows × 18 columns
In [51]:
pos = mt.positions_total()
if pos > 0:
open_positions = mt.positions_get()
trail_factor = 0.5
entry_price = open_positions[0].price_open
tp_current = open_positions[0].tp
current_price = open_positions[0].price_current
profit = open_positions[0].profit
profit = current_price - entry_price
if profit > 0:
new_tp = current_price - entry_price * trail_factor
if profit < 0:
new_tp = current_price - profit * trail_factor
if new_tp > current_price:
#update tp from open pos
new_tp
print(f"Trailing +TP for BUY {symbol}: {new_tp} CP: {current_price}")
if new_tp < current_price:
#update tp from open pos
new_tp
print(f"Trailing -TP for BUY {symbol}: {new_tp} CP: {current_price}")
#open_positions
In [52]:
def update_trailing_sl_tp(pos, atr, rrr=2.0, atr_mult=1.5, max_retries=2):
"""
Aktualisiert SL und TP für offene Positionen:
- ATR-basiertes Trailing
- Gewinn-stufenweises Nachziehen
- Dynamische Anpassung mit Validierung
- Fallback-Mechanismus bei RETCODE 1016
"""
symbol = pos.symbol
info = mt.symbol_info(symbol)
digits = info.digits
point = info.point
stops_level = info.trade_stops_level * point # Mindestabstand vom Broker
entry_price = pos.price_open
current_tick = mt.symbol_info_tick(symbol)
bid, ask = current_tick.bid, current_tick.ask
current_price = bid if pos.type == 0 else ask
pos_type = pos.type # 0 = BUY, 1 = SELL
# Gewinn in ATR berechnen
if pos_type == 0: # LONG
profit_atr = (current_price - entry_price) / atr
base_sl = current_price - atr_mult * atr
new_sl = max(pos.sl or 0, base_sl)
if profit_atr > 2:
new_sl = max(new_sl, entry_price + 1.5 * atr)
elif profit_atr > 1:
new_sl = max(new_sl, entry_price + 1.0 * atr)
elif profit_atr > 0.5:
new_sl = max(new_sl, entry_price + 0.5 * atr)
new_tp = entry_price + (entry_price - new_sl) * rrr
# Validierung BUY
if new_sl >= bid - stops_level:
new_sl = bid - stops_level
if new_tp <= ask + stops_level:
new_tp = ask + stops_level
else: # SHORT
profit_atr = (entry_price - current_price) / atr
base_sl = current_price + atr_mult * atr
new_sl = min(pos.sl or 999999, base_sl)
if profit_atr > 2:
new_sl = min(new_sl, entry_price - 1.5 * atr)
elif profit_atr > 1:
new_sl = min(new_sl, entry_price - 1.0 * atr)
elif profit_atr > 0.5:
new_sl = min(new_sl, entry_price - 0.5 * atr)
new_tp = entry_price - (new_sl - entry_price) * rrr
# Validierung SELL
if new_sl <= ask + stops_level:
new_sl = ask + stops_level
if new_tp >= bid - stops_level:
new_tp = bid - stops_level
# Runden auf gültige Stellen
new_sl = round(new_sl, digits)
new_tp = round(new_tp, digits)
# --- Nur updaten, wenn sich Werte geändert haben ---
if (pos.sl is None or abs(new_sl - pos.sl) > point) or \
(pos.tp is None or abs(new_tp - pos.tp) > point):
for attempt in range(max_retries):
request = {
"action": mt.TRADE_ACTION_SLTP,
"symbol": symbol,
"sl": new_sl,
"tp": new_tp,
"position": pos.ticket
}
result = mt.order_send(request)
if result.retcode == mt.TRADE_RETCODE_DONE:
print(f"🔄 Updated {symbol} | SL: {new_sl:.5f} | TP: {new_tp:.5f}")
break
elif result.retcode == mt.TRADE_RETCODE_INVALID_STOPS:
# Fallback: Stops korrigieren
print(f"⚠️ RETCODE 1016 (Invalid stops) – Versuch {attempt+1}/{max_retries}")
adjust = 2 * stops_level # mehr Abstand
if pos_type == 0: # BUY
new_sl = bid - adjust
new_tp = ask + adjust
else: # SELL
new_sl = ask + adjust
new_tp = bid - adjust
new_sl = round(new_sl, digits)
new_tp = round(new_tp, digits)
continue # retry
else:
print(f"❌ SL/TP Update Fehler: {result.retcode} ({result.comment})")
break
return {"new_sl": new_sl, "new_tp": new_tp, "profit_atr": profit_atr}
In [ ]:
def get_m5_trade_signals(symbol=symbol, atr_mult=1.5, base_rrr=2.0, atr_min=0.0005, slope_factor=1.5):
"""
M5 Trade Signal mit dynamischem Seitwärtsfilter + dynamischer RRR-Berechnung:
- ATR-Minimum prüft ob Markt volatil genug ist
- Linear Regression ersetzt Savitzky-Golay für Trenddetektion
- adaptive Filterung nach ATR und Trend-Slope
- dynamisches RRR (Chance-Risiko-Verhältnis) auf Basis von ATR + Slope
"""
global trend_dict, periods_dict, pause_trading, volume_dict
set_trend()
# if pause_trading == 1:
# pos = mt.positions_total()
# if pos > 0:
# open_positions = mt.positions_get()
# print(f"⚠️ Trading pausiert, offene Positionen auf {symbol} werden geschlossen...")
# for pos in open_positions:
# market_order(symbol, volume_dict[symbol], "sell")
# return {"signal": 0, "reason": "Trading paused"}
# --- Hole die letzten M5-Daten ---
df = get_rates('m5').iloc[-200:]
# --- ATR Berechnung ---
df["hl"] = df["high"] - df["low"]
df["hc"] = (df["high"] - df["close"].shift()).abs()
df["lc"] = (df["low"] - df["close"].shift()).abs()
df["tr"] = df[["hl","hc","lc"]].max(axis=1)
df["atr"] = df["tr"].rolling(14).mean()
atr = df["atr"].iloc[-1]
if atr < atr_min:
return {"signal": 0, "reason": "ATR too low → sideways"}
# --- Trendrichtung per Linear Regression ---
def linreg_slope(series):
X = np.arange(len(series)).reshape(-1, 1)
y = series.values.reshape(-1, 1)
model = LinearRegression().fit(X, y)
return model.coef_[0][0]
slope_long = linreg_slope(df["close"].iloc[-50:]) # 50 Balken (~4h)
slope_short = linreg_slope(df["close"].iloc[-15:]) # 15 Balken (~1h)
# --- Dynamischer Seitwärtsfilter ---
slope_threshold = slope_factor * atr / df["close"].iloc[-1]
if abs(slope_long) < slope_threshold and abs(slope_short) < slope_threshold:
return {"signal": 0, "reason": "Trend flat → sideways"}
# --- Trendlogik ---
trend_standard = "uptrend" if slope_long > 0 else "downtrend"
trend_fast = "uptrend" if slope_short > 0 else "downtrend"
current_price = df["close"].iloc[-1]
# --- Dynamische RRR-Berechnung ---
atr_norm = atr / current_price # relative Volatilität
slope_strength = abs(slope_short) # Trendstärke aus Regression
rrr_base = 1.2
rrr_from_vol = atr_norm * 1500 # skaliert ATR-Einfluss
rrr_from_slope = slope_strength / slope_threshold # Slope-Einfluss
rrr = rrr_base + rrr_from_vol + rrr_from_slope
rrr = max(1.2, min(rrr, 3.0)) # Begrenzung
print(f"[RRR-DEBUG] ATR_norm={atr_norm:.5f} | slope_strength={slope_strength:.5f} | "
f"rrr_vol={rrr_from_vol:.2f} | rrr_slope={rrr_from_slope:.2f} | FINAL_RRR={rrr:.2f}")
# --- Signale ---
signal, stop_loss, takeprofit = 0, None, None
if trend_standard == "uptrend" and trend_fast == "uptrend":
signal = 1
stop_loss = current_price - atr_mult * atr
takeprofit = current_price + atr_mult * atr * rrr
print(f"BUY {symbol} @ {current_price} | TP: {takeprofit} | SL: {stop_loss}")
market_order(symbol, volume_dict[symbol], "buy", stoploss=stop_loss, take_profit=takeprofit)
elif trend_standard == "downtrend" and trend_fast == "downtrend":
signal = -1
stop_loss = current_price + atr_mult * atr
takeprofit = current_price - atr_mult * atr * rrr
print(f"SELL {symbol} @ {current_price} | TP: {takeprofit} | SL: {stop_loss}")
market_order(symbol, volume_dict[symbol], "sell", stoploss=stop_loss, take_profit=takeprofit)
# --- SL/TP sowohl mit ATR als auch mit Gewinn-Trailing
# manage_open_trades()
open_positions = mt.positions_get(symbol=symbol)
for pos in open_positions:
atr_value = df["atr"].iloc[-1]
update_trailing_sl_tp(pos, atr=atr_value, rrr=rrr, atr_mult=atr_mult)
# --- speichern ---
trend_dict['m5'] = {"standard": trend_standard, "fast": trend_fast, "signal": signal}
return {
"signal": signal,
"price": current_price,
"stop_loss": stop_loss,
"take_profit": takeprofit,
"Risk Reward": rrr,
"trends": trend_dict['m5'],
"atr": atr,
"slope_long": slope_long,
"slope_short": slope_short
}
In [60]:
get_m5_trade_signals()Out [60]:
m5 Pause Trading: 1
{'signal': 0, 'reason': 'Trading paused'}In [ ]: