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

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

1class arLCbot:

1    def __init__(self, L:int, sTrain:float=0.6):

1    def __init__(self, L:int, sTrain:float=0.6):
2        self.L = L
3        self.sTrain = sTrain
4        self.TrainLogReturn = []
5        self.TrainLogLoss = []
6        self.TrainLogRegularization = []

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

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):

1    def ProcessData(self):
2        self.DF['LC'] = np.log(self.DF['Close'] / self.DF['Open'])

1    def ProcessData(self):
2        self.DF['LC'] = np.log(self.DF['Close'] / self.DF['Open'])
3        for i in range(self.L):
4            self.DF[f'LC(t-{i})'] = self.DF['LC'].shift(i)

1    def ProcessData(self):
2        self.DF['LC'] = np.log(self.DF['Close'] / self.DF['Open'])
3        for i in range(self.L):
4            self.DF[f'LC(t-{i})'] = self.DF['LC'].shift(i)
5        self.DF['FP'] = self.DF['Open'].shift(-1)

1    def ProcessData(self):
2        self.DF['LC'] = np.log(self.DF['Close'] / self.DF['Open'])
3        for i in range(self.L):
4            self.DF[f'LC(t-{i})'] = self.DF['LC'].shift(i)
5        self.DF['FP'] = self.DF['Open'].shift(-1)
6        self.DF.dropna(inplace=True)
7        self.nD = len(self.DF)
8        self.nDtr = round(self.sTrain * self.nD)
9        self.nDte = self.nD - self.nDtr
10        self.trDF = self.DF.iloc[:self.nDtr, :]
11        self.teDF = self.DF.iloc[self.nDtr:, :]

1    def Trade(self, TH:float, W:np.ndarray, DF:pd.core.frame.DataFrame):
2        nD = len(DF)

1    def Trade(self, TH:float, W:np.ndarray, DF:pd.core.frame.DataFrame):
2        nD = len(DF)
3        Moneys = np.zeros(nD)
4        Shares = np.zeros(nD)
5        Values = np.zeros(nD)

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

1    def Trade(self, TH:float, W:np.ndarray, DF:pd.core.frame.DataFrame):
2        nD = len(DF)
3        Moneys = np.zeros(nD)
4        Shares = np.zeros(nD)
5        Values = np.zeros(nD)
6        Buys = {'Time':[], 'Price':[]}
7        Sells = {'Time':[], 'Price':[]}
8        X = DF.iloc[:, -self.L - 1:-1].to_numpy()

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

1    def Trade(self, TH:float, W:np.ndarray, DF:pd.core.frame.DataFrame):
2        nD = len(DF)
3        Moneys = np.zeros(nD)
4        Shares = np.zeros(nD)
5        Values = np.zeros(nD)
6        Buys = {'Time':[], 'Price':[]}
7        Sells = {'Time':[], 'Price':[]}
8        X = DF.iloc[:, -self.L - 1:-1].to_numpy()
9        Signals = np.dot(X, W)
10        Signals[np.abs(Signals) < TH] = 0

1    def Trade(self, TH:float, W:np.ndarray, DF:pd.core.frame.DataFrame):
2        nD = len(DF)
3        Moneys = np.zeros(nD)
4        Shares = np.zeros(nD)
5        Values = np.zeros(nD)
6        Buys = {'Time':[], 'Price':[]}
7        Sells = {'Time':[], 'Price':[]}
8        X = DF.iloc[:, -self.L - 1:-1].to_numpy()
9        Signals = np.dot(X, W)
10        Signals[np.abs(Signals) < TH] = 0
11        money = 1
12        share = 0
13        for i in range(nD):

1    def Trade(self, TH:float, W:np.ndarray, DF:pd.core.frame.DataFrame):
2        nD = len(DF)
3        Moneys = np.zeros(nD)
4        Shares = np.zeros(nD)
5        Values = np.zeros(nD)
6        Buys = {'Time':[], 'Price':[]}
7        Sells = {'Time':[], 'Price':[]}
8        X = DF.iloc[:, -self.L - 1:-1].to_numpy()
9        Signals = np.dot(X, W)
10        Signals[np.abs(Signals) < TH] = 0
11        money = 1
12        share = 0
13        for i in range(nD):
14            fp = DF['FP'][i]
15            signal = Signals[i]

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

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

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

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

1    def Loss(self, P:np.ndarray, DF:pd.core.frame.DataFrame, Lambda:float=8e-2):

1    def Loss(self, P:np.ndarray, DF:pd.core.frame.DataFrame, Lambda:float=8e-2):
2        TH = P[0]
3        W = P[1:]

1    def Loss(self, P:np.ndarray, DF:pd.core.frame.DataFrame, Lambda:float=8e-2):
2        TH = P[0]
3        W = P[1:]
4        R = self.Trade(TH, W, DF)[-1]

1    def Loss(self, P:np.ndarray, DF:pd.core.frame.DataFrame, Lambda:float=8e-2):
2        TH = P[0]
3        W = P[1:]
4        R = self.Trade(TH, W, DF)[-1]
5        G = Lambda * np.power(W, 2).sum()
6        L = -R + G

1    def Loss(self, P:np.ndarray, DF:pd.core.frame.DataFrame, Lambda:float=8e-2):
2        TH = P[0]
3        W = P[1:]
4        R = self.Trade(TH, W, DF)[-1]
5        G = Lambda * np.power(W, 2).sum()
6        L = -R + G
7        self.TrainLogReturn.append(R)
8        self.TrainLogLoss.append(L)
9        self.TrainLogRegularization.append(G)
10        return L

1    def Train(self, SS:int, MI:int):

1    def Train(self, SS:int, MI:int):
2        lb = -1 * np.ones(self.L + 1)
3        ub = +1 * np.ones(self.L + 1)

1    def Train(self, SS:int, MI:int):
2        lb = -1 * np.ones(self.L + 1)
3        ub = +1 * np.ones(self.L + 1)
4        lb[0] = 0.005
5        ub[0] = 0.15

1    def Train(self, SS:int, MI:int):
2        lb = -1 * np.ones(self.L + 1)
3        ub = +1 * np.ones(self.L + 1)
4        lb[0] = 0.005
5        ub[0] = 0.15
6        self.P, self.BestLoss = ps.pso(self.Loss,
7                                       lb,
8                                       ub,
9                                       args=(self.trDF, ),
10                                       swarmsize=SS,
11                                       maxiter=MI)

1    def Train(self, SS:int, MI:int):
2        lb = -1 * np.ones(self.L + 1)
3        ub = +1 * np.ones(self.L + 1)
4        lb[0] = 0.005
5        ub[0] = 0.15
6        self.P, self.BestLoss = ps.pso(self.Loss,
7                                       lb,
8                                       ub,
9                                       args=(self.trDF, ),
10                                       swarmsize=SS,
11                                       maxiter=MI)
12        self.TH = self.P[0]
13        self.W = self.P[1:]

1    def Train(self, SS:int, MI:int):
2        lb = -1 * np.ones(self.L + 1)
3        ub = +1 * np.ones(self.L + 1)
4        lb[0] = 0.005
5        ub[0] = 0.15
6        self.P, self.BestLoss = ps.pso(self.Loss,
7                                       lb,
8                                       ub,
9                                       args=(self.trDF, ),
10                                       swarmsize=SS,
11                                       maxiter=MI)
12        self.TH = self.P[0]
13        self.W = self.P[1:]
14        self.BestReturn = self.Trade(self.TH,
15                                     self.W,
16                                     self.trDF)[-1]

1    def Train(self, SS:int, MI:int):
2        lb = -1 * np.ones(self.L + 1)
3        ub = +1 * np.ones(self.L + 1)
4        lb[0] = 0.005
5        ub[0] = 0.15
6        self.P, self.BestLoss = ps.pso(self.Loss,
7                                       lb,
8                                       ub,
9                                       args=(self.trDF, ),
10                                       swarmsize=SS,
11                                       maxiter=MI)
12        self.TH = self.P[0]
13        self.W = self.P[1:]
14        self.BestReturn = self.Trade(self.TH,
15                                     self.W,
16                                     self.trDF)[-1]
17        print('_' * 60)
18        print('Optimization Result:')
19        print(f'\tTH:        {round(self.TH, 4)}')
20        for i in range(self.L):
21            print(f'\tW(Lag={i}): {round(self.P[i + 1], 4)}')
22        print(f'\tReturn:    {round(self.BestReturn, 4)} %')
23        print('_' * 60)

1    def PlotWeights(self):
2        T = np.arange(start=0, stop=self.L, step=1)
3        plt.bar(T,
4                self.W,
5                width=0.4,
6                color='crimson')
7        plt.axhline(lw=1, c='k')
8        plt.title('Weight of Lags')
9        plt.xlabel('Lag')
10        plt.ylabel('Weight')
11        plt.show()

1    def PlotTrainLog(self):
2        plt.plot(self.TrainLogReturn,
3                 lw=0.8,
4                 c='teal',
5                 label='Return')
6        plt.plot(self.TrainLogLoss,
7                 lw=0.8,
8                 c='crimson',
9                 label='Loss')
10        plt.plot(self.TrainLogRegularization,
11                 lw=0.8,
12                 c='lime',
13                 label='Regularization')
14        plt.title('Bot Return, Loss & Regularization Over Training Iterations')
15        plt.xlabel('Function Evaluation')
16        plt.ylabel('Value')
17        plt.legend()
18        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        O = self.Trade(self.TH, self.W, DF)
7        Values, bReturn = O[2], O[7]
8        plt.subplot(1, 2, 1)
9        hReturn = 100 * ((DF['Close'][-1] / DF['Close'][0])**(1 / (len(DF) - 1)) - 1)
10        T = np.arange(start=0, stop=Values.size, step=1)
11        hMeanValue = DF['Close'][0] * (1 + hReturn / 100)**T
12        plt.plot(DF.index,
13                 DF['Close'],
14                 lw=0.8,
15                 c='crimson',
16                 label='Close')
17        plt.plot(DF.index,
18                 hMeanValue,
19                 ls='--',
20                 lw=1,
21                 c='k',
22                 label=f'Mean Values (ADR: {round(hReturn, 4)} %)')
23        plt.title(f'Price Over Time ({Data})')
24        plt.xlabel('Time (Day)')
25        plt.ylabel('Price ($)')
26        plt.yscale('log')
27        plt.legend()
28        plt.subplot(1, 2, 2)
29        bMeanValue = (1 + bReturn / 100)**T
30        plt.plot(DF.index,
31                 Values,
32                 lw=0.8,
33                 c='crimson',
34                 label='Real Values')
35        plt.plot(DF.index,
36                 bMeanValue,
37                 ls='--',
38                 lw=1,
39                 c='k',
40                 label=f'Mean Values (ADR: {round(bReturn, 4)} %)')
41        plt.title(f'Value Over Time ({Data})')
42        plt.xlabel('Time (Day)')
43        plt.ylabel('Value')
44        plt.yscale('log')
45        plt.legend()
46        plt.show()

1hReturn = 100 * ((DF['Close'][-1] / DF['Close'][0])**(1 / (len(DF) - 1)) - 1)

1    def PlotSignal(self, Data:str):
2        if Data == 'Train':
3            DF = self.trDF
4        elif Data == 'Test':
5            DF = self.teDF
6        O = self.Trade(self.TH, self.W, DF)
7        Signals, Buys, Sells = O[3], O[4], O[5]
8        plt.subplot(3, 1, (1, 2))
9        plt.plot(DF.index,
10                 DF['Close'].to_numpy(),
11                 lw=0.7,
12                 c='k',
13                 label='Close')
14        plt.scatter(DF.index[Buys['Time']],
15                    Buys['Price'],
16                    s=24,
17                    c='lime',
18                    marker='o',
19                    label='Buy')
20        plt.scatter(DF.index[Sells['Time']],
21                    Sells['Price'],
22                    s=24,
23                    c='crimson',
24                    marker='o',
25                    label='Sell')
26        plt.title(f'Price Over Time ({Data})')
27        plt.ylabel('Price ($)')
28        plt.yscale('log')
29        plt.tight_layout()
30        plt.legend()
31        plt.subplot(3, 1, 3)
32        Z = np.zeros_like(Signals)
33        plt.plot(DF.index,
34                 Signals,
35                 lw=0.5,
36                 c='k')
37        plt.fill_between(DF.index,
38                         Signals,
39                         Z,
40                         where=(Signals > Z),
41                         color='lime',
42                         alpha=0.9,
43                         label='Buy Signal')
44        plt.fill_between(DF.index,
45                         Signals,
46                         Z,
47                         where=(Signals < Z),
48                         color='crimson',
49                         alpha=0.9,
50                         label='Sell Signal')
51        plt.axhline(lw=0.8, c='b')
52        plt.title(f'Signal Over Time ({Data})')
53        plt.xlabel('Time (Day)')
54        plt.ylabel('Value')
55        plt.tight_layout()
56        plt.legend()
57        plt.show()

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

1Trader = arLCbot(10)

1Trader.GetData('BTC-USD', '2018-01-01', '2022-05-01')

1Trader.ProcessData()

1Trader.Train(SS=40, MI=15)

1Trader.PlotWeights()

1Trader.PlotTrainLog()

1Trader.PlotValue('Train')

1Trader.PlotSignal('Train')

1trWR = Trader.Trade(Trader.TH,
2                    Trader.W,
3                    Trader.trDF)[-2]
4teWR = Trader.Trade(Trader.TH,
5                    Trader.W,
6                    Trader.teDF)[-2]

1print(f'Train Win Rate: {round(trWR, 2)} %')
2print(f'Test Win Rate:  {round(teWR, 2)} %')