"""After-tax portfolio value: "what do I keep if I sell everything today?" Model (simplified DAS) ---------------------- The account starts at 1.0 (growth-ratio units; no fixed capital). - Holdings are valued on *adjusted* prices, which already assume distributions are reinvested. Distributions therefore flow through as: dividend income -> taxed at `div_rate` capital gain dist-> taxed at `lt_rate` The after-tax remainder increases each lot's cost basis proportionally. - On rebalance, sells are FIFO. A lot is long-term if held more than 365 days at sale, else short-term; realized gains/losses are taxed at `lt_rate` / `st_rate`. - THE headline number, `equity[t]`, is the after-tax value if you LIQUIDATE the entire account on day t: equity[t] = market value + cash - tax_on_unrealized_gains(t) where tax_on_unrealized_gains marks every remaining lot to market, classifies it LT/ST by age, nets losses against gains, and applies the rates. This is what you would actually have in your pocket after selling everything and filing your taxes. Deliberately NOT modeled: loss carryover, wash sales, state rates, brackets. """ from __future__ import annotations from dataclasses import dataclass import numpy as np import pandas as pd ST_WINDOW_DAYS = 365 @dataclass class _Lot: units: float cost: float date: pd.Timestamp @dataclass class TaxResult: equity: pd.Series # after-tax value if you sell everything today (ratio) liq_tax: pd.Series # tax that a full liquidation today would owe (unrealized) taxes: pd.DataFrame # daily taxes actually PAID: div_tax, capg_tax, realized_tax, total realized: pd.DataFrame # daily realized (at rebalances): lt_gain, st_gain, lt_loss, st_loss lots_outstanding: int # final lot count (sanity check) def _liquidation_tax(lt_gain: float, lt_loss: float, st_gain: float, st_loss: float, lt_rate: float, st_rate: float) -> float: """Tax on selling all lots now; losses offset gains (same/other type).""" lt_net = lt_gain - lt_loss st_net = st_gain - st_loss if lt_net >= 0 and st_net >= 0: return lt_net * lt_rate + st_net * st_rate if lt_net < 0 and st_net < 0: return 0.0 if st_net < 0: # ST losses offset LT gains return max(lt_net + st_net, 0.0) * lt_rate return max(st_net + lt_net, 0.0) * st_rate def after_tax_portfolio(adj: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFrame, weights: dict[str, float], rebalance: str | None = None, cost_bps: float = 0.0, lt_rate: float = 0.20, st_rate: float = 0.15, div_rate: float = 0.15, start: str | None = None, end: str | None = None ) -> TaxResult: syms = [s for s in weights if s in adj.columns] p = adj[syms] d = div[[s for s in syms if s in div.columns]].reindex(index=p.index, columns=syms).fillna(0.0) c = capg[[s for s in syms if s in capg.columns]].reindex(index=p.index, columns=syms).fillna(0.0) if start or end: p = p.loc[start:end] p = p.dropna() d, c = d.reindex(p.index).fillna(0.0), c.reindex(p.index).fillna(0.0) if len(p) < 2: raise ValueError("no overlapping data for the given symbols/period") idx = p.index pv = p.values dv = d.values cv = c.values n = len(p) w0 = np.array([weights[s] for s in syms]) w0 = w0 / w0.sum() cost = cost_bps / 1e4 # state (account = 1.0 at start) units = w0 / pv[0] lots: list[list[_Lot]] = [[_Lot(units[i], float(units[i] * pv[0][i]), idx[0])] if units[i] > 0 else [] for i in range(len(syms))] cash = -float(w0.sum() * cost) # initial purchase cost tax_cols = ["div_tax", "capg_tax", "realized_tax", "total"] tax_rows = np.zeros((n, len(tax_cols))) real_cols = ["lt_gain", "st_gain", "lt_loss", "st_loss"] real_rows = np.zeros((n, len(real_cols))) equity = np.empty(n) liq_tax = np.zeros(n) # rebalance schedule (skip first day) from portfolio import _rebalance_dates rebal = (_rebalance_dates(idx, rebalance) - {idx[0]}) if rebalance else set() for t in range(n): prices = pv[t] market_value = float(np.dot(units, prices)) # --- distributions (taxed, net flows back into basis) ----------- dinc = units * dv[t] cinc = units * cv[t] d_tax = float(dinc.sum() * div_rate) c_tax = float(cinc.sum() * lt_rate) cash -= d_tax + c_tax tax_rows[t, 0] = d_tax tax_rows[t, 1] = c_tax for i in range(len(syms)): # grow cost basis with reinvested net net = (dv[t, i] * (1 - div_rate) + cv[t, i] * (1 - lt_rate)) * units[i] if net <= 0 or not lots[i]: continue tot = sum(l.units for l in lots[i]) for l in lots[i]: l.cost += net * (l.units / tot) # --- rebalance to target weights -------------------------------- if idx[t] in rebal: value = market_value target_val = w0 * value for i in range(len(syms)): cur_val = units[i] * prices[i] trade = target_val[i] - cur_val # + buy, - sell if abs(trade) < 1e-9: continue c_cost = cost * abs(trade) cash -= c_cost if trade < 0: # sell |trade| at price, FIFO to_sell = -trade / prices[i] for l in lots[i]: if to_sell <= 1e-12: break u = min(l.units, to_sell) gain = u * prices[i] - (l.cost * u / l.units) held = (idx[t] - l.date).days li = 1 if held > ST_WINDOW_DAYS else 2 if gain >= 0: tax = gain * (lt_rate if li == 1 else st_rate) real_rows[t, li - 1] += gain else: tax = gain * (lt_rate if li == 1 else st_rate) real_rows[t, 3 if li == 1 else 2] += -gain cash -= tax tax_rows[t, 2] += tax l.units -= u l.cost *= (l.units / (l.units + u)) if l.units > 0 else 0.0 to_sell -= u lots[i] = [l for l in lots[i] if l.units > 1e-12] units[i] = cur_val / prices[i] + trade / prices[i] else: # buy spend = trade + c_cost u = trade / prices[i] lots[i].append(_Lot(u, spend, idx[t])) units[i] += u # --- tax owed if we liquidate everything today ------------------ lt_gain = lt_loss = st_gain = st_loss = 0.0 for i in range(len(syms)): px = prices[i] for l in lots[i]: gain = l.units * px - l.cost if (idx[t] - l.date).days > ST_WINDOW_DAYS: if gain >= 0: lt_gain += gain else: lt_loss += -gain else: if gain >= 0: st_gain += gain else: st_loss += -gain liq_tax[t] = _liquidation_tax(lt_gain, lt_loss, st_gain, st_loss, lt_rate, st_rate) equity[t] = market_value + cash - liq_tax[t] tax_rows[t, 3] = tax_rows[t, 0] + tax_rows[t, 1] + tax_rows[t, 2] return TaxResult( equity=pd.Series(equity, index=idx, name="after_tax_liquidation"), liq_tax=pd.Series(liq_tax, index=idx, name="liquidation_tax"), taxes=pd.DataFrame(tax_rows, index=idx, columns=tax_cols), realized=pd.DataFrame(real_rows, index=idx, columns=real_cols), lots_outstanding=sum(len(l) for l in lots), )