gmp/f
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Tax engine: fix loss-pool corruption, wash double-subtract, progressive keep

Browse Source 
- 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.
This commit is contained in:
Greg Pomerantz 2026-10-23 10:21:43 -04:00
parent 2b7d545dc7
commit 2c48d94f69
2 changed files with 150 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

176
tax.py
View File

@ -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. remove/add ops in overrides/corrections/ where a symbol is analyzed.
Losses realized along the way are carried forward (loss pools) and offset 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 later gains; repurchases within WASH_WINDOW_DAYS of a loss sale of the
same security disallow the loss (wash sale). Progressive federal brackets same security disallow the loss (wash sale). The daily liquidation is a
are used when `progressive` is on (marginal rate at the investor's taxable PURE function of the simulated state (it never consumes loss pools or
income position, starting at `fed_base`, less the `salt` deduction); wash-sale candidates). Progressive federal brackets are used when
otherwise the flat lt/st/div rates are used. `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 from __future__ import annotations
@ -54,7 +57,7 @@ import numpy as np
import pandas as pd import pandas as pd
ST_WINDOW_DAYS = 365 ST_WINDOW_DAYS = 365
WASh_WINDOW_DAYS = 30 WASH_WINDOW_DAYS = 30
# Federal ordinary / LTCG brackets (2025, MFJ) for the progressive mode: # Federal ordinary / LTCG brackets (2025, MFJ) for the progressive mode:
# marginal federal rate at a given taxable ordinary income. LTCG: 0% to # 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: def fed_marginal_rate(income: float, brackets: list[tuple[float, float]]) -> float:
"""Marginal federal rate at a given taxable ordinary income.""" """Marginal federal rate at a given taxable ordinary income."""
rate = brackets[0][1] rate = brackets[0][1]
for limit, r in sorted(brackets): for limit, r in brackets: # brackets are sorted by limit
if income > limit: if income > limit:
rate = r rate = r
return rate 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.""" before (or on) a loss sale on day `ts` - wash sale, loss disallowed."""
for b in buys[i]: for b in buys[i]:
delta = ts - b 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 True
return False 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 @dataclass
class _Lot: class _Lot:
units: float 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 = np.array([weights[s] for s in syms])
w0 = w0 / w0.sum() w0 = w0 / w0.sum()
cost = cost_bps / 1e4 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 # progressive federal brackets: when on, the federal rate for each
# flow is the marginal rate at the investor's taxable income position # 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 # loss carryover pools: unused realized losses, offset by later gains
lt_pool = 0.0 lt_pool = 0.0
st_pool = 0.0 st_pool = 0.0
# open losses awaiting a possible wash sale: [loss, ts, is_lt, consumed] # realized losses still open to a wash-sale disallowance (chronological)
open_losses: dict[int, list[list]] = {} open_losses: list[_LossEvent] = []
buys: list[list[pd.Timestamp]] = [[] for _ in syms] buys: list[list[pd.Timestamp]] = [[] for _ in syms]
def _mark_losses_consumed(i: int, is_lt: bool, amount: float) -> None: def _mark_losses_consumed(is_lt: bool, amount: float) -> None:
"""Mark up to `amount` of this symbol's unconsumed losses (of the """Mark up to `amount` of the unconsumed losses of the given type
given type) as consumed - i.e. offset by a gain. Consumed losses as offset by a gain, in chronological order (any symbol - losses
must no longer be wash-sale candidates.""" net globally). Offset losses must no longer be wash-sale
evs = open_losses.get(i) candidates."""
if not evs or amount <= 0: if amount <= 0:
return return
remaining = amount remaining = amount
for e in evs: for e in open_losses:
if remaining <= 0: if remaining <= 0:
break break
loss, ts, lt, consumed = e if e.consumed or e.is_lt != is_lt or e.loss <= 0:
if consumed or lt != is_lt or loss <= 0:
continue continue
take = min(loss, remaining) take = min(e.loss, remaining)
if take >= loss: e.loss -= take
e[3] = True remaining -= take
remaining -= loss if e.loss <= 0:
else: e.consumed = True
e[0] = loss - take open_losses[:] = [e for e in open_losses if not e.consumed]
remaining -= take
open_losses[i] = [e for e in evs if not e[3]]
# state (account = 1.0 at start) # state (account = 1.0 at start)
units = w0 / pv[0] 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. # reinvested units, so a second deduction would double-count it.
dinc = units * dv[t] dinc = units * dv[t]
cinc = units * cv[t] cinc = units * cv[t]
fed_div = _fed("ord", div_rate) + niit + sl_rate # the rates used to REINVEST must equal the rates the tax is
fed_c = _fed("ltcg", lt_rate) + niit + sl_rate # recorded at (they diverge in progressive mode)
d_tax = float(dinc.sum() * fed_div) d_rate = _fed("ord", div_rate) + niit + sl_rate
c_tax = float(cinc.sum() * fed_c) 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, 0] = d_tax
tax_rows[t, 1] = c_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)): for i in range(len(syms)):
net = (dv[t, i] * d_keep + cv[t, i] * c_keep) * units[i] net = (dv[t, i] * d_keep + cv[t, i] * c_keep) * units[i]
if net <= 0: 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]) tot = sum(l.units for l in lots[i])
for l in lots[i]: for l in lots[i]:
l.cost += net * (l.units / tot) l.cost += net * (l.units / tot)
lt_rf = _fed("ltcg", lt_rate) lt_fed = _fed("ltcg", lt_rate)
st_rf = _fed("ord", st_rate) st_fed = _fed("ord", st_rate)
if progressive: if progressive:
taxable_income += float(dinc.sum()) 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) gain = u * prices[i] - (l.cost * u / l.units)
held = (idx[t] - l.date).days held = (idx[t] - l.date).days
li = 1 if held > ST_WINDOW_DAYS else 2 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: if gain >= 0:
pool = lt_pool if li == 1 else st_pool pool = lt_pool if li == 1 else st_pool
net = gain - 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 tax = net * r_eff
real_rows[t, li - 1] += gain real_rows[t, li - 1] += gain
# the whole carried pool was consumed by this # the whole carried pool was consumed by this
# gain -> mark those losses as used # gain -> those losses are no longer wash
_mark_losses_consumed(i, li == 1, pool) # candidates
_mark_losses_consumed(li == 1, pool)
if li == 1: if li == 1:
lt_pool = 0.0 lt_pool = 0.0
else: else:
st_pool = 0.0 st_pool = 0.0
else: else:
# the gain is absorbed by carried losses; the # the gain is absorbed by carried losses; the
# remainder is a loss # remainder stays in the pool
real_rows[t, li - 1] += gain real_rows[t, li - 1] += gain
_mark_losses_consumed(i, li == 1, gain) _mark_losses_consumed(li == 1, gain)
if li == 1: if li == 1:
lt_pool = pool - gain lt_pool = pool - gain
else: else:
st_pool = pool - gain st_pool = pool - gain
tax = 0.0 tax = 0.0
if progressive and li == 2:
taxable_income += max(net, 0.0)
cash -= tax cash -= tax
tax_rows[t, 2] += tax tax_rows[t, 2] += tax
else: else:
@ -303,8 +323,8 @@ def after_tax_portfolio(close: pd.DataFrame, div: pd.DataFrame, capg: pd.DataFra
lt_pool += loss lt_pool += loss
else: else:
st_pool += loss st_pool += loss
open_losses.setdefault(i, []).append( open_losses.append(
[loss, idx[t], li == 1, False]) _LossEvent(i, loss, idx[t], li == 1))
l.units -= u l.units -= u
l.cost *= (l.units / (l.units + u)) if l.units > 0 else 0.0 l.cost *= (l.units / (l.units + u)) if l.units > 0 else 0.0
to_sell -= u 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])) lots[i].append(_Lot(u, spend, idx[t]))
units[i] += u units[i] += u
buys[i].append(idx[t]) buys[i].append(idx[t])
# wash sale: a repurchase within WASh_WINDOW_DAYS after while buys[i] and \
# a loss sale of the same security disallows that loss (idx[t] - buys[i][0]).days > WASH_WINDOW_DAYS:
evs = open_losses.get(i) buys[i].pop(0)
if evs: # wash sale: a repurchase within WASH_WINDOW_DAYS
keep: list[list] = [] # after a loss sale of the same security disallows
for e in evs: # that loss (only its remaining, unoffset amount)
loss, ts, is_lt, consumed = e disallowed = [e for e in open_losses
if (not consumed) and ts < idx[t] and \ if not e.consumed and e.sym == i
(idx[t] - ts).days <= WASh_WINDOW_DAYS: and e.ts < idx[t]
if is_lt: and (idx[t] - e.ts).days <= WASH_WINDOW_DAYS]
lt_pool -= loss if disallowed:
else: for e in disallowed:
st_pool -= loss if e.is_lt:
lt_pool -= e.loss
else: else:
keep.append(e) st_pool -= e.loss
open_losses[i] = keep open_losses[:] = [e for e in open_losses
if e not in disallowed]
# --- tax owed if we liquidate everything today ------------------ # --- 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 lt_gain = lt_loss = st_gain = st_loss = 0.0
for i in range(len(syms)): for i in range(len(syms)):
px = prices[i] px = prices[i]
sym_lt_gain = sym_lt_loss = sym_st_gain = sym_st_loss = 0.0
for l in lots[i]: for l in lots[i]:
gain = l.units * px - l.cost gain = l.units * px - l.cost
if (idx[t] - l.date).days > ST_WINDOW_DAYS: if (idx[t] - l.date).days > ST_WINDOW_DAYS:
if gain >= 0: if gain >= 0:
sym_lt_gain += gain lt_gain += gain
else: else:
sym_lt_loss += -gain lt_loss += -gain
else: else:
if gain >= 0: if gain >= 0:
sym_st_gain += gain st_gain += gain
else: else:
sym_st_loss += -gain st_loss += -gain
sym_lt_avail = sym_lt_loss + lt_pool lt_net = lt_gain - min(lt_gain, lt_loss + lt_pool) if lt_gain > 0 else 0.0
sym_lt_consumed = min(sym_lt_gain, sym_lt_avail) if sym_lt_gain > 0 else 0.0 st_net = st_gain - min(st_gain, st_loss + st_pool) if st_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, liq_tax[t] = _liquidation_tax(lt_net, st_net,
lt_rf + niit + sl_rate, lt_fed + niit + sl_rate,
st_rf + niit + sl_rate) st_fed + niit + sl_rate)
equity[t] = market_value + cash - liq_tax[t] equity[t] = market_value + cash - liq_tax[t]
tax_rows[t, 3] = tax_rows[t, 0] + tax_rows[t, 1] + tax_rows[t, 2] tax_rows[t, 3] = tax_rows[t, 0] + tax_rows[t, 1] + tax_rows[t, 2]

58
tests/test_tax.py
View File

@ -187,6 +187,64 @@ def main() -> int:
- (float(r7c.realized["st_gain"][d7[2]]) - (float(r7c.realized["st_gain"][d7[2]])
- float(r7c.realized["st_loss"][d7[1]])) * 0.15) < 1e-9) - 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") print(f"\n{PASS} passed, {FAIL} failed")
return 1 if FAIL else 0 return 1 if FAIL else 0