Project 3: Personal Finance Tracker with Database0%

Project 3: Personal Finance Tracker with Database

Beginner12 min read•Updated: Jul 11, 2026
Study Materials

Capstone Project: Personal Finance Tracker with Database

Managing financial transactions, portfolio allocations, and monthly expense budgets requires data consistency, auditability, and mathematical precision. A single rounding error or failed transfer between accounts can lead to corrupt ledgers and unaccounted funds.

In this capstone project, we will construct a production-ready Personal Finance & Investment Ledger Engine. Built on modern SQLAlchemy 2.0, it implements atomic double-entry transfer transactions, monthly budget tracking, functional cashflow aggregation with itertools and functools, and an automated Pytest verification suite.

1. System Architecture

The ledger engine enforces strict relational integrity and transactional boundaries:

Output
Personal Finance Management Service Layer
│
┌──────────────────────────┴──────────────────────────┐
▼ ▼
Direct Expense / Income Double-Entry Transfer
Transaction (Account A ──► Account B)
│ │
▼ ▼
Assign Category & Account Atomic Transaction Scope
│ (with session.begin():)
▼ │
Check Category Budget ┌───────┴───────┐
Threshold Level ▼ ▼
│ Deduct from A Credit to B
▼ │ │
Commit Record to Database └───────┬───────┘
▼
Audit Record Committed

2. Production Implementation

Python
from datetime import datetime
from decimal import Decimal
import functools
import itertools
import json
import operator
from typing import Any, Dict, List, Optional
from sqlalchemy import Date, ForeignKey, Integer, Numeric, String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship, selectinload, sessionmaker
 
# -------------------------------------------------------------
# 1. SQLAlchemy 2.0 Relational Data Schema
# -------------------------------------------------------------
class Base(DeclarativeBase):
pass
 
class Account(Base):
__tablename__ = "accounts"
 
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
account_type: Mapped[str] = mapped_column(String(20), nullable=False) # CHECKING, SAVINGS, INVESTMENT
balance: Mapped[Decimal] = mapped_column(Numeric(12, 2), default=Decimal("0.00"))
 
transactions: Mapped[List["Transaction"]] = relationship(back_populates="account", cascade="all, delete-orphan")
 
def __repr__(self) -> str:
return f"Account(name={self.name!r}, balance={self.balance})"
 
class Category(Base):
__tablename__ = "categories"
 
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
monthly_budget: Mapped[Decimal] = mapped_column(Numeric(10, 2), default=Decimal("0.00"))
 
transactions: Mapped[List["Transaction"]] = relationship(back_populates="category")
 
def __repr__(self) -> str:
return f"Category(name={self.name!r}, budget={self.monthly_budget})"
 
class Transaction(Base):
__tablename__ = "transactions"
 
id: Mapped[int] = mapped_column(primary_key=True)
amount: Mapped[Decimal] = mapped_column(Numeric(10, 2), nullable=False)
transaction_type: Mapped[str] = mapped_column(String(10), nullable=False) # INCOME, EXPENSE, TRANSFER
description: Mapped[str] = mapped_column(String(200), nullable=False)
date: Mapped[datetime] = mapped_column(default=datetime.utcnow)
 
account_id: Mapped[int] = mapped_column(ForeignKey("accounts.id"), nullable=False)
account: Mapped["Account"] = relationship(back_populates="transactions")
 
category_id: Mapped[Optional[int]] = mapped_column(ForeignKey("categories.id"), nullable=True)
category: Mapped[Optional["Category"]] = relationship(back_populates="transactions")
 
def __repr__(self) -> str:
return f"Transaction({self.transaction_type}, ${self.amount}, desc={self.description!r})"
 
# -------------------------------------------------------------
# 2. Finance Ledger Business Service
# -------------------------------------------------------------
class FinanceService:
"""Provides atomic operations for managing accounts, categories, and transactions."""
 
def __init__(self, session_factory: sessionmaker[Session]) -> None:
self.session_factory = session_factory
 
def create_account(self, name: str, account_type: str, initial_balance: Decimal) -> int:
with self.session_factory() as session:
with session.begin():
account = Account(name=name, account_type=account_type, balance=initial_balance)
session.add(account)
session.flush()
return account.id
 
def create_category(self, name: str, monthly_budget: Decimal) -> int:
with self.session_factory() as session:
with session.begin():
category = Category(name=name, monthly_budget=monthly_budget)
session.add(category)
session.flush()
return category.id
 
def record_expense(self, account_name: str, category_name: str, amount: Decimal, description: str) -> int:
"""Records an expense and checks if category budget has been exceeded."""
with self.session_factory() as session:
with session.begin():
account = session.scalars(select(Account).where(Account.name == account_name)).one()
category = session.scalars(select(Category).where(Category.name == category_name)).one()
 
if account.balance < amount:
raise ValueError(f"Insufficient funds in '{account_name}'. Current balance: ${account.balance}")
 
# Deduct account balance
account.balance -= amount
 
# Record transaction
tx = Transaction(
amount=amount,
transaction_type="EXPENSE",
description=description,
account_id=account.id,
category_id=category.id
)
session.add(tx)
session.flush()
 
# Calculate cumulative monthly spending in this category using select
month_start = datetime(datetime.utcnow().year, datetime.utcnow().month, 1)
tx_stmt = select(Transaction).where(
Transaction.category_id == category.id,
Transaction.transaction_type == "EXPENSE",
Transaction.date >= month_start
)
cat_txs = list(session.scalars(tx_stmt))
# Sum spending via functools.reduce and operator.add
total_spent = functools.reduce(
operator.add,
(t.amount for t in cat_txs),
Decimal("0.00")
)
 
if category.monthly_budget > 0 and total_spent > category.monthly_budget:
print(
f"[BUDGET ALERT] Spending in '{category.name}' (${total_spent}) "
f"exceeded monthly limit (${category.monthly_budget})!"
)
 
return tx.id
 
def execute_transfer(self, source_account: str, target_account: str, amount: Decimal, description: str) -> bool:
"""Atomic double-entry transfer between two accounts with rollback guarantee."""
with self.session_factory() as session:
with session.begin():
src = session.scalars(select(Account).where(Account.name == source_account)).one()
dst = session.scalars(select(Account).where(Account.name == target_account)).one()
 
if src.balance < amount:
raise ValueError(f"Transfer rejected: Insufficient funds in {source_account}.")
 
# Double-entry balance adjustments
src.balance -= amount
dst.balance += amount
 
# Record transfer transaction pair
tx_out = Transaction(
amount=amount,
transaction_type="TRANSFER",
description=f"Transfer to {target_account}: {description}",
account_id=src.id
)
tx_in = Transaction(
amount=amount,
transaction_type="TRANSFER",
description=f"Transfer from {source_account}: {description}",
account_id=dst.id
)
session.add_all([tx_out, tx_in])
 
print(f"[TRANSFER SUCCESS] Transferred ${amount} from {source_account} to {target_account}.")
return True
 
def generate_financial_statement(self) -> Dict[str, Any]:
"""Generates aggregate balance statements and category breakdowns."""
with self.session_factory() as session:
accounts = list(session.scalars(select(Account).options(selectinload(Account.transactions))))
categories = list(session.scalars(select(Category).options(selectinload(Category.transactions))))
 
net_worth = sum((acc.balance for acc in accounts), Decimal("0.00"))
 
# Group all expenses by category name
all_expenses = session.scalars(
select(Transaction).where(Transaction.transaction_type == "EXPENSE").options(selectinload(Transaction.category))
).all()
 
# Sort before grouping for itertools.groupby
sorted_expenses = sorted(all_expenses, key=lambda t: t.category.name if t.category else "Uncategorized")
category_spending = {}
for cat_name, group in itertools.groupby(sorted_expenses, key=lambda t: t.category.name if t.category else "Uncategorized"):
total = sum((t.amount for t in group), Decimal("0.00"))
category_spending[cat_name] = str(total)
 
return {
"net_worth": str(net_worth),
"accounts": {acc.name: str(acc.balance) for acc in accounts},
"category_spending": category_spending
}

3. Verification & Execution Benchmark

Python
def main():
print("=====================================================")
print(" INITIALIZING PERSONAL FINANCE LEDGER TEST ")
print("=====================================================")
 
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
SessionFactory = sessionmaker(bind=engine)
 
service = FinanceService(SessionFactory)
 
# 1. Initialize Accounts & Budgets
service.create_account("Primary Checking", "CHECKING", Decimal("2500.00"))
service.create_account("High-Yield Savings", "SAVINGS", Decimal("10000.00"))
service.create_account("Investment Brokerage", "INVESTMENT", Decimal("5000.00"))
 
service.create_category("Groceries", Decimal("400.00"))
service.create_category("Cloud Services", Decimal("150.00"))
 
print("[SYSTEM] Accounts and budget categories established.")
 
# 2. Record Expenses (Including Budget Overflow Trigger)
service.record_expense("Primary Checking", "Groceries", Decimal("120.50"), "Weekly supermarket run")
service.record_expense("Primary Checking", "Cloud Services", Decimal("99.00"), "AWS hosting bill")
# This expense breaches the Cloud Services $150 budget limit ($99 + $80 = $179 > $150)
service.record_expense("Primary Checking", "Cloud Services", Decimal("80.00"), "Database cluster hosting")
 
# 3. Double-Entry Transfer
print("\n--- Executing Atomic Transfer ---")
service.execute_transfer(
source_account="Primary Checking",
target_account="Investment Brokerage",
amount=Decimal("500.00"),
description="Monthly index fund DCA contribution"
)
 
# 4. Generate Financial Statement
print("\n=====================================================")
print(" CONSOLIDATED NET WORTH REPORT ")
print("=====================================================")
statement = service.generate_financial_statement()
print(json.dumps(statement, indent=2))
print("=====================================================")
 
if __name__ == "__main__":
main()

4. Key Architectural Patterns

  1. 1
    Decimal for Exact Financial Math: Floating-point types (float) introduce precision errors like 0.1 + 0.2 != 0.3. Using Python's Decimal backed by SQLAlchemy's Numeric(12, 2) guarantees exact cent-level precision.
  2. 2
    Double-Entry Bookkeeping: Transfers update source and target accounts together in an atomic with session.begin(): block. If either account update fails, both are rolled back.
  3. 3
    Budget Overflow Alerts: The system calculates running month-to-date category totals via functools.reduce upon each expense recording, providing immediate over-budget detection.

Multiple Choice Questions

1.

Why must financial applications use Python's Decimal type instead of standard float for monetary calculations? A. Decimal runs faster on GPUs. B. Standard binary floating-point numbers cannot accurately represent base-10 fractions (e.g. 0.1), causing accumulated rounding discrepancies, whereas Decimal provides exact base-10 precision. C. float values cannot be stored in SQL databases. D. Decimal automatically converts currencies.

Answer: B
Explanation:In binary floating-point representation, numbers like 0.1 and 0.2 have repeating fractional values that introduce precision bugs. Decimal uses fixed-point base-10 arithmetic, guaranteeing exact financial accuracy.

2.

How does the FinanceService.execute_transfer method guarantee that funds are never debited from the source account without being credited to the destination account? A. By placing both balance updates and transaction records inside an atomic with session.begin(): block that rolls back completely if any step fails. B. By writing to a text file. C. By locking the operating system kernel. D. By delaying transfers by 24 hours.

Answer: A
Explanation:Wrapping the transfer in a single database transaction guarantees Atomicity: either both the debit and the credit commit successfully, or both are discarded.

3.

What role does selectinload(Account.transactions) play in generate_financial_statement? A. It calculates the interest rate. B. It pre-fetches all associated transactions in a single bulk query, eliminating the N+1 query problem when iterating over accounts. C. It deletes duplicate transactions. D. It sorts transactions alphabetically.

Answer: B
Explanation:selectinload prevents the N+1 problem by eagerly preloading child collections in a bulk query before iterations begin.

4.

What is the effect of cascade="all, delete-orphan" on Account.transactions? A. Transactions cannot be deleted. B. Deleting an Account automatically deletes all its child Transaction rows, preventing dangling foreign key references. C. It hides transactions from reports. D. It encrypts transaction descriptions.

Answer: B
Explanation:The delete-orphan cascade ensures that if an account is removed, all of its child transaction records are automatically purged from the database.

5.

Which standard functional tool is used to aggregate total category expenses into a cumulative sum? A. functools.partial B. functools.reduce(operator.add, (t.amount for t in cat_txs), Decimal("0.00")) C. itertools.cycle D. itertools.permutations

Answer: B
Explanation:functools.reduce combines an accumulator function (operator.add) across an iterable of Decimal amounts starting from an initial value of zero.

Related Lessons

Previous LessonNext Lesson
Project 2: API-based Weather DashboardNone

Practice Quiz

Test your understanding of this lesson with 5 questions. Each question has one correct answer.

Previous
Prev