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

fix corporate action ledger replay invariants

Browse Source
This commit is contained in:
genichin
2026-09-13 14:04:30 +09:00
parent 48735bbb4a
commit 14c118f503
4 changed files with 59 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/repositories/postgres.py
+37 -8
View File
@@ -1648,10 +1648,31 @@ class PostgresPortfolioRepository:
self.ensure_schema()
with self._connect() as conn:
with conn.cursor() as cur:
cur.execute("SELECT asset_quantity, cash_balance, profit_buffer, asset_id FROM portfolios WHERE portfolio_id=%s FOR UPDATE", (portfolio_id,))
cur.execute("SELECT asset_quantity, cash_balance, profit_buffer, asset_id, status FROM portfolios WHERE portfolio_id=%s FOR UPDATE", (portfolio_id,))
row = cur.fetchone()
if row is None:
raise ValueError("portfolio not found")
if row[4] != "enabled":
raise ValueError("CORPORATE_ACTION_PORTFOLIO_DISABLED")
if row[3] != action.asset_id:
raise ValueError("CORPORATE_ACTION_ASSET_MISMATCH")
if Decimal(str(row[1])) != action.cash_balance or Decimal(str(row[2])) != action.profit_buffer:
raise ValueError("CORPORATE_ACTION_INVARIANT_MISMATCH")
cur.execute("SELECT 1 FROM portfolio_ledger WHERE portfolio_id=%s AND side IN ('buy','sell') AND executed_at::date > %s LIMIT 1", (portfolio_id, action.effective_date))
if cur.fetchone() is not None:
raise ValueError("CORPORATE_ACTION_HISTORICAL_CONFLICT")
# Derive WAC from the authoritative locked ledger, never portfolios.price.
cur.execute("SELECT side, quantity, price, fee_amount, tax_amount, executed_at FROM portfolio_ledger WHERE portfolio_id=%s ORDER BY executed_at ASC NULLS FIRST, ledger_entry_id ASC", (portfolio_id,))
history = [{"side": r[0], "quantity": r[1], "price": r[2], "fee_amount": r[3], "tax_amount": r[4], "executed_at": r[5]} for r in cur.fetchall()]
from app.domain.pnl import calculate_current_wac
wac = calculate_current_wac(history)
if Decimal(str(row[0])) != action.quantity_before or Decimal(str(wac.current_quantity)) != action.quantity_before:
raise ValueError("CORPORATE_ACTION_STALE_QUANTITY")
action.average_price_before = Decimal(str(wac.current_wac or 0))
action.total_cost_basis = (action.average_price_before * action.quantity_before).quantize(Decimal("0.00000001"))
action.average_price_after = (action.total_cost_basis / action.quantity_after).quantize(Decimal("0.00000001"))
if action.average_price_after * action.quantity_after != action.total_cost_basis:
raise ValueError("FRACTIONAL_RESULT_UNSUPPORTED")
cur.execute("SELECT corporate_action_id, payload_hash, ledger_entry_id, recorded_at FROM portfolio_corporate_actions WHERE portfolio_id=%s AND idempotency_key=%s", (portfolio_id, idempotency_key))
prior = cur.fetchone()
if prior:
@@ -1662,7 +1683,7 @@ class PostgresPortfolioRepository:
return action
if Decimal(str(row[0])) != action.quantity_before:
raise ValueError("CORPORATE_ACTION_STALE_QUANTITY")
cur.execute("UPDATE portfolios SET asset_quantity=%s WHERE portfolio_id=%s", (float(action.quantity_after), portfolio_id))
cur.execute("UPDATE portfolios SET asset_quantity=%s, updated_at=NOW() WHERE portfolio_id=%s", (float(action.quantity_after), portfolio_id))
cur.execute("INSERT INTO portfolio_ledger (portfolio_id, recorded_at, side, source, quantity, price, cash_balance_after, asset_quantity_after, profit_buffer_after, executed_at, note) VALUES (%s,NOW(),'stock_consolidation','corporate_action',0,0,%s,%s,%s,NOW(),%s) RETURNING ledger_entry_id, recorded_at", (portfolio_id, float(action.cash_balance), float(action.quantity_after), float(action.profit_buffer), action.evidence_ref))
lid, recorded = cur.fetchone()
action.ledger_entry_id, action.recorded_at = int(lid), recorded
@@ -1777,16 +1798,24 @@ class PostgresPortfolioRepository:
executed_at, recorded_at,
cash_balance_after::float8, asset_quantity_after::float8,
total_value_after::float8, profit_buffer_after::float8,
strategy_set_id, note
FROM portfolio_ledger
WHERE portfolio_id = %s
AND side IN ('buy', 'sell')
ORDER BY executed_at ASC, ledger_entry_id ASC
strategy_set_id, note,
ca.corporate_action_id, ca.effective_date, ca.quantity_before,
ca.quantity_after, ca.average_price_after
FROM portfolio_ledger AS ledger
LEFT JOIN portfolio_corporate_actions ca ON ca.ledger_entry_id = ledger.ledger_entry_id
WHERE ledger.portfolio_id = %s
ORDER BY ledger.executed_at ASC, ledger.ledger_entry_id ASC
""",
(portfolio_id,),
)
rows = cur.fetchall()
return [self._row_to_ledger_entry(row) for row in rows]
entries = []
for row in rows:
entry = self._row_to_ledger_entry(row)
if len(row) > 17 and row[17] is not None:
entry.corporate_action = {"corporate_action_id": str(row[17]), "effective_date": row[18], "quantity_before": Decimal(str(row[19])), "quantity_after": Decimal(str(row[20])), "average_price_after": Decimal(str(row[21]))}
entries.append(entry)
return entries
def insert_ledger_trade(
self,