基于时间序列的气温预测系统

from statsmodels.tsa.arima_model import ARIMA import warnings import pandas as pd import numpy import datetime # create a seasonal differenced series def difference(dataset, interval=1): diff = list() for i in range(interval, len(dataset)): value = dataset[i] - dataset[i - interval] diff.append(value) return numpy.array(diff) # invert differenced value def inverse_difference(history, yhat, interval=1): return yhat + history[-interval] def fahrenheit_to_celsius(fahrenheit): celsius = (fahrenheit - 32) / 1.8 return celsius def change_to_int(data_list): data = list() for t in range(len(data_list)): data.append(int(round(data_list[t]))) return data def get_latest_date(place): file = 'dataset/' + place + '.csv' dataset = open(file) file_str = '' while True: try: file_str = next(dataset) except: break date = file_str.split(',')[0].split('\"')[1].split('-') year = int(date[0]) month = int(date[1]) day = int(date[2]) return datetime.date(year, month, day) class Forecast: dict_f_max = {'beijing': list(), 'chengdu': list(), 'shanghai': list(), 'xi\'an': list()} dict_f_min = {'beijing': list(), 'chengdu': list(), 'shanghai': list(), 'xi\'an': list()} dict_d_max = {'beijing': list(), 'chengdu': list(), 'shanghai': list(), 'xi\'an': list()} dict_d_min = {'beijing': list(), 'chengdu': list(), 'shanghai': list(), 'xi\'an': list()} def get_forecast_max(self, place, period): if len(self.dict_f_max[place]) == 0: warnings.filterwarnings("ignore") file = 'dataset/' + place + '.csv' series = pd.read_csv(file, header=0) data = series['MAX'].values diff_data = difference(data, 365) # do forecast if place == 'beijing': order = (1, 0, 2) elif place == 'shanghai': order = (2, 0, 2) elif place == 'chengdu': order = (2, 0, 1) elif place == 'xi\'an': order = (3, 0, 1) else: return model = ARIMA(diff_data, order=order) model_fit = model.fit(disp=0) predictions = model_fit.forecast(steps=15)[0] history = [x for x in data] forecast = list() for yhat in predictions: inverted = inverse_difference(history, yhat, 365) history.append(inverted) c = fahrenheit_to_celsius(inverted) forecast.append(c) forecast = change_to_int(forecast) date = list() date_latest = get_latest_date(place) for t in range(1, 16): date_latest += datetime.timedelta(days=1) date.append(date_latest.__format__("%m-%d")) self.dict_f_max[place] = [date, forecast] date = self.dict_f_max[place][0][:int(period)] forecast = self.dict_f_max[place][1][:int(period)] return [date, forecast] def get_forecast_min(self, place, period): if len(self.dict_f_min[place]) == 0: warnings.filterwarnings("ignore") file = 'dataset/' + place + '.csv' series = pd.read_csv(file, header=0) data = series['MIN'].values diff_data = difference(data, 365) # do forecast if place == 'beijing': order = (3, 0, 2) elif place == 'shanghai': order = (3, 0, 3) elif place == 'chengdu': order = (2, 0, 2) elif place == 'xi\'an': order = (3, 0, 1) else: return model = ARIMA(diff_data, order=order) model_fit = model.fit(disp=0) predictions = model_fit.forecast(steps=15)[0] history = [x for x in data] forecast = list() for yhat in predictions: inverted = inverse_difference(history, yhat, 365) history.append(inverted) c = fahrenheit_to_celsius(inverted) forecast.append(c) forecast = change_to_int(forecast) date = list() date_latest = get_latest_date(place) for t in range(1, 16): date_latest += datetime.timedelta(days=1) date.append(date_latest.__format__("%m-%d")) self.dict_f_min[place] = [date, forecast] date = self.dict_f_min[place][0][:int(period)] forecast = self.dict_f_min[place][1][:int(period)] return [date, forecast] def get_data_max(self, place, period): if len(self.dict_d_max[place]) == 0: warnings.filterwarnings("ignore") file = 'dataset/' + place + '.csv' series = pd.read_csv(file, header=0) data = series['MAX'].values f = list() for t in range(15): c = fahrenheit_to_celsius(data[len(data) - 15 + t]) f.append(c) f = change_to_int(f) date = list() date_latest = get_latest_date(place) for t in range(-14, 1): date_now = date_latest + datetime.timedelta(days=t) date.append(date_now.__format__("%m-%d")) self.dict_d_max[place] = [date, f] date = self.dict_d_max[place][0][-int(period):] data = self.dict_d_max[place][1][-int(period):] return [date, data] def get_data_min(self, place, period): if len(self.dict_d_min[place]) == 0: warnings.filterwarnings("ignore") file = 'dataset/' + place + '.csv' series = pd.read_csv(file, header=0) data = series['MIN'].values f = list() for t in range(15): c = fahrenheit_to_celsius(data[len(data) - 15 + t]) f.append(c) f = change_to_int(f) date = list() date_latest = get_latest_date(place) for t in range(-14, 1): date_now = date_latest + datetime.timedelta(days=t) date.append(date_now.__format__("%m-%d")) self.dict_d_min[place] = [date, f] date = self.dict_d_min[place][0][-int(period):] data = self.dict_d_min[place][1][-int(period):] return [date, data]