From 2c48d94f699741f94b5a031d01e9bd1b12cdea54 Mon Sep 17 00:00:00 2001 From: Greg Pomerantz Date: Fri, 23 Oct 2026 10:21:43 -0400 Subject: [PATCH] Tax engine: fix loss-pool corruption, wash double-subtract, progressive keep - liquidation is now a pure function of state: the daily mark-to-market netting no longer consumes the carried loss pools or the open wash-sale candidates (unrealized gains/losses were leaking into the carryover pools, so equity[t] depended on past unrealized values) - loss offsetting is now global (chronological, by type) instead of per-symbol: a symbol A gain consuming a symbol B loss marks B's event consumed, so a later B repurchase cannot subtract the same loss a second time (pools could go negative -> over-taxed gains) - progressive mode: the reinvested after-tax remainder now uses the SAME marginal rates the tax is recorded at (was flat d_keep/c_keep, so recorded tax and cash flow diverged); the bracket position also advances by net short-term gains - renames: WASh->WASH, lt_rf/st_rf->lt_fed/st_fed (rf is the project's risk-free-rate convention); buys pruned to the 30d wash window; dropped the per-call sorted() on the constant bracket tables tests: wash-sale disallowance; liquidation purity (no-trade day with unrealized gains must not eat the pool); cross-symbol consumption (no double subtract). All scenarios verified to fail on the pre-fix code. --- tax.py | 176 ++++++++++++++++++++++++---------------------- tests/test_tax.py | 58 +++++++++++++++ 2 files changed, 150 insertions(+), 84 deletions(-) diff --git a/tax.py b/tax.py index 975466f..15c81c9 100644 --- a/tax.py +++ b/tax.py @@ -39,11 +39,14 @@ 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. +later gains; repurchases within WASH_WINDOW_DAYS of a loss sale of the +same security disallow the loss (wash sale). The daily liquidation is a +PURE function of the simulated state (it never consumes loss pools or +wash-sale candidates). 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, advanced by +dividend income and net short-term gains); otherwise the flat lt/st/div +rates are used. """ from __future__ import annotations @@ -54,7 +57,7 @@ import numpy as np import pandas as pd ST_WINDOW_DAYS = 365 -WASh_WINDOW_DAYS = 30 +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 @@ -71,7 +74,7 @@ FED_BRACKETS_LT: list[tuple[float, float]] = [ 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): + for limit, r in brackets: # brackets are sorted by limit if income > limit: rate = r return rate @@ -83,11 +86,28 @@ def _loss_wash_disallowed(i: int, ts: pd.Timestamp, 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): + if pd.Timedelta(0) <= delta <= pd.Timedelta(days=WASH_WINDOW_DAYS): return True return False +@dataclass +class _LossEvent: + """A realized loss still open to wash-sale disallowance. + + `loss` is the REMAINING (not-yet-offset) amount; `consumed` means it + was fully offset by a later gain and can no longer be disallowed. + Losses net globally (not per security), so offsetting is matched in + chronological order by type across all symbols - a symbol A gain + consuming a symbol B loss marks B's event consumed, so a later B + repurchase must not subtract it a second time.""" + sym: int + loss: float + ts: pd.Timestamp + is_lt: bool + consumed: bool = False + + @dataclass class _Lot: units: float @@ -152,10 +172,6 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra 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 @@ -172,32 +188,29 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra # 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]] = {} + # realized losses still open to a wash-sale disallowance (chronological) + open_losses: list[_LossEvent] = [] 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: + def _mark_losses_consumed(is_lt: bool, amount: float) -> None: + """Mark up to `amount` of the unconsumed losses of the given type + as offset by a gain, in chronological order (any symbol - losses + net globally). Offset losses must no longer be wash-sale + candidates.""" + if amount <= 0: return remaining = amount - for e in evs: + for e in open_losses: if remaining <= 0: break - loss, ts, lt, consumed = e - if consumed or lt != is_lt or loss <= 0: + if e.consumed or e.is_lt != is_lt or e.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]] + take = min(e.loss, remaining) + e.loss -= take + remaining -= take + if e.loss <= 0: + e.consumed = True + open_losses[:] = [e for e in open_losses if not e.consumed] # state (account = 1.0 at start) units = w0 / pv[0] @@ -225,12 +238,16 @@ 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] - 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) + # the rates used to REINVEST must equal the rates the tax is + # recorded at (they diverge in progressive mode) + d_rate = _fed("ord", div_rate) + niit + sl_rate + c_rate = _fed("ltcg", lt_rate) + niit + sl_rate + d_tax = float(dinc.sum() * d_rate) + c_tax = float(cinc.sum() * c_rate) tax_rows[t, 0] = d_tax tax_rows[t, 1] = c_tax + d_keep = max(0.0, 1.0 - d_rate) + c_keep = max(0.0, 1.0 - c_rate) for i in range(len(syms)): net = (dv[t, i] * d_keep + cv[t, i] * c_keep) * units[i] if net <= 0: @@ -242,8 +259,8 @@ 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) + lt_fed = _fed("ltcg", lt_rate) + st_fed = _fed("ord", st_rate) if progressive: taxable_income += float(dinc.sum()) @@ -269,7 +286,7 @@ 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_rf if li == 1 else st_rf)) + niit + sl_rate + r_eff = (lt_fed if li == 1 else st_fed) + niit + sl_rate if gain >= 0: pool = lt_pool if li == 1 else st_pool net = gain - pool @@ -277,22 +294,25 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra 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) + # gain -> those losses are no longer wash + # candidates + _mark_losses_consumed(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 + # remainder stays in the pool real_rows[t, li - 1] += gain - _mark_losses_consumed(i, li == 1, gain) + _mark_losses_consumed(li == 1, gain) if li == 1: lt_pool = pool - gain else: st_pool = pool - gain tax = 0.0 + if progressive and li == 2: + taxable_income += max(net, 0.0) cash -= tax tax_rows[t, 2] += tax else: @@ -303,8 +323,8 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra lt_pool += loss else: st_pool += loss - open_losses.setdefault(i, []).append( - [loss, idx[t], li == 1, False]) + open_losses.append( + _LossEvent(i, loss, idx[t], li == 1)) l.units -= u l.cost *= (l.units / (l.units + u)) if l.units > 0 else 0.0 to_sell -= u @@ -316,61 +336,49 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra 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 + while buys[i] and \ + (idx[t] - buys[i][0]).days > WASH_WINDOW_DAYS: + buys[i].pop(0) + # wash sale: a repurchase within WASH_WINDOW_DAYS + # after a loss sale of the same security disallows + # that loss (only its remaining, unoffset amount) + disallowed = [e for e in open_losses + if not e.consumed and e.sym == i + and e.ts < idx[t] + and (idx[t] - e.ts).days <= WASH_WINDOW_DAYS] + if disallowed: + for e in disallowed: + if e.is_lt: + lt_pool -= e.loss else: - keep.append(e) - open_losses[i] = keep + st_pool -= e.loss + open_losses[:] = [e for e in open_losses + if e not in disallowed] # --- tax owed if we liquidate everything today ------------------ + # PURE: netting against the carried pools is done on local copies + # only - the liquidation is counterfactual and must never consume + # the pools or the open (wash-sale-eligible) losses. 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 + lt_gain += gain else: - sym_lt_loss += -gain + lt_loss += -gain else: if gain >= 0: - sym_st_gain += gain + 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 + st_loss += -gain + lt_net = lt_gain - min(lt_gain, lt_loss + lt_pool) if lt_gain > 0 else 0.0 + st_net = st_gain - min(st_gain, st_loss + st_pool) if st_gain > 0 else 0.0 liq_tax[t] = _liquidation_tax(lt_net, st_net, - lt_rf + niit + sl_rate, - st_rf + niit + sl_rate) + lt_fed + niit + sl_rate, + st_fed + 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 ee9efc2..e00a48b 100644 --- a/tests/test_tax.py +++ b/tests/test_tax.py @@ -187,6 +187,64 @@ def main() -> int: - (float(r7c.realized["st_gain"][d7[2]]) - float(r7c.realized["st_loss"][d7[1]])) * 0.15) < 1e-9) + # --- wash sale: a loss sale of `a` on day 3, then `a` is repurchased + # on day 4 (within 30d) -> the carried loss is disallowed and the day-5 + # gain is taxed in FULL (no offset). + dws = pd.to_datetime(["2024-01-0%d" % i for i in (1, 2, 3, 4, 5)] + + ["2024-01-08"]) + close_ws = pd.DataFrame({"a": [10.0, 10.0, 12.0, 9.0, 9.0, 12.0], + "b": [10.0, 5.0, 5.0, 4.0, 3.0, 3.0]}, index=dws) + zero_ws = pd.DataFrame(0.0, index=dws, columns=["a", "b"]) + rws = tax.after_tax_portfolio(close_ws, zero_ws, zero_ws, + {"a": 0.5, "b": 0.5}, rebalance="1D", + lt_rate=0.20, st_rate=0.15, div_rate=0.10) + check("wash sale: repurchase within 30d disallows the carried loss", + float(rws.realized["st_loss"][dws[3]]) > 0 + and float(rws.realized["st_gain"][dws[5]]) > 0 + and abs(float(rws.taxes["realized_tax"][dws[5]]) + - float(rws.realized["st_gain"][dws[5]]) * 0.15) < 1e-9) + + # --- liquidation is pure: day 2 realizes an ST loss into the pool; + # day 3 is a NO-TRADE day (all symbols up 1.3x proportionally) with big + # UNREALIZED gains - the liquidation math must not consume the pool; + # day 4's realized gain of `a` is still offset exactly by the carried + # loss (old code under-taxed it: the pool had been eaten by the marks). + dlp = pd.to_datetime(["2024-01-0%d" % i for i in (1, 2, 3, 4)]) + clp = pd.DataFrame({"a": [10.0, 5.0, 6.5, 14.0], + "b": [10.0, 8.0, 10.4, 10.4], + "c": [10.0, 10.0, 13.0, 13.0]}, index=dlp) + zero_lp = pd.DataFrame(0.0, index=dlp, columns=["a", "b", "c"]) + rlp = tax.after_tax_portfolio(clp, zero_lp, zero_lp, + {"a": 1 / 3, "b": 1 / 3, "c": 1 / 3}, + rebalance="1D", lt_rate=0.20, st_rate=0.15, + div_rate=0.10) + check("liquidation is pure: unrealized gains do not consume the pool", + float(rlp.realized["st_loss"][dlp[1]]) > 0 + and float(rlp.realized["st_gain"][dlp[3]]) > 0 + and abs(float(rlp.taxes["realized_tax"][dlp[3]]) + - (float(rlp.realized["st_gain"][dlp[3]]) + - float(rlp.realized["st_loss"][dlp[1]])) * 0.15) < 1e-9) + + # --- global loss offset vs per-symbol wash bookkeeping: b's loss (day 2) + # is offset by a's gain (day 3, `a` is processed first) -> b's loss + # event is consumed, so the b repurchase on day 3 must NOT subtract the + # loss a second time. Day 4's b gain is therefore taxed in full + # (no phantom extra offset from a negative pool). + dcs = pd.to_datetime(["2024-01-0%d" % i for i in (1, 2, 3, 4)]) + ccs = pd.DataFrame({"a": [10.0, 6.0, 15.0, 10.0], + "b": [10.0, 8.0, 6.0, 12.0], + "c": [10.0, 6.0, 6.0, 10.0]}, index=dcs) + zero_cs = pd.DataFrame(0.0, index=dcs, columns=["a", "b", "c"]) + rcs = tax.after_tax_portfolio(ccs, zero_cs, zero_cs, + {"a": 1 / 3, "b": 1 / 3, "c": 1 / 3}, + rebalance="1D", lt_rate=0.20, st_rate=0.15, + div_rate=0.10) + check("cross-symbol offset marks the loss consumed (no double subtract)", + float(rcs.realized["st_loss"][dcs[1]]) > 0 + and float(rcs.realized["st_gain"][dcs[3]]) > 0 + and abs(float(rcs.taxes["realized_tax"][dcs[3]]) + - float(rcs.realized["st_gain"][dcs[3]]) * 0.15) < 1e-9) + print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0