fix corporate action replay invariants

app/api/v1_portfolios.py

@@ -1324,11 +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, canonical_payload_hash
+        from app.domain.corporate_actions import 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)
+        # The repository derives all WAC and immutable snapshot fields under its row lock.
+        action = CorporateAction(str(uuid.uuid4()), portfolio_id, portfolio.asset_id, payload.action_type, payload.effective_date, payload.quantity_before, payload.quantity_after, Decimal("0"), Decimal("0"), Decimal("0"), 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__)
     except ValueError as exc:

app/repositories/postgres.py

@@ -445,24 +445,7 @@ class PostgresPortfolioRepository:
                 ON portfolios (broker, status, updated_at DESC)
                 """
             )
-            cur.execute(
+
-                """
-                CREATE TABLE IF NOT EXISTS portfolio_corporate_actions (
-                    corporate_action_id UUID PRIMARY KEY,
-                    portfolio_id TEXT NOT NULL,
-                    asset_id UUID NOT NULL,
-                    action_type TEXT NOT NULL CHECK (action_type = 'stock_consolidation'),
-                    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,
-                    UNIQUE (portfolio_id, idempotency_key)
-                )
-                """
-            )
             cur.execute(
                 """
                 CREATE TABLE IF NOT EXISTS portfolio_comments (
@@ -1140,6 +1123,8 @@ class PostgresPortfolioRepository:
         with self._connect().cursor() as cur:
             cur.execute(_SELECT_PORTFOLIO + "WHERE p.portfolio_id = %s", (portfolio_id,))
             row = cur.fetchone()
+            if row is None:
+                raise ValueError("포트폴리오를 찾을 수 없습니다")
             portfolio = self._row_to_portfolio(row)
             self._attach_tags(cur, [portfolio])
         return portfolio
@@ -1658,7 +1643,7 @@ class PostgresPortfolioRepository:
                     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))
+                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.
@@ -1673,12 +1658,19 @@ class PostgresPortfolioRepository:
                 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))
+                cur.execute("SELECT 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, 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:
+                    if prior[8] != 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.corporate_action_id, action.portfolio_id, action.asset_id, action.action_type,
+                     action.effective_date, action.actor_user_id, action.evidence_ref,
+                     action.idempotency_key, action.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,
+                     ledger_id, action.recorded_at) = prior
+                    action.corporate_action_id = str(action.corporate_action_id)
+                    action.ledger_entry_id = int(ledger_id)
                     action.idempotent_replay = True
                     return action
                 if Decimal(str(row[0])) != action.quantity_before:
@@ -1798,9 +1790,12 @@ 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,
+                    strategy_set_id, note, source,
-                    ca.corporate_action_id, ca.effective_date, ca.quantity_before,
+                    ca.corporate_action_id, ca.action_type, ca.effective_date,
-                    ca.quantity_after, ca.average_price_after
+                    ca.quantity_before, ca.quantity_after, ca.average_price_before,
+                    ca.average_price_after, ca.total_cost_basis, ca.cash_balance,
+                    ca.profit_buffer, ca.evidence_ref, ca.actor_user_id,
+                    ca.idempotency_key, ca.payload_hash, ca.ledger_entry_id, ca.recorded_at
                 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
@@ -1812,8 +1807,16 @@ class PostgresPortfolioRepository:
         entries = []
         for row in rows:
             entry = self._row_to_ledger_entry(row)
-            if len(row) > 17 and row[17] is not None:
+            if len(row) > 18 and row[18] 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]))}
+                entry.corporate_action = {
+                    "corporate_action_id": str(row[18]), "action_type": row[19], "effective_date": row[20],
+                    "quantity_before": Decimal(str(row[21])), "quantity_after": Decimal(str(row[22])),
+                    "average_price_before": Decimal(str(row[23])), "average_price_after": Decimal(str(row[24])),
+                    "total_cost_basis": Decimal(str(row[25])), "cash_balance": Decimal(str(row[26])),
+                    "profit_buffer": Decimal(str(row[27])), "evidence_ref": row[28], "actor_user_id": row[29],
+                    "idempotency_key": row[30], "payload_hash": row[31], "ledger_entry_id": int(row[32]),
+                    "recorded_at": row[33],
+                }
             entries.append(entry)
         return entries
 

tests/test_issue34_regressions.py

@@ -0,0 +1,40 @@
+from datetime import date, datetime, timezone
+from decimal import Decimal
+from pathlib import Path
+
+from app.domain.corporate_actions import canonical_payload_hash
+from app.domain.pnl import calculate_current_wac, calculate_realized_pnl
+from app.schemas.corporate_actions import CorporateActionRequest
+
+
+def test_buy_consolidation_sell_replays_wac_and_realized_pnl():
+    rows = [
+        {"side": "buy", "quantity": 10, "price": 100, "fee_amount": 0, "tax_amount": 0, "executed_at": datetime(2026, 1, 1, tzinfo=timezone.utc)},
+        {"side": "stock_consolidation", "quantity": 0, "price": 0, "quantity_before": 10, "quantity_after": 5, "average_price_after": 200, "executed_at": datetime(2026, 1, 2, tzinfo=timezone.utc)},
+        {"side": "sell", "quantity": 2, "price": 250, "fee_amount": 0, "tax_amount": 0, "executed_at": datetime(2026, 1, 3, tzinfo=timezone.utc)},
+    ]
+    assert calculate_current_wac(rows).current_wac == 200
+    result = calculate_realized_pnl(rows)
+    assert result.items[0].cost_basis_per_unit == 200
+    assert result.total_realized_pnl == 100
+
+
+def test_canonical_hash_same_payload_and_changed_payload_differ():
+    a = CorporateActionRequest(effective_date=date(2026, 1, 2), quantity_before=Decimal("10"), quantity_after=Decimal("5"), evidence_ref="ev")
+    b = CorporateActionRequest(effective_date=date(2026, 1, 2), quantity_before=Decimal("10"), quantity_after=Decimal("4"), evidence_ref="ev")
+    assert canonical_payload_hash(a) == canonical_payload_hash(a.model_copy())
+    assert canonical_payload_hash(a) != canonical_payload_hash(b)
+
+
+def test_runtime_schema_has_one_corporate_action_ddl_and_replay_mapping_starts_at_18():
+    source = Path("app/repositories/postgres.py").read_text()
+    assert source.count("CREATE TABLE IF NOT EXISTS portfolio_corporate_actions") == 1
+    assert "strategy_set_id, note, source," in source
+    assert "ca.corporate_action_id, ca.action_type" in source
+    assert 'row[18] is not None' in source
+
+
+def test_joined_corporate_action_metadata_is_complete_in_mapping_contract():
+    source = Path("app/repositories/postgres.py").read_text()
+    for field in ("action_type", "quantity_before", "quantity_after", "average_price_before", "average_price_after", "total_cost_basis", "cash_balance", "profit_buffer", "evidence_ref", "actor_user_id", "idempotency_key", "payload_hash", "recorded_at"):
+        assert f'"{field}"' in source