app.cashflow.calc_schedule() - Code Metrics - whahn1983/pycashflow - Measure and Improve Code Quality continuously with Scrutinizer

app.cashflow.calc_schedule() F
last analyzed 2025-06-25 16:54 UTC

↳ Parent: app.cashflow

Complexity

Conditions

Size

Total Lines	210
Code Lines	169

Duplication

Lines	24
Ratio	11.43 %

Importance

Changes

Metric	Value
cc	46
eloc	169
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 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_dict = {}

    # loop through the schedule and create transactions in a table out to the future number of years
    todaydate = datetime.today().date()
    for i in df.itertuples(index=False):
        format = '%Y-%m-%d'
        name = i.name
        startdate = i.startdate
        firstdate = i.firstdate
        frequency = i.frequency
        amount = i.amount
        type = i.type
        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_dict[len(total_dict)] = new_row

                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_dict[len(total_dict)] = new_row
        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_dict[len(total_dict)] = new_row
        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_dict[len(total_dict)] = new_row
        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_dict[len(total_dict)] = new_row
        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_dict[len(total_dict)] = new_row
        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_dict[len(total_dict)] = new_row
    db.session.commit()

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

    # add the skip items
    df = pd.read_sql('SELECT * FROM skip;', engine)
    for i in df.itertuples(index=False):
        format = '%Y-%m-%d'
        name = i.name
        amount = i.amount
        type = i.type
        date = i.date
        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_dict[len(total_dict)] = new_row

    total = pd.DataFrame.from_dict(total_dict, orient="index")

    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_dict = {}
    # collect the next 60 days of transactions for the transactions table
    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_dict[len(trans_dict)] = new_row


    trans = pd.DataFrame.from_dict(trans_dict, orient="index")

    # for schedules marked as expenses, make the value negative for the sum
    for i in df.itertuples():
        amount = i.amount
        exp_type = i.type
        if exp_type == 'Expense':
            amount = float(amount) * -1
            df.loc[i.Index, 'amount'] = amount
        elif exp_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_dict = {}
    # Create a new row
    new_row = {
        'amount': runbalance,
        'date': datetime.today().date()
    }
    # Append the row to the DataFrame
    run_dict[len(run_dict)] = new_row
    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_dict[len(run_dict)] = new_row

    run = pd.DataFrame.from_dict(run_dict, orient="index")

    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 = go.Figure(data=go.Scatter(x=df['date'].values.tolist(), y=df['amount'].values.tolist(), mode='lines', line=dict(shape='spline', smoothing=0.8)))
    fig.update_layout(layout)
    fig.update_xaxes(title_text='Date')
    fig.update_yaxes(title_text='Amount')
    fig.update_layout(paper_bgcolor="PaleTurquoise")
    fig.update_layout(title="Cash Flow")
    fig.update_layout(xaxis_type='date')
    fig.update_layout(yaxis_tickformat='$,.2f')

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

whahn1983 / pycashflow

app.cashflow.calc_schedule() F last analyzed 2025-06-25 16:54 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

app.cashflow.calc_schedule() F
last analyzed 2025-06-25 16:54 UTC