From 14c118f50350d343d1544148a965bc5cd719d56c Mon Sep 17 00:00:00 2001 From: Justin Ko Date: Sun, 13 Sep 2026 14:04:30 +0900 Subject: [PATCH] fix corporate action ledger replay invariants --- app/api/v1_portfolios.py | 9 ++++--- app/cli/commands/portfolio.py | 12 ++++++--- app/domain/corporate_actions.py | 8 ++++++ app/repositories/postgres.py | 45 +++++++++++++++++++++++++++------ 4 files changed, 59 insertions(+), 15 deletions(-) diff --git a/app/api/v1_portfolios.py b/app/api/v1_portfolios.py index eb8041c..22aef26 100644 --- a/app/api/v1_portfolios.py +++ b/app/api/v1_portfolios.py @@ -1313,6 +1313,8 @@ def get_holdings_cost_basis( @router.post("/{portfolio_id}/corporate-actions", response_model=CorporateActionResponse, status_code=status.HTTP_201_CREATED) def create_corporate_action(portfolio_id: str, payload: CorporateActionRequest, response: Response, auth: AuthContext = Depends(require_trade_write), idempotency_key: str = Header(..., alias="Idempotency-Key")): + if not idempotency_key or not idempotency_key.strip(): + raise HTTPException(status_code=422, detail="Idempotency-Key must not be empty") repo = _get_repo() try: portfolio = repo.get(portfolio_id) @@ -1322,9 +1324,10 @@ def create_corporate_action(portfolio_id: str, payload: CorporateActionRequest, raise HTTPException(status_code=403, detail="Portfolio ownership mismatch") if portfolio.managed_by_daemon_id and not payload.confirm_daemon_managed: raise HTTPException(status_code=409, detail="DAEMON_MANAGED_CONFIRMATION_REQUIRED") - from app.domain.corporate_actions import calculate_stock_consolidation - calc = calculate_stock_consolidation(payload.quantity_before, payload.quantity_after, Decimal(str(getattr(portfolio, "price", 0)))) - digest = hashlib.sha256(payload.model_dump_json().encode()).hexdigest() + from app.domain.corporate_actions import calculate_stock_consolidation, canonical_payload_hash + # The repository validates and derives WAC from the locked ledger replay. + calc = calculate_stock_consolidation(payload.quantity_before, payload.quantity_after, Decimal("0")) + digest = canonical_payload_hash(payload) action = CorporateAction(str(uuid.uuid4()), portfolio_id, portfolio.asset_id, payload.action_type, payload.effective_date, calc.quantity_before, calc.quantity_after, calc.average_price_before, calc.average_price_after, calc.total_cost_basis, Decimal(str(portfolio.cash_balance)), Decimal(str(portfolio.profit_buffer)), payload.evidence_ref, idempotency_key, digest, 0, auth.user_id) saved = repo.record_corporate_action(portfolio_id, action=action, idempotency_key=idempotency_key, payload_hash=digest) return CorporateActionResponse(**saved.__dict__) diff --git a/app/cli/commands/portfolio.py b/app/cli/commands/portfolio.py index 460c930..09d5c43 100644 --- a/app/cli/commands/portfolio.py +++ b/app/cli/commands/portfolio.py @@ -290,7 +290,7 @@ def corporate_action_portfolio( """Record an auditable stock consolidation without a trade.""" from datetime import date from decimal import Decimal - from app.domain.corporate_actions import calculate_stock_consolidation + from app.domain.corporate_actions import calculate_stock_consolidation, canonical_payload_hash from app.domain.models import CorporateAction if action_type != "stock_consolidation": ctx.obj.console.print("Error: INVALID_CORPORATE_ACTION_TYPE") @@ -307,9 +307,13 @@ def corporate_action_portfolio( def _callback(service, repository, app_context): portfolio_id = _resolve_portfolio_id(repository, name, user) portfolio = repository.get(portfolio_id) - calc = calculate_stock_consolidation(Decimal(str(quantity_before)), Decimal(str(quantity_after)), Decimal(str(portfolio.price))) - action = CorporateAction("", portfolio_id, portfolio.asset_id, action_type, day, calc.quantity_before, calc.quantity_after, calc.average_price_before, calc.average_price_after, calc.total_cost_basis, Decimal(str(portfolio.cash_balance)), Decimal(str(portfolio.profit_buffer)), evidence_ref, idempotency_key, "", 0, user) - return repository.record_corporate_action(portfolio_id, action=action, idempotency_key=idempotency_key, payload_hash="") + if not idempotency_key.strip(): + raise ValueError("Idempotency-Key must not be empty") + payload = {"action_type": action_type, "effective_date": day.isoformat(), "quantity_before": str(quantity_before), "quantity_after": str(quantity_after), "evidence_ref": evidence_ref, "confirm_daemon_managed": False} + digest = canonical_payload_hash(payload) + calc = calculate_stock_consolidation(Decimal(str(quantity_before)), Decimal(str(quantity_after)), Decimal("0")) + action = CorporateAction("", portfolio_id, portfolio.asset_id, action_type, day, calc.quantity_before, calc.quantity_after, calc.average_price_before, calc.average_price_after, calc.total_cost_basis, Decimal(str(portfolio.cash_balance)), Decimal(str(portfolio.profit_buffer)), evidence_ref, idempotency_key, digest, 0, user) + return repository.record_corporate_action(portfolio_id, action=action, idempotency_key=idempotency_key, payload_hash=digest) detail = _with_service_and_repo(ctx, _callback) render_output(ctx.obj.console, ctx.obj.output_format, detail) diff --git a/app/domain/corporate_actions.py b/app/domain/corporate_actions.py index 189e499..82b4302 100644 --- a/app/domain/corporate_actions.py +++ b/app/domain/corporate_actions.py @@ -2,6 +2,14 @@ from __future__ import annotations from dataclasses import dataclass from decimal import Decimal, ROUND_HALF_UP +import hashlib +import json + +def canonical_payload_hash(payload: object) -> str: + if hasattr(payload, "model_dump"): + payload = payload.model_dump(mode="json", exclude_none=True) + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + return hashlib.sha256(encoded).hexdigest() class CorporateActionError(ValueError): code = "INVALID_CORPORATE_ACTION_RATIO" diff --git a/app/repositories/postgres.py b/app/repositories/postgres.py index c50e85e..1f90d99 100644 --- a/app/repositories/postgres.py +++ b/app/repositories/postgres.py @@ -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,