trex/portfolio-hub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(portfolios): replay corporate-action cost basis (#38)

Browse Source
This commit is contained in:
genichin
2026-09-19 17:07:14 +09:00
committed by GitHub
parent f79f3c79b5
commit 567d662913
12 changed files with 352 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/v1_portfolios.py
+37 -16
View File
@@ -1282,27 +1282,48 @@ def get_holdings_cost_basis(
finally:
_close_repo(repo)
# Build rows for domain function
rows = [
{
"side": e.side,
"quantity": e.quantity,
"price": e.price,
"fee_amount": e.fee_amount,
"executed_at": e.executed_at,
# Build rows for domain function, preserving joined corporate-action metadata.
rows = []
corporate_action_fields = (
"corporate_action_id",
"action_type",
"quantity_before",
"quantity_after",
"average_price_after",
"total_cost_basis",
)
for entry in ledger_entries:
row = {
"side": entry.side,
"quantity": entry.quantity,
"price": entry.price,
"fee_amount": entry.fee_amount,
"executed_at": entry.executed_at,
"ledger_entry_id": entry.ledger_entry_id,
}
for e in ledger_entries
]
if entry.corporate_action:
row.update(
{
field: entry.corporate_action.get(field)
for field in corporate_action_fields
}
)
rows.append(row)
wac_result = calculate_current_wac(rows)
current_quantity = portfolio.asset_quantity
cost_basis_per_unit = wac_result.current_wac if current_quantity > 0 else None
total_cost_basis = (
cost_basis_per_unit * current_quantity
if cost_basis_per_unit is not None and current_quantity > 0
else None
)
if wac_result.total_cost_basis is not None and wac_result.current_quantity > 0:
current_quantity = wac_result.current_quantity
total_cost_basis = wac_result.total_cost_basis
cost_basis_per_unit = total_cost_basis / current_quantity
else:
cost_basis_per_unit = wac_result.current_wac if current_quantity > 0 else None
total_cost_basis = (
cost_basis_per_unit * current_quantity
if cost_basis_per_unit is not None and current_quantity > 0
else None
)
return HoldingsCostBasisResponse(
portfolio_id=portfolio_id,
app/domain/pnl.py
+59 -33
View File
@@ -57,6 +57,9 @@ class CurrentWacResult:
data_completeness_warning: str | None
"""Non-None when ledger history is incomplete (sell before buy, oversell, etc.)."""
total_cost_basis: float | None = None
"""Authoritative current total after a corporate action, when available."""
def calculate_realized_pnl(
rows: list[dict[str, Any]],
@@ -204,29 +207,32 @@ def calculate_realized_pnl(
def calculate_current_wac(rows: list[dict[str, Any]]) -> CurrentWacResult:
"""Calculate the current holding WAC (weighted average cost) from a time-ordered ledger.
Replays the full ledger history (buy/sell) and returns the running WAC state,
representing the cost basis of currently held (unsold) shares.
Replays the full ledger history (buy/sell/corporate action) and returns the
running WAC state representing the cost basis of currently held shares.
Args:
rows: List of portfolio_ledger row dicts, each containing:
side (str), quantity (float), price (float),
executed_at (datetime).
Must be sorted by executed_at ASC.
rows: Time-ordered portfolio-ledger dictionaries. BUY/SELL rows contain
side, quantity, price, fee_amount and executed_at. A
stock_consolidation row must also contain a positive
quantity_after and authoritative total_cost_basis.
Rows must be sorted by executed_at ASC (and ledger_entry_id ASC
for ties) by the repository.
Returns:
CurrentWacResult with:
- current_wac: WAC per unit (None when current_quantity == 0)
- current_quantity: remaining quantity after all trades
- current_wac: WAC per unit (None for no holding or unusable
corporate-action metadata)
- current_quantity: remaining replay quantity
- cost_basis_from: earliest buy date used
- data_completeness_warning: non-None when history is incomplete
- total_cost_basis: authoritative current total after a corporate
action, otherwise None for the legacy BUY/SELL path
WAC Rules (same as calculate_realized_pnl, REQ-041, buy-fee-in-cost-basis):
buy: running_wac = (running_wac * running_qty + price * qty + buy_fee) / (running_qty + qty)
running_qty += qty
(buy_fee = fee_amount of the buy row, incorporated into cost basis)
sell: running_qty -= qty (WAC per unit unchanged on sell)
Incomplete history (sell with running_qty=0, or sell qty > running_qty):
data_completeness_warning is set; calculation continues with best estimate.
WAC rules preserve the existing BUY/SELL behavior. A stock consolidation
replaces the replay quantity with quantity_after, preserves
total_cost_basis as authoritative, and derives unit cost by division.
Missing required corporate-action metadata fails closed instead of falling
back to the pre-action WAC.
"""
if not rows:
return CurrentWacResult(
@@ -240,6 +246,8 @@ def calculate_current_wac(rows: list[dict[str, Any]]) -> CurrentWacResult:
running_qty: float = 0.0
cost_basis_from: datetime | None = None
warning_triggered: bool = False
corporate_action_metadata_missing: bool = False
authoritative_total_cost_basis: float | None = None
for row in rows:
side = str(row.get("side", "")).lower()
@@ -252,38 +260,47 @@ def calculate_current_wac(rows: list[dict[str, Any]]) -> CurrentWacResult:
# Include buy fee in cost basis (매수 수수료 편입)
new_total_qty = running_qty + qty
if new_total_qty > 0:
running_wac = (running_wac * running_qty + price * qty + fee) / new_total_qty
if authoritative_total_cost_basis is not None:
authoritative_total_cost_basis += price * qty + fee
running_wac = authoritative_total_cost_basis / new_total_qty
else:
running_wac = (
running_wac * running_qty + price * qty + fee
) / new_total_qty
running_qty = new_total_qty
if cost_basis_from is None and executed_at is not None:
cost_basis_from = executed_at
elif side == "sell":
if running_qty <= 0:
warning_triggered = True
elif qty > running_qty + 1e-9:
if running_qty <= 0 or qty > running_qty + 1e-9:
warning_triggered = True
running_qty -= qty
if running_qty < 0:
running_qty = 0.0
if authoritative_total_cost_basis is not None:
authoritative_total_cost_basis = running_wac * running_qty
elif side == "stock_consolidation":
before = float(row.get("quantity_before", running_qty))
after = float(row.get("quantity_after", row.get("asset_quantity_after", 0.0)))
if before <= 0 or after <= 0 or running_qty <= 0:
warning_triggered = True
continue
if abs(running_qty - before) > 1e-9:
warning_triggered = True
running_qty = after
quantity_after = row.get("quantity_after")
total_cost_basis = row.get("total_cost_basis")
if total_cost_basis is not None:
running_wac = float(total_cost_basis) / after
else:
running_wac = float(row.get("average_price_after", running_wac * before / after))
if quantity_after is None or total_cost_basis is None:
corporate_action_metadata_missing = True
continue
after = float(quantity_after)
if after <= 0:
corporate_action_metadata_missing = True
continue
running_qty = after
authoritative_total_cost_basis = float(total_cost_basis)
running_wac = authoritative_total_cost_basis / after
warning: str | None = None
if warning_triggered:
if corporate_action_metadata_missing:
warning = (
"기업행동 metadata 불완전: 양수 quantity_after와 total_cost_basis가 필요합니다."
)
elif warning_triggered:
warning = (
"이력 불완전: 보유 수량보다 많은 매도 또는 buy 이력 없는 sell이 감지되어 "
"cost_basis 산출이 부정확할 수 있습니다. "
@@ -291,10 +308,19 @@ def calculate_current_wac(rows: list[dict[str, Any]]) -> CurrentWacResult:
)
return CurrentWacResult(
current_wac=running_wac if running_qty > 0 else None,
current_wac=(
running_wac
if running_qty > 0 and not corporate_action_metadata_missing
else None
),
current_quantity=running_qty,
cost_basis_from=cost_basis_from,
data_completeness_warning=warning,
total_cost_basis=(
authoritative_total_cost_basis
if running_qty > 0 and not corporate_action_metadata_missing
else None
),
)