diff --git a/app.py b/app.py index cb854fd..45e2c7b 100644 --- a/app.py +++ b/app.py @@ -143,6 +143,18 @@ sl_rate = st.sidebar.number_input( "State + local % (NY+NYC: capital gains taxed at ORDINARY rates, " "no preferential cap-gain rate)", 0.0, 49.0, float(_settings.get("sl_rate", 0.0))) / 100 +progressive = st.sidebar.checkbox( + "Progressive federal brackets", value=bool(_settings.get("progressive", False)), + key="progressive", help=("Use marginal federal rates from the tax " + "brackets (LTCG vs ordinary) instead of the flat " + "LT/ST/div rates above.")) +fed_base = st.sidebar.number_input( + "Taxable income base, fed (MFJ) $", 0.0, 10_000_000.0, + float(_settings.get("fed_base", 0.0)), key="fed_base", + help="Starting federal taxable ordinary income for the bracket math.") +salt = st.sidebar.number_input( + "SALT deduction, federal + state/local $ (itemized, MFJ ~$10k cap)", 0.0, 10_000.0, + float(_settings.get("salt", 0.0)), key="salt") # statistics available in the Statistics tab (names = metrics.summary keys) _STATS_VALID = ("total_return", "return", "vol", "sharpe", "sortino", @@ -234,6 +246,7 @@ if _end_ts == "bad": _remember(spec=spec, bench_spec=bench_spec, scheme_index=_scheme_labels.index(scheme), cost_bps=cost_bps, lt_rate=lt_rate * 100, st_rate=st_rate * 100, div_rate=div_rate * 100, niit=niit * 100, sl_rate=sl_rate * 100, + progressive=progressive, fed_base=fed_base, salt=salt, period=period, stats_order=stats_order, date_window=win, range_start=range_start, range_end=range_end) @@ -303,25 +316,29 @@ if saved: @st.cache_data(show_spinner=False) def _compute_portfolio(w_key: tuple, scheme: str | None, cost_bps: float, start: str, lt: float, st_r: float, div: float, - n: float, sl: float, root: str, gen: int): + n: float, sl: float, root: str, gen: int, + prog: bool, fed_b: float, salt_d: float): w = dict(w_key) r = portfolio_returns(bundle.adj, w, rebalance=scheme, cost_bps=cost_bps, start=start) t = after_tax_portfolio(bundle.close, bundle.div, bundle.capg, w, rebalance=scheme, cost_bps=cost_bps, lt_rate=lt, st_rate=st_r, - div_rate=div, niit=n, sl_rate=sl, start=start) + div_rate=div, niit=n, sl_rate=sl, start=start, + progressive=prog, fed_base=fed_b, salt=salt_d) return r, t -def build_result(name: str, w: dict, scheme: str | None, cost: float) -> dict: +def build_result(name: str, w: dict, scheme: str | None, cost: float, + prog: bool, fed_b: float, salt_d: float) -> dict: r, t = _compute_portfolio(tuple(sorted(w.items())), scheme, cost, start, lt_rate, st_rate, div_rate, niit, sl_rate, - str(root), generation(Path(root))) + str(root), generation(Path(root)), prog, fed_b, salt_d) return {"name": name, "weights": w, "res": r, "tax": t} results = [] for k, w in enumerate(items): name = "Current" if len(items) == 1 else ", ".join(w) - results.append(build_result(name, w, freq, cost_bps)) + results.append(build_result(name, w, freq, cost_bps, + progressive, fed_base, salt)) for p in saved: if p.name not in compare_names: continue @@ -330,7 +347,8 @@ for p in saved: if not w: st.sidebar.warning(f"'{p.name}': no known symbols, skipped.") continue - results.append(build_result(p.name, w, p.scheme, p.cost_bps)) + results.append(build_result(p.name, w, p.scheme, p.cost_bps, + progressive, fed_base, salt)) except Exception as e: st.sidebar.warning(f"'{p.name}': {e}") @@ -356,7 +374,8 @@ for i, item in enumerate(bench_spec.split(), 1): try: r, t = _compute_portfolio(tuple(sorted(w.items())), freq, cost_bps, start, lt_rate, st_rate, div_rate, niit, sl_rate, - str(root), generation(Path(root))) + str(root), generation(Path(root)), progressive, + fed_base, salt) except ValueError as e: st.sidebar.warning(f"Benchmark {i} ignored: {e}") continue diff --git a/tax.py b/tax.py index 5fcf35a..975466f 100644 --- a/tax.py +++ b/tax.py @@ -38,8 +38,12 @@ 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. -Deliberately NOT modeled: loss carryover, wash sales, progressive -brackets (flat marginal rates), SALT-deduction interaction. +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 @@ -50,6 +54,38 @@ 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 @@ -68,12 +104,11 @@ class TaxResult: lots_outstanding: int # final lot count (sanity check) -def _liquidation_tax(lt_gain: float, lt_loss: float, - st_gain: float, st_loss: float, +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 = lt_gain - lt_loss - st_net = st_gain - st_loss + """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: @@ -89,7 +124,9 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra 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 + 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 @@ -120,6 +157,48 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra 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])] @@ -146,8 +225,10 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra # reinvested units, so a second deduction would double-count it. dinc = units * dv[t] cinc = units * cv[t] - d_tax = float(dinc.sum() * (div_rate + niit + sl_rate)) - c_tax = float(cinc.sum() * (lt_rate + niit + sl_rate)) + 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)): @@ -161,6 +242,10 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra 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 @@ -184,15 +269,42 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra 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_rate if li == 1 else st_rate) + niit + sl_rate + r_eff = ((lt_rf if li == 1 else st_rf)) + niit + sl_rate if gain >= 0: - tax = gain * r_eff - real_rows[t, li - 1] += gain + 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: - tax = gain * r_eff - real_rows[t, 3 if li == 1 else 2] += -gain - cash -= tax - tax_rows[t, 2] += tax + 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 @@ -203,26 +315,62 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra 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: - lt_gain += gain + sym_lt_gain += gain else: - lt_loss += -gain + sym_lt_loss += -gain else: if gain >= 0: - st_gain += gain + sym_st_gain += gain else: - st_loss += -gain - liq_tax[t] = _liquidation_tax(lt_gain, lt_loss, st_gain, st_loss, - lt_rate + niit + sl_rate, - st_rate + niit + sl_rate) + 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] diff --git a/tests/test_tax.py b/tests/test_tax.py index c75d90c..ee9efc2 100644 --- a/tests/test_tax.py +++ b/tests/test_tax.py @@ -149,6 +149,44 @@ def main() -> int: abs(float(r7.equity[dlt2[1]]) - (34.0 / 30.0 - (0.20 + 0.038 + 0.14) * 4.0 / 30.0)) < 1e-9) + # --- progressive federal brackets: with progressive=True the fed rate + # for each flow is the marginal rate at (fed_base - salt); flat mode + # ignores fed_base/salt and uses the flat rates. + r8 = tax.after_tax_portfolio(close, div2, capg2, {"f": 1.0}, lt_rate=0.20, + st_rate=0.15, div_rate=0.10, niit=0.038, + sl_rate=0.14, progressive=True, fed_base=500_000.0) + check("progressive: LTCG taxed at 15% marginal rate", + abs(r8.taxes["capg_tax"][t5] - (0.15 + 0.038 + 0.14) * 0.6 / 30.0) < 1e-12) + check("progressive: ordinary taxed at 32% marginal rate", + abs(r8.taxes["div_tax"][t5] - (0.32 + 0.038 + 0.14) * 0.4 / 30.0) < 1e-12) + r9 = tax.after_tax_portfolio(close, div2, capg2, {"f": 1.0}, lt_rate=0.20, + st_rate=0.15, div_rate=0.10, niit=0.038, + sl_rate=0.14, progressive=True, fed_base=800_000.0, + salt=10_000.0) + check("progressive: high income -> LTCG 20%, ordinary top rate 37%", + abs(r9.taxes["capg_tax"][t5] - (0.20 + 0.038 + 0.14) * 0.6 / 30.0) < 1e-12 + and abs(r9.taxes["div_tax"][t5] - (0.37 + 0.038 + 0.14) * 0.4 / 30.0) < 1e-12) + r10 = tax.after_tax_portfolio(close, div2, capg2, {"f": 1.0}, lt_rate=0.20, + st_rate=0.15, div_rate=0.10, niit=0.038, + sl_rate=0.14, fed_base=1_000_000.0, salt=50_000.0) + check("flat mode ignores fed_base/salt (uses flat rates)", + abs(r10.taxes["capg_tax"][t5] - (0.20 + 0.038 + 0.14) * 0.6 / 30.0) < 1e-12 + and abs(r10.taxes["div_tax"][t5] - (0.10 + 0.038 + 0.14) * 0.4 / 30.0) < 1e-12) + + # --- loss carryover: a realized loss is carried in the loss pool and + # offsets a later realized gain; only the net is taxed. + d7 = pd.DatetimeIndex([pd.Timestamp("2024-01-02"), pd.Timestamp("2024-01-03"), + pd.Timestamp("2024-02-01")]) + close7 = pd.DataFrame({"a": [30.0, 20.0, 20.0], "b": [40.0, 30.0, 45.0]}, index=d7) + zero7 = pd.DataFrame(0.0, index=d7, columns=["a", "b"]) + r7c = tax.after_tax_portfolio(close7, zero7, zero7, {"a": 0.5, "b": 0.5}, rebalance="1D", + lt_rate=0.20, st_rate=0.15, div_rate=0.10, niit=0.0, sl_rate=0.0) + check("loss carryover: realized loss offsets later gain (net of losses)", + float(r7c.realized["st_loss"].sum()) > 0 + and abs(float(r7c.taxes["realized_tax"][d7[2]]) + - (float(r7c.realized["st_gain"][d7[2]]) + - float(r7c.realized["st_loss"][d7[1]])) * 0.15) < 1e-9) + print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0