399 lines
12 KiB
Python
399 lines
12 KiB
Python
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)
|