forked from BloomTech-Labs/budget-blocks-ds
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtransactionhist.py
158 lines (138 loc) · 6.7 KB
/
transactionhist.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
150
151
152
153
154
155
156
157
158
from pydantic import BaseModel
from fastapi import HTTPException
from pydantic import BaseModel
def reCatHelper(trans: dict, newCat: str, temp_totals: dict):
"""
Helper function to retitle the Plaid categories from 'category' to 'plaid_category' and add a single budget blocks category, 'budget_blocks_category'
Also adds the BB categories to the new corresponding key
Inputs: trans - dict of a single transaction
newCat - str corresponding to BB category
temp_totals - dict where keys are the user's categories and values are each category's total
Returns: the edited JSON object of a transaction
"""
temp = trans['category'].copy()
trans['plaid_category'] = temp
# [:] Means everything in the list
del trans['category']
trans['budget_blocks_category'] = []
trans['budget_blocks_category'].append(newCat)
# Increment totals
temp_totals[newCat] += trans['amount']
return trans
class TransactionHistory(BaseModel):
transactions: list
user_id: int
# Example JSON for the transaction request
class Config:
schema_extra = {
"example": {
"transactions": [
{
"account_id": "k9VvjL1Eq3Cnk8gaQXDXt3aRVe7DaGiWekXgz",
"account_owner": 0,
"amount": 25,
"authorized_date": 0,
"category": [
"Payment",
"Credit Card"
],
"category_id": "16001000",
"date": "2020-05-15",
"iso_currency_code": "USD",
"location": {
"address": 0,
"city": 0,
"country": 0,
"lat": 0,
"lon": 0,
"postal_code": 0,
"region": 0,
"store_number": 0
},
"name": "CREDIT CARD 3333 PAYMENT *//",
"payment_channel": "other",
"payment_meta": {
"by_order_of": 0,
"payee": 0,
"payer": 0,
"payment_method": 0,
"payment_processor": 0,
"ppd_id": 0,
"reason": 0,
"reference_number": 0
},
"pending": False,
"pending_transaction_id": 0,
"transaction_code": 0,
"transaction_id": "47E1jBQLoNhA6ajvnLeLCX4NwvKQmnFd7wQ8K",
"transaction_type": "special",
"unofficial_currency_code": 0
}
],
"user_id": 1
}
}
def getCats(self, cats_dict: dict):
"""
Function to go through the transactions and categorize them
Inouts: self- a TransactionHistory object,
cats_dict - a dictionary whose keys are the budget blocks categories and the values are its corresponding Plaid categories
Output: a JSON object of all the transactions with the budget blocks categorizations, BB totals, and a user ID
"""
transactions = self.transactions
# Dictionary to store the totals of each BB category
temp_totals = {}
# Create a key in totals for each BB category so we can do += later
for cat in list(cats_dict.keys()):
temp_totals[cat] = 0
# Index into each transaction dict
for trans in transactions:
# Need to create a copy so that it keeps the original value and
# doesn't get updated by getCats
cat_list = trans['category'].copy()
numb_of_cats = len(cat_list)
# Cash Advance is the only Plaid main category with no sub
# categories, and they all are remapped to "Income"
if (numb_of_cats == 1) & (cat_list == ["Cash Advance"]):
trans = reCatHelper(trans, "Income", temp_totals)
elif (numb_of_cats == 1) & (cat_list == ["Payment"]):
trans = reCatHelper(trans, "Debt", temp_totals)
elif (numb_of_cats == 1) & (cat_list != ["Cash Advance"] or cat_list != ['Payment']):
raise HTTPException(
status_code=500,
detail=f"Contact the DS team: There was only a single category, {cat_list}, and it was not Cash Advance")
elif (numb_of_cats >= 2):
# Making it so cat_list doesn't include the first index (main
# category from Plaid)
cat_list_sliced = cat_list[1:]
dupl_test = 0
# Iterate through the lists in the dict, check if cat_list is
# in each list, and use reCatHelper to change the category
for key in list(cats_dict.keys()):
for i in cats_dict[key]:
if cat_list_sliced == i:
dupl_test += 1
# Custom exception to report if a cat_list is in 2
# or more cat_dicts lists
if dupl_test >= 2:
raise HTTPException(
status_code=500,
detail=f"Contact the DS team: A cat_list, {cat_list_sliced}, is in two or more cat_dicts lists, {trans['transactions'][0]['category'][0]} and {key}")
trans = reCatHelper(trans, key, temp_totals)
# Custom error exception to tell us if there is a plaid category we didn't account for
# Transportation is the only case where Plaid cat == Budget
# Blocks cat
if ('category' in trans):
raise HTTPException(
status_code=500,
detail=f"Contact the DS team: One of the categories from this list: {cat_list} is not accounted for")
temp_dict = {"transactions": transactions,
"user_id": self.user_id,
"totals": temp_totals}
# Make savings equal to the income minus all of the spending
total_spending = 0
for key in list(temp_totals.keys()):
if (key != "Income") & (key != "Savings"):
total_spending += temp_totals[key]
temp_totals["Savings"] = abs(temp_totals["Income"]) - total_spending
return temp_dict