Add a New Statistics Tab

expense_tracker/app.py

@@ -1,13 +1,12 @@
-from expense_tracker.core.repositories import (
-    TransactionRepository,
-    MerchantCategoryRepository,
-)
 from tkinter import Tk, ttk
 from expense_tracker.gui.main_window import MainWindow
+
 from expense_tracker.version import versions
 from expense_tracker.utils.path import get_database_path
 from expense_tracker.utils.migration import migrate_legacy_databases
-
+from expense_tracker.core.merchant_repository import MerchantCategoryRepository
+from expense_tracker.core.transaction_repository import TransactionRepository
+from expense_tracker.services.statistics import StatisticsService
 
 def main():
     """Start the Expense Tracker application."""
@@ -25,6 +24,8 @@ def main():
     merchant_repo = MerchantCategoryRepository(
         str(get_database_path("merchant_categories.db"))
     )
+    statistics_service = StatisticsService(transaction_repo)
+
     root = Tk()
     root.title("Expense Tracker")
     root.geometry("1200x700")
@@ -34,6 +35,6 @@ def main():
         tb.Style("darkly")
     except Exception:
         ttk.Style()
-    MainWindow(root, transaction_repo, merchant_repo)
+    MainWindow(root, transaction_repo, merchant_repo, statistics_service)
     root.focus_force()
     root.mainloop()

expense_tracker/core/merchant_repository.py

@@ -0,0 +1,64 @@
+import logging
+import sqlite3
+
+from expense_tracker.core.models import MerchantCategory
+
+logger = logging.getLogger(__name__)
+
+class MerchantCategoryRepository:
+    """
+    A repository for managing merchant categories.
+    """
+
+    def __init__(self, db_path: str):
+        self.conn = sqlite3.connect(db_path, check_same_thread=False)
+        self.conn.row_factory = sqlite3.Row
+        self._init_schema()
+        logger.info("Initialized merchant category database schema")
+
+    def _init_schema(self) -> None:
+        self.conn.executescript("""
+            CREATE TABLE IF NOT EXISTS merchant_categories (
+                merchant_key TEXT PRIMARY KEY,
+                category TEXT NOT NULL
+            );
+        """)
+
+    def _row_to_merchant_category(
+        self, row: sqlite3.Row | None
+    ) -> MerchantCategory | None:
+        if row is None:
+            return None
+        return MerchantCategory(
+            merchant_key=row["merchant_key"],
+            category=row["category"],
+        )
+
+    def set_category(self, merchant_category: MerchantCategory) -> None:
+        """Sets or updates the category for a given merchant key."""
+        self.conn.execute(
+            """
+            INSERT INTO merchant_categories (merchant_key, category)
+            VALUES (?, ?)
+            ON CONFLICT(merchant_key) DO UPDATE SET category=excluded.category
+        """,
+            (merchant_category.merchant_key, merchant_category.category),
+        )
+        self.conn.commit()
+
+    def get_category(self, merchant_key: str) -> MerchantCategory | None:
+        """Retrieves the category for a given merchant key."""
+        row = self.conn.execute(
+            "SELECT * FROM merchant_categories WHERE merchant_key = ?", (merchant_key,)
+        )
+        return self._row_to_merchant_category(row.fetchone())
+
+    def get_all_merchants(self) -> list[MerchantCategory]:
+        """Retrieves all merchant categories."""
+        rows = self.conn.execute("SELECT * FROM merchant_categories")
+        merchants = []
+        for row in rows.fetchall():
+            merchant = self._row_to_merchant_category(row)
+            if merchant:
+                merchants.append(merchant)
+        return merchants

expense_tracker/core/repositories.py → expense_tracker/core/transaction_repository.py

@@ -3,7 +3,7 @@
 from dataclasses import replace
 from datetime import date
 
-from expense_tracker.core.models import Transaction, MerchantCategory
+from expense_tracker.core.models import Transaction
 
 logger = logging.getLogger(__name__)
 
@@ -170,18 +170,11 @@ def update_transaction(self, transaction_id: int, data: dict) -> None:
         self.conn.execute(query, values)
         self.conn.commit()
 
-    def get_daily_spending_for_month(self, year: int, month: int) -> dict[int, float]:
+    def get_daily_spending_range(self, start_date: date, end_date: date) -> dict[int, float]:
         """
         Returns a dictionary mapping day-of-month (1-31) to total spending.
         Only includes expenses (negative amounts).
         """
-        # Create date range for the month
-        start_date = date(year, month, 1)
-        if month == 12:
-            end_date = date(year + 1, 1, 1)
-        else:
-            end_date = date(year, month + 1, 1)
-
         rows = self.conn.execute(
             """
             SELECT CAST(strftime('%d', date) AS INTEGER) as day,
@@ -198,6 +191,70 @@ def get_daily_spending_for_month(self, year: int, month: int) -> dict[int, float
         for row in rows.fetchall():
             result[row["day"]] = row["total"]
         return result
+
+    def get_monthly_cashflow_trend(self, num_months: int) -> list[tuple[int, int, float]]:
+        """
+        Returns a list of (year, month, net_amount) tuples for the past num_months.
+        Net amount is total income minus total expenses for each month.
+        Ordered by year and month ascending.
+        """
+        rows = self.conn.execute(
+            """
+            SELECT 
+                CAST(strftime('%Y', date) AS INTEGER) as year,
+                CAST(strftime('%m', date) AS INTEGER) as month,
+                SUM(amount) as net_amount
+            FROM transactions
+            GROUP BY year, month
+            ORDER BY year DESC, month DESC
+            LIMIT ?
+            """,
+            (num_months,),
+        )
+
+        result: list[tuple[int, int, float]] = []
+        for row in reversed(rows.fetchall()):
+            result.append((row["year"], row["month"], row["net_amount"]))
+        return result
+
+    def get_monthly_net_income(self, start_date: date, end_date: date) -> float:
+        """
+        Returns the net income (total income minus total expenses) for a specific month.
+        Positive amount means more income than expenses, negative means more expenses than income.
+        """
+        # Create date range for the month
+        row = self.conn.execute(
+            """
+            SELECT SUM(amount) as net_income
+            FROM transactions
+            WHERE date >= ? AND date < ?
+            """,
+            (start_date.isoformat(), end_date.isoformat()),
+        )
+        result = row.fetchone()
+        return result["net_income"] if result["net_income"] is not None else 0.0
+
+    def get_top_spending_category(self, start_date: date, end_date: date) -> tuple[str, float] | None:
+        """
+        Returns the category with the highest spending (sum of negative amounts) for a specific month.
+        Returns tuple of (category_name, total_spending) or None if no expenses exist.
+        """
+        rows = self.conn.execute(
+            """
+            SELECT category, SUM(ABS(amount)) as total
+            FROM transactions
+            WHERE date >= ? AND date < ?
+              AND amount < 0
+            GROUP BY category
+            ORDER BY total DESC
+            LIMIT 1
+            """,
+            (start_date.isoformat(), end_date.isoformat()),
+        )
+        result = rows.fetchone()
+        if result is None:
+            return None
+        return (result["category"], result["total"])
 
     def get_transactions_for_date(self, target_date: date) -> list[Transaction]:
         """
@@ -214,6 +271,48 @@ def get_transactions_for_date(self, target_date: date) -> list[Transaction]:
             if transaction:
                 transactions.append(transaction)
         return transactions
+
+    def get_latest_month_with_data(self) -> tuple[int, int]:
+        """
+        Get the most recent month that has transaction data.
+        Falls back to current month if no transactions exist.
+        """
+        # Query for all months with transactions (ordered by most recent first)
+        rows = self.conn.execute(
+            """
+            SELECT DISTINCT
+                CAST(strftime('%Y', date) AS INTEGER) as year,
+                CAST(strftime('%m', date) AS INTEGER) as month
+            FROM transactions
+            ORDER BY year DESC, month DESC
+            LIMIT 1
+            """
+        )
+        result = rows.fetchone()
+
+        if result is None:
+            # No transactions exist, default to current month
+            today = date.today()
+            return (today.year, today.month)
+
+        return (result["year"], result["month"])
+
+
+    def get_all_months_with_data(self) -> list[tuple[int, int]]:
+        """
+        Returns a list of (year, month) tuples for all months that have transaction data.
+        Ordered by year and month descending (most recent first).
+        """
+        rows = self.conn.execute(
+            """
+            SELECT DISTINCT
+                CAST(strftime('%Y', date) AS INTEGER) as year,
+                CAST(strftime('%m', date) AS INTEGER) as month
+            FROM transactions
+            ORDER BY year DESC, month DESC
+            """
+        )
+        return {(row["year"], row["month"]) for row in rows.fetchall()}
 
     def get_months_with_expenses(self) -> list[tuple[int, int]]:
         """
@@ -237,60 +336,3 @@ def get_months_with_expenses(self) -> list[tuple[int, int]]:
         return result
 
 
-class MerchantCategoryRepository:
-    """
-    A repository for managing merchant categories.
-    """
-
-    def __init__(self, db_path: str):
-        self.conn = sqlite3.connect(db_path, check_same_thread=False)
-        self.conn.row_factory = sqlite3.Row
-        self._init_schema()
-        logger.info("Initialized merchant category database schema")
-
-    def _init_schema(self) -> None:
-        self.conn.executescript("""
-            CREATE TABLE IF NOT EXISTS merchant_categories (
-                merchant_key TEXT PRIMARY KEY,
-                category TEXT NOT NULL
-            );
-        """)
-
-    def _row_to_merchant_category(
-        self, row: sqlite3.Row | None
-    ) -> MerchantCategory | None:
-        if row is None:
-            return None
-        return MerchantCategory(
-            merchant_key=row["merchant_key"],
-            category=row["category"],
-        )
-
-    def set_category(self, merchant_category: MerchantCategory) -> None:
-        """Sets or updates the category for a given merchant key."""
-        self.conn.execute(
-            """
-            INSERT INTO merchant_categories (merchant_key, category)
-            VALUES (?, ?)
-            ON CONFLICT(merchant_key) DO UPDATE SET category=excluded.category
-        """,
-            (merchant_category.merchant_key, merchant_category.category),
-        )
-        self.conn.commit()
-
-    def get_category(self, merchant_key: str) -> MerchantCategory | None:
-        """Retrieves the category for a given merchant key."""
-        row = self.conn.execute(
-            "SELECT * FROM merchant_categories WHERE merchant_key = ?", (merchant_key,)
-        )
-        return self._row_to_merchant_category(row.fetchone())
-
-    def get_all_merchants(self) -> list[MerchantCategory]:
-        """Retrieves all merchant categories."""
-        rows = self.conn.execute("SELECT * FROM merchant_categories")
-        merchants = []
-        for row in rows.fetchall():
-            merchant = self._row_to_merchant_category(row)
-            if merchant:
-                merchants.append(merchant)
-        return merchants

expense_tracker/gui/dialogs/add_expense.py

@@ -3,7 +3,7 @@
 from tkinter import ttk, messagebox
 
 from expense_tracker.core.models import Transaction
-from expense_tracker.core.repositories import TransactionRepository
+from expense_tracker.core.transaction_repository import TransactionRepository
 
 
 class AddExpenseDialog(tk.Toplevel):

expense_tracker/gui/dialogs/edit_expense.py

@@ -1,10 +1,9 @@
 import logging
 import tkinter as tk
 from tkinter import ttk, messagebox
-from expense_tracker.core.repositories import (
-    TransactionRepository,
-    MerchantCategoryRepository,
-)
+
+from expense_tracker.core.transaction_repository import TransactionRepository
+from expense_tracker.core.merchant_repository import MerchantCategoryRepository
 from expense_tracker.services.merchant import MerchantCategoryService
 from expense_tracker.utils.merchant_normalizer import normalize_merchant
 

expense_tracker/gui/dialogs/upload.py

@@ -3,10 +3,8 @@
 from datetime import datetime, date
 
 from expense_tracker.core.models import Transaction
-from expense_tracker.core.repositories import (
-    TransactionRepository,
-    MerchantCategoryRepository,
-)
+from expense_tracker.core.transaction_repository import TransactionRepository
+from expense_tracker.core.merchant_repository import MerchantCategoryRepository
 from expense_tracker.utils.extract import parse_bofa_statement_pdf
 from expense_tracker.services.merchant import MerchantCategoryService
 from expense_tracker.utils.merchant_normalizer import normalize_merchant

expense_tracker/gui/main_window.py

@@ -2,14 +2,15 @@
 from datetime import date
 from tkinter import ttk
 
-from expense_tracker.gui.tabs import TransactionsTab, HeatmapTab
+from expense_tracker.gui.tabs import TransactionsTab, HeatmapTab, StatisticsTab
 
 
 class MainWindow(tk.Frame):
-    def __init__(self, master, transaction_repo, merchant_repo):
+    def __init__(self, master, transaction_repo, merchant_repo, statistics_service):
         super().__init__(master)
         self.transaction_repo = transaction_repo
         self.merchant_repo = merchant_repo
+        self.statistics_service = statistics_service
         self.master = master
         self._active_dialog: tk.Toplevel | None = None
 
@@ -24,11 +25,15 @@ def __init__(self, master, transaction_repo, merchant_repo):
             self.notebook, transaction_repo, merchant_repo, self
         )
 
+        # Create Statistics tab
+        self.statistics_tab = StatisticsTab(self.notebook, statistics_service)
+
         # Create Heatmap tab
-        self.heatmap_tab = HeatmapTab(self.notebook, transaction_repo, self)
+        self.heatmap_tab = HeatmapTab(self.notebook, statistics_service, self)
 
         # Add tabs to notebook
         self.notebook.add(self.transactions_tab, text="Transactions")
+        self.notebook.add(self.statistics_tab, text="Statistics")
         self.notebook.add(self.heatmap_tab, text="Heatmap")
 
         # Bind tab change event for lazy loading
@@ -64,7 +69,9 @@ def _on_tab_changed(self, event):
         """Refresh tab content when user switches tabs."""
         current_tab = self.notebook.select()
         tab_index = self.notebook.index(current_tab)
-        if tab_index == 1:  # Heatmap tab
+        if tab_index == 1:  # Statistics tab
+            self.statistics_tab.refresh()
+        elif tab_index == 2:  # Heatmap tab
             self.heatmap_tab.refresh()
 
     def show_transactions_for_date(self, target_date: date):

expense_tracker/gui/tabs/__init__.py

@@ -1,4 +1,5 @@
 from .transactions_tab import TransactionsTab
 from .heatmap_tab import HeatmapTab
+from .statistics_tab import StatisticsTab
 
-__all__ = ["TransactionsTab", "HeatmapTab"]
+__all__ = ["TransactionsTab", "HeatmapTab", "StatisticsTab"]