[verified] Task 4: lot- and cash-constrained portfolio rebalancer

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

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()