from datetime import timedelta from decimal import Decimal from django.http import HttpRequest from django.shortcuts import get_object_or_404, render from django.utils import timezone from .models import ( Account, AccountHolding, AccountSnapshot, AccountTransaction, IngestionRun, ) INCOME_TRANSACTION_TYPES = {"dividend", "interest", "income", "distribution"} REALIZED_SALE_TRANSACTION_TYPES = {"sell", "sale"} def _format_currency(value): return f"{value:,.2f}" DATE_RANGE_OPTIONS = [ ("7d", "7D"), ("30d", "30D"), ("90d", "90D"), ("1y", "1Y"), ("all", "All"), ] def _get_selected_range(request: HttpRequest): allowed_values = {value for value, _ in DATE_RANGE_OPTIONS} selected_range = request.GET.get("range", "all") if selected_range not in allowed_values: return "all" return selected_range def _get_range_cutoff(selected_range: str): now = timezone.now() if selected_range == "7d": return now - timedelta(days=7) if selected_range == "30d": return now - timedelta(days=30) if selected_range == "90d": return now - timedelta(days=90) if selected_range == "1y": return now - timedelta(days=365) return None def _filter_timeline_by_range(timeline, selected_range: str): cutoff = _get_range_cutoff(selected_range) if cutoff is None: return timeline filtered = [point for point in timeline if point["timestamp"] >= cutoff] if filtered: return filtered return timeline[-1:] if timeline else [] def _build_chart_model(timeline, width=100, height=100): if not timeline: return { "path": "", "points": [], "min_label": _format_currency(0), "max_label": _format_currency(0), "start_label": "", "end_label": "", } values = [point["value"] for point in timeline] min_value = min(values) max_value = max(values) span = max_value - min_value if span == 0: span = 1 chart_points = [] coordinates = [] point_count = len(timeline) for index, point in enumerate(timeline): if point_count == 1: x = width / 2 y = height / 2 else: x = (index / (point_count - 1)) * width normalized = (point["value"] - min_value) / span y = height - (normalized * height) change = None change_display = "N/A" if index > 0: change = point["value"] - timeline[index - 1]["value"] sign = "+" if change >= 0 else "-" change_display = f"{sign}${_format_currency(abs(change))}" chart_points.append( { "x": f"{x:.2f}", "y": f"{y:.2f}", "label": point["label"], "display_value": point["display_value"], "change_display": change_display, } ) normalized = (point["value"] - min_value) / span coordinates.append(f"{x:.2f},{y:.2f}") if len(coordinates) == 1: path = f"M 0 {height / 2:.2f} L {width:.2f} {height / 2:.2f}" else: path = "M " + " L ".join(coordinates) return { "path": path, "points": chart_points, "min_label": _format_currency(min_value), "max_label": _format_currency(max_value), "start_label": timeline[0]["short_label"], "end_label": timeline[-1]["short_label"], } def _filter_transactions_by_range(transactions, selected_range: str): cutoff = _get_range_cutoff(selected_range) if cutoff is None: return transactions return transactions.filter(date__gte=cutoff) def _build_income_chart(transactions, width=100, height=100): monthly_totals = {} for transaction in transactions: if transaction.transaction_type.lower() not in INCOME_TRANSACTION_TYPES: continue if transaction.value <= 0: continue month_key = transaction.date.strftime("%Y-%m") if month_key not in monthly_totals: monthly_totals[month_key] = { "label": transaction.date.strftime("%b %Y"), "short_label": transaction.date.strftime("%b"), "value": 0.0, } monthly_totals[month_key]["value"] += float(transaction.value) months = [ { "label": value["label"], "short_label": value["short_label"], "value": round(value["value"], 2), "display_value": _format_currency(value["value"]), } for _, value in sorted(monthly_totals.items()) ] if not months: return { "bars": [], "max_label": _format_currency(0), "min_label": _format_currency(0), } max_value = max(month["value"] for month in months) if max_value == 0: max_value = 1 bar_width = width / max(len(months), 1) bars = [] for index, month in enumerate(months): bar_height = (month["value"] / max_value) * height x = index * bar_width + (bar_width * 0.15) y = height - bar_height bars.append( { "x": f"{x:.2f}", "y": f"{y:.2f}", "width": f"{bar_width * 0.7:.2f}", "height": f"{bar_height:.2f}", "label": month["label"], "short_label": month["short_label"], "display_value": month["display_value"], } ) return { "bars": bars, "months": months, "max_label": _format_currency(max(month["value"] for month in months)), "min_label": _format_currency(0), } def _get_realized_gain_years(transactions): current_year = timezone.now().year years = [current_year] for transaction in transactions: if transaction.transaction_type.lower() not in REALIZED_SALE_TRANSACTION_TYPES: continue year = transaction.date.year if year not in years: years.append(year) return sorted(years, reverse=True) def _get_selected_realized_gain_year(request: HttpRequest, available_years): requested_year = request.GET.get("realized_year") if requested_year: try: requested_year = int(requested_year) except ValueError: requested_year = None if requested_year in available_years: return requested_year if available_years: return available_years[0] return timezone.now().year def _build_realized_gain_summary(transactions, selected_year): now = timezone.now() realized_sales = [] selected_year_total = 0.0 selected_year_sales_count = 0 for transaction in transactions: if transaction.transaction_type.lower() not in REALIZED_SALE_TRANSACTION_TYPES: continue realized_gain = transaction.realized_gain is_selected_year = transaction.date.year == selected_year if is_selected_year: selected_year_sales_count += 1 if is_selected_year and realized_gain is not None: selected_year_total += float(realized_gain) if is_selected_year: realized_sales.append( { "date": transaction.date, "name": transaction.name, "asset": transaction.asset, "quantity": transaction.quantity, "proceeds": abs(float(transaction.value)), "proceeds_display": _format_currency(abs(float(transaction.value))), "realized_gain": realized_gain, "realized_gain_display": ( _format_currency(realized_gain) if realized_gain is not None else None ), "is_gain": realized_gain is not None and realized_gain >= 0, } ) return { "sales": realized_sales, "total": selected_year_total, "total_display": _format_currency(selected_year_total), "is_gain": selected_year_total >= 0, "label": _realized_gain_label(selected_year_total, selected_year), "selected_year": selected_year, "sales_count": selected_year_sales_count, "has_sales": selected_year_sales_count > 0, } def _realized_gain_label(value, year): if value > 0: return f"{year} Realized Gain" if value < 0: return f"{year} Realized Loss" return f"{year} Realized Gain/Loss" def _build_portfolio_timeline(): latest_balances = {} timeline = [] runs = IngestionRun.objects.order_by("started_at", "created_at").prefetch_related( "accountsnapshot_set__account" ) for run in runs: snapshots = list(run.accountsnapshot_set.all()) if not snapshots: continue for snapshot in snapshots: latest_balances[snapshot.account_id] = Decimal(snapshot.balance) total_value = sum(latest_balances.values(), Decimal("0")) timestamp = run.started_at or run.created_at timeline.append( { "timestamp": timestamp, "label": timestamp.strftime("%Y-%m-%d %H:%M"), "short_label": timestamp.strftime("%b %d"), "value": float(total_value), "display_value": _format_currency(total_value), } ) return timeline def _build_account_timeline(account): snapshots = AccountSnapshot.objects.filter(account=account).order_by("snapshot_at") timeline = [] for snapshot in snapshots: timeline.append( { "timestamp": snapshot.snapshot_at, "label": snapshot.snapshot_at.strftime("%Y-%m-%d %H:%M"), "short_label": snapshot.snapshot_at.strftime("%b %d"), "value": float(snapshot.balance), "display_value": _format_currency(snapshot.balance), } ) return timeline def index(request): accounts = Account.objects.all().order_by("name") selected_range = _get_selected_range(request) portfolio_timeline = _filter_timeline_by_range( _build_portfolio_timeline(), selected_range, ) latest_portfolio_point = portfolio_timeline[-1] if portfolio_timeline else None chart_model = _build_chart_model(portfolio_timeline) context = { "accounts": accounts, "date_range_options": DATE_RANGE_OPTIONS, "selected_range": selected_range, "portfolio_timeline": portfolio_timeline, "latest_portfolio_point": latest_portfolio_point, "portfolio_chart": chart_model, } return render(request, "overviewbase.html", context) def portfolio(request, account_id): account = get_object_or_404(Account, pk=account_id) holdings = AccountHolding.objects.filter(account=account).select_related("asset") transactions = AccountTransaction.objects.filter(account=account).select_related( "asset" ) selected_range = _get_selected_range(request) active_tab = request.GET.get("tab", "holdings") if active_tab not in {"holdings", "transactions", "income", "realized_gain"}: active_tab = "holdings" account_timeline = _filter_timeline_by_range( _build_account_timeline(account), selected_range, ) latest_account_point = account_timeline[-1] if account_timeline else None chart_model = _build_chart_model(account_timeline) ranged_transactions = _filter_transactions_by_range(transactions, selected_range) income_chart = _build_income_chart(ranged_transactions) realized_gain_years = _get_realized_gain_years(transactions) selected_realized_gain_year = _get_selected_realized_gain_year( request, realized_gain_years, ) realized_gain_summary = _build_realized_gain_summary( transactions, selected_realized_gain_year, ) context = { "account": account, "date_range_options": DATE_RANGE_OPTIONS, "selected_range": selected_range, "active_tab": active_tab, "holdings": holdings, "transactions": transactions, "income_chart": income_chart, "realized_gain_years": realized_gain_years, "selected_realized_gain_year": selected_realized_gain_year, "realized_gain_summary": realized_gain_summary, "account_timeline": account_timeline, "latest_account_point": latest_account_point, "account_chart": chart_model, } return render(request, "portfolio.html", context)