mirror of
https://github.com/wahyd4/links.git
synced 2026-08-09 05:06:16 +10:00
OpenClaw Sub-agent
8d05a11eb6
- Portfolio management (MOMO, IBKR personal, IBKR Yanhua) - Stock holdings with transaction history (buy/sell) - Weekly AI report generation - Real-time price cache via Finnhub API - Dashboard with tree view and performance charts - REST API with AI-friendly batch update endpoint - Management command for scheduled report generation For大哥's personal investment advisor system.
381 lines
13 KiB
Python
381 lines
13 KiB
Python
"""
|
|
Service layer for the invest app.
|
|
All write operations go through here to ensure atomicity and consistent derived-field updates.
|
|
"""
|
|
import logging
|
|
from decimal import Decimal
|
|
from datetime import timedelta
|
|
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
|
|
from .models import Portfolio, Stock, Transaction, PriceCache
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PRICE_CACHE_TTL_MINUTES = 15
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Holdings helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _recompute_holding(stock: Stock) -> None:
|
|
"""
|
|
Recompute shares_held and weighted-average cost from all transactions.
|
|
Must be called inside a transaction.atomic() block.
|
|
"""
|
|
buys = stock.transactions.filter(tx_type=Transaction.TX_BUY)
|
|
sells = stock.transactions.filter(tx_type=Transaction.TX_SELL)
|
|
|
|
total_bought = sum((t.shares for t in buys), Decimal('0'))
|
|
total_sold = sum((t.shares for t in sells), Decimal('0'))
|
|
shares_held = total_bought - total_sold
|
|
|
|
# Weighted average cost based only on buy transactions
|
|
total_cost = sum((t.shares * t.price_per_share + t.fee for t in buys), Decimal('0'))
|
|
avg_cost = total_cost / total_bought if total_bought > 0 else Decimal('0')
|
|
|
|
stock.shares_held = max(shares_held, Decimal('0'))
|
|
stock.avg_cost = avg_cost
|
|
stock.save(update_fields=['shares_held', 'avg_cost', 'updated_at'])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Portfolio CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def create_portfolio(name: str, description: str = '', account_id: str = '', base_currency: str = 'USD') -> Portfolio:
|
|
return Portfolio.objects.create(
|
|
name=name,
|
|
description=description,
|
|
account_id=account_id,
|
|
base_currency=base_currency,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stock CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def upsert_stock(
|
|
portfolio: Portfolio,
|
|
ticker: str,
|
|
exchange: str = '',
|
|
company_name: str = '',
|
|
quote_currency: str = 'USD',
|
|
notes: str = '',
|
|
) -> tuple[Stock, bool]:
|
|
"""Create or update a stock holding. Returns (stock, created)."""
|
|
stock, created = Stock.objects.get_or_create(
|
|
portfolio=portfolio,
|
|
ticker=ticker.upper(),
|
|
exchange=exchange.upper(),
|
|
defaults={
|
|
'company_name': company_name,
|
|
'quote_currency': quote_currency,
|
|
'notes': notes,
|
|
},
|
|
)
|
|
if not created:
|
|
update_fields = []
|
|
if company_name and stock.company_name != company_name:
|
|
stock.company_name = company_name
|
|
update_fields.append('company_name')
|
|
if quote_currency and stock.quote_currency != quote_currency:
|
|
stock.quote_currency = quote_currency
|
|
update_fields.append('quote_currency')
|
|
if notes and stock.notes != notes:
|
|
stock.notes = notes
|
|
update_fields.append('notes')
|
|
if update_fields:
|
|
stock.save(update_fields=update_fields + ['updated_at'])
|
|
return stock, created
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Transaction CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def add_transaction(
|
|
stock: Stock,
|
|
tx_type: str,
|
|
date,
|
|
price_per_share: Decimal,
|
|
shares: Decimal,
|
|
fee: Decimal = Decimal('0'),
|
|
notes: str = '',
|
|
source: str = 'manual',
|
|
idempotency_key: str | None = None,
|
|
) -> Transaction:
|
|
"""
|
|
Add a buy or sell transaction and recompute holdings atomically.
|
|
If idempotency_key is provided, skip if already recorded.
|
|
"""
|
|
with transaction.atomic():
|
|
if idempotency_key:
|
|
existing = Transaction.objects.filter(idempotency_key=idempotency_key).first()
|
|
if existing:
|
|
logger.info("Transaction with idempotency_key=%s already exists, skipping.", idempotency_key)
|
|
return existing
|
|
|
|
tx = Transaction.objects.create(
|
|
stock=stock,
|
|
tx_type=tx_type,
|
|
date=date,
|
|
price_per_share=price_per_share,
|
|
shares=shares,
|
|
fee=fee,
|
|
notes=notes,
|
|
source=source,
|
|
idempotency_key=idempotency_key or None,
|
|
)
|
|
_recompute_holding(stock)
|
|
return tx
|
|
|
|
|
|
def delete_transaction(tx: Transaction) -> None:
|
|
"""Delete a transaction and recompute holdings atomically."""
|
|
with transaction.atomic():
|
|
stock = tx.stock
|
|
tx.delete()
|
|
_recompute_holding(stock)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AI batch update
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def ai_batch_update(operations: list[dict], dry_run: bool = False) -> list[dict]:
|
|
"""
|
|
Apply a list of AI-driven operations atomically.
|
|
|
|
Supported operation types:
|
|
- upsert_portfolio: {type, name, description, account_id, base_currency}
|
|
- upsert_stock: {type, portfolio_name, ticker, exchange, company_name, quote_currency, notes}
|
|
- add_transaction: {type, portfolio_name, ticker, exchange, tx_type, date, price_per_share,
|
|
shares, fee, notes, idempotency_key}
|
|
|
|
Returns a list of per-operation results.
|
|
"""
|
|
results = []
|
|
|
|
try:
|
|
with transaction.atomic():
|
|
for i, op in enumerate(operations):
|
|
op_type = op.get('type')
|
|
try:
|
|
result = _process_op(op)
|
|
results.append({'index': i, 'type': op_type, 'status': 'ok', 'detail': result})
|
|
except Exception as exc:
|
|
results.append({'index': i, 'type': op_type, 'status': 'error', 'detail': str(exc)})
|
|
raise # bubble up to abort the atomic block
|
|
|
|
if dry_run:
|
|
raise _DryRunAbort()
|
|
|
|
except _DryRunAbort:
|
|
pass # Rollback on dry_run is expected
|
|
|
|
return results
|
|
|
|
|
|
class _DryRunAbort(Exception):
|
|
pass
|
|
|
|
|
|
def _process_op(op: dict) -> str:
|
|
op_type = op.get('type')
|
|
|
|
if op_type == 'upsert_portfolio':
|
|
portfolio, created = Portfolio.objects.update_or_create(
|
|
name=op['name'],
|
|
defaults={
|
|
'description': op.get('description', ''),
|
|
'account_id': op.get('account_id', ''),
|
|
'base_currency': op.get('base_currency', 'USD'),
|
|
},
|
|
)
|
|
return f"Portfolio '{portfolio.name}' {'created' if created else 'updated'}"
|
|
|
|
elif op_type == 'upsert_stock':
|
|
portfolio = Portfolio.objects.get(name=op['portfolio_name'])
|
|
stock, created = upsert_stock(
|
|
portfolio=portfolio,
|
|
ticker=op['ticker'],
|
|
exchange=op.get('exchange', ''),
|
|
company_name=op.get('company_name', ''),
|
|
quote_currency=op.get('quote_currency', 'USD'),
|
|
notes=op.get('notes', ''),
|
|
)
|
|
return f"Stock '{stock.ticker}' in '{portfolio.name}' {'created' if created else 'updated'}"
|
|
|
|
elif op_type == 'add_transaction':
|
|
from datetime import date as date_cls
|
|
portfolio = Portfolio.objects.get(name=op['portfolio_name'])
|
|
stock = Stock.objects.get(
|
|
portfolio=portfolio,
|
|
ticker=op['ticker'].upper(),
|
|
exchange=op.get('exchange', '').upper(),
|
|
)
|
|
date_val = op['date']
|
|
if isinstance(date_val, str):
|
|
from datetime import datetime
|
|
date_val = datetime.strptime(date_val, '%Y-%m-%d').date()
|
|
|
|
tx = add_transaction(
|
|
stock=stock,
|
|
tx_type=op['tx_type'].upper(),
|
|
date=date_val,
|
|
price_per_share=Decimal(str(op['price_per_share'])),
|
|
shares=Decimal(str(op['shares'])),
|
|
fee=Decimal(str(op.get('fee', 0))),
|
|
notes=op.get('notes', ''),
|
|
source='ai',
|
|
idempotency_key=op.get('idempotency_key'),
|
|
)
|
|
return f"Transaction {tx.id} recorded"
|
|
|
|
else:
|
|
raise ValueError(f"Unknown operation type: {op_type!r}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Price quotes (Finnhub)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_quotes(tickers_and_exchanges: list[tuple[str, str]], force_refresh: bool = False) -> dict:
|
|
"""
|
|
Fetch quotes for a list of (ticker, exchange) tuples.
|
|
Uses PriceCache with 15-minute TTL.
|
|
Returns {ticker: {price, change_percent, prev_close, currency, cached}}.
|
|
"""
|
|
import requests
|
|
from django.conf import settings
|
|
|
|
api_key = getattr(settings, 'FINNHUB_API_KEY', '')
|
|
cutoff = timezone.now() - timedelta(minutes=PRICE_CACHE_TTL_MINUTES)
|
|
results = {}
|
|
|
|
for ticker, exchange in tickers_and_exchanges:
|
|
cache_entry = (
|
|
PriceCache.objects
|
|
.filter(ticker=ticker, exchange=exchange, fetched_at__gte=cutoff)
|
|
.order_by('-fetched_at')
|
|
.first()
|
|
)
|
|
if cache_entry and not force_refresh:
|
|
results[ticker] = {
|
|
'price': float(cache_entry.price),
|
|
'change_percent': float(cache_entry.change_percent or 0),
|
|
'prev_close': float(cache_entry.prev_close or 0),
|
|
'currency': cache_entry.currency,
|
|
'cached': True,
|
|
}
|
|
continue
|
|
|
|
# Determine Finnhub symbol
|
|
if exchange.upper() == 'HKG':
|
|
symbol = f"{ticker}.HK"
|
|
currency = 'HKD'
|
|
else:
|
|
symbol = ticker
|
|
currency = 'USD'
|
|
|
|
try:
|
|
resp = requests.get(
|
|
'https://finnhub.io/api/v1/quote',
|
|
params={'symbol': symbol, 'token': api_key},
|
|
timeout=5,
|
|
)
|
|
data = resp.json()
|
|
price = Decimal(str(data.get('c', 0)))
|
|
prev_close = Decimal(str(data.get('pc', 0)))
|
|
change_pct = (
|
|
((price - prev_close) / prev_close * 100)
|
|
if prev_close and prev_close != 0
|
|
else Decimal('0')
|
|
)
|
|
|
|
PriceCache.objects.create(
|
|
ticker=ticker,
|
|
exchange=exchange,
|
|
price=price,
|
|
currency=currency,
|
|
change_percent=change_pct,
|
|
prev_close=prev_close,
|
|
)
|
|
results[ticker] = {
|
|
'price': float(price),
|
|
'change_percent': float(change_pct),
|
|
'prev_close': float(prev_close),
|
|
'currency': currency,
|
|
'cached': False,
|
|
}
|
|
except Exception as exc:
|
|
logger.warning("Failed to fetch quote for %s: %s", symbol, exc)
|
|
results[ticker] = {'price': None, 'change_percent': None, 'prev_close': None,
|
|
'currency': currency, 'cached': False, 'error': str(exc)}
|
|
|
|
return results
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Portfolio summary
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def portfolio_summary(portfolio: Portfolio) -> dict:
|
|
"""Build a full summary dict with current prices and P&L."""
|
|
stocks = list(portfolio.stocks.prefetch_related('transactions'))
|
|
tickers = [(s.ticker, s.exchange) for s in stocks if s.shares_held > 0]
|
|
quotes = get_quotes(tickers) if tickers else {}
|
|
|
|
stock_summaries = []
|
|
total_cost = Decimal('0')
|
|
total_market_value = Decimal('0')
|
|
|
|
for stock in stocks:
|
|
cost_basis = stock.shares_held * stock.avg_cost
|
|
quote = quotes.get(stock.ticker, {})
|
|
current_price = Decimal(str(quote.get('price') or 0))
|
|
market_value = stock.shares_held * current_price
|
|
unrealized_pnl = market_value - cost_basis
|
|
unrealized_pnl_pct = (unrealized_pnl / cost_basis * 100) if cost_basis else Decimal('0')
|
|
|
|
stock_summaries.append({
|
|
'id': stock.id,
|
|
'ticker': stock.ticker,
|
|
'exchange': stock.exchange,
|
|
'company_name': stock.company_name,
|
|
'shares_held': float(stock.shares_held),
|
|
'avg_cost': float(stock.avg_cost),
|
|
'quote_currency': stock.quote_currency,
|
|
'current_price': float(current_price),
|
|
'cost_basis': float(cost_basis),
|
|
'market_value': float(market_value),
|
|
'unrealized_pnl': float(unrealized_pnl),
|
|
'unrealized_pnl_pct': float(unrealized_pnl_pct),
|
|
'change_percent': quote.get('change_percent'),
|
|
'is_active': stock.is_active,
|
|
})
|
|
|
|
total_cost += cost_basis
|
|
total_market_value += market_value
|
|
|
|
total_pnl = total_market_value - total_cost
|
|
total_pnl_pct = (total_pnl / total_cost * 100) if total_cost else Decimal('0')
|
|
|
|
return {
|
|
'portfolio': {
|
|
'id': portfolio.id,
|
|
'name': portfolio.name,
|
|
'account_id': portfolio.account_id,
|
|
'base_currency': portfolio.base_currency,
|
|
},
|
|
'stocks': stock_summaries,
|
|
'total_cost': float(total_cost),
|
|
'total_market_value': float(total_market_value),
|
|
'total_unrealized_pnl': float(total_pnl),
|
|
'total_unrealized_pnl_pct': float(total_pnl_pct),
|
|
}
|