-
Notifications
You must be signed in to change notification settings - Fork 1
/
WebApp.py
92 lines (72 loc) · 3.26 KB
/
WebApp.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
"""
Author: Mikhail Filippov (University of Mannheim)
Version: 15.07.2024
"""
from flask import Flask, request, jsonify
import joblib
import pandas as pd
POLICY_PRICES = {
1: 72.24,
2: 84.98,
3: 95.50,
4: 105.00,
}
class WebApp:
def __init__(self):
self.app = Flask(__name__)
self.model_class = joblib.load('flight_delay_classifier_v1.pkl')
self.model_reg = joblib.load('flight_delay_regressor_v1.pkl')
def calculate_insurance_price(self, is_delayed, delay_duration, standard_price):
if is_delayed:
# Calculate the number of 10 minute intervals in the delay duration
intervals = delay_duration // 10
# Increase the insurance price by 10% for each 10 minute interval
insurance_price = standard_price * (1 + 0.1 * intervals)
else:
insurance_price = standard_price
return round(insurance_price, 2)
def predict(self):
@self.app.route('/predict', methods=['POST'])
def predict():
# Get the parameters from the request
params = request.get_json(force=True)
# Extract the POLICY_ID from the parameters
policy_id = params.pop('POLICY_ID', None)
# Get the standard price for the policy
standard_price = POLICY_PRICES.get(policy_id, 72.24)
# Convert the parameters to a DataFrame
df_params = pd.DataFrame([params])
df_params['FL_DATE'] = pd.to_datetime(df_params['FL_DATE'])
df_params['Month'] = df_params['FL_DATE'].dt.month_name()
df_params['Weekday'] = df_params['FL_DATE'].dt.day_name()
df_params['Year'] = df_params['FL_DATE'].dt.year
df_params['Planned_departure_hour'] = df_params['CRS_DEP_TIME'] // 100
# Drop unused columns
selected_cols = ['AIRLINE_CODE', 'ORIGIN_CITY', 'DEST_CITY', 'Planned_departure_hour', 'Month',
'Weekday', 'Year']
df_params = df_params[selected_cols]
# Convert categorical variables to dummy variables
df_params = pd.get_dummies(df_params)
# Ensure the DataFrame has the same columns as the training data
columns_without_delayed = [col for col in self.model_class.feature_names_in_]
df_params = df_params.reindex(columns=columns_without_delayed, fill_value=0)
# Predict if the flight will be delayed
delay_proba = self.model_class.predict_proba(df_params)[:, 1][0]
is_delayed = delay_proba > 0.05
# If the flight is predicted to be delayed, predict the duration of the delay
if is_delayed:
delay_duration = self.model_reg.predict(df_params)[0]
else:
delay_duration = 0
# Calculate the insurance price for the flight
insurance_price = self.calculate_insurance_price(is_delayed, delay_duration, standard_price)
# Return the predictions and insurance price
return jsonify({
'insurance_price': insurance_price
})
def run(self):
self.app.run(host='0.0.0.0', port=5000)
# Create an instance of the class and run the server
predictor = WebApp()
predictor.predict()
predictor.run()