DLE-190520-Impl polynomial model (bonus)

Author: Anh-Duy Le

__main__.py

@@ -59,7 +59,7 @@ def mainProcess(self):
 
                 elif(option == 3):
                     self.validateInput(option, checkedDate, baseCurrency, toCurrency)
-                    checkedDaysPerMonth = 6
+                    checkedDaysPerMonth = 20
 
                     service = ExchangeRateService()
                     data = service.predicted_long_exrate_bytrainnedmodel(self.ApiGetExcRateConfig, self.checkedDate, self.baseCurrency, self.toCurrency, checkedDaysPerMonth)

core_utils/util_logic.py

@@ -36,6 +36,7 @@ def calLinearRegressionOfY(self, lstSampleValueX, lstSampleValueY, xVal):
             intercept = (sumY - slope*sumX) / numberOfVal
 
             #calculate approximate Y value (predict Y) for specific X value by regression equation: y = a + bx            
+            print("Formula Equation: y = {0}x + {1}".format(slope, intercept))
             equationResult = intercept + slope*xVal
 
             return equationResult
@@ -60,3 +61,37 @@ def calLinearRegressionOfListY(self, lstSampleValueX, lstSampleValueY, lstXVal):
 
         return None
 
+    def calPolynomialRegressionOfY(self, lstSampleValueX, lstSampleValueY, xVal):
+        try:
+            lstSpX = 0
+            lstSpY = 0
+
+            if lstSampleValueX and len(lstSampleValueX) > 0:
+                lstSpX = lstSampleValueX
+            if lstSampleValueY and len(lstSampleValueY) > 0:
+                lstSpY = lstSampleValueY
+            
+            numberOfVal = (len(lstSpX) > len(lstSpY)) and len(lstSpX) or len(lstSpY) 
+            sumX = self.calTotalSumOfSeries(lstSpX)
+            sumY = self.calTotalSumOfSeries(lstSpY)
+            sumXY = sum([x * y for (x, y) in zip(lstSpX, lstSpY)])
+            sumXX = sum([x1 * x2 for (x1, x2) in zip(lstSpX, lstSpX)])
+            sumXXX = sum([x1 * x2 * x3 for (x1, x2, x3) in zip(lstSpX, lstSpX, lstSpX)])
+            sumXXXX = sum([x1 * x2 * x3 * x4 for (x1, x2, x3, x4) in zip(lstSpX, lstSpX, lstSpX, lstSpX)])
+            sumXXY = sum([x1 * x2 * y for (x1, x2, y) in zip(lstSpX, lstSpX, lstSpY)])
+
+            #the more list of value X and Y passed in, the more calculation for coefficients be matched
+            coefficientA = (sumXXY * sumXX - sumXY * sumXXX) / (sumXX * sumXXXX - pow(sumXXX, 2))
+            coefficientB = (sumXY * sumXXXX - sumXXY * sumXXX) / (sumXX * sumXXXX - pow(sumXXX, 2))
+            coefficientC = (sumY/numberOfVal) - (coefficientB * (sumX/numberOfVal)) - (coefficientA * (sumXX/numberOfVal))
+
+            #calculate approximate Y value (predict Y) for specific X value by quadratic equation: y = ax^2 + bx + c
+            print("Formula Equation: y = {0}x^2 + {1}x + {2}".format(coefficientA, coefficientB, coefficientC))
+            equationResult = coefficientA*pow(xVal, 2) + coefficientB*xVal + coefficientC
+
+            return equationResult
+
+        except ValueError as ex:
+            print('Error: ', ex)
+        
+        return 0

service/banking/exchangerate_service.py

@@ -1,11 +1,15 @@
 import os, sys
 import inspect, types
 from datetime import timedelta, date, datetime
+from array import *
 
 import matplotlib.pyplot as plt
 
 import sklearn.utils._cython_blas
+from sklearn import linear_model
 from sklearn.linear_model import LinearRegression
+from sklearn.preprocessing import PolynomialFeatures
+from sklearn.pipeline import Pipeline
 from sklearn.model_selection import train_test_split
 import joblib
 from pathlib import Path
@@ -106,12 +110,13 @@ def display_graph(self, listdata):
         plt.show()
 
 
-    def training_linear_model(self, listdata):
+    def training_linear_model(self, listdata, valueNeedPredict):
         date = [[x.OnDate] for x in listdata]
         rate = [[x.RateValue] for x in listdata]
 
         #Use 80% of data as training, rest 20% to Test model
         x_train, x_test, y_train, y_test = train_test_split(date, rate, test_size=0.2)
+
         # training model
         linear = LinearRegression()
         linear.fit(x_train, y_train)
@@ -125,23 +130,67 @@ def training_linear_model(self, listdata):
         joblib.dump(linear, modelPathDir)
 
         # loading model
+        not valueNeedPredict and x_test or x_test.insert(0, [valueNeedPredict])
         clf = joblib.load(modelPathDir)
         predicted = clf.predict(x_test)#linear.predict(x_test)
+        print("###Predicted by Linear Model")
+        print("Predicted Max:", predicted.max())
+        print("Predicted Min:", predicted.min())
+        print("Predicted: ", predicted)
+
+        return predicted
+
+    
+    def training_polynomial_model(self, listdata, valueNeedPredict):
+        date = [[x.OnDate] for x in listdata]
+        rate = [[x.RateValue] for x in listdata]
+
+        #Use 80% of data as training, rest 20% to Test model
+        x_train, x_test, y_train, y_test = train_test_split(date, rate, test_size=0.2)
+
+        # training model
+        poly = Pipeline([('poly', PolynomialFeatures(interaction_only=True, degree=2)),
+                       ('linear', linear_model.LinearRegression(fit_intercept=False))])
+        poly.fit(x_train, y_train)
+
+        # evaluating model
+        score_trained = poly.score(x_test, y_test)
+        print("Model scored:", score_trained)
+       
+        # saving model
+        modelPathDir = Path(ROOT_DIR + r'/model_trained/poly_model.pkl')
+        joblib.dump(poly, modelPathDir)
+
+        # loading model
+        not valueNeedPredict and x_test or x_test.insert(0, [valueNeedPredict])
+        clf = joblib.load(modelPathDir)
+        predicted = clf.predict(x_test)#poly.predict(x_test)
+        print("###Predicted by Polynomial Model")
         print("Predicted Max:", predicted.max())
         print("Predicted Min:", predicted.min())
         print("Predicted: ", predicted)
 
         return predicted
 
 
-    def calculate_exrate_forSpecificDate(self, listdata, checkedDate):
+    def calculate_exrate_bypurelinearmodel(self, listdata, checkedDate):
         date = [x.OnDate for x in listdata]
         rate = [x.RateValue for x in listdata]
         chkDate = datetime.timestamp(checkedDate)
 
         predictedRate = self.util_logic.calLinearRegressionOfY(lstSampleValueX=date, lstSampleValueY=rate, xVal=chkDate)
 
         return predictedRate
+
+
+    def calculate_exrate_bypurepolynomialmodel(self, listdata, checkedDate):
+        date = [x.OnDate for x in listdata]
+        rate = [x.RateValue for x in listdata]
+        chkDate = datetime.timestamp(checkedDate)
+
+        predictedRate = self.util_logic.calPolynomialRegressionOfY(lstSampleValueX=date, lstSampleValueY=rate, xVal=chkDate)
+
+        return predictedRate
 
 
     def predicted_quick_exrate_bytrainnedmodel(self, apiConfig, strPredictedDate, baseCurrency, toCurrency):
@@ -154,7 +203,8 @@ def predicted_quick_exrate_bytrainnedmodel(self, apiConfig, strPredictedDate, ba
 
         data = self.get_specific_exrate_byDateRange(apiConfig, dateLastMonth, dateLastYear, checkedDate, baseCurrency, toCurrency)
 
-        self.training_linear_model(data)
+        self.training_linear_model(data, datetime.timestamp(predictedDate))
+        self.training_polynomial_model(data, datetime.timestamp(predictedDate))
 
         return data
 
@@ -170,7 +220,8 @@ def predicted_long_exrate_bytrainnedmodel(self, apiConfig, strPredictedDate, bas
 
         data = self.get_specific_exrate_byDateRange(apiConfig, dateLastMonth, dateLastYear, checkedDate, baseCurrency, toCurrency)
 
-        self.training_linear_model(data)
+        self.training_linear_model(data, datetime.timestamp(predictedDate))
+        self.training_polynomial_model(data, datetime.timestamp(predictedDate))
 
         return data
 
@@ -184,8 +235,10 @@ def predicted_basic_exrate(self, apiConfig, strPredictedDate, baseCurrency, toCu
 
         data = self.get_specific_exrate_byDateRange(apiConfig, dateLastMonth, dateLastYear, checkedDate, baseCurrency, toCurrency)
 
-        predictedRate = self.calculate_exrate_forSpecificDate(data, predictedDate)
-        print("Predicted: ", predictedRate)
+        predictedRate = self.calculate_exrate_bypurelinearmodel(data, predictedDate)
+        print("Predicted with Linear Regression Model: ", predictedRate)
+        predictedRate2 = self.calculate_exrate_bypurepolynomialmodel(data, predictedDate)
+        print("Predicted with Polynomial Regression Model: ", predictedRate2)
 
         return data
 

unit_tests/test_service.py

@@ -2,6 +2,7 @@
 import sys, os
 import urllib
 import urllib.parse
+import datetime
 
 
 ROOT_DIR = os.getcwd()  # Get root directory
@@ -58,7 +59,7 @@ def test_exrate__get_specific_exrate_byDateRange(self):
         data = service.get_specific_exrate_byDateRange(ServiceTestCase.ApiCheckExcRateConfig, fromDate, toDate, checkedDate, baseCurrency, toCurrency)
 
         # service.display_graph(data)
-        service.training_linear_model(data)
+        service.training_linear_model(data, datetime.datetime.timestamp( datetime.datetime(year=2017, month=1, day=checkedDate[0]) ))
 
         self.assertIsNotNone(data, '###Error Message: No data')
 

unit_tests/test_utility.py

@@ -75,6 +75,17 @@ def test_logic__calLinearRegressionOfListY(self):
 
         self.assertIsNotNone(data, '###Error Message: No data')
 
+    def test_logic__calPolynomialRegressionOfY(self):
+                
+        lstX = [-6, -4, -2, 0, 2, 4, 6]
+        lstY = [2.28, 1.42, 0.67, 0, -0.57, -1.05, -1.42]
+        x = -7.72
+
+        data = UtilityTestCase.UtilLogic.calPolynomialRegressionOfY(lstX, lstY, x)
+        print(data)
+
+        self.assertIsNotNone(data, '###Error Message: No data')
+
 
 
 ###########################################################################################################