feat: insight page

app/apps/insights/utils/sankey.py

@@ -1,4 +1,3 @@
-from django.db.models import Sum
 from django.utils.translation import gettext_lazy as _
 from decimal import Decimal
 from typing import Dict, List, TypedDict
@@ -17,41 +16,64 @@ class SankeyFlow(TypedDict):
     percentage: float
 
 
-def generate_sankey_data(transactions_queryset):
+def generate_sankey_data_by_account(transactions_queryset):
     """
-    Generates Sankey diagram data from transaction queryset.
-    Uses a 1-5 scale for flows based on percentages.
+    Generates Sankey diagram data from transaction queryset using account as intermediary.
     """
-    nodes: Dict[str, SankeyNode] = {}
+    nodes: Dict[str, Dict] = {}
     flows: List[SankeyFlow] = []
 
     # Aggregate transactions
     income_data = {}  # {(category, currency, account) -> amount}
     expense_data = {}  # {(category, currency, account) -> amount}
-    total_amount = Decimal("0")
+    total_income_by_currency = {}  # {currency -> amount}
+    total_expense_by_currency = {}  # {currency -> amount}
+    total_volume_by_currency = {}  # {currency -> amount}
 
     for transaction in transactions_queryset:
         currency = transaction.account.currency
         account = transaction.account
         category = transaction.category or _("Uncategorized")
-
         key = (category, currency, account)
+        amount = transaction.amount
 
         if transaction.type == "IN":
-            income_data[key] = income_data.get(key, Decimal("0")) + transaction.amount
+            income_data[key] = income_data.get(key, Decimal("0")) + amount
+            total_income_by_currency[currency] = (
+                total_income_by_currency.get(currency, Decimal("0")) + amount
+            )
         else:
-            expense_data[key] = expense_data.get(key, Decimal("0")) + transaction.amount
+            expense_data[key] = expense_data.get(key, Decimal("0")) + amount
+            total_expense_by_currency[currency] = (
+                total_expense_by_currency.get(currency, Decimal("0")) + amount
+            )
 
-        total_amount += transaction.amount
+        total_volume_by_currency[currency] = (
+            total_volume_by_currency.get(currency, Decimal("0")) + amount
+        )
+
+    def get_node_id(node_type: str, name: str, account_id: int) -> str:
+        """Generate unique node ID."""
+        return f"{node_type}_{name}_{account_id}".lower().replace(" ", "_")
+
+    def add_node(node_id: str, display_name: str) -> None:
+        """Add node with both ID and display name."""
+        nodes[node_id] = {"id": node_id, "name": display_name}
+
+    def add_flow(
+        from_node_id: str, to_node_id: str, amount: Decimal, currency, is_income: bool
+    ) -> None:
+        """
+        Add flow with percentage based on total transaction volume for the specific currency.
+        """
+        total_volume = total_volume_by_currency.get(currency, Decimal("0"))
+        percentage = (amount / total_volume) * 100 if total_volume else 0
+        scaled_flow = percentage / 100
 
-    # Function to add flow
-    def add_flow(from_node, to_node, amount, currency):
-        percentage = (amount / total_amount) * 100 if total_amount else 0
-        scaled_flow = 1 + min(percentage / 20, 4)  # Scale 1-5, capping at 100%
         flows.append(
             {
-                "from_node": from_node,
-                "to_node": to_node,
+                "from_node": from_node_id,
+                "to_node": to_node_id,
                 "flow": float(scaled_flow),
                 "currency": {
                     "code": currency.code,
@@ -63,40 +85,164 @@ def generate_sankey_data(transactions_queryset):
                 "percentage": float(percentage),
             }
         )
-        nodes[from_node] = {"name": from_node}
-        nodes[to_node] = {"name": to_node}
 
     # Process income
     for (category, currency, account), amount in income_data.items():
-        category_name = f"{category} ({currency.code})"
-        account_name = f"{account.name} ({currency.code})"
-        add_flow(category_name, account_name, amount, currency)
+        category_node_id = get_node_id("income", category, account.id)
+        account_node_id = get_node_id("account", account.name, account.id)
+        add_node(category_node_id, str(category))
+        add_node(account_node_id, account.name)
+        add_flow(category_node_id, account_node_id, amount, currency, is_income=True)
 
     # Process expenses
     for (category, currency, account), amount in expense_data.items():
-        category_name = f"{category} ({currency.code})"
-        account_name = f"{account.name} ({currency.code})"
-        add_flow(account_name, category_name, amount, currency)
+        category_node_id = get_node_id("expense", category, account.id)
+        account_node_id = get_node_id("account", account.name, account.id)
+        add_node(category_node_id, str(category))
+        add_node(account_node_id, account.name)
+        add_flow(account_node_id, category_node_id, amount, currency, is_income=False)
 
     # Calculate and add savings flows
     savings_data = {}  # {(account, currency) -> amount}
-
     for (category, currency, account), amount in income_data.items():
         key = (account, currency)
         savings_data[key] = savings_data.get(key, Decimal("0")) + amount
-
     for (category, currency, account), amount in expense_data.items():
         key = (account, currency)
         savings_data[key] = savings_data.get(key, Decimal("0")) - amount
 
     for (account, currency), amount in savings_data.items():
         if amount > 0:
-            account_name = f"{account.name} ({currency.code})"
-            savings_name = f"{_('Savings')} ({currency.code})"
-            add_flow(account_name, savings_name, amount, currency)
+            account_node_id = get_node_id("account", account.name, account.id)
+            savings_node_id = get_node_id("savings", _("Saved"), account.id)
+            add_node(savings_node_id, str(_("Saved")))
+            add_flow(account_node_id, savings_node_id, amount, currency, is_income=True)
+
+    # Calculate total across all currencies (for reference only)
+    total_amount = sum(float(amount) for amount in total_income_by_currency.values())
 
     return {
         "nodes": list(nodes.values()),
         "flows": flows,
-        "total_amount": float(total_amount),
+        "total_amount": total_amount,
+        "total_by_currency": {
+            curr.code: float(amount)
+            for curr, amount in total_income_by_currency.items()
+        },
+    }
+
+
+def generate_sankey_data_by_currency(transactions_queryset):
+    """
+    Generates Sankey diagram data from transaction queryset, using currency as intermediary.
+    """
+    nodes: Dict[str, Dict] = {}
+    flows: List[SankeyFlow] = []
+
+    # Aggregate transactions
+    income_data = {}  # {(category, currency) -> amount}
+    expense_data = {}  # {(category, currency) -> amount}
+    total_income_by_currency = {}  # {currency -> amount}
+    total_expense_by_currency = {}  # {currency -> amount}
+    total_volume_by_currency = {}  # {currency -> amount}
+
+    for transaction in transactions_queryset:
+        currency = transaction.account.currency
+        category = transaction.category or _("Uncategorized")
+        key = (category, currency)
+        amount = transaction.amount
+
+        if transaction.type == "IN":
+            income_data[key] = income_data.get(key, Decimal("0")) + amount
+            total_income_by_currency[currency] = (
+                total_income_by_currency.get(currency, Decimal("0")) + amount
+            )
+        else:
+            expense_data[key] = expense_data.get(key, Decimal("0")) + amount
+            total_expense_by_currency[currency] = (
+                total_expense_by_currency.get(currency, Decimal("0")) + amount
+            )
+
+        total_volume_by_currency[currency] = (
+            total_volume_by_currency.get(currency, Decimal("0")) + amount
+        )
+
+    def get_node_id(node_type: str, name: str, currency_id: int) -> str:
+        """Generate unique node ID including currency information."""
+        return f"{node_type}_{name}_{currency_id}".lower().replace(" ", "_")
+
+    def add_node(node_id: str, display_name: str) -> None:
+        """Add node with both ID and display name."""
+        nodes[node_id] = {"id": node_id, "name": display_name}
+
+    def add_flow(
+        from_node_id: str, to_node_id: str, amount: Decimal, currency, is_income: bool
+    ) -> None:
+        """
+        Add flow with percentage based on total transaction volume for the specific currency.
+        """
+        total_volume = total_volume_by_currency.get(currency, Decimal("0"))
+        percentage = (amount / total_volume) * 100 if total_volume else 0
+        scaled_flow = percentage / 100
+
+        flows.append(
+            {
+                "from_node": from_node_id,
+                "to_node": to_node_id,
+                "flow": float(scaled_flow),
+                "currency": {
+                    "code": currency.code,
+                    "name": currency.name,
+                    "prefix": currency.prefix,
+                    "suffix": currency.suffix,
+                    "decimal_places": currency.decimal_places,
+                },
+                "original_amount": float(amount),
+                "percentage": float(percentage),
+            }
+        )
+
+    # Process income
+    for (category, currency), amount in income_data.items():
+        category_node_id = get_node_id("income", category, currency.id)
+        currency_node_id = get_node_id("currency", currency.name, currency.id)
+        add_node(category_node_id, str(category))
+        add_node(currency_node_id, currency.name)
+        add_flow(category_node_id, currency_node_id, amount, currency, is_income=True)
+
+    # Process expenses
+    for (category, currency), amount in expense_data.items():
+        category_node_id = get_node_id("expense", category, currency.id)
+        currency_node_id = get_node_id("currency", currency.name, currency.id)
+        add_node(category_node_id, str(category))
+        add_node(currency_node_id, currency.name)
+        add_flow(currency_node_id, category_node_id, amount, currency, is_income=False)
+
+    # Calculate and add savings flows
+    savings_data = {}  # {currency -> amount}
+    for (category, currency), amount in income_data.items():
+        savings_data[currency] = savings_data.get(currency, Decimal("0")) + amount
+    for (category, currency), amount in expense_data.items():
+        savings_data[currency] = savings_data.get(currency, Decimal("0")) - amount
+
+    for currency, amount in savings_data.items():
+        if amount > 0:
+            currency_node_id = get_node_id("currency", currency.name, currency.id)
+            savings_node_id = get_node_id("savings", _("Saved"), currency.id)
+            add_node(savings_node_id, str(_("Saved")))
+            add_flow(
+                currency_node_id, savings_node_id, amount, currency, is_income=True
+            )
+
+    # Calculate total across all currencies (for reference only)
+    total_amount = sum(float(amount) for amount in total_income_by_currency.values())
+
+    return {
+        "nodes": list(nodes.values()),
+        "flows": flows,
+        "total_amount": total_amount,
+        "total_by_currency": {
+            curr.name: float(amount)
+            for curr, amount in total_income_by_currency.items()
+        },
     }

app/apps/insights/utils/transactions.py

@@ -0,0 +1,96 @@
+from django.db.models import Q
+from django.utils import timezone
+
+from dateutil.relativedelta import relativedelta
+
+from apps.transactions.models import Transaction
+from apps.insights.forms import (
+    SingleMonthForm,
+    SingleYearForm,
+    MonthRangeForm,
+    YearRangeForm,
+    DateRangeForm,
+)
+
+
+def get_transactions(request, include_unpaid=True, include_silent=False):
+    transactions = Transaction.objects.all()
+
+    filter_type = request.GET.get("type", None)
+
+    if filter_type is not None:
+        if filter_type == "month":
+            form = SingleMonthForm(request.GET)
+
+            if form.is_valid():
+                month = form.cleaned_data["month"].replace(day=1)
+            else:
+                month = timezone.localdate(timezone.now()).replace(day=1)
+
+            transactions = transactions.filter(
+                reference_date__month=month.month, reference_date__year=month.year
+            )
+        elif filter_type == "year":
+            form = SingleYearForm(request.GET)
+            if form.is_valid():
+                year = form.cleaned_data["year"].replace(day=1, month=1)
+            else:
+                year = timezone.localdate(timezone.now()).replace(day=1, month=1)
+
+            transactions = transactions.filter(reference_date__year=year.year)
+        elif filter_type == "month-range":
+            form = MonthRangeForm(request.GET)
+            if form.is_valid():
+                month_from = form.cleaned_data["month_from"].replace(day=1)
+                month_to = form.cleaned_data["month_to"].replace(day=1)
+            else:
+                month_from = timezone.localdate(timezone.now()).replace(day=1)
+                month_to = (
+                    timezone.localdate(timezone.now()) + relativedelta(months=1)
+                ).replace(day=1)
+
+            transactions = transactions.filter(
+                reference_date__gte=month_from,
+                reference_date__lte=month_to,
+            )
+        elif filter_type == "year-range":
+            form = YearRangeForm(request.GET)
+            if form.is_valid():
+                year_from = form.cleaned_data["year_from"].replace(day=1, month=1)
+                year_to = form.cleaned_data["year_to"].replace(day=31, month=12)
+            else:
+                year_from = timezone.localdate(timezone.now()).replace(day=1, month=1)
+                year_to = (
+                    timezone.localdate(timezone.now()) + relativedelta(years=1)
+                ).replace(day=31, month=12)
+
+            transactions = transactions.filter(
+                reference_date__gte=year_from,
+                reference_date__lte=year_to,
+            )
+        elif filter_type == "date-range":
+            form = DateRangeForm(request.GET)
+            if form.is_valid():
+                date_from = form.cleaned_data["date_from"]
+                date_to = form.cleaned_data["date_to"]
+            else:
+                date_from = timezone.localdate(timezone.now())
+                date_to = timezone.localdate(timezone.now()) + relativedelta(months=1)
+
+            transactions = transactions.filter(
+                date__gte=date_from,
+                date__lte=date_to,
+            )
+    else:  # Default to current month
+        month = timezone.localdate(timezone.now())
+        transactions = transactions.filter(
+            reference_date__month=month.month, reference_date__year=month.year
+        )
+
+    if not include_unpaid:
+        transactions = transactions.filter(is_paid=True)
+
+    if not include_silent:
+        transactions = transactions.exclude(Q(category__mute=True) & ~Q(category=None))
+
+    return transactions