Greg Pomerantz
d8703a7a63
- single spec grammar for symbol and benchmark fields: commas join one portfolio (MSFT:0.6,V:0.4), spaces separate distinct symbols/portfolios; both fields accept one or many entries - benchmarks simulated with the same scheme/cost/tax rules; per-benchmark beta/alpha columns; after-tax benchmark curves - global Curve mode (pre/after/both) above the tabs; clean names in single-curve mode - live updates: field commits on Enter/blur, page recomputes per rerun; portfolio+tax sims cached (st.cache_data); plotly.js from CDN (4.6MB -> browser-cached) with F_INLINE_PLOTLY=1 offline fallback - chart: legend underneath, solid lines, pan sticks to data edges (width-preserving), zoom edge-clamped - inputs persist in settings.json across reloads/restarts/devices - tests: tests/test_app.py (AppTest) + tests/test_e2e_browser.py (Playwright) via ./run_tests.sh
201 lines
8.0 KiB
Python
201 lines
8.0 KiB
Python
"""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),
|
|
)
|