fix weekly overview: use asof snapshot lookup and 5-day gap for last_week

This commit is contained in:
2026-04-18 23:03:33 +10:00
parent e7d24eb781
commit 8209ac34c7
+46 -26
View File
@@ -94,25 +94,53 @@ def _get_snapshot_total(date) -> Optional[float]:
def get_weekly_overview() -> dict:
"""
Compute overview from the two most recent Saturday snapshots.
Returns totals, week-over-week change, and per-portfolio rows.
Compute overview from the two most recent weekly snapshots.
'This week' = most recent snapshot date.
'Last week' = most recent snapshot date at least 5 days earlier (ensuring different week).
Per-portfolio values use as-of lookups (latest snapshot on or before the target date).
"""
# Find the 2 most recent distinct snapshot dates
seen_days: list = []
for dt in (PortfolioSnapshot.objects
.values_list('captured_at', flat=True)
.order_by('-captured_at')):
day = dt.date() if hasattr(dt, 'date') else dt
if day not in seen_days:
seen_days.append(day)
if len(seen_days) == 2:
break
latest_ts = (
PortfolioSnapshot.objects.order_by('-captured_at')
.values_list('captured_at', flat=True)
.first()
)
if not latest_ts:
return {
'this_week_total': None, 'last_week_total': None,
'this_week_date': None, 'last_week_date': None,
'week_gain': None, 'week_change_pct': None,
'portfolio_rows': [], 'portfolio_count': Portfolio.objects.count(),
}
this_week_date = seen_days[0] if len(seen_days) >= 1 else None
last_week_date = seen_days[1] if len(seen_days) >= 2 else None
this_week_date = latest_ts.date() if hasattr(latest_ts, 'date') else latest_ts
last_week_cutoff = this_week_date - timedelta(days=5)
this_week_total = _get_snapshot_total(this_week_date) if this_week_date else None
last_week_total = _get_snapshot_total(last_week_date) if last_week_date else None
prev_ts = (
PortfolioSnapshot.objects
.filter(captured_at__date__lte=last_week_cutoff)
.order_by('-captured_at')
.values_list('captured_at', flat=True)
.first()
)
last_week_date = (prev_ts.date() if hasattr(prev_ts, 'date') else prev_ts) if prev_ts else None
def _snap_asof(portfolio, date):
"""Most recent snapshot for portfolio on or before date."""
if not date:
return None
s = (
PortfolioSnapshot.objects
.filter(portfolio=portfolio, captured_at__date__lte=date)
.order_by('-captured_at')
.first()
)
return float(s.total_value) if s else None
portfolios = list(Portfolio.objects.all())
this_week_total = sum(v for p in portfolios if (v := _snap_asof(p, this_week_date)) is not None) or None
last_week_total = sum(v for p in portfolios if (v := _snap_asof(p, last_week_date)) is not None) if last_week_date else None
if last_week_total == 0:
last_week_total = None
week_gain = None
week_change_pct = None
@@ -130,16 +158,8 @@ def get_weekly_overview() -> dict:
]
portfolio_rows = []
for idx, portfolio in enumerate(Portfolio.objects.all()):
def _snap(date):
if not date:
return None
s = PortfolioSnapshot.objects.filter(
portfolio=portfolio, captured_at__date=date
).first()
return float(s.total_value) if s else None
this_val = _snap(this_week_date)
last_val = _snap(last_week_date)
this_val = _snap_asof(portfolio, this_week_date)
last_val = _snap_asof(portfolio, last_week_date)
change = change_pct = None
if this_val is not None and last_val is not None and last_val > 0: