-
Notifications
You must be signed in to change notification settings - Fork 0
/
credit_model.py
149 lines (120 loc) · 4.43 KB
/
credit_model.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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
from pathlib import Path
import feast
import joblib
import pandas as pd
from sklearn import tree
from sklearn.exceptions import NotFittedError
from sklearn.preprocessing import OrdinalEncoder
from sklearn.utils.validation import check_is_fitted
class CreditScoringModel:
categorical_features = [
"person_home_ownership",
"loan_intent",
"city",
"state",
"location_type",
]
feast_features = [
"zipcode_features:city",
"zipcode_features:state",
"zipcode_features:location_type",
"zipcode_features:tax_returns_filed",
"zipcode_features:population",
"zipcode_features:total_wages",
"credit_history:credit_card_due",
"credit_history:mortgage_due",
"credit_history:student_loan_due",
"credit_history:vehicle_loan_due",
"credit_history:hard_pulls",
"credit_history:missed_payments_2y",
"credit_history:missed_payments_1y",
"credit_history:missed_payments_6m",
"credit_history:bankruptcies",
]
target = "loan_status"
model_filename = "model.bin"
encoder_filename = "encoder.bin"
def __init__(self):
# load model
if Path(self.model_filename).exists():
self.classifier = joblib.load(self.model_filename)
else:
self.classifier = tree.DecisionTreeClassifier()
#load ordinal encoder
if Path(self.encoder_filename).exists():
self.encoder = joblib.load(self.encoder_filename)
else:
self.encoder = OrdinalEncoder()
#set up feature store
self.fs = feast.FeatureStore(repo_path="feature_repo")
def _get_training_features(self, loans):
"""get training feature from feast
Args:
loans (_type_): entity dataset
Returns:
train_X: training dataset
train_Y: va
"""
training_df = self.fs.get_historical_features(
entity_df=loans, features=self.feast_features
).to_df()
self._fit_ordinal_encoder(training_df)
self._apply_ordinal_encoding(training_df)
train_X = training_df[
training_df.columns.drop(self.target)
.drop("event_timestamp")
.drop("created_timestamp")
.drop("loan_id")
.drop("zipcode")
.drop("dob_ssn")
]
train_X = train_X.reindex(sorted(train_X.columns), axis=1)
train_Y = training_df.loc[:, self.target]
return train_X, train_Y
def train(self, loans):
train_X, train_Y = self._get_training_features(loans)
self.classifier.fit(train_X[sorted(train_X)], train_Y)
joblib.dump(self.classifier, self.model_filename)
def _fit_ordinal_encoder(self, requests):
self.encoder.fit(requests[self.categorical_features])
joblib.dump(self.encoder, self.encoder_filename)
def _apply_ordinal_encoding(self, requests):
requests[self.categorical_features] = self.encoder.transform(
requests[self.categorical_features]
)
def _get_online_features_from_feast(self, requests):
"""get online feature from Feast, prepare to serving
Args:
requests (_type_): data sample
Returns:
feaures: dict()
"""
zipcode = requests["zipcode"][0]
dob_ssn = requests["dob_ssn"][0]
return self.fs.get_online_features(
entity_rows=[{
"zipcode": zipcode,
"dob_ssn": dob_ssn
}],
features=self.feast_features,
).to_dict()
def predict(self, requests):
#get online features from Feast
feature_vector = self._get_online_features_from_feast(requests)
#join features to request features
features = requests.copy()
features.update(feature_vector)
features_df = pd.DataFrame.from_dict(features)
#apply ordinal encoding to categorical features
self._apply_ordinal_encoding(features_df)
#sort column
features_df = features_df[features_df.columns.drop("zipcode").drop("dob_ssn")]
#prediction
features_df["prediction"] = self.classifier.predict(features_df)
return features_df["prediction"].iloc[0]
def is_model_trained(self):
try:
check_is_fitted(self.classifier, "tree_")
except NotFittedError:
return False
return True