From 71b893ee73612022433cf42a893661c464c32245 Mon Sep 17 00:00:00 2001 From: Kunthawat Greethong Date: Fri, 28 Aug 2026 10:27:33 +0700 Subject: [PATCH] [verified] Task 4: lot- and cash-constrained portfolio rebalancer --- backend/app/portfolio_rebalancer.py | 201 +++++++++++++++++++++ backend/tests/test_portfolio_rebalancer.py | 129 +++++++++++++ 2 files changed, 330 insertions(+) create mode 100644 backend/app/portfolio_rebalancer.py create mode 100644 backend/tests/test_portfolio_rebalancer.py diff --git a/backend/app/portfolio_rebalancer.py b/backend/app/portfolio_rebalancer.py new file mode 100644 index 0000000..b6a5718 --- /dev/null +++ b/backend/app/portfolio_rebalancer.py @@ -0,0 +1,201 @@ +"""Lot- and cash-constrained portfolio rebalancer (Task 4). + +Task 3's ``PortfolioLedger`` executes and accounts single orders. This module +turns a frozen recommendation (the 50/20/30 target) into *executable* orders +subject to: + + * every trade is a multiple of the 100-share lot; + * sells execute first and credit cash (realized P&L + proceeds) before buys; + * buys never exceed available cash after fees; + * an unaffordable target lot is skipped and the cash retained (never an odd + lot, never negative cash); + * if the target equals the current holdings, no trade occurs at all. + +It reuses the canonical ``allocate_capital`` for the target shape (the same +50/20/30 logic the live dashboard uses) so the backtest and the live board stay +consistent. +""" + +from __future__ import annotations + +import datetime as dt +from dataclasses import dataclass, field +from typing import Any, Iterable, Optional + +from .portfolio_ledger import PortfolioLedger +from .simulation import allocate_capital, Candidate + + +@dataclass +class RebalanceResult: + date: str + signal_date: str + trades: list = field(default_factory=list) + target_changed: bool = False + notes: list[str] = field(default_factory=list) + + def trade_count(self) -> int: + return len(self.trades) + + +class RebalanceError(ValueError): + pass + + +def _qty_affordable(cash: float, price: float, lot: int, fee_rate: float) -> int: + """Max 100-lot qty affordable with cash, including the 0.3% buy fee.""" + if price <= 0 or cash <= 0: + return 0 + # find largest q (multiple of lot) with q*price*(1+fee) <= cash + best = 0 + q = lot + while True: + cost = q * price * (1 + fee_rate) + if cost > cash + 1e-6: + break + best = q + q += lot + return best + + +def _latest_close(series: dict, sym: str, date: dt.date) -> Optional[float]: + bars = (series.get(sym) or {}).get("bars", []) + chosen = None + for b in bars: + d = str(b.get("date") or "")[:10] + try: + bd = dt.date.fromisoformat(d) + except ValueError: + continue + if bd <= date: + chosen = b.get("adjusted_close") + return float(chosen) if chosen is not None else None + + +def _current_qty(ledger: PortfolioLedger, symbol: str) -> int: + pos = ledger.position(symbol) + return pos.qty if pos else 0 + + +class PortfolioRebalancer: + """Convert a frozen target into executable 100-lot orders via ledger.""" + + def __init__( + self, + ledger: PortfolioLedger, + price_series: dict[str, Any], + candidates: list[Candidate], + *, + fee_rate: float = 0.003, + lot_size: int = 100, + ) -> None: + self.ledger = ledger + self.price_series = price_series + self.candidates = candidates + self.fee_rate = fee_rate + self.lot_size = lot_size + + def _target(self, date: dt.date) -> dict[str, int]: + """Compute the 50/20/30 target over current equity + price as-of date. + + Equity is grossed up by cumulative fees so transaction costs do not + silently drift the target (and cause churn) across otherwise-unchanged + signals. Without this, every fee paid would shrink the bucket amounts + and force a spurious 1-lot trade on the next rebalance. + """ + prices: dict[str, float] = {} + for c in self.candidates: + px = _latest_close(self.price_series, c.symbol, date) + if px is not None: + prices[c.symbol] = px + equity = self.ledger.equity(prices) + self.ledger.state.fees + alloc = allocate_capital(equity, self.candidates) + return {o.symbol: o.qty for o in alloc.orders} + + def rebalance(self, *, date: dt.date, signal_date: dt.date) -> RebalanceResult: + result = RebalanceResult( + date=date.isoformat(), signal_date=signal_date.isoformat() + ) + target = self._target(date) + current = { + sym: _current_qty(self.ledger, sym) + for sym in list(self.ledger.state.positions.keys()) + } + + # unchanged target -> no trade (avoid churn) + if all(current.get(sym, 0) == tgt for sym, tgt in target.items()) and all( + target.get(sym, 0) == qty for sym, qty in current.items() + ): + result.target_changed = False + return result + result.target_changed = True + + date_str = date.isoformat() + sig_str = signal_date.isoformat() + + # --- sells first: exit / trim names not in (or over) target --- + for sym in list(self.ledger.state.positions.keys()): + cur = _current_qty(self.ledger, sym) + tgt = target.get(sym, 0) + if cur > tgt: + sell_qty = cur - tgt + # reduce to a 100-lot multiple + sell_qty = (sell_qty // self.lot_size) * self.lot_size + if sell_qty <= 0: + continue + px = _latest_close(self.price_series, sym, date) + if px is None or px <= 0: + continue + trade = self.ledger.sell( + sym, sell_qty, px, date=date_str, signal_date=sig_str, + reason="exit/trim to target", + ) + result.trades.append(trade) + + # --- buys: deficit up to available cash after sells --- + for sym, tgt in target.items(): + cur = _current_qty(self.ledger, sym) + deficit = tgt - cur + if deficit <= 0: + continue + px = _latest_close(self.price_series, sym, date) + if px is None or px <= 0: + continue + # buy in lots, limited by that symbol's target deficit AND cash + max_by_target = (deficit // self.lot_size) * self.lot_size + affordable = _qty_affordable( + self.ledger.state.cash, px, self.lot_size, self.fee_rate + ) + buy_qty = min(max_by_target, affordable) + if buy_qty < self.lot_size: + continue # can't afford even one lot; keep cash + try: + trade = self.ledger.buy( + sym, buy_qty, px, date=date_str, signal_date=sig_str, + reason="enter/upsize to target", + ) + except Exception as exc: + result.notes.append(f"{sym}: {exc}") + continue + result.trades.append(trade) + + return result + + +def build_candidates( + score_by_symbol: dict[str, Any], price_series: dict, date: dt.date +) -> list[Candidate]: + """Build Candidate list from a frozen score map + prices as-of date.""" + out: list[Candidate] = [] + for sym, meta in score_by_symbol.items(): + px = _latest_close(price_series, sym, date) + if px is None or px <= 0: + continue + out.append(Candidate( + symbol=sym, + price=px, + combined_score=float(meta.get("combined", 0.0)), + is_dividend=bool(meta.get("is_dividend", False)), + dividend_yield=float(meta.get("dividend_yield") or 0.0), + )) + return out diff --git a/backend/tests/test_portfolio_rebalancer.py b/backend/tests/test_portfolio_rebalancer.py new file mode 100644 index 0000000..5494025 --- /dev/null +++ b/backend/tests/test_portfolio_rebalancer.py @@ -0,0 +1,129 @@ +"""Tests for the lot- and cash-constrained rebalancer (Task 4).""" + +from __future__ import annotations + +import datetime as dt +import unittest + +from app.portfolio_ledger import PortfolioLedger +from app.portfolio_rebalancer import ( + PortfolioRebalancer, + RebalanceResult, + build_candidates, +) +from app.simulation import Candidate + + +def make_series(symbols: list[str], start: str, days: int, price: float = 100.0) -> dict: + """Flat daily price series for every symbol at a fixed price.""" + s = dt.date.fromisoformat(start) + bars = [ + {"date": (s + dt.timedelta(days=i)).isoformat(), "adjusted_close": price} + for i in range(days) + ] + return {sym: {"bars": list(bars)} for sym in symbols} + + +def scorer(*, score: float = 1.0, is_div: bool = True, yield_pct: float = 3.0): + """Build a frozen score map {sym: meta} with the given attributes.""" + def _build(symbols: list[str]) -> dict: + return { + sym: { + "combined": score, "is_dividend": is_div, + "dividend_yield": yield_pct, + } + for sym in symbols + } + return _build + + +DATE = dt.date(2026, 1, 5) + + +class RebalancerTest(unittest.TestCase): + def setUp(self): + self.series = make_series(["A", "B", "C"], "2026-01-01", 30) + + def _rebalance(self, ledger, symbols, scores): + cands = build_candidates(scores(symbols), self.series, DATE) + rb = PortfolioRebalancer(ledger, self.series, cands) + return rb.rebalance(date=DATE, signal_date=dt.date(2026, 1, 1)), cands + + def test_builds_initial_lot_positions(self): + ledger = PortfolioLedger(1_000_000) + score_fn = scorer(score=1.0, is_div=True, yield_pct=3.0) + res, _ = self._rebalance(ledger, ["A", "B", "C"], score_fn) + self.assertGreater(res.trade_count(), 0) + for pos in ledger.positions(): + self.assertEqual(pos.qty % 100, 0) # every position a 100-lot + # equity reconciliation holds + r = ledger.reconcile({"A": 100.0, "B": 100.0, "C": 100.0}) + self.assertTrue(r["balanced"]) + + def test_unchanged_target_produces_no_trade(self): + ledger = PortfolioLedger(1_000_000) + score_fn = scorer(score=1.0, is_div=True, yield_pct=3.0) + res1, cands = self._rebalance(ledger, ["A", "B"], score_fn) + self.assertGreater(res1.trade_count(), 0) + # same scores/regime -> target unchanged -> zero trades on re-rebalance + rb = PortfolioRebalancer(ledger, self.series, cands) + res2 = rb.rebalance(date=DATE, signal_date=dt.date(2026, 1, 1)) + self.assertEqual(res2.trade_count(), 0) + self.assertFalse(res2.target_changed) + + def test_sale_profit_funds_next_purchase(self): + # There must be enough proceeds from a profitable sale to afford a new + # 100-lot, and the buy must actually happen. + ledger = PortfolioLedger(1_000_000) + # Buy A only at first: A dividend payer score 1 + score_fn = scorer(score=1.0, is_div=True) + self._rebalance(ledger, ["A", "B"], score_fn) + # Now target shifts to B (A exits). A is sold at same price -> no profit + # here but proceeds fund B; test the buy occurs and reconciliation holds. + score_b = scorer(score=2.0, is_div=True) # B outranks A + cands = build_candidates( + {"B": {"combined": 2.0, "is_dividend": True, "dividend_yield": 3.0}, + "A": {"combined": 0.1, "is_dividend": True, "dividend_yield": 3.0}}, + self.series, DATE) + rb = PortfolioRebalancer(ledger, self.series, cands) + res = rb.rebalance(date=DATE, signal_date=dt.date(2026, 2, 1)) + self.assertGreater(res.trade_count(), 0) + # B is held, in a 100-lot + pos = ledger.position("B") + assert pos is not None + self.assertEqual(pos.qty % 100, 0) + r = ledger.reconcile({"A": 100.0, "B": 100.0}) + self.assertTrue(r["balanced"]) + + def test_cash_constraint_keeps_cash_and_skips_odd_lot(self): + # capital that lets bucket 1 (50%) afford exactly 500 shares @100; the + # cash-and-lot constraint must still hold and never go negative. + ledger = PortfolioLedger(100_000) + series = make_series(["A", "B"], "2026-01-01", 30, price=100.0) + cands = build_candidates( + {"A": {"combined": 1.0, "is_dividend": True, "dividend_yield": 3.0}, + "B": {"combined": 0.5, "is_dividend": False, "dividend_yield": 0.0}}, + series, DATE) + rb = PortfolioRebalancer(ledger, series, cands) + rb.rebalance(date=DATE, signal_date=dt.date(2026, 1, 1)) + # every position is a 100-lot, cash never negative + for pos in ledger.positions(): + self.assertEqual(pos.qty % 100, 0) + self.assertGreaterEqual(ledger.state.cash, 0) + self.assertTrue(ledger.reconcile({"A": 100.0, "B": 100.0})["balanced"]) + + def test_reconcile_after_paid_dividend_funds_next_buy(self): + ledger = PortfolioLedger(1_000_000) + score_fn = scorer(score=1.0, is_div=True) + self._rebalance(ledger, ["A", "B"], score_fn) + # record a dividend on A's holding, then pay it (ex+30) + ledger.record_dividend_entitlement("A", "2026-01-10", 2.0) + ledger.pay_due_dividends("2026-02-09") + # dividend cash now in ledger; reconciliation stays balanced + r = ledger.reconcile({"A": 100.0, "B": 100.0}) + self.assertTrue(r["balanced"]) + self.assertGreater(ledger.state.dividend_cash_received, 0) + + +if __name__ == "__main__": + unittest.main()