Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
# coding: utf-8
# In[3]:
import sys
import pandas as pd
import numpy as np
import pyodbc
import statistics
from datetime import datetime, timedelta
# In[5]:
import datalib.config as CFG
from datalib.db_utils import getSQLfromFile
from datalib.db_utils import check_value
from datalib.db_utils import restore_revmaxMonitor_from_file
from datalib.db_utils import update_db
from datalib.db_utils import get_frames
# In[6]:
import re
import requests
import datetime
import matplotlib.pyplot as plt
import seaborn as sns
from dateutil import rrule
from datetime import datetime, timedelta
from bs4 import BeautifulSoup
from fbprophet import Prophet
from sqlitedict import SqliteDict
# In[7]:
from datetime import datetime, timedelta
yesterday = - timedelta(days=50)
YESTERDAYS_DATE = yesterday.strftime('%Y-%m-%d')
begin_month = datetime(yesterday.year, yesterday.month, 1)
BEGIN_MONTH_DATE = begin_month.strftime('%Y-%m-%d')
today = - timedelta(days=0)
yesterday = - timedelta(days=1)
two_days_ago = - timedelta(days=2)
TODAYS_DATE = today.strftime('%Y-%m-%d')
YESTERDAYS_DATE = yesterday.strftime('%Y-%m-%d')
TWO_DAYS_AGO_DATE = two_days_ago.strftime('%Y-%m-%d')
weather_dict = SqliteDict('./storage/rateMyView_db.sqlite', autocommit=True)
# In[9]:
is_it_goin_to_rain = ['Low', 'Medium', 'High', 'Nil']
weather_categories = {'Clear':5,
'Mainly sunny':5,
'A few clouds':4,
'A mix of sun and cloud':4,
'Mainly cloudy':3,
'Chance of showers':3,
'Chance of showers or drizzle':3,
'Chance of showers. Risk of freezing rain':3,
'Periods of drizzle':2,
'Periods of rain':2,
'Chance of flurries':3,
'Periods of light snow':3,
'Light snow':2,
'Light snow. Risk of freezing rain':2,
'Snow. Risk of freezing rain':1,
'Snow at times heavy':1}
# In[10]:
ACCESS_SUMMARY_query = f"""SELECT CAST(AccessDate AS DATE) AS ds, COUNT(*) as y FROM dbo.AccessDailySummary ads JOIN Location l ON l.LocationCode = ads.AccessLocationCode JOIN AccessProductType apt ON apt.AccessProductTypeCode = ads.AccessProductTypeCode JOIN AccessRule ar on ar.AccessRule = SUBSTRING(ads.AccessCode, 4, 4) JOIN AccessRuleCategory arc ON ar.AccessRuleCategoryCode = arc.AccessRuleCategoryCode WHERE CAST(AccessDate AS DATE) BETWEEN '2015-01-01' AND {YESTERDAYS_DATE!r} AND ads.AccessLocationCode IN (1901, 1902) GROUP BY CAST(AccessDate AS DATE) ORDER BY 1"""
# In[11]:
rtp_cnxn = pyodbc.connect(CFG.rtp_stringz)
df_daily_scan = pd.read_sql_query(ACCESS_SUMMARY_query, rtp_cnxn)
# In[12]:
monte_carlo_simulator = Prophet() #seasonality_mode='multiplicative')
future_df = monte_carlo_simulator.make_future_dataframe(periods=3)
# In[13]:
passenger_forecast = monte_carlo_simulator.predict(future_df)
pf = passenger_forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']]
# In[14]:
#pf[pf['ds'] >'%Y-%m-%d')][:12][['yhat']]
# In[15]:
#pf[pf['ds'] >'%Y-%m-%d')][:12][['yhat']].plot()
# In[16]:
predictions = pf[pf['ds'] =='%Y-%m-%d')]
# In[17]:
def check_weather_forecast_gc_ca():
page = requests.get("")
soup = BeautifulSoup(page.content, 'html.parser')
weather_forecast = soup.find_all('p')[0]
#weather_forecast.getText(), weather_categories[weather_forecast.getText()]
return weather_categories[weather_forecast.getText()]
# In[18]:
def get_visibility_rating():
#visibility score what is happening now
yesterday = - timedelta(days=0)
YESTERDAYS_DATE = yesterday.strftime('%Y-%m-%d')
_, visibility_rating = extract_average_weather_rating(YESTERDAYS_DATE, weather_dict)
print(YESTERDAYS_DATE, visibility_rating)
#update_db_weather(YESTERDAYS_DATE, int(visibility_rating), "Gondola")
return visibility_rating
print("Bad Value or Visibility Not Found")
return 0
# In[19]:
def sunny_previously():
Pent up demand evaluation
For example if 3 bad days of visibility will make PAX closer to
yhat_upper predictions.
So False means a spike if its clear.
_, visibility_rating1 = extract_average_weather_rating(TODAYS_DATE, weather_dict)
_, visibility_rating2 = extract_average_weather_rating(YESTERDAYS_DATE, weather_dict)
_, visibility_rating3 = extract_average_weather_rating(TWO_DAYS_AGO_DATE, weather_dict)
return statistics.mean([visibility_rating1,visibility_rating2,visibility_rating3])>=4
print("Bad Value or Visibility Not Found")
return True
# In[20]:
def fine_tune_prediction_past(predictions):
if sunny_previously() == False:
if get_visibility_rating() >= 4:
#print("Sunny Pent Up")
return predictions.yhat_upper.values[0]
#print("NOT Sunny Pent Up")
return predictions.yhat_lower.values[0]
elif sunny_previously() == True:
if get_visibility_rating() >= 4:
#print("Sunny NOT Pent Up")
return predictions.yhat.values[0]
#print("Terrible NOT Pent Up")
return predictions.yhat_lower.values[0]
#print("BAD BAD BAD")
return predictions.yhat.values[0]/2
# In[22]:
predictions.yhat_lower.values[0],((predictions.yhat.values + predictions.yhat_lower.values)/2)[0], predictions.yhat.values[0], ((predictions.yhat.values + predictions.yhat_upper.values)/2)[0],predictions.yhat_upper.values[0]
# In[23]:
prediction_past = fine_tune_prediction_past(predictions)
# In[24]:
def fine_tune_prediction_future(prediction_past, predictions):
forecast_score = check_weather_forecast_gc_ca()
if forecast_score == 5:
return predictions.yhat_upper.values
elif forecast_score == 4:
return predictions.yhat.values
elif forecast_score == 3:
return prediction_past
elif forecast_score == 2:
return predictions.yhat.values[0]/2
elif forecast_score == 1:
return predictions.yhat.values[0]/2
# In[25]:
fine_tune_prediction_future(prediction_past, predictions)
# In[26]:
# In[27]:
from matplotlib import pyplot as plt
plt.plot(monte_carlo_simulator.changepoints.tolist(), [ passenger_forecast[passenger_forecast.ds == x].values[0][1:] for x in monte_carlo_simulator.changepoints.tolist()], 'r*')
# In[28]:
fig1 = monte_carlo_simulator.plot(passenger_forecast)
fig2 = monte_carlo_simulator.plot_components(passenger_forecast)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.