Files
Place-Order-Trading-Bot/detecting_price_trends.ipynb
T
2025-05-27 13:23:26 +02:00

173 KiB

Import Libaries

In [1]:
import pandas as pd
import pandas_ta as ta
import MetaTrader5 as mt
import keyring as kr

Login

In [2]:
# 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 [2]:
True
In [3]:
project = "trading-" + server[-4::1]
project = project.lower()
project
Out [3]:
'trading-demo'
In [4]:
symbols = ['XAUUSD']
In [5]:
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 [6]:
symbol = 'XAUUSD'
In [ ]:
In [107]:
def get_rates():

    ohlc = mt.copy_rates_from_pos(symbol, mt.TIMEFRAME_M5, 0, 200)
    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 df

Plotting Price Data

In [10]:
%matplotlib inline

import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
from scipy.signal import savgol_filter
from scipy.signal import find_peaks
from IPython import display
from IPython.display import HTML
pd.set_option('mode.chained_assignment', None)
In [20]:
#df2 = df.iloc[0:500]
df2 = df

fig, ax = plt.subplots()
plt.xticks(rotation=-30)
price, = ax.plot(df2.index, df2.close, c='grey', lw=2, alpha=0.5, zorder=5)

plt.show()
In [15]:
#df2 = df.iloc[0:500]
df2 = df

df2["close_smooth"] = savgol_filter(df2.close, 49, 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)

plt.show()

Detecting Extrema

In [112]:
#df2 = df.iloc[0:500]
df2 = get_rates()

df2["close_smooth"] = savgol_filter(df2.close, 49, 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 = 15, 
        width = 3, prominence=atr)

troughs_idx, _ = find_peaks(-1*df2.close_smooth, distance = 15, 
        width = 3, 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")
else:
    print("uptrend")
downtrend

Finding Runs

In [89]:
df2 = df.iloc[0:100]
#df2 = df.tail(50)

df2["close_smooth"] = savgol_filter(df2.close, 49, 5)

fig, ax = plt.subplots()
plt.xticks(rotation=-30)
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 = 15, 
        width = 3, prominence=atr)

troughs_idx, _ = find_peaks(-1*df2.close_smooth, distance = 15, 
        width = 3, 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)


print(peaks_idx, troughs_idx, peaks, troughs)

up_run_length = 0
up_run = True
down_run_length = 0
while up_run:
    if 2 + up_run_length > len(peaks_idx) or 2 + up_run_length > len(troughs_idx):
        break
    if df2.close_smooth.iloc[peaks_idx[-1 - up_run_length]] > df2.close_smooth.iloc[peaks_idx[-2 - up_run_length]] and \
       df2.close_smooth.iloc[troughs_idx[-1 - up_run_length]] > df2.close_smooth.iloc[troughs_idx[-2 - up_run_length]]:
        up_run_length += 1
    else:
        up_run = False 

# down_run_length = 0
# down_run = True
# while down_run:
#     if 2 + down_run_length > len(peaks_idx) or 2 + down_run_length > len(troughs_idx):
#         break
        
#     if df2.close_smooth.iloc[peaks_idx[-1 - down_run_length]] <  df2.close_smooth.iloc[peaks_idx[-2 - down_run_length]] and \
#         df2.close_smooth.iloc[troughs_idx[-1 - down_run_length]] <  + df2.close_smooth.iloc[troughs_idx[-2 - down_run_length]]:
#         down_run_length += 1
#     else:
#         down_run = False
        

if up_run_length > 0:
    ax.set_facecolor((150/255, 255/255, 159/255, 0.3))
    pause_trading = 0
else:# down_run_length > 0:
    ax.set_facecolor((255/255, 255/255, 80/255, 0.3))
    pause_trading = 1
        
print(pause_trading, up_run_length)

plt.show()
[21] [85] Line2D(_child2) Line2D(_child3)
1 0
In [55]:
#if df2.close_smooth.iloc[peaks_idx[-1 - down_run_length]] <  df2.close_smooth.iloc[peaks_idx[-2 - down_run_length]] :
#\
if df2.close_smooth.iloc[troughs_idx[-1 - down_run_length]] <  df2.close_smooth.iloc[troughs_idx[-1 - down_run_length]]:
    print(True)
In [ ]:
!pip install ffmpeg-python  
In [ ]:
In [31]:
import matplotlib
matplotlib.rcParams['animation.embed_limit'] = 2**128

bars_in_frame = 300


fig, ax = plt.subplots()

# increase video quality
#fig, ax = plt.subplots(figsize=(8,8), dpi=300)

price, = ax.plot([], c='grey', lw=3, alpha=0.5, zorder=5)
price_smooth, = ax.plot([], c='b', lw=5, zorder=5)
peaks, = ax.plot([], c="r", linestyle='None', markersize = 15.0, marker = "o", zorder=10)
troughs, = ax.plot([], c="g", linestyle='None', markersize = 15.0, marker = "o", zorder=10)


## Might turn the yaxis off, can be annoying
ax.set_ylim(15000,20000)
ax.set_xlim(0,bars_in_frame)

def animate(frame):
    
    frames_behind = 30

    df2 = df.iloc[-len(df) + frame + 30:frame+bars_in_frame + 30]
    df2["close_smooth"] = savgol_filter(df2.close, 49, 5)

    df2 = df2.iloc[frames_behind:]
    
    x_coords = [x for x in range(len(df2.close_smooth))]
    
    price.set_data((x_coords, df2.close))
    price_smooth.set_data((x_coords, df2.close_smooth))
    
    first_atr = df2.atr.iloc[0]
    
    ax.set_ylim(df2.close_smooth.min() - 10*first_atr, df2.close_smooth.max() + 10*first_atr)
    
    peaks_idx, _ = find_peaks(df2.close_smooth, distance = 15, 
                width = 3, prominence=first_atr)

    troughs_idx, _ = find_peaks(-1*df2.close_smooth, distance = 15, 
                width = 3, prominence=first_atr)
    
    up_run_length = 0
    up_run = True
    while up_run:
        if 2 + up_run_length > len(peaks_idx) or 2 + up_run_length > len(troughs_idx):
            break

        if df2.close_smooth.iloc[peaks_idx[-1 - up_run_length]] >  df2.close_smooth.iloc[peaks_idx[-2 - up_run_length]] and \
           df2.close_smooth.iloc[troughs_idx[-1 - up_run_length]] >  df2.close_smooth.iloc[troughs_idx[-2 - up_run_length]]:
            up_run_length += 1
        else:
            up_run = False

    down_run_length = 0
    down_run = True
    while down_run:
        if 2 + down_run_length > len(peaks_idx) or 2 + down_run_length > len(troughs_idx):
            break
            
        if df2.close_smooth.iloc[peaks_idx[-1 - down_run_length]] <  df2.close_smooth.iloc[peaks_idx[-2 - down_run_length]] and \
           df2.close_smooth.iloc[troughs_idx[-1 - down_run_length]] <  + df2.close_smooth.iloc[troughs_idx[-2 - down_run_length]]:
            down_run_length += 1
        else:
            down_run = False
            
    peaks.set_data((peaks_idx, df2.close_smooth.iloc[peaks_idx]))
    troughs.set_data((troughs_idx, df2.close_smooth.iloc[troughs_idx]))
        
    if up_run_length > 0:
        
        ax.set_facecolor((150/255, 255/255, 159/255, 0.3))
    elif down_run_length > 0:
        ax.set_facecolor((255/255, 150/255, 150/255, 0.3))
    else:
        ax.set_facecolor("white")
    
    return price, price_smooth, peaks, troughs


anim = FuncAnimation(fig, animate, frames=len(df)-bars_in_frame, interval=40, blit=True)
video = HTML(anim.to_html5_video())
display.display(video)
c:\ProgramData\anaconda3\Lib\site-packages\matplotlib\animation.py:1740: UserWarning: Can not start iterating the frames for the initial draw. This can be caused by passing in a 0 length sequence for *frames*.

If you passed *frames* as a generator it may be exhausted due to a previous display or save.
  warnings.warn(
---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
Cell In[31], line 85
     81     return price, price_smooth, peaks, troughs
     84 anim = FuncAnimation(fig, animate, frames=len(df)-bars_in_frame, interval=40, blit=True)
---> 85 video = HTML(anim.to_html5_video())
     86 display.display(video)

File c:\ProgramData\anaconda3\Lib\site-packages\matplotlib\animation.py:1284, in Animation.to_html5_video(self, embed_limit)
   1281 path = Path(tmpdir, "temp.m4v")
   1282 # We create a writer manually so that we can get the
   1283 # appropriate size for the tag
-> 1284 Writer = writers[mpl.rcParams['animation.writer']]
   1285 writer = Writer(codec='h264',
   1286                 bitrate=mpl.rcParams['animation.bitrate'],
   1287                 fps=1000. / self._interval)
   1288 self.save(str(path), writer=writer)

File c:\ProgramData\anaconda3\Lib\site-packages\matplotlib\animation.py:148, in MovieWriterRegistry.__getitem__(self, name)
    146 if self.is_available(name):
    147     return self._registered[name]
--> 148 raise RuntimeError(f"Requested MovieWriter ({name}) not available")

RuntimeError: Requested MovieWriter (ffmpeg) not available

Saving our animation

In [ ]:
with open('video-hd.html', 'w') as f:
    f.write(video.data)
In [ ]: