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, IngestionRun 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 _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") selected_range = _get_selected_range(request) 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) context = { "account": account, "date_range_options": DATE_RANGE_OPTIONS, "selected_range": selected_range, "holdings": holdings, "account_timeline": account_timeline, "latest_account_point": latest_account_point, "account_chart": chart_model, } return render(request, "portfolio.html", context)