62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
"""Very small solver prototype.
|
|
|
|
Provides a deterministic greedy warm-start optimizer for harvesting losses.
|
|
For Phase-0 we avoid external MILP dependencies and implement a clear,
|
|
deterministic algorithm suitable for unit testing. Future versions will
|
|
replace or augment this with MILP (e.g., using pulp or ortools).
|
|
"""
|
|
from datetime import date
|
|
from typing import List, Tuple
|
|
from .ir import TaxLot, HarvestAction, PlanDelta, AuditLog
|
|
|
|
|
|
def _lot_loss(lot: TaxLot) -> float:
|
|
"""Return (market_value - basis) * quantity. Positive=gain, negative=loss.
|
|
|
|
Note: if market_value is None treat as 0 to keep deterministic behavior.
|
|
"""
|
|
mv = lot.market_value if lot.market_value is not None else 0.0
|
|
return (mv - lot.basis) * lot.quantity
|
|
|
|
|
|
def optimize_harvest(lots: List[TaxLot], target_loss: float) -> Tuple[PlanDelta, AuditLog]:
|
|
"""Greedy harvest: choose lots with largest losses first until reaching
|
|
target_loss (in absolute terms). target_loss should be positive number
|
|
indicating total loss to realize (e.g., harvest $10,000 of losses).
|
|
|
|
Returns a PlanDelta and AuditLog.
|
|
"""
|
|
# Work only with lots that currently show a loss
|
|
losses = []
|
|
for l in lots:
|
|
loss = -_lot_loss(l) if _lot_loss(l) < 0 else 0.0
|
|
if loss > 0:
|
|
losses.append((loss, l))
|
|
|
|
# Sort descending by loss magnitude (largest losses first)
|
|
losses.sort(key=lambda x: x[0], reverse=True)
|
|
|
|
accumulated = 0.0
|
|
actions = []
|
|
audit = AuditLog()
|
|
|
|
for loss_amt, lot in losses:
|
|
if accumulated >= target_loss:
|
|
break
|
|
# sell entire lot for simplicity in this prototype
|
|
sell_qty = lot.quantity
|
|
expected_gain = _lot_loss(lot)
|
|
action = HarvestAction(lot_id=lot.id, sell_qty=sell_qty, date=date.today(), expected_gain=expected_gain)
|
|
actions.append(action)
|
|
accumulated += loss_amt
|
|
audit.entries.append({
|
|
"lot_id": lot.id,
|
|
"realized_loss": loss_amt,
|
|
"basis": lot.basis,
|
|
"market_value": lot.market_value,
|
|
})
|
|
|
|
metadata = {"target_loss": target_loss, "realized_loss": accumulated}
|
|
plan = PlanDelta(actions=actions, metadata=metadata)
|
|
return plan, audit
|