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

1import numpy as np
2import pandas as pd
3import pyswarm as ps
4import pandas_datareader as pdt
5import matplotlib.pyplot as plt
6
7class smaLR:
8    def __init__(self):
9        self.TrainLog = []
10    def GetData(self, Ticker:str, Start:str, End:str):
11        self.Ticker = Ticker
12        self.Start = Start
13        self.End = End
14        self.DF = pdt.DataReader(Ticker,
15                                 data_source='yahoo',
16                                 start=Start,
17                                 end=End)
18        self.DF.drop(labels=['High', 'Low', 'Volume', 'Adj Close'],
19                     axis=1,
20                     inplace=True)
21    def ProcessData(self, L:int):
22        self.L = L
23        self.DF[f'SMA({L})'] = self.DF['Close'].rolling(L).mean()
24        self.DF[f'Log-Ratio({L})'] = np.log(self.DF['Close'] / self.DF[f'SMA({L})'])
25        self.DF['FP'] = self.DF['Open'].shift(-1)
26        self.DF.dropna(inplace=True)
27        self.nD = len(self.DF)
28    def Trade(self, Parameters:np.ndarray):
29        Moneys = np.zeros(self.nD)
30        Shares = np.zeros(self.nD)
31        Values = np.zeros(self.nD)
32        Signals = np.zeros(self.nD)
33        Buys = {'Time':[], 'Price':[]}
34        Sells = {'Time':[], 'Price':[]}
35        money = 1
36        share = 0
37        for i in range(self.nD):
38            fp = self.DF['FP'][i]
39            lr = self.DF[f'Log-Ratio({self.L})'][i]
40            signal = Parameters[0] + Parameters[1] * lr
41            if signal > 0 and share == 0:
42                share = money / fp
43                money = 0
44                Buys['Time'].append(i)
45                Buys['Price'].append(fp)
46            elif signal < 0 and share > 0:
47                money = share * fp
48                share = 0
49                Sells['Time'].append(i)
50                Sells['Price'].append(fp)
51            Moneys[i] = money
52            Shares[i] = share
53            Values[i] = money + share * fp
54            Signals[i] = signal
55        Return = 100 * ((Values[-1] / Values[0])**(1 / (self.nD - 1)) - 1)
56        return Moneys, Shares, Values, Signals, Buys, Sells, Return
57    def Loss(self, P:np.ndarray):
58        Return = self.Trade(P)[-1]
59        self.TrainLog.append(Return)
60        return -Return
61    def Train(self, MaxIteration:int, SwarmSize:int):
62        lb = -1 * np.ones(2)
63        ub = +1 * np.ones(2)
64        self.P, BestLoss = ps.pso(self.Loss,
65                                  lb,
66                                  ub,
67                                  swarmsize=SwarmSize,
68                                  maxiter=MaxIteration)
69        BestReturn = -BestLoss
70        print('_'*50)
71        print('Optimization Result:')
72        print(f'\tBest Parameters: {self.P}')
73        print(f'\tBest Return: {BestReturn} %')
74        print('_'*50)

1    def __init__(self, sTrain:float=0.8):

1    def __init__(self, sTrain:float=0.8):
2        self.TrainLog = []
3        self.sTrain = sTrain

1    def ProcessData(self, L:int):
2        self.L = L
3        self.DF[f'SMA({L})'] = self.DF['Close'].rolling(L).mean()
4        self.DF[f'Log-Ratio({L})'] = np.log(self.DF['Close'] / self.DF[f'SMA({L})'])
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

1    def ProcessData(self, L:int):
2        self.L = L
3        self.DF[f'SMA({L})'] = self.DF['Close'].rolling(L).mean()
4        self.DF[f'Log-Ratio({L})'] = np.log(self.DF['Close'] / self.DF[f'SMA({L})'])
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, Parameters:np.ndarray, DF:pd.core.frame.DataFrame):

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

1    def Trade(self, Parameters: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        Signals = np.zeros(nD)
7        Buys = {'Time':[], 'Price':[]}
8        Sells = {'Time':[], 'Price':[]}

1    def Trade(self, Parameters: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        Signals = np.zeros(nD)
7        Buys = {'Time':[], 'Price':[]}
8        Sells = {'Time':[], 'Price':[]}
9        money = 1
10        share = 0
11        for i in range(nD):

1    def Trade(self, Parameters: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        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            lr = DF[f'Log-Ratio({self.L})'][i]
14            signal = Parameters[0] + Parameters[1] * lr

1    def Trade(self, Parameters: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        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            lr = DF[f'Log-Ratio({self.L})'][i]
14            signal = Parameters[0] + Parameters[1] * lr
15            if signal > 0 and share == 0:
16                share = money / fp
17                money = 0
18                Buys['Time'].append(i)
19                Buys['Price'].append(fp)
20            elif signal < 0 and share > 0:
21                money = share * fp
22                share = 0
23                Sells['Time'].append(i)
24                Sells['Price'].append(fp)
25            Moneys[i] = money
26            Shares[i] = share
27            Values[i] = money + share * fp
28            Signals[i] = signal

1    def Trade(self, Parameters: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        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            lr = DF[f'Log-Ratio({self.L})'][i]
14            signal = Parameters[0] + Parameters[1] * lr
15            if signal > 0 and share == 0:
16                share = money / fp
17                money = 0
18                Buys['Time'].append(i)
19                Buys['Price'].append(fp)
20            elif signal < 0 and share > 0:
21                money = share * fp
22                share = 0
23                Sells['Time'].append(i)
24                Sells['Price'].append(fp)
25            Moneys[i] = money
26            Shares[i] = share
27            Values[i] = money + share * fp
28            Signals[i] = signal
29        Return = 100 * ((Values[-1] / Values[0])**(1 / (nD - 1)) - 1)
30        return Moneys, Shares, Values, Signals, Buys, Sells, Return

1    def Loss(self, P:np.ndarray, DF:pd.core.frame.DataFrame):

1    def Loss(self, P:np.ndarray, DF:pd.core.frame.DataFrame):
2        Return = self.Trade(P, DF)[-1]
3        self.TrainLog.append(Return)
4        return -Return

1    def Train(self, MaxIteration:int, SwarmSize:int):
2        lb = -1 * np.ones(2)
3        ub = +1 * np.ones(2)
4        self.P, BestLoss = ps.pso(self.Loss,
5                                  lb,
6                                  ub,
7                                  swarmsize=SwarmSize,
8                                  maxiter=MaxIteration,
9                                  args=(self.trDF, ))
10        BestReturn = -BestLoss
11        print('_'*50)
12        print('Optimization Result:')
13        print(f'\tBest Parameters: {self.P}')
14        print(f'\tBest Return: {BestReturn} %')
15        print('_'*50)

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

1Trader = smaLR()

1Trader.GetData('BTC-USD', '2017-01-01', '2022-04-01')
2
3Trader.ProcessData(40)
4
5Trader.Train(30, 40)

1Optimization Result:
2        Best Parameters: [-0.00424805  0.34820413]
3        Best Return: 0.30380404578569387 %

1plt.plot(Trader.TrainLog, ls='-', lw=0.8, c='crimson')
2plt.title('Model Return Over Function Evaluations')
3plt.xlabel('Function Evaluation')
4plt.ylabel('Average Daily Return (%)')
5plt.show()

1_, _, trValues, trSignals, trBuys, trSells, trReturn = Trader.Trade(Trader.P, Trader.trDF)
2_, _, teValues, teSignals, teBuys, teSells, teReturn = Trader.Trade(Trader.P, Trader.teDF)

1plt.subplot(1, 2, 1)
2plt.semilogy(Trader.trDF['Close'].to_numpy(), ls='-', lw=0.8, c='k', label='Close')
3plt.semilogy(Trader.trDF[f'SMA({Trader.L})'].to_numpy(), ls='-', lw=1, c='b', label=f'SMA({Trader.L})')
4plt.title('Price Over Time (Train)')
5plt.xlabel('Time (Day)')
6plt.ylabel('Price ($)')
7plt.legend()
8
9plt.subplot(1, 2, 2)
10trMeanValue = (1 + trReturn / 100)**np.arange(start=0, stop=trValues.size, step=1)
11plt.semilogy(trValues, ls='-', lw=0.8, c='crimson', label='Real Values')
12plt.semilogy(trMeanValue, ls='--', lw=1, c='k', label=f'Mean Values (Return: {round(trReturn, 4)} %)')
13plt.title('Value Over Time (Train)')
14plt.xlabel('Time (Day)')
15plt.ylabel('Value')
16plt.legend()
17
18plt.show()

1plt.subplot(3, 1, (1, 2))
2plt.semilogy(Trader.trDF['Close'].to_numpy(), ls='-', lw=0.8, c='k', label='Close')
3plt.semilogy(Trader.trDF[f'SMA({Trader.L})'].to_numpy(), ls='-', lw=1, c='b', label=f'SMA({Trader.L})')
4plt.title('Price Over Time (Train)')
5plt.ylabel('Price ($)')
6plt.legend()
7
8plt.subplot(3, 1, 3)
9trZ = np.zeros_like(trSignals)
10plt.plot(trSignals, ls='-', lw=0.7, c='k')
11plt.fill_between(np.arange(trSignals.size), trSignals, trZ, where=(trSignals > trZ), color='lime', alpha=0.9, label='Buy Signal')
12plt.fill_between(np.arange(trSignals.size), trSignals, trZ, where=(trSignals < trZ), color='crimson', alpha=0.9, label='Sell Signal')
13plt.axhline(ls='-', lw=1, c='b')
14plt.title('Signal Over Time (Train)')
15plt.xlabel('Time (Day)')
16plt.ylabel('Value')
17plt.legend()
18
19plt.show()

1plt.semilogy(Trader.trDF['Close'].to_numpy(), ls='-', lw=0.8, c='k', label='Close')
2plt.semilogy(Trader.trDF[f'SMA({Trader.L})'].to_numpy(), ls='-', lw=1, c='b', label=f'SMA({Trader.L})')
3plt.scatter(trBuys['Time'], trBuys['Price'], s=20, c='g', marker='o', label='Buy')
4plt.scatter(trSells['Time'], trSells['Price'], s=20, c='r', marker='o', label='Sell')
5plt.title('Price Over Time (Train)')
6plt.xlabel('Time (Day)')
7plt.ylabel('Price ($)')
8plt.legend()
9plt.show()