DLE-190520-Impl a pure LinearRegression method not use 3rd party library

Author: Anh-Duy Le

.gitignore

@@ -1,4 +1,6 @@
 __pycache__
-.vscode/settings.json
+.vscode
 temp
 *.pkl
+app_build/model_trained
+app_build/ExchangeRatesPrediction.exe

__main__.py

@@ -27,15 +27,16 @@ def mainProcess(self):
 
             print("Welcome to Exchange Rates Service!")
             option = input("Which services do you want to use?: \
-                \n1. Predict tight range of matching (Quick process) \
-                \n2. Predict wide range of matching (Long process) \
-                \n3. Check exchange rates of specific currency \
-                \n4. Check all exchange rates \
+                \n1 - Predict basic matching (Extreme Quick process) \
+                \n2 - Predict tight range of matching (Quick process) \
+                \n3 - Predict wide range of matching (Long process) \
+                \n4 - Check exchange rates of specific currency \
+                \n5 - Check all exchange rates \
                 \nAny key to quit... \
                 \nPlease input your choice's number: ").strip()
 
 
-            while ( option.isdigit() and int(option) > 0 and int(option) < 5 ):
+            while ( option.isdigit() and int(option) > 0 and int(option) < 6 ):
                 option = int(option)
 
                 checkedDate = input("Which date do you want to check? (YYYY-MM-DD): ").strip()
@@ -46,25 +47,32 @@ def mainProcess(self):
                     self.validateInput(option, checkedDate, baseCurrency, toCurrency)
 
                     service = ExchangeRateService()
-                    data = service.predicted_quick_exrate(self.ApiGetExcRateConfig, self.checkedDate, self.baseCurrency, self.toCurrency)
+                    data = service.predicted_basic_exrate(self.ApiGetExcRateConfig, self.checkedDate, self.baseCurrency, self.toCurrency)
                     service.display_graph(data)
 
-                elif(option == 2):
+                if(option == 2):
                     self.validateInput(option, checkedDate, baseCurrency, toCurrency)
-                    checkedDaysPerMonth = 10
 
                     service = ExchangeRateService()
-                    data = service.predicted_long_exrate(self.ApiGetExcRateConfig, self.checkedDate, self.baseCurrency, self.toCurrency, checkedDaysPerMonth)
+                    data = service.predicted_quick_exrate_bytrainnedmodel(self.ApiGetExcRateConfig, self.checkedDate, self.baseCurrency, self.toCurrency)
                     service.display_graph(data)
 
                 elif(option == 3):
                     self.validateInput(option, checkedDate, baseCurrency, toCurrency)
+                    checkedDaysPerMonth = 6
+
+                    service = ExchangeRateService()
+                    data = service.predicted_long_exrate_bytrainnedmodel(self.ApiGetExcRateConfig, self.checkedDate, self.baseCurrency, self.toCurrency, checkedDaysPerMonth)
+                    service.display_graph(data)
+
+                elif(option == 4):
+                    self.validateInput(option, checkedDate, baseCurrency, toCurrency)
 
                     service = ExchangeRateService()
                     data = service.get_specific_exrate_byDate(self.ApiGetExcRateConfig, self.checkedDate, self.baseCurrency, self.toCurrency)
                     print(str(data.Currency) + " - " +  str(data.RateValue))
 
-                elif(option == 4):
+                elif(option == 5):
                     self.validateInput(option, checkedDate, baseCurrency, toCurrency)
 
                     service = ExchangeRateService()
@@ -77,10 +85,11 @@ def mainProcess(self):
                 print("-"*30 + "PROCESS DONE" + "-"*30)
 
                 option = input("Which services do you want to use?: \
-                \n1. Predict tight range of matching (Quick process) \
-                \n2. Predict wide range of matching (Long process) \
-                \n3. Check exchange rates of specific currency \
-                \n4. Check all exchange rates \
+                \n1 - Predict basic matching (Extreme Quick process) \
+                \n2 - Predict tight range of matching (Quick process) \
+                \n3 - Predict wide range of matching (Long process) \
+                \n4 - Check exchange rates of specific currency \
+                \n5 - Check all exchange rates \
                 \nAny key to quit... \
                 \nPlease input your choice's number: ").strip()
 
@@ -93,14 +102,15 @@ def mainProcess(self):
     def validateInput(self, option, checkedDate, baseCurrency, toCurrency):
         self.baseCurrency = (not baseCurrency) and 'USD' or str(baseCurrency).upper()
 
-        currentDate = datetime.date.today()
+        currentDate = datetime.datetime.today()
         if ExchangeRateService().UtilCommon().validateDateFormat(checkedDate):
-            checkedDate = ExchangeRateService().UtilCommon().parseStringToDateTime(checkedDate).date()
+            checkedDate = ExchangeRateService().UtilCommon().parseStringToDateTime(checkedDate)
             self.checkedDate = (checkedDate >= currentDate) and currentDate or checkedDate
         else:                
             self.checkedDate = currentDate
 
-        if(option == 1 or option == 2 or option == 3):
+        optionNeedToCurreny = [1,2,3,4]
+        if(option in optionNeedToCurreny):
             self.toCurrency = toCurrency
             countFail = 0
             while not toCurrency:
@@ -121,7 +131,8 @@ def validateInput(self, option, checkedDate, baseCurrency, toCurrency):
 
 
 ### BUILD APP
-# Step 1: Put all images folder to "dist" folder (folder to deploy app) 
-# Step 2: Open CMD/Terminal and change directory path to main python script
-# Step 3: Run command in first time => pyinstaller --clean --distpath=./app_build --workpath=./temp --onefile --name ExchangeRatesPrediction ./__main__.py
+# Step 1: Put all neccessary media/config files to build folder (folder to deploy app) 
+# Step 2: Open CMD/Terminal and change directory path to main python script (__main__.py)
+# Step 3: Run command pip install these packages: pyinstaller, tornado, pypiwin32
+# Step 4: Run command in first time => pyinstaller --clean --distpath=./app_build --workpath=./temp --onefile --name ExchangeRatesPrediction ./__main__.py
 # Or Run this command when already have .spec file => pyinstaller --clean --distpath=./app_build --workpath=./temp --add-data="config.json;." --add-data="/model_trained/linear_model.pkl;." --onefile ExchangeRatesPrediction.spec

app_build/ExchangeRatesPrediction.exe

@@ -22,7 +22,7 @@
                     "param_defvalue": "USD"
                 }
             ],
-            "is_using": "False"
+            "is_using": "True"
         },
         {       
             "type": "api",
@@ -37,7 +37,7 @@
                     "param_defvalue": "USD"
                 }
             ],
-            "is_using": "True"
+            "is_using": "False"
         },
         {                                
             "type": "api",

app_build/model_trained/linear_model.pkl

@@ -78,12 +78,14 @@ def generateRandomDateInMonth(self, year, month, totalRandom = 1):
         currentTotalDays = calendar.monthrange(year, month)[1]
 
         if totalRandom <= currentTotalDays:            
-            dt = random.randint(1, currentTotalDays)
+            rdt = random.randint(1, currentTotalDays)
+            d = datetime(year, month, rdt)
             # check if random number not exist in list and list size must smaller than total random times
             while(len(lstDate) < totalRandom):
-                if not dt in lstDate:
-                    lstDate.append(dt)
-                dt = random.randint(1, currentTotalDays)
+                if not d in lstDate:
+                    lstDate.append(d)
+                rdt = random.randint(1, currentTotalDays)
+                d = datetime(year, month, rdt)
 
         return lstDate            
 

config.json

@@ -0,0 +1,62 @@
+
+class Core_UtilityLogic():
+
+
+    def __init__(self):
+        pass
+
+    def calTotalSumOfSeries(self, lstValue):
+        try:
+            if lstValue and len(lstValue) > 0:
+                return sum(lstValue)
+        except ValueError as ex:
+            print('Error: ', ex)
+        
+        return 0
+
+    
+    def calLinearRegressionOfY(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([x * y for (x, y) in zip(lstSpX, lstSpX)])
+
+            #the more list of value X and Y passed in, the more calculation for accurate slope and intercept be matched
+            slope = (numberOfVal*sumXY - sumX*sumY) / (numberOfVal*sumXX - pow(sumX,2))            
+            intercept = (sumY - slope*sumX) / numberOfVal
+
+            #calculate approximate Y value (predict Y) for specific X value by regression equation: y = a + bx            
+            equationResult = intercept + slope*xVal
+
+            return equationResult
+
+        except ValueError as ex:
+            print('Error: ', ex)
+        
+        return 0
+
+    def calLinearRegressionOfListY(self, lstSampleValueX, lstSampleValueY, lstXVal):
+        try:
+            if lstXVal and len(lstXVal) > 0:
+                lstYVal = []
+                for x in lstXVal:
+                    y = self.calLinearRegressionOfY(lstSampleValueX, lstSampleValueY, x)
+                    lstYVal.append(y)
+
+                return lstYVal
+
+        except ValueError as ex:
+            print('Error: ', ex)
+        
+        return None
+

core_utils/util_common.py

@@ -27,7 +27,8 @@ def __init__(self):
         self.base_service = BaseService()        
         self.domain_factory = self.base_service.DomainFactory()
         self.util_common = self.base_service.UtilCommon()
-        self.util_data = self.base_service.UtilData()        
+        self.util_data = self.base_service.UtilData()
+        self.util_logic = self.base_service.UtilLogic()       
 
         self.ConfigApiModel = self.domain_factory.init_ModelClass('ConfigApiModel')
         self.ConfigApiQueryModel = self.domain_factory.init_ModelClass('ConfigApiQueryModel')
@@ -43,7 +44,8 @@ def get_exrate_byDate(self, apiConfig, dateReport, baseCurrency):
         lstCustomizeQuery = []
         lstCustomizeQuery.append(modelBaseCurParam)
 
-        apiUrl = self.base_service.Config.initApiUrl(apiConfig, dateReport, lstCustomizeQuery)
+        dateReport = self.util_common.parseStringToDateTime(dateReport)
+        apiUrl = self.base_service.Config.initApiUrl(apiConfig, dateReport.date(), lstCustomizeQuery)
         if apiUrl:
             content = self.util_data.readJsonFromUrl(apiUrl)
             lstRates = []
@@ -74,20 +76,21 @@ def get_specific_exrate_byDateRange(self, apiConfig, fromDate, toDate, checkedDa
         # print('Checked Date: Exchange Rates')  
         lstExcRates = []
         for single_date in self.util_common.dateRange(fromDate, toDate):
-            if any(da == single_date.day for da in checkedDate):             
+            if any(da == single_date.day for da in [d.day for d in checkedDate]):             
                 data = self.get_specific_exrate_byDate(apiConfig, single_date.strftime("%Y-%m-%d"), baseCurrency, toCurrency)  
                 # print(single_date.strftime("%Y-%m-%d") + ': ' + str(data.RateValue))         
 
                 model = self.ExchangeRateModel()
                 model.BaseCurrency = baseCurrency
                 model.ConvertedCurrency = toCurrency
-                model.OnDate = single_date.month#datetime.timestamp(single_date)#convert datetime to timestamp
+                model.OnDate = datetime.timestamp(single_date) #single_date.month
                 model.RateValue = data.RateValue
 
                 lstExcRates.append(model)
 
         return len(lstExcRates) > 0 and lstExcRates or None
 
+
     def display_graph(self, listdata):       
 
         date = [x.OnDate for x in listdata]
@@ -102,6 +105,7 @@ def display_graph(self, listdata):
         plt.ylabel("Rates")
         plt.show()
 
+
     def training_linear_model(self, listdata):
         date = [[x.OnDate] for x in listdata]
         rate = [[x.RateValue] for x in listdata]
@@ -128,14 +132,24 @@ def training_linear_model(self, listdata):
         print("Predicted: ", predicted)
 
         return predicted
+
 
+    def calculate_exrate_forSpecificDate(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)
 
-    def predicted_quick_exrate(self, apiConfig, strPredictedDate, baseCurrency, toCurrency):
+        return predictedRate
+    
+
+    def predicted_quick_exrate_bytrainnedmodel(self, apiConfig, strPredictedDate, baseCurrency, toCurrency):
         predictedDate = self.util_common.parseStringToDateTime(strPredictedDate)
         dateLastMonth = self.util_common.getDateByYearCount(strPredictedDate, -1)
         dateLastYear = self.util_common.getDateByMonthCount(strPredictedDate, -1)
         checkedDate = []
-        checkedDate.append(predictedDate.day)
+        checkedDate.append(predictedDate)
 
 
         data = self.get_specific_exrate_byDateRange(apiConfig, dateLastMonth, dateLastYear, checkedDate, baseCurrency, toCurrency)
@@ -144,13 +158,14 @@ def predicted_quick_exrate(self, apiConfig, strPredictedDate, baseCurrency, toCu
 
         return data
 
-    def predicted_long_exrate(self, apiConfig, strPredictedDate, baseCurrency, toCurrency, checkedDaysPerMonth=5):
+
+    def predicted_long_exrate_bytrainnedmodel(self, apiConfig, strPredictedDate, baseCurrency, toCurrency, checkedDaysPerMonth=5):
         predictedDate = self.util_common.parseStringToDateTime(strPredictedDate)
         dateLastMonth = self.util_common.getDateByYearCount(strPredictedDate, -1)
         dateLastYear = self.util_common.getDateByMonthCount(strPredictedDate, -1)
         checkedDate = self.util_common.generateRandomDateInMonth(year=predictedDate.year, month=predictedDate.month, totalRandom=checkedDaysPerMonth)
-        if not predictedDate.day in checkedDate:
-            checkedDate.append(predictedDate.day)
+        if not predictedDate in checkedDate:
+            checkedDate.append(predictedDate)
 
 
         data = self.get_specific_exrate_byDateRange(apiConfig, dateLastMonth, dateLastYear, checkedDate, baseCurrency, toCurrency)
@@ -159,6 +174,22 @@ def predicted_long_exrate(self, apiConfig, strPredictedDate, baseCurrency, toCur
 
         return data
 
+    
+    def predicted_basic_exrate(self, apiConfig, strPredictedDate, baseCurrency, toCurrency):
+        predictedDate = self.util_common.parseStringToDateTime(strPredictedDate)
+        dateLastMonth = self.util_common.getDateByYearCount(strPredictedDate, -1)
+        dateLastYear = self.util_common.getDateByMonthCount(strPredictedDate, -1)
+        checkedDate = []
+        checkedDate.append(predictedDate)        
+
+        data = self.get_specific_exrate_byDateRange(apiConfig, dateLastMonth, dateLastYear, checkedDate, baseCurrency, toCurrency)
+
+        predictedRate = self.calculate_exrate_forSpecificDate(data, predictedDate)
+        print("Predicted: ", predictedRate)
+
+        return data
+
+    
 
 
 

core_utils/util_logic.py

@@ -8,6 +8,7 @@
 
 from core_utils.util_common import Core_UtilityCommon
 from core_utils.util_data import Core_UtilityData
+from core_utils.util_logic import Core_UtilityLogic
 
 from domains.domain_factory import DomainFactory
 from config import Configuration
@@ -31,6 +32,9 @@ def UtilCommon(self):
     def UtilData(self):
         return Core_UtilityData()
 
+    def UtilLogic(self):
+        return Core_UtilityLogic()
+