ساخت ربات معامله گر رمزارز با استفاده از اسیلاتور استوکستیک — پیاده سازی در پایتون

1import numpy as np
2import pandas as pd
3import pandas_datareader as pdt
4import matplotlib.pyplot as plt

1np.random.seed(0)
2plt.style.use('ggplot')

1class stcBot:

1    def __init__(self, Ld:int=5, MinL:int=2, MaxL:int=200, sTrain:float=0.6):

1    def __init__(self, Ld:int=4, MinL:int=2, MaxL:int=200, sTrain:float=0.6):
2        self.Ld = Ld
3        self.MinL = MinL
4        self.MaxL = MaxL
5        self.sTrain = sTrain

1    def GetData(self, Ticker:str, Start:str, End:str):

1    def GetData(self, Ticker:str, Start:str, End:str):
2        self.Ticker = Ticker
3        self.Start = Start
4        self.End = End
5        self.DF = pdt.DataReader(Ticker,
6                                 data_source='yahoo',
7                                 start=Start,
8                                 end=End)

1    def GetData(self, Ticker:str, Start:str, End:str):
2        self.Ticker = Ticker
3        self.Start = Start
4        self.End = End
5        self.DF = pdt.DataReader(Ticker,
6                                 data_source='yahoo',
7                                 start=Start,
8                                 end=End)
9        self.DF.drop(labels=['Volume', 'Adj Close'],
10                     axis=1,
11                     inplace=True)

1    def ProcessData(self):
2        for L in range(self.MinL, self.MaxL+1):

1    def ProcessData(self):
2        for L in range(self.MinL, self.MaxL+1):
3            self.DF['LL'] = self.DF['Low'].rolling(L).min()
4            self.DF['HH'] = self.DF['High'].rolling(L).max()

1    def ProcessData(self):
2        for L in range(self.MinL, self.MaxL+1):
3            self.DF['LL'] = self.DF['Low'].rolling(L).min()
4            self.DF['HH'] = self.DF['High'].rolling(L).max()
5            self.DF['K'] = (self.DF['Close'] - self.DF['LL']) / (self.DF['HH'] - self.DF['LL'])
6            self.DF['D'] = self.DF[f'K'].rolling(self.Ld).mean()
7            self.DF[f'Signal({L})'] = self.DF['K'] - self.DF['D']

1    def ProcessData(self):
2        for L in range(self.MinL, self.MaxL+1):
3            self.DF['LL'] = self.DF['Low'].rolling(L).min()
4            self.DF['HH'] = self.DF['High'].rolling(L).max()
5            self.DF['K'] = (self.DF['Close'] - self.DF['LL']) / (self.DF['HH'] - self.DF['LL'])
6            self.DF['D'] = self.DF[f'K'].rolling(self.Ld).mean()
7            self.DF[f'Signal({L})'] = self.DF['K'] - self.DF['D']
8            self.DF.drop(['LL', 'HH', 'K', 'D'], axis=1, inplace=True)

1    def ProcessData(self):
2        for L in range(self.MinL, self.MaxL+1):
3            self.DF['LL'] = self.DF['Low'].rolling(L).min()
4            self.DF['HH'] = self.DF['High'].rolling(L).max()
5            self.DF['K'] = (self.DF['Close'] - self.DF['LL']) / (self.DF['HH'] - self.DF['LL'])
6            self.DF['D'] = self.DF[f'K'].rolling(self.Ld).mean()
7            self.DF[f'Signal({L})'] = self.DF['K'] - self.DF['D']
8            self.DF.drop(['LL', 'HH', 'K', 'D'], axis=1, inplace=True)
9        self.DF['FP'] = self.DF['Open'].shift(-1)

1    def ProcessData(self):
2        for L in range(self.MinL, self.MaxL+1):
3            self.DF['LL'] = self.DF['Low'].rolling(L).min()
4            self.DF['HH'] = self.DF['High'].rolling(L).max()
5            self.DF['K'] = (self.DF['Close'] - self.DF['LL']) / (self.DF['HH'] - self.DF['LL'])
6            self.DF['D'] = self.DF[f'K'].rolling(self.Ld).mean()
7            self.DF[f'Signal({L})'] = self.DF['K'] - self.DF['D']
8            self.DF.drop(['LL', 'HH', 'K', 'D'], axis=1, inplace=True)
9        self.DF['FP'] = self.DF['Open'].shift(-1)
10        self.DF.dropna(inplace=True)

1    def ProcessData(self):
2        for L in range(self.MinL, self.MaxL+1):
3            self.DF['LL'] = self.DF['Low'].rolling(L).min()
4            self.DF['HH'] = self.DF['High'].rolling(L).max()
5            self.DF['K'] = (self.DF['Close'] - self.DF['LL']) / (self.DF['HH'] - self.DF['LL'])
6            self.DF['D'] = self.DF[f'K'].rolling(self.Ld).mean()
7            self.DF[f'Signal({L})'] = self.DF['K'] - self.DF['D']
8            self.DF.drop(['LL', 'HH', 'K', 'D'], axis=1, inplace=True)
9        self.DF['FP'] = self.DF['Open'].shift(-1)
10        self.DF.dropna(inplace=True)
11        self.nD = len(self.DF)
12        self.nDtr = round(self.sTrain * self.nD)
13        self.nDte = self.nD - self.nDtr
14        self.trDF = self.DF.iloc[:self.nDtr, :]
15        self.teDF = self.DF.iloc[self.nDtr:, :]

1    def Trade(self, DF:pd.core.frame.DataFrame, L:int):

1    def Trade(self, DF:pd.core.frame.DataFrame, L:int):
2        nD = len(DF)
3        Moneys = np.zeros(nD)
4        Shares = np.zeros(nD)
5        Values = np.zeros(nD)
6        Signals = np.zeros(nD)
7        Buys = {'Time':[], 'Price':[]}
8        Sells = {'Time':[], 'Price':[]}

1    def Trade(self, DF:pd.core.frame.DataFrame, L:int):
2        nD = len(DF)
3        Moneys = np.zeros(nD)
4        Shares = np.zeros(nD)
5        Values = np.zeros(nD)
6        Signals = np.zeros(nD)
7        Buys = {'Time':[], 'Price':[]}
8        Sells = {'Time':[], 'Price':[]}
9        money = 1
10        share = 0

1    def Trade(self, DF:pd.core.frame.DataFrame, L:int):
2        nD = len(DF)
3        Moneys = np.zeros(nD)
4        Shares = np.zeros(nD)
5        Values = np.zeros(nD)
6        Signals = np.zeros(nD)
7        Buys = {'Time':[], 'Price':[]}
8        Sells = {'Time':[], 'Price':[]}
9        money = 1
10        share = 0
11        for i in range(nD):
12            fp = DF['FP'][i]
13            signal = DF[f'Signal({L})'][i]

1    def Trade(self, DF:pd.core.frame.DataFrame, L:int):
2        nD = len(DF)
3        Moneys = np.zeros(nD)
4        Shares = np.zeros(nD)
5        Values = np.zeros(nD)
6        Signals = np.zeros(nD)
7        Buys = {'Time':[], 'Price':[]}
8        Sells = {'Time':[], 'Price':[]}
9        money = 1
10        share = 0
11        for i in range(nD):
12            fp = DF['FP'][i]
13            signal = DF[f'Signal({L})'][i]
14            if signal > 0 and share == 0:
15                share = money / fp
16                money = 0
17                Buys['Time'].append(i)
18                Buys['Price'].append(fp)
19            elif signal < 0 and share > 0:
20                money = share * fp
21                share = 0
22                Sells['Time'].append(i)
23                Sells['Price'].append(fp)

1    def Trade(self, DF:pd.core.frame.DataFrame, L:int):
2        nD = len(DF)
3        Moneys = np.zeros(nD)
4        Shares = np.zeros(nD)
5        Values = np.zeros(nD)
6        Signals = np.zeros(nD)
7        Buys = {'Time':[], 'Price':[]}
8        Sells = {'Time':[], 'Price':[]}
9        money = 1
10        share = 0
11        for i in range(nD):
12            fp = DF['FP'][i]
13            signal = DF[f'Signal({L})'][i]
14            if signal > 0 and share == 0:
15                share = money / fp
16                money = 0
17                Buys['Time'].append(i)
18                Buys['Price'].append(fp)
19            elif signal < 0 and share > 0:
20                money = share * fp
21                share = 0
22                Sells['Time'].append(i)
23                Sells['Price'].append(fp)
24            Moneys[i] = money
25            Shares[i] = share
26            Values[i] = money + share * fp
27            Signals[i] = signal
28        Return = 100 * ((Values[-1] / Values[0])**(1 / (nD - 1)) - 1)
29        return Moneys, Shares, Values, Signals, Buys, Sells, Return

1    def Train(self):
2        self.Ls = np.arange(start=self.MinL, stop=self.MaxL + 1, step=1)
3        self.Rs = []

1    def Train(self):
2        self.Ls = np.arange(start=self.MinL, stop=self.MaxL + 1, step=1)
3        self.Rs = []
4        for L in self.Ls:
5            R = self.Trade(self.trDF, L)[-1]
6            self.Rs.append(R)

1    def Train(self):
2        self.Ls = np.arange(start=self.MinL, stop=self.MaxL + 1, step=1)
3        self.Rs = []
4        for L in self.Ls:
5            R = self.Trade(self.trDF, L)[-1]
6            self.Rs.append(R)
7        self.Rs = np.array(self.Rs)
8        self.BestReturn = self.Rs.max()
9        self.L = self.Ls[self.Rs.argmax()]
10        print('_' * 50)
11        print('Optimization Result:')
12        print(f'\tBest L: {self.L}')
13        print(f'\tBest Return: {self.BestReturn} %')
14        print('_' * 50)

1    def PlotRs(self):
2        plt.plot(self.Ls, self.Rs, ls='-', lw=0.8, c='teal', marker='o', ms=3, label='Points')
3        plt.scatter(self.L, self.BestReturn, s=50, marker='o', color='crimson', label='Best')
4        plt.title('Bot Return for Different Values of L')
5        plt.xlabel('L')
6        plt.ylabel('Average Daily Return (%)')
7        plt.legend()
8        plt.show()

1    def PlotValue(self, Data:str):
2        if Data == 'Train':
3            DF = self.trDF
4        elif Data == 'Test':
5            DF = self.teDF
6        _, _, Values, _, _, _, bReturn = self.Trade(DF, self.L)

1    def PlotValue(self, Data:str):
2        if Data == 'Train':
3            DF = self.trDF
4        elif Data == 'Test':
5            DF = self.teDF
6        _, _, Values, _, _, _, bReturn = self.Trade(DF, self.L)
7        plt.subplot(1, 2, 1)
8        hReturn = 100 * ((DF['Close'][-1] / DF['Close'][0])**(1 / (len(DF) - 1)) - 1)
9        hMeanValue = DF['Close'][0] * (1 + hReturn / 100)**np.arange(start=0, stop=Values.size, step=1)
10        plt.plot(DF.index, DF['Close'], ls='-', lw=0.8, c='crimson', label='Close')
11        plt.plot(DF.index, hMeanValue, ls='--', lw=1, c='k', label=f'Mean Values (ADR: {round(hReturn, 4)} %)')
12        plt.title(f'Price Over Time ({Data})')
13        plt.xlabel('Time (Day)')
14        plt.ylabel('Price ($)')
15        plt.yscale('log')
16        plt.legend()
17        plt.subplot(1, 2, 2)
18        bMeanValue = (1 + bReturn / 100)**np.arange(start=0, stop=Values.size, step=1)
19        plt.plot(DF.index, Values, ls='-', lw=0.8, c='crimson', label='Real Values')
20        plt.plot(DF.index, bMeanValue, ls='--', lw=1, c='k', label=f'Mean Values (ADR: {round(bReturn, 4)} %)')
21        plt.title(f'Value Over Time ({Data})')
22        plt.xlabel('Time (Day)')
23        plt.ylabel('Value')
24        plt.yscale('log')
25        plt.legend()
26        plt.show()

1    def PlotSignal(self, Data:str):
2        if Data == 'Train':
3            DF = self.trDF
4        elif Data == 'Test':
5            DF = self.teDF
6        _, _, _, Signals, Buys, Sells, _ = self.Trade(DF, self.L)
7        plt.subplot(3, 1, (1, 2))
8        plt.plot(DF.index, DF['Close'].to_numpy(), ls='-', lw=0.7, c='k', label='Close')
9        plt.scatter(DF.index[Buys['Time']], Buys['Price'], s=24, c='lime', marker='o', label='Buy')
10        plt.scatter(DF.index[Sells['Time']], Sells['Price'], s=24, c='crimson', marker='o', label='Sell')
11        plt.title(f'Price Over Time ({Data})')
12        plt.ylabel('Price ($)')
13        plt.yscale('log')
14        plt.tight_layout()
15        plt.legend()
16        plt.subplot(3, 1, 3)
17        Z = np.zeros_like(Signals)
18        plt.plot(DF.index, Signals, ls='-', lw=0.7, c='k')
19        plt.fill_between(DF.index, Signals, Z, where=(Signals > Z), color='lime', alpha=0.9, label='Buy Signal')
20        plt.fill_between(DF.index, Signals, Z, where=(Signals < Z), color='crimson', alpha=0.9, label='Sell Signal')
21        plt.axhline(ls='-', lw=0.8, c='b')
22        plt.title(f'Signal Over Time ({Data})')
23        plt.xlabel('Time (Day)')
24        plt.ylabel('Value')
25        plt.tight_layout()
26        plt.legend()
27        plt.show()

1Trader = stcBot()

1Trader.GetData('BTC-USD', '2009-01-01', '2022-04-24')
2
3Trader.ProcessData()

1Trader.Train()

1Optimization Result:
2        Best L: 80
3        Best Return: 0.2914800215035429 %

1Trader.PlotRs()

1Trader.PlotValue('Train')
2Trader.PlotValue('Test')

1Trader.PlotSignal('Train')
2Trader.PlotSignal('Test')

1def WR(Buys:dict, Sells:dict):

1def WR(Buys:dict, Sells:dict):
2    nT = 0
3    nW = 0

1def WR(Buys:dict, Sells:dict):
2    nT = 0
3    nW = 0
4    for b, s in zip(Buys['Price'], Sells['Price']):
5        nT += 1
6        if s > b:
7            nW += 1

1def WR(Buys:dict, Sells:dict):
2    nT = 0
3    nW = 0
4    for b, s in zip(Buys['Price'], Sells['Price']):
5        nT += 1
6        if s > b:
7            nW += 1
8    wr = 100 * nW / nT
9    return wr

1    def Trade(self, DF:pd.core.frame.DataFrame, L:int):
2        nD = len(DF)
3        Moneys = np.zeros(nD)
4        Shares = np.zeros(nD)
5        Values = np.zeros(nD)
6        Signals = np.zeros(nD)
7        Buys = {'Time':[], 'Price':[]}
8        Sells = {'Time':[], 'Price':[]}
9        money = 1
10        share = 0
11        for i in range(nD):
12            fp = DF['FP'][i]
13            signal = DF[f'Signal({L})'][i]
14            if signal > 0 and share == 0:
15                share = money / fp
16                money = 0
17                Buys['Time'].append(i)
18                Buys['Price'].append(fp)
19            elif signal < 0 and share > 0:
20                money = share * fp
21                share = 0
22                Sells['Time'].append(i)
23                Sells['Price'].append(fp)
24            Moneys[i] = money
25            Shares[i] = share
26            Values[i] = money + share * fp
27            Signals[i] = signal
28        Return = 100 * ((Values[-1] / Values[0])**(1 / (nD - 1)) - 1)
29        wr = WR(Buys, Sells)
30        return Moneys, Shares, Values, Signals, Buys, Sells, wr, Return

1_, _, _, _, _, _, trWR, _ = Trader.Trade(Trader.trDF, Trader.L)
2_, _, _, _, _, _, teWR, _ = Trader.Trade(Trader.teDF, Trader.L)
3
4print(f'Train Win Rate: {round(trWR, 2)} %')
5print(f'Test Win Rate:  {round(teWR, 2)} %')