mirror of
https://github.com/wahyd4/links.git
synced 2026-08-09 05:06:16 +10:00
433 lines
16 KiB
Python
433 lines
16 KiB
Python
"""
|
|
Service layer for the invest app.
|
|
Prices fetched from Yahoo Finance on demand via yfinance.
|
|
No cost basis or P&L tracking.
|
|
"""
|
|
import json
|
|
import logging
|
|
from decimal import Decimal
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
|
|
from django.db.models import Sum
|
|
|
|
from .models import Portfolio, Stock, PortfolioSnapshot
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# In-process price cache (5 min TTL)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_price_cache: dict[str, tuple[float, datetime]] = {}
|
|
_PRICE_CACHE_TTL_SECONDS = 300
|
|
|
|
|
|
def _get_yfinance_price(stock_code: str) -> Optional[float]:
|
|
try:
|
|
import yfinance as yf
|
|
ticker = yf.Ticker(stock_code)
|
|
hist = ticker.history(period="1d")
|
|
if hist.empty:
|
|
return None
|
|
return float(hist["Close"].iloc[-1])
|
|
except Exception as exc:
|
|
logger.warning("yfinance failed for %s: %s", stock_code, exc)
|
|
return None
|
|
|
|
|
|
def get_current_price(stock_code: str) -> Optional[float]:
|
|
now = datetime.now()
|
|
cached = _price_cache.get(stock_code)
|
|
if cached:
|
|
price, cached_at = cached
|
|
if (now - cached_at).total_seconds() < _PRICE_CACHE_TTL_SECONDS:
|
|
return price
|
|
price = _get_yfinance_price(stock_code)
|
|
if price is not None:
|
|
_price_cache[stock_code] = (price, now)
|
|
return price
|
|
if cached:
|
|
logger.info("Using stale cached price for %s", stock_code)
|
|
return cached[0]
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Portfolio value (live prices, no cost tracking)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_portfolio_value(portfolio: Portfolio) -> dict:
|
|
"""Return live holdings with current prices and total value."""
|
|
holdings = []
|
|
total_value = Decimal('0')
|
|
|
|
for stock in portfolio.stocks.filter(quantity__gt=0):
|
|
price = get_current_price(stock.stock_code) or 0.0
|
|
value = Decimal(str(price)) * stock.quantity
|
|
holdings.append({
|
|
'stock_code': stock.stock_code,
|
|
'quantity': float(stock.quantity),
|
|
'current_price': price,
|
|
'current_value': float(value),
|
|
})
|
|
total_value += value
|
|
|
|
return {
|
|
'portfolio_id': portfolio.id,
|
|
'portfolio_name': portfolio.name,
|
|
'holdings': holdings,
|
|
'total_value': float(total_value),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Weekly snapshot overview
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _get_snapshot_total(date) -> Optional[float]:
|
|
result = PortfolioSnapshot.objects.filter(
|
|
captured_at__date=date
|
|
).aggregate(total=Sum('total_value'))['total']
|
|
return float(result) if result is not None else None
|
|
|
|
|
|
def get_weekly_overview() -> dict:
|
|
"""
|
|
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).
|
|
"""
|
|
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 = latest_ts.date() if hasattr(latest_ts, 'date') else latest_ts
|
|
last_week_cutoff = this_week_date - timedelta(days=5)
|
|
|
|
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
|
|
if this_week_total is not None and last_week_total is not None and last_week_total > 0:
|
|
week_gain = this_week_total - last_week_total
|
|
week_change_pct = round((week_gain / last_week_total) * 100, 2)
|
|
|
|
# Per-portfolio breakdown
|
|
_palette = [
|
|
{'badge': 'bg-indigo-100 text-indigo-800', 'row': 'bg-indigo-50', 'border': 'border-indigo-200'},
|
|
{'badge': 'bg-emerald-100 text-emerald-800', 'row': 'bg-emerald-50', 'border': 'border-emerald-200'},
|
|
{'badge': 'bg-amber-100 text-amber-800', 'row': 'bg-amber-50', 'border': 'border-amber-200'},
|
|
{'badge': 'bg-rose-100 text-rose-800', 'row': 'bg-rose-50', 'border': 'border-rose-200'},
|
|
{'badge': 'bg-sky-100 text-sky-800', 'row': 'bg-sky-50', 'border': 'border-sky-200'},
|
|
]
|
|
portfolio_rows = []
|
|
for idx, portfolio in enumerate(Portfolio.objects.all()):
|
|
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:
|
|
change = this_val - last_val
|
|
change_pct = round((change / last_val) * 100, 2)
|
|
|
|
portfolio_rows.append({
|
|
'portfolio': portfolio,
|
|
'this_week_value': this_val,
|
|
'last_week_value': last_val,
|
|
'change': change,
|
|
'change_pct': change_pct,
|
|
'position_count': portfolio.stocks.filter(quantity__gt=0).count(),
|
|
'colors': _palette[idx % len(_palette)],
|
|
})
|
|
|
|
return {
|
|
'this_week_total': this_week_total,
|
|
'last_week_total': last_week_total,
|
|
'this_week_date': this_week_date,
|
|
'last_week_date': last_week_date,
|
|
'week_gain': week_gain,
|
|
'week_change_pct': week_change_pct,
|
|
'portfolio_rows': portfolio_rows,
|
|
'portfolio_count': Portfolio.objects.count(),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Holdings sync (AI / manual)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_all_holdings() -> list[dict]:
|
|
"""
|
|
Return live holdings for every portfolio, grouped for dashboard display.
|
|
Each entry: portfolio, portfolio_color_class, holdings (list), total_value
|
|
"""
|
|
# Assign a distinct Tailwind color set per portfolio (cycled if more than defined)
|
|
palette = [
|
|
{'badge': 'bg-indigo-100 text-indigo-800', 'row': 'bg-indigo-50', 'border': 'border-indigo-200'},
|
|
{'badge': 'bg-emerald-100 text-emerald-800', 'row': 'bg-emerald-50', 'border': 'border-emerald-200'},
|
|
{'badge': 'bg-amber-100 text-amber-800', 'row': 'bg-amber-50', 'border': 'border-amber-200'},
|
|
{'badge': 'bg-rose-100 text-rose-800', 'row': 'bg-rose-50', 'border': 'border-rose-200'},
|
|
{'badge': 'bg-sky-100 text-sky-800', 'row': 'bg-sky-50', 'border': 'border-sky-200'},
|
|
]
|
|
|
|
result = []
|
|
for idx, portfolio in enumerate(Portfolio.objects.all()):
|
|
colors = palette[idx % len(palette)]
|
|
data = get_portfolio_value(portfolio)
|
|
result.append({
|
|
'portfolio': portfolio,
|
|
'colors': colors,
|
|
'holdings': data['holdings'],
|
|
'total_value': data['total_value'],
|
|
})
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Performance chart data (cumulative % from first snapshot + benchmarks)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Simple in-process cache — benchmarks don't need to refresh every page load
|
|
_chart_cache: dict = {}
|
|
_CHART_CACHE_TTL = 900 # 15 minutes
|
|
|
|
|
|
def get_performance_chart_data() -> Optional[str]:
|
|
"""
|
|
Build Chart.js-ready JSON with cumulative % return from the earliest snapshot.
|
|
Base week = 0%. Each portfolio gets a series; S&P 500 (SPY) and QQQ added as benchmarks.
|
|
Returns a JSON string (safe to pass directly to the template) or None if no snapshots.
|
|
"""
|
|
now = datetime.now()
|
|
cached = _chart_cache.get('performance')
|
|
if cached:
|
|
data, cached_at = cached
|
|
if (now - cached_at).total_seconds() < _CHART_CACHE_TTL:
|
|
return data
|
|
|
|
result = _build_performance_chart_data()
|
|
_chart_cache['performance'] = (result, now)
|
|
return result
|
|
|
|
|
|
def _build_performance_chart_data() -> Optional[str]:
|
|
# Collect all snapshots, deduplicate by ISO week (keep latest date per portfolio per week)
|
|
# This merges HK-market Friday dates with US-market Monday dates for the same week.
|
|
from collections import defaultdict
|
|
|
|
all_snaps = list(
|
|
PortfolioSnapshot.objects.select_related('portfolio').order_by('captured_at')
|
|
)
|
|
if not all_snaps:
|
|
return None
|
|
|
|
# Group each portfolio's snapshots by ISO year-week, keep last per week
|
|
portfolio_weekly: dict = {} # portfolio_id -> {iso_week_key -> (date, value)}
|
|
for snap in all_snaps:
|
|
d = snap.captured_at.date() if hasattr(snap.captured_at, 'date') else snap.captured_at
|
|
key = d.isocalendar()[:2] # (year, week)
|
|
pid = snap.portfolio_id
|
|
if pid not in portfolio_weekly:
|
|
portfolio_weekly[pid] = {}
|
|
existing = portfolio_weekly[pid].get(key)
|
|
# keep the later date within the same week
|
|
if existing is None or d > existing[0]:
|
|
portfolio_weekly[pid][key] = (d, float(snap.total_value))
|
|
|
|
# Build the union of all week keys, sorted chronologically
|
|
all_week_keys = sorted(
|
|
{wk for pw in portfolio_weekly.values() for wk in pw}
|
|
)
|
|
if not all_week_keys:
|
|
return None
|
|
|
|
# Representative label date: latest date seen in that week across all portfolios
|
|
week_label_date: dict = {}
|
|
for pw in portfolio_weekly.values():
|
|
for wk, (d, _) in pw.items():
|
|
if wk not in week_label_date or d > week_label_date[wk]:
|
|
week_label_date[wk] = d
|
|
|
|
# Per-portfolio cumulative % series — each portfolio bases off its own first week
|
|
portfolio_colors = ['#2563EB', '#7C3AED', '#0D9488', '#DB2777', '#EA580C']
|
|
datasets = []
|
|
|
|
for idx, portfolio in enumerate(Portfolio.objects.all()):
|
|
pw = portfolio_weekly.get(portfolio.id, {})
|
|
if not pw:
|
|
continue
|
|
# base = value in the earliest week available for this portfolio
|
|
first_week = min(pw.keys())
|
|
base_val = pw[first_week][1]
|
|
if not base_val:
|
|
continue
|
|
|
|
data_pts = [
|
|
round((pw[wk][1] - base_val) / base_val * 100, 2) if wk in pw else None
|
|
for wk in all_week_keys
|
|
]
|
|
color = portfolio_colors[idx % len(portfolio_colors)]
|
|
datasets.append({
|
|
'label': portfolio.name,
|
|
'data': data_pts,
|
|
'borderColor': color,
|
|
'backgroundColor': color,
|
|
'borderWidth': 2,
|
|
'pointRadius': 5,
|
|
'pointHoverRadius': 7,
|
|
'tension': 0.3,
|
|
'borderDash': [],
|
|
'fill': False,
|
|
})
|
|
|
|
# Benchmark series — start 7 days before the earliest label date
|
|
earliest_date = week_label_date[all_week_keys[0]]
|
|
latest_date = week_label_date[all_week_keys[-1]]
|
|
start_str = (earliest_date - timedelta(days=7)).isoformat()
|
|
end_str = (latest_date + timedelta(days=5)).isoformat()
|
|
|
|
def _benchmark(ticker: str, label: str, color: str) -> Optional[dict]:
|
|
from .models import BenchmarkPrice
|
|
import datetime as dt
|
|
today = dt.date.today()
|
|
|
|
# Check DB coverage — refresh if we have no rows or latest price > 7 days stale
|
|
qs = BenchmarkPrice.objects.filter(ticker=ticker, date__gte=earliest_date - timedelta(days=7))
|
|
latest_db_date = qs.order_by('-date').values_list('date', flat=True).first()
|
|
need_refresh = latest_db_date is None or (today - latest_db_date).days > 7
|
|
|
|
if need_refresh:
|
|
try:
|
|
import yfinance as yf
|
|
hist = yf.Ticker(ticker).history(start=start_str, end=end_str)
|
|
if not hist.empty:
|
|
rows = []
|
|
for d, v in hist['Close'].items():
|
|
date_val = d.date() if hasattr(d, 'date') else d
|
|
rows.append(BenchmarkPrice(ticker=ticker, date=date_val, close=round(float(v), 4)))
|
|
BenchmarkPrice.objects.bulk_create(rows, update_conflicts=True,
|
|
unique_fields=['ticker', 'date'],
|
|
update_fields=['close'])
|
|
logger.info("invest: cached %d prices for %s", len(rows), ticker)
|
|
except Exception as exc:
|
|
logger.warning("benchmark %s yfinance fetch failed: %s", ticker, exc)
|
|
|
|
try:
|
|
closes = {
|
|
row.date: float(row.close)
|
|
for row in BenchmarkPrice.objects.filter(
|
|
ticker=ticker,
|
|
date__gte=earliest_date - timedelta(days=7),
|
|
date__lte=latest_date + timedelta(days=5),
|
|
).order_by('date')
|
|
}
|
|
if not closes:
|
|
return None
|
|
sorted_trading_days = sorted(closes.keys())
|
|
|
|
def closest_close(target):
|
|
candidates = [td for td in sorted_trading_days if td <= target]
|
|
return closes[candidates[-1]] if candidates else None
|
|
|
|
base_price = closest_close(earliest_date)
|
|
if not base_price:
|
|
return None
|
|
data_pts = [
|
|
round((closest_close(week_label_date[wk]) - base_price) / base_price * 100, 2)
|
|
if closest_close(week_label_date[wk]) is not None else None
|
|
for wk in all_week_keys
|
|
]
|
|
return {
|
|
'label': label,
|
|
'data': data_pts,
|
|
'borderColor': color,
|
|
'backgroundColor': color,
|
|
'borderWidth': 1.5,
|
|
'pointRadius': 3,
|
|
'pointHoverRadius': 5,
|
|
'tension': 0.3,
|
|
'borderDash': [5, 5],
|
|
'fill': False,
|
|
}
|
|
except Exception as exc:
|
|
logger.warning("benchmark %s failed: %s", ticker, exc)
|
|
return None
|
|
|
|
spy = _benchmark('SPY', 'S&P 500', '#D97706')
|
|
qqq = _benchmark('QQQ', 'QQQ', '#16A34A')
|
|
if spy:
|
|
datasets.append(spy)
|
|
if qqq:
|
|
datasets.append(qqq)
|
|
|
|
labels = [week_label_date[wk].strftime('%b %-d') for wk in all_week_keys]
|
|
return json.dumps({'labels': labels, 'datasets': datasets})
|
|
|
|
|
|
def ai_update_holdings(portfolio: Portfolio, holdings: list[dict], reset: bool = False) -> dict:
|
|
"""Update Stock records. No cost/price tracking."""
|
|
from django.db import transaction as db_transaction
|
|
|
|
results = []
|
|
with db_transaction.atomic():
|
|
if reset:
|
|
portfolio.stocks.all().delete()
|
|
|
|
for item in holdings:
|
|
stock_code = item['stock_code']
|
|
quantity = Decimal(str(item['quantity']))
|
|
|
|
stock, created = Stock.objects.update_or_create(
|
|
portfolio=portfolio,
|
|
stock_code=stock_code,
|
|
defaults={'quantity': quantity},
|
|
)
|
|
results.append({
|
|
'stock_code': stock_code,
|
|
'quantity': float(quantity),
|
|
'created': created,
|
|
})
|
|
|
|
return {
|
|
'portfolio_id': portfolio.id,
|
|
'portfolio_name': portfolio.name,
|
|
'reset': reset,
|
|
'results': results,
|
|
}
|