"""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 RAW (close) prices, in raw share units. Distributions (per-share dollars from the event files) are taxed on their event date and the after-tax remainder is REINVESTED at the same day's raw close (the ex-div price): a fund paying d drops by d, the investor keeps d(1-tau) and buys it back cheaper. The tax's effect is carried by the (smaller) reinvested units; it is recorded in `taxes` but not separately deducted from cash (that would double-count it). dividend income -> taxed at (div_rate + niit + sl_rate) capital gain dist-> taxed at (lt_rate + niit + sl_rate) `niit` = federal 3.8% Net Investment Income Tax (0 if MAGI is under the threshold). `sl_rate` = STATE+LOCAL marginal rate (e.g. NY+NYC) applied at ORDINARY rates to every flow: state and local have no preferential capital-gains rate, so a NYC resident pays city + state on top of the federal rate for the same dollar. The after-tax remainder also increases each lot's cost basis proportionally (standard after-tax-IRR convention, so a liquidation doesn't re-tax the distribution). - 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. NOTE: distribution event dates must be EX-DIV dates for the reinvestment pricing to be right (the raw close on that day already reflects the drop). Yahoo usually dates events on the ex-div date; the record-date misalignments found by scripts/scan_adj_misalign.py are corrected by remove/add ops in overrides/corrections/ where a symbol is analyzed. Losses realized along the way are carried forward (loss pools) and offset later gains; repurchases within WASh_WINDOW_DAYS of a loss sale of the same security disallow the loss (wash sale). Progressive federal brackets are used when `progressive` is on (marginal rate at the investor's taxable income position, starting at `fed_base`, less the `salt` deduction); otherwise the flat lt/st/div rates are used. """ from __future__ import annotations from dataclasses import dataclass import numpy as np import pandas as pd ST_WINDOW_DAYS = 365 WASh_WINDOW_DAYS = 30 # Federal ordinary / LTCG brackets (2025, MFJ) for the progressive mode: # marginal federal rate at a given taxable ordinary income. LTCG: 0% to # $96,700 / 15% to ~$599k / 20% above. FED_BRACKETS_ORD: list[tuple[float, float]] = [ (0.0, 0.10), (23625.0, 0.12), (95350.0, 0.22), (206700.0, 0.24), (394600.0, 0.32), (684950.0, 0.35), (751600.0, 0.37), ] FED_BRACKETS_LT: list[tuple[float, float]] = [ (0.0, 0.00), (96700.0, 0.15), (599000.0, 0.20), ] def fed_marginal_rate(income: float, brackets: list[tuple[float, float]]) -> float: """Marginal federal rate at a given taxable ordinary income.""" rate = brackets[0][1] for limit, r in sorted(brackets): if income > limit: rate = r return rate def _loss_wash_disallowed(i: int, ts: pd.Timestamp, buys: list[list[pd.Timestamp]]) -> bool: """True if the security `i` was repurchased within WASh_WINDOW_DAYS before (or on) a loss sale on day `ts` - wash sale, loss disallowed.""" for b in buys[i]: delta = ts - b if pd.Timedelta(0) <= delta <= pd.Timedelta(days=WASh_WINDOW_DAYS): return True return False @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_net: float, st_net: float, lt_rate: float, st_rate: float) -> float: """Tax on selling all lots now; losses offset gains (same/other type). lt_net / st_net are net of realized losses (and carried loss pools).""" 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(close: 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, niit: float = 0.0, sl_rate: float = 0.0, start: str | None = None, end: str | None = None, progressive: bool = False, fed_base: float = 0.0, salt: float = 0.0 ) -> TaxResult: """`close` must be RAW close prices (bundle.close), not adj: the per-share distribution dollars are raw, so units must be raw shares. All rates are decimals; realized gains are taxed at (st/lt_rate + niit + sl_rate).""" syms = [s for s in weights if s in close.columns] p = close[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 # full NYC-style stacks: distributions keep (1 - federal - NIIT - state); # a sale (liquidation) gains are taxed at (federal + NIIT + state) d_keep = max(0.0, 1.0 - div_rate - niit - sl_rate) c_keep = max(0.0, 1.0 - lt_rate - niit - sl_rate) # progressive federal brackets: when on, the federal rate for each # flow is the marginal rate at the investor's taxable income position # (starts at fed_base, less the SALT deduction); ordinary flows push # the position up. Flat lt/st/div rates are used when off. taxable_income = float(fed_base) - float(salt) if progressive else 0.0 def _fed(kind: str, flat: float) -> float: if not progressive: return flat brackets = FED_BRACKETS_LT if kind == "ltcg" else FED_BRACKETS_ORD return fed_marginal_rate(taxable_income, brackets) # loss carryover pools: unused realized losses, offset by later gains lt_pool = 0.0 st_pool = 0.0 # open losses awaiting a possible wash sale: [loss, ts, is_lt, consumed] open_losses: dict[int, list[list]] = {} buys: list[list[pd.Timestamp]] = [[] for _ in syms] def _mark_losses_consumed(i: int, is_lt: bool, amount: float) -> None: """Mark up to `amount` of this symbol's unconsumed losses (of the given type) as consumed - i.e. offset by a gain. Consumed losses must no longer be wash-sale candidates.""" evs = open_losses.get(i) if not evs or amount <= 0: return remaining = amount for e in evs: if remaining <= 0: break loss, ts, lt, consumed = e if consumed or lt != is_lt or loss <= 0: continue take = min(loss, remaining) if take >= loss: e[3] = True remaining -= loss else: e[0] = loss - take remaining -= take open_losses[i] = [e for e in evs if not e[3]] # 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] # --- distributions: pay tax, reinvest the after-tax remainder at # today's close. The tax is recorded in tax_rows but NOT also # deducted from cash: it is already reflected in the (smaller) # reinvested units, so a second deduction would double-count it. dinc = units * dv[t] cinc = units * cv[t] fed_div = _fed("ord", div_rate) + niit + sl_rate fed_c = _fed("ltcg", lt_rate) + niit + sl_rate d_tax = float(dinc.sum() * fed_div) c_tax = float(cinc.sum() * fed_c) tax_rows[t, 0] = d_tax tax_rows[t, 1] = c_tax for i in range(len(syms)): net = (dv[t, i] * d_keep + cv[t, i] * c_keep) * units[i] if net <= 0: continue px_i = float(prices[i]) if px_i > 0: units[i] += net / px_i # reinvest the after-tax cash if lots[i]: # grow cost basis tot = sum(l.units for l in lots[i]) for l in lots[i]: l.cost += net * (l.units / tot) lt_rf = _fed("ltcg", lt_rate) st_rf = _fed("ord", st_rate) if progressive: taxable_income += float(dinc.sum()) market_value = float(np.dot(units, prices)) # post-reinvestment # --- 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 r_eff = ((lt_rf if li == 1 else st_rf)) + niit + sl_rate if gain >= 0: pool = lt_pool if li == 1 else st_pool net = gain - pool if net > 0: tax = net * r_eff real_rows[t, li - 1] += gain # the whole carried pool was consumed by this # gain -> mark those losses as used _mark_losses_consumed(i, li == 1, pool) if li == 1: lt_pool = 0.0 else: st_pool = 0.0 else: # the gain is absorbed by carried losses; the # remainder is a loss real_rows[t, li - 1] += gain _mark_losses_consumed(i, li == 1, gain) if li == 1: lt_pool = pool - gain else: st_pool = pool - gain tax = 0.0 cash -= tax tax_rows[t, 2] += tax else: loss = -gain real_rows[t, 2 if li == 1 else 3] += loss if not _loss_wash_disallowed(i, idx[t], buys): if li == 1: lt_pool += loss else: st_pool += loss open_losses.setdefault(i, []).append( [loss, idx[t], li == 1, False]) 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 buys[i].append(idx[t]) # wash sale: a repurchase within WASh_WINDOW_DAYS after # a loss sale of the same security disallows that loss evs = open_losses.get(i) if evs: keep: list[list] = [] for e in evs: loss, ts, is_lt, consumed = e if (not consumed) and ts < idx[t] and \ (idx[t] - ts).days <= WASh_WINDOW_DAYS: if is_lt: lt_pool -= loss else: st_pool -= loss else: keep.append(e) open_losses[i] = keep # --- 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] sym_lt_gain = sym_lt_loss = sym_st_gain = sym_st_loss = 0.0 for l in lots[i]: gain = l.units * px - l.cost if (idx[t] - l.date).days > ST_WINDOW_DAYS: if gain >= 0: sym_lt_gain += gain else: sym_lt_loss += -gain else: if gain >= 0: sym_st_gain += gain else: sym_st_loss += -gain sym_lt_avail = sym_lt_loss + lt_pool sym_lt_consumed = min(sym_lt_gain, sym_lt_avail) if sym_lt_gain > 0 else 0.0 lt_gain += sym_lt_gain lt_loss += sym_lt_loss if sym_lt_consumed > 0: _mark_losses_consumed(i, True, sym_lt_consumed) sym_st_avail = sym_st_loss + st_pool sym_st_consumed = min(sym_st_gain, sym_st_avail) if sym_st_gain > 0 else 0.0 st_gain += sym_st_gain st_loss += sym_st_loss if sym_st_consumed > 0: _mark_losses_consumed(i, False, sym_st_consumed) lt_avail = lt_loss + lt_pool lt_net = lt_gain - min(lt_gain, lt_avail) if lt_gain > 0 else 0.0 lt_pool = lt_avail - min(lt_gain, lt_avail) if lt_gain > 0 else lt_pool st_avail = st_loss + st_pool st_net = st_gain - min(st_gain, st_avail) if st_gain > 0 else 0.0 st_pool = st_avail - min(st_gain, st_avail) if st_gain > 0 else st_pool liq_tax[t] = _liquidation_tax(lt_net, st_net, lt_rf + niit + sl_rate, st_rf + niit + sl_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), )