app.cashflow.calc_schedule() - Code Metrics - Inspection of "Greatly improved performance by using pandas data..." - whahn1983/pycashflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — master (73df59)

by William

created 2025-02-23 23:26 UTC

app.cashflow.calc_schedule() F

↳ Parent: app.cashflow

Complexity

Conditions

Size

Total Lines	210
Code Lines	168

Duplication

Lines	24
Ratio	11.43 %

Importance

Changes

Metric	Value
cc	46
eloc	168
nop	0
dl	24
loc	210
rs	0
c	0
b	0
f	0

How to fix Long Method Complexity

from app import db
from .models import Schedule, Skip
from datetime import datetime, date
import pandas as pd
import json
import plotly
import plotly.express as px
import os
from dateutil.relativedelta import relativedelta
from natsort import index_natsorted
import numpy as np
import decimal
import plotly.graph_objs as go


def update_cash(balance):
    # calculate total events for the year amount
    total = calc_schedule()

    # calculate sum of running transactions
    trans, run = calc_transactions(balance, total)

    return trans, run


def calc_schedule():
    months = 13
    weeks = 53
    years = 1
    quarters = 4
    biweeks = 27

    try:
        engine = db.create_engine(os.environ.get('DATABASE_URL')).connect()
    except:
        engine = db.create_engine('sqlite:///db.sqlite').connect()

    # pull the schedule information
    df = pd.read_sql('SELECT * FROM schedule;', engine)
    total = pd.DataFrame(columns=['type', 'name', 'amount', 'date'])

    # loop through the schedule and create transactions in a table out to the future number of years
    todaydate = datetime.today().date()
    for i in range(len(df.index)):
        format = '%Y-%m-%d'
        name = df['name'][i]
        startdate = df['startdate'][i]
        firstdate = df['firstdate'][i]
        frequency = df['frequency'][i]
        amount = df['amount'][i]
        type = df['type'][i]
        existing = Schedule.query.filter_by(name=name).first()
        if not firstdate:
            existing.firstdate = datetime.strptime(startdate, format).date()
            firstdate = existing.firstdate.strftime(format)
            db.session.commit()
        if frequency == 'Monthly':
            for k in range(months):
                futuredate = datetime.strptime(startdate, format).date() + relativedelta(months=k)
                futuredateday = futuredate.day
                firstdateday = datetime.strptime(firstdate, format).date().day
                if firstdateday > futuredateday:
                    try:
                        for m in range(3):
                            futuredateday += 1
                            if firstdateday >= futuredateday:
                                futuredate = futuredate.replace(day=futuredateday)
                    except ValueError:
                        pass
                if futuredate <= todaydate and datetime.today().weekday() < 5:

                    existing.startdate = futuredate + relativedelta(months=1)
                    daycheckdate = futuredate + relativedelta(months=1)
                    daycheck = daycheckdate.day
                    if firstdateday > daycheck:
                        try:
                            for m in range(3):
                                daycheck += 1
                                if firstdateday >= daycheck:
                                    existing.startdate = daycheckdate.replace(day=daycheck)
                        except ValueError:
                            pass
                if type == 'Income':
                    rollbackdate = datetime.combine(futuredate, datetime.min.time())

                    # Create a new row
                    new_row = {
                        'type': type,
                        'name': name,
                        'amount': amount,
                        'date': pd.tseries.offsets.BDay(1).rollback(rollbackdate).date()
                    }
                    # Append the row to the DataFrame
                    total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
                else:
                    # Create a new row
                    new_row = {
                        'type': type,
                        'name': name,
                        'amount': amount,
                        'date': (futuredate - pd.tseries.offsets.BDay(0)).date()
                    }
                    # Append the row to the DataFrame
                    total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
        elif frequency == 'Weekly':
            for k in range(weeks):
                futuredate = datetime.strptime(startdate, format).date() + relativedelta(weeks=k)
                if futuredate <= todaydate and datetime.today().weekday() < 5:
                    existing.startdate = futuredate + relativedelta(weeks=1)
                # Create a new row
                new_row = {
                    'type': type,
                    'name': name,
                    'amount': amount,
                    'date': (futuredate - pd.tseries.offsets.BDay(0)).date()
                }
                # Append the row to the DataFrame
                total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
        elif frequency == 'Yearly':
            for k in range(years):
                futuredate = datetime.strptime(startdate, format).date() + relativedelta(years=k)
                if futuredate <= todaydate and datetime.today().weekday() < 5:
                    existing.startdate = futuredate + relativedelta(years=1)
                # Create a new row
                new_row = {
                    'type': type,
                    'name': name,
                    'amount': amount,
                    'date': (futuredate - pd.tseries.offsets.BDay(0)).date()
                }

                # Append the row to the DataFrame
                total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
        elif frequency == 'Quarterly':
            for k in range(quarters):
                futuredate = datetime.strptime(startdate, format).date() + relativedelta(months=3 * k)
                futuredateday = futuredate.day
                firstdateday = datetime.strptime(firstdate, format).date().day
                if firstdateday > futuredateday:
                    try:
                        for m in range(3):
                            futuredateday += 1
                            if firstdateday >= futuredateday:
                                futuredate = futuredate.replace(day=futuredateday)
                    except ValueError:
                        pass
                if futuredate <= todaydate and datetime.today().weekday() < 5:

                    existing.startdate = futuredate + relativedelta(months=3)
                    daycheckdate = futuredate + relativedelta(months=3)
                    daycheck = daycheckdate.day
                    if firstdateday > daycheck:
                        try:
                            for m in range(3):
                                daycheck += 1
                                if firstdateday >= daycheck:
                                    existing.startdate = daycheckdate.replace(day=daycheck)
                        except ValueError:
                            pass
                # Create a new row
                new_row = {
                    'type': type,
                    'name': name,
                    'amount': amount,
                    'date': (futuredate - pd.tseries.offsets.BDay(0)).date()
                }
                # Append the row to the DataFrame
                total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
        elif frequency == 'BiWeekly':
            for k in range(biweeks):
                futuredate = datetime.strptime(startdate, format).date() + relativedelta(weeks=2 * k)
                if futuredate <= todaydate and datetime.today().weekday() < 5:
                    existing.startdate = futuredate + relativedelta(weeks=2)
                # Create a new row
                new_row = {
                    'type': type,
                    'name': name,
                    'amount': amount,
                    'date': (futuredate - pd.tseries.offsets.BDay(0)).date()
                }
                # Append the row to the DataFrame
                total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
        elif frequency == 'Onetime':
            futuredate = datetime.strptime(startdate, format).date()
            if futuredate < todaydate:
                db.session.delete(existing)
            else:
                # Create a new row
                new_row = {
                    'type': type,
                    'name': name,
                    'amount': amount,
                    'date': futuredate
                }
                # Append the row to the DataFrame
                total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
    db.session.commit()

    # add the hold items
    df = pd.read_sql('SELECT * FROM hold;', engine)
    for i in range(len(df.index)):
        name = df['name'][i]
        amount = df['amount'][i]
        type = df['type'][i]
        # Create a new row
        new_row = {
            'type': type,
            'name': name,
            'amount': amount,
            'date': todaydate + relativedelta(days=1)
        }
        # Append the row to the DataFrame
        total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)

    # add the skip items
    df = pd.read_sql('SELECT * FROM skip;', engine)
    for i in range(len(df.index)):
        format = '%Y-%m-%d'
        name = df['name'][i]
        amount = df['amount'][i]
        type = df['type'][i]
        date = df['date'][i]
        if datetime.strptime(date, format).date() < todaydate:
            skip = Skip.query.filter_by(name=name).first()
            db.session.delete(skip)
        else:
            # Create a new row
            new_row = {
                'type': type,
                'name': name,
                'amount': amount,
                'date': datetime.strptime(date, format).date()
            }
            # Append the row to the DataFrame
            total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)

    return total


def calc_transactions(balance, total):
    # retrieve the total future transactions
    df = total.sort_values(by="date", key=lambda x: np.argsort(index_natsorted(total["date"])))
    trans = pd.DataFrame(columns=['name', 'type', 'amount', 'date'])
    # collect the next 60 days of transactions for the transactions table
    format = '%Y-%m-%d'
    todaydate = datetime.today().date()
    todaydateplus = todaydate + relativedelta(months=2)
    for i in df.itertuples(index=False):
        if todaydateplus > \
                i.date > todaydate and "(SKIP)" not in i.name:
            # Create a new row from i[1]
            new_row = {
                'name': i.name,  # Accessing the 4th column value
                'type': i.type,
                'amount': i.amount,
                'date': i.date
            }
            # Append the row to the DataFrame
            trans = pd.concat([trans, pd.DataFrame([new_row])], ignore_index=True)

    # for schedules marked as expenses, make the value negative for the sum
    for i in df.itertuples(index=False):
        name = i.name
        amount = i.amount
        type = i.type
        if type == 'Expense':
            amount = float(amount) * -1
            df.loc[df['name'] == name, 'amount'] = amount
        elif type == 'Income':
            pass

    # group total transactions by date and sum the amounts for each date
    df = df.groupby("date")['amount'].sum().reset_index()

    # loop through the total transactions by date and add the sums to the total balance amount
    runbalance = balance
    run = pd.DataFrame(columns=['amount', 'date'])
    # Create a new row
    new_row = {
        'amount': runbalance,
        'date': datetime.today().date()
    }
    # Append the row to the DataFrame
    run = pd.concat([run, pd.DataFrame([new_row])], ignore_index=True)
    for i in df.itertuples(index=False):
        rundate = i.date
        amount = i.amount
        if i.date > todaydate:
            runbalance += amount
            # Create a new row
            new_row = {
                'amount': runbalance,
                'date': rundate
            }
            # Append the row to the DataFrame
            run = pd.concat([run, pd.DataFrame([new_row])], ignore_index=True)

    return trans, run


def plot_cash(run):
    # plot the running balances by date on a line plot
    df = run.sort_values(by='date', ascending=False)
    minbalance = df['amount'].min()
    minbalance = decimal.Decimal(str(minbalance)).quantize(decimal.Decimal('.01'))
    if float(minbalance) >= 0:
        minrange = 0
    else:
        minrange = float(minbalance) * 1.1
    maxbalance = 0
    todaydate = datetime.today().date()
    todaydateplus = todaydate + relativedelta(months=2)
    for i in df.itertuples(index=False):
        if todaydateplus > i.date > todaydate:
            if i.amount > maxbalance:
                maxbalance = i.amount
    maxrange = maxbalance * 1.1
    start_date = str(datetime.today().date())
    end_date = str(datetime.today().date() + relativedelta(months=2))
    layout = go.Layout(yaxis=dict(range=[minrange, maxrange]), xaxis=dict(range=[start_date, end_date]),
                       margin=dict(l=5, r=20, t=35, b=5), dragmode='pan')
    fig = px.line(df, x="date", y="amount", template="plotly", title="Cash Flow", line_shape="spline")
    fig.update_layout(layout)
    fig.update_xaxes(title_text='Date')
    fig.update_yaxes(title_text='Amount')
    fig.update_layout(paper_bgcolor="PaleTurquoise")

    graphJSON = json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)

    return minbalance, graphJSON

1		from app import db
2		from .models import Schedule, Skip
3		from datetime import datetime, date
4		import pandas as pd
5		import json
6		import plotly
7		import plotly.express as px
8		import os
9		from dateutil.relativedelta import relativedelta
10		from natsort import index_natsorted
11		import numpy as np
12		import decimal
13		import plotly.graph_objs as go
14
15
16		def update_cash(balance):
17		# calculate total events for the year amount
18		total = calc_schedule()
19
20		# calculate sum of running transactions
21		trans, run = calc_transactions(balance, total)
22
23		return trans, run
24
25
26		def calc_schedule():
27		months = 13
28		weeks = 53
29		years = 1
30		quarters = 4
31		biweeks = 27
32
33		try:
34		engine = db.create_engine(os.environ.get('DATABASE_URL')).connect()
35		except:
36		engine = db.create_engine('sqlite:///db.sqlite').connect()
37
38		# pull the schedule information
39		df = pd.read_sql('SELECT * FROM schedule;', engine)
40		total = pd.DataFrame(columns=['type', 'name', 'amount', 'date'])
41
42		# loop through the schedule and create transactions in a table out to the future number of years
43		todaydate = datetime.today().date()
44		for i in range(len(df.index)):
45		format = '%Y-%m-%d'
46		name = df['name'][i]
47		startdate = df['startdate'][i]
48		firstdate = df['firstdate'][i]
49		frequency = df['frequency'][i]
50		amount = df['amount'][i]
51		type = df['type'][i]
52		existing = Schedule.query.filter_by(name=name).first()
53		if not firstdate:
54		existing.firstdate = datetime.strptime(startdate, format).date()
55		firstdate = existing.firstdate.strftime(format)
56		db.session.commit()
57		if frequency == 'Monthly':
58		for k in range(months):
59		futuredate = datetime.strptime(startdate, format).date() + relativedelta(months=k)
60		futuredateday = futuredate.day
61		firstdateday = datetime.strptime(firstdate, format).date().day
62		if firstdateday > futuredateday:
63		try:
64		for m in range(3):
65		futuredateday += 1
66		if firstdateday >= futuredateday:
67		futuredate = futuredate.replace(day=futuredateday)
68		except ValueError:
69		pass
70	View Code Duplication	if futuredate <= todaydate and datetime.today().weekday() < 5:
		0 ignored issues – show Duplication introduced 2024-02-09 16:12 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
71		existing.startdate = futuredate + relativedelta(months=1)
72		daycheckdate = futuredate + relativedelta(months=1)
73		daycheck = daycheckdate.day
74		if firstdateday > daycheck:
75		try:
76		for m in range(3):
77		daycheck += 1
78		if firstdateday >= daycheck:
79		existing.startdate = daycheckdate.replace(day=daycheck)
80		except ValueError:
81		pass
82		if type == 'Income':
83		rollbackdate = datetime.combine(futuredate, datetime.min.time())
84
85		# Create a new row
86		new_row = {
87		'type': type,
88		'name': name,
89		'amount': amount,
90		'date': pd.tseries.offsets.BDay(1).rollback(rollbackdate).date()
91		}
92		# Append the row to the DataFrame
93		total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
94		else:
95		# Create a new row
96		new_row = {
97		'type': type,
98		'name': name,
99		'amount': amount,
100		'date': (futuredate - pd.tseries.offsets.BDay(0)).date()
101		}
102		# Append the row to the DataFrame
103		total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
104		elif frequency == 'Weekly':
105		for k in range(weeks):
106		futuredate = datetime.strptime(startdate, format).date() + relativedelta(weeks=k)
107		if futuredate <= todaydate and datetime.today().weekday() < 5:
108		existing.startdate = futuredate + relativedelta(weeks=1)
109		# Create a new row
110		new_row = {
111		'type': type,
112		'name': name,
113		'amount': amount,
114		'date': (futuredate - pd.tseries.offsets.BDay(0)).date()
115		}
116		# Append the row to the DataFrame
117		total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
118		elif frequency == 'Yearly':
119		for k in range(years):
120		futuredate = datetime.strptime(startdate, format).date() + relativedelta(years=k)
121		if futuredate <= todaydate and datetime.today().weekday() < 5:
122		existing.startdate = futuredate + relativedelta(years=1)
123		# Create a new row
124		new_row = {
125		'type': type,
126		'name': name,
127		'amount': amount,
128		'date': (futuredate - pd.tseries.offsets.BDay(0)).date()
129		}
130
131		# Append the row to the DataFrame
132		total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
133		elif frequency == 'Quarterly':
134		for k in range(quarters):
135		futuredate = datetime.strptime(startdate, format).date() + relativedelta(months=3 * k)
136		futuredateday = futuredate.day
137		firstdateday = datetime.strptime(firstdate, format).date().day
138		if firstdateday > futuredateday:
139		try:
140		for m in range(3):
141		futuredateday += 1
142		if firstdateday >= futuredateday:
143		futuredate = futuredate.replace(day=futuredateday)
144		except ValueError:
145		pass
146	View Code Duplication	if futuredate <= todaydate and datetime.today().weekday() < 5:
		0 ignored issues – show Duplication introduced 2024-02-09 16:12 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
147		existing.startdate = futuredate + relativedelta(months=3)
148		daycheckdate = futuredate + relativedelta(months=3)
149		daycheck = daycheckdate.day
150		if firstdateday > daycheck:
151		try:
152		for m in range(3):
153		daycheck += 1
154		if firstdateday >= daycheck:
155		existing.startdate = daycheckdate.replace(day=daycheck)
156		except ValueError:
157		pass
158		# Create a new row
159		new_row = {
160		'type': type,
161		'name': name,
162		'amount': amount,
163		'date': (futuredate - pd.tseries.offsets.BDay(0)).date()
164		}
165		# Append the row to the DataFrame
166		total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
167		elif frequency == 'BiWeekly':
168		for k in range(biweeks):
169		futuredate = datetime.strptime(startdate, format).date() + relativedelta(weeks=2 * k)
170		if futuredate <= todaydate and datetime.today().weekday() < 5:
171		existing.startdate = futuredate + relativedelta(weeks=2)
172		# Create a new row
173		new_row = {
174		'type': type,
175		'name': name,
176		'amount': amount,
177		'date': (futuredate - pd.tseries.offsets.BDay(0)).date()
178		}
179		# Append the row to the DataFrame
180		total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
181		elif frequency == 'Onetime':
182		futuredate = datetime.strptime(startdate, format).date()
183		if futuredate < todaydate:
184		db.session.delete(existing)
185		else:
186		# Create a new row
187		new_row = {
188		'type': type,
189		'name': name,
190		'amount': amount,
191		'date': futuredate
192		}
193		# Append the row to the DataFrame
194		total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
195		db.session.commit()
196
197		# add the hold items
198		df = pd.read_sql('SELECT * FROM hold;', engine)
199		for i in range(len(df.index)):
200		name = df['name'][i]
201		amount = df['amount'][i]
202		type = df['type'][i]
203		# Create a new row
204		new_row = {
205		'type': type,
206		'name': name,
207		'amount': amount,
208		'date': todaydate + relativedelta(days=1)
209		}
210		# Append the row to the DataFrame
211		total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
212
213		# add the skip items
214		df = pd.read_sql('SELECT * FROM skip;', engine)
215		for i in range(len(df.index)):
216		format = '%Y-%m-%d'
217		name = df['name'][i]
218		amount = df['amount'][i]
219		type = df['type'][i]
220		date = df['date'][i]
221		if datetime.strptime(date, format).date() < todaydate:
222		skip = Skip.query.filter_by(name=name).first()
223		db.session.delete(skip)
224		else:
225		# Create a new row
226		new_row = {
227		'type': type,
228		'name': name,
229		'amount': amount,
230		'date': datetime.strptime(date, format).date()
231		}
232		# Append the row to the DataFrame
233		total = pd.concat([total, pd.DataFrame([new_row])], ignore_index=True)
234
235		return total
236
237
238		def calc_transactions(balance, total):
239		# retrieve the total future transactions
240		df = total.sort_values(by="date", key=lambda x: np.argsort(index_natsorted(total["date"])))
241		trans = pd.DataFrame(columns=['name', 'type', 'amount', 'date'])
242		# collect the next 60 days of transactions for the transactions table
243		format = '%Y-%m-%d'
244		todaydate = datetime.today().date()
245		todaydateplus = todaydate + relativedelta(months=2)
246		for i in df.itertuples(index=False):
247		if todaydateplus > \
248		i.date > todaydate and "(SKIP)" not in i.name:
249		# Create a new row from i[1]
250		new_row = {
251		'name': i.name, # Accessing the 4th column value
252		'type': i.type,
253		'amount': i.amount,
254		'date': i.date
255		}
256		# Append the row to the DataFrame
257		trans = pd.concat([trans, pd.DataFrame([new_row])], ignore_index=True)
258
259		# for schedules marked as expenses, make the value negative for the sum
260		for i in df.itertuples(index=False):
261		name = i.name
262		amount = i.amount
263		type = i.type
264		if type == 'Expense':
265		amount = float(amount) * -1
266		df.loc[df['name'] == name, 'amount'] = amount
267		elif type == 'Income':
268		pass
269
270		# group total transactions by date and sum the amounts for each date
271		df = df.groupby("date")['amount'].sum().reset_index()
272
273		# loop through the total transactions by date and add the sums to the total balance amount
274		runbalance = balance
275		run = pd.DataFrame(columns=['amount', 'date'])
276		# Create a new row
277		new_row = {
278		'amount': runbalance,
279		'date': datetime.today().date()
280		}
281		# Append the row to the DataFrame
282		run = pd.concat([run, pd.DataFrame([new_row])], ignore_index=True)
283		for i in df.itertuples(index=False):
284		rundate = i.date
285		amount = i.amount
286		if i.date > todaydate:
287		runbalance += amount
288		# Create a new row
289		new_row = {
290		'amount': runbalance,
291		'date': rundate
292		}
293		# Append the row to the DataFrame
294		run = pd.concat([run, pd.DataFrame([new_row])], ignore_index=True)
295
296		return trans, run
297
298
299		def plot_cash(run):
300		# plot the running balances by date on a line plot
301		df = run.sort_values(by='date', ascending=False)
302		minbalance = df['amount'].min()
303		minbalance = decimal.Decimal(str(minbalance)).quantize(decimal.Decimal('.01'))
304		if float(minbalance) >= 0:
305		minrange = 0
306		else:
307		minrange = float(minbalance) * 1.1
308		maxbalance = 0
309		todaydate = datetime.today().date()
310		todaydateplus = todaydate + relativedelta(months=2)
311		for i in df.itertuples(index=False):
312		if todaydateplus > i.date > todaydate:
313		if i.amount > maxbalance:
314		maxbalance = i.amount
315		maxrange = maxbalance * 1.1
316		start_date = str(datetime.today().date())
317		end_date = str(datetime.today().date() + relativedelta(months=2))
318		layout = go.Layout(yaxis=dict(range=[minrange, maxrange]), xaxis=dict(range=[start_date, end_date]),
319		margin=dict(l=5, r=20, t=35, b=5), dragmode='pan')
320		fig = px.line(df, x="date", y="amount", template="plotly", title="Cash Flow", line_shape="spline")
321		fig.update_layout(layout)
322		fig.update_xaxes(title_text='Date')
323		fig.update_yaxes(title_text='Amount')
324		fig.update_layout(paper_bgcolor="PaleTurquoise")
325
326		graphJSON = json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
327
328		return minbalance, graphJSON

whahn1983 / pycashflow

Branch — master (73df59)

app.cashflow.calc_schedule() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like