feat(portfolio): add atomic corporate action persistence

app/repositories/postgres.py

@@ -7,7 +7,7 @@ import json
 import logging
 import uuid
 from dataclasses import dataclass
-from datetime import UTC, datetime
+from datetime import UTC, date, datetime
 from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
 from typing import Any
 
@@ -22,6 +22,7 @@ from app.domain.errors import (
     StrategyNotFoundError,
 )
 from app.domain.models import (
+    CorporateAction,
     Asset,
     AuthToken,
     BrokerCredential,
@@ -332,6 +333,23 @@ class PostgresPortfolioRepository:
             )
             cur.execute("UPDATE portfolios SET name = portfolio_id WHERE name IS NULL OR name = ''")
             cur.execute("ALTER TABLE portfolios ALTER COLUMN name SET NOT NULL")
+            cur.execute(
+                """
+                CREATE TABLE IF NOT EXISTS portfolio_corporate_actions (
+                    corporate_action_id UUID PRIMARY KEY, portfolio_id TEXT NOT NULL,
+                    asset_id UUID, action_type TEXT NOT NULL, effective_date DATE NOT NULL,
+                    recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), actor_user_id TEXT,
+                    evidence_ref TEXT NOT NULL, idempotency_key TEXT NOT NULL,
+                    payload_hash TEXT NOT NULL, quantity_before NUMERIC(20,8) NOT NULL,
+                    quantity_after NUMERIC(20,8) NOT NULL, average_price_before NUMERIC(20,8) NOT NULL,
+                    average_price_after NUMERIC(20,8) NOT NULL, total_cost_basis NUMERIC(20,8) NOT NULL,
+                    cash_balance NUMERIC(20,8) NOT NULL, profit_buffer NUMERIC(20,8) NOT NULL,
+                    ledger_entry_id BIGINT NOT NULL,
+                    UNIQUE(portfolio_id, idempotency_key)
+                )
+                """
+            )
+            cur.execute("CREATE INDEX IF NOT EXISTS idx_corporate_actions_portfolio ON portfolio_corporate_actions (portfolio_id, effective_date)")
             cur.execute(
                 """
                 CREATE TABLE IF NOT EXISTS portfolio_ledger (
@@ -1616,6 +1634,32 @@ class PostgresPortfolioRepository:
                     ),
                 )
 
+    def record_corporate_action(self, portfolio_id: str, *, action: CorporateAction, idempotency_key: str, payload_hash: str) -> CorporateAction:
+        """Persist an immutable action and snapshot atomically."""
+        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,))
+                row = cur.fetchone()
+                if row is None:
+                    raise ValueError("portfolio not found")
+                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:
+                    if prior[1] != payload_hash:
+                        raise ValueError("IDEMPOTENCY_KEY_REUSE")
+                    action.corporate_action_id, action.ledger_entry_id, action.recorded_at = str(prior[0]), int(prior[2]), prior[3]
+                    action.idempotent_replay = True
+                    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("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
+                cur.execute("INSERT INTO portfolio_corporate_actions (corporate_action_id,portfolio_id,asset_id,action_type,effective_date,actor_user_id,evidence_ref,idempotency_key,payload_hash,quantity_before,quantity_after,average_price_before,average_price_after,total_cost_basis,cash_balance,profit_buffer,ledger_entry_id) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", (action.corporate_action_id,portfolio_id,action.asset_id,action.action_type,action.effective_date,action.actor_user_id,action.evidence_ref,idempotency_key,payload_hash,action.quantity_before,action.quantity_after,action.average_price_before,action.average_price_after,action.total_cost_basis,action.cash_balance,action.profit_buffer,lid))
+        return action
+
     def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]:
         self.ensure_schema()
         with self._connect().cursor() as cur: