feat(portfolio): add stock consolidation domain and schema

app/db/alembic/versions/20260913_0031_corporate_actions.py

@@ -0,0 +1,27 @@
+"""Additive storage for auditable, idempotent stock consolidations."""
+from alembic import op
+from sqlalchemy import text
+revision = "20260913_0031"
+down_revision = "20260820_0030"
+branch_labels = None
+depends_on = None
+
+def upgrade() -> None:
+    op.get_bind().execute(text("""
+    CREATE TABLE IF NOT EXISTS portfolio_corporate_actions (
+      corporate_action_id UUID PRIMARY KEY,
+      portfolio_id TEXT NOT NULL REFERENCES portfolios(portfolio_id) ON DELETE RESTRICT,
+      asset_id UUID NOT NULL REFERENCES assets(asset_id) ON DELETE RESTRICT,
+      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)
+    )"""))
+
+def downgrade() -> None:
+    op.get_bind().execute(text("DROP TABLE IF EXISTS portfolio_corporate_actions"))

app/domain/corporate_actions.py

@@ -0,0 +1,37 @@
+"""Pure domain rules for portfolio stock consolidations."""
+from __future__ import annotations
+from dataclasses import dataclass
+from decimal import Decimal, ROUND_HALF_UP
+
+class CorporateActionError(ValueError):
+    code = "INVALID_CORPORATE_ACTION_RATIO"
+    def __init__(self, message: str, code: str | None = None):
+        super().__init__(message); self.code = code or self.code
+
+@dataclass(frozen=True)
+class StockConsolidation:
+    quantity_before: Decimal
+    quantity_after: Decimal
+    average_price_before: Decimal
+    average_price_after: Decimal
+    total_cost_basis: Decimal
+    ratio: Decimal
+
+def calculate_stock_consolidation(quantity_before: Decimal, quantity_after: Decimal, average_price_before: Decimal) -> StockConsolidation:
+    q0, q1, wac = map(Decimal, (quantity_before, quantity_after, average_price_before))
+    if q0 <= 0 or q1 <= 0 or wac < 0:
+        raise CorporateActionError("quantities must be positive and price non-negative")
+    ratio = q1 / q0
+    if ratio >= 1 or ratio <= 0:
+        raise CorporateActionError("stock consolidation must reduce quantity")
+    if q1 != q1.to_integral_value() or q0 != q0.to_integral_value():
+        raise CorporateActionError("fractional result unsupported", "FRACTIONAL_RESULT_UNSUPPORTED")
+    total = (q0 * wac).quantize(Decimal("0.00000001"), rounding=ROUND_HALF_UP)
+    after = (total / q1).quantize(Decimal("0.00000001"), rounding=ROUND_HALF_UP)
+    if after * q1 != total:
+        raise CorporateActionError("fractional result unsupported", "FRACTIONAL_RESULT_UNSUPPORTED")
+    return StockConsolidation(q0, q1, wac, after, total, ratio)
+
+def apply_corporate_action_to_wac(quantity: Decimal, wac: Decimal, quantity_after: Decimal) -> tuple[Decimal, Decimal]:
+    result = calculate_stock_consolidation(quantity, quantity_after, wac)
+    return result.quantity_after, result.average_price_after

app/repositories/postgres.py

@@ -418,6 +418,24 @@ 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 (

app/schemas/corporate_actions.py

@@ -0,0 +1,38 @@
+from datetime import date, datetime
+from decimal import Decimal
+from typing import Literal
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+# Corporate actions (stock_consolidation-only, issue #34)
+class CorporateActionRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    action_type: Literal["stock_consolidation"] = "stock_consolidation"
+    effective_date: date
+    quantity_before: Decimal = Field(..., gt=0, max_digits=20, decimal_places=8)
+    quantity_after: Decimal = Field(..., gt=0, max_digits=20, decimal_places=8)
+    evidence_ref: str = Field(..., min_length=1, max_length=500)
+    confirm_daemon_managed: bool = False
+
+    @model_validator(mode="after")
+    def valid_ratio(self):
+        if self.quantity_after >= self.quantity_before:
+            raise ValueError("quantity_after must be less than quantity_before")
+        return self
+
+class CorporateActionResponse(BaseModel):
+    corporate_action_id: str
+    ledger_entry_id: int
+    portfolio_id: str
+    action_type: Literal["stock_consolidation"]
+    effective_date: date
+    quantity_before: Decimal
+    quantity_after: Decimal
+    average_price_before: Decimal
+    average_price_after: Decimal
+    total_cost_basis: Decimal
+    cash_balance: Decimal
+    profit_buffer: Decimal
+    evidence_ref: str
+    actor_user_id: str | None = None
+    recorded_at: datetime
+    idempotent_replay: bool = False

tests/test_corporate_actions.py

@@ -0,0 +1,15 @@
+from decimal import Decimal
+from app.domain.corporate_actions import calculate_stock_consolidation
+
+def test_stock_consolidation_preserves_cost_and_scales_wac():
+    result = calculate_stock_consolidation(Decimal("1452"), Decimal("726"), Decimal("1086.23"))
+    assert result.average_price_after == Decimal("2172.46")
+    assert result.total_cost_basis == Decimal("1577205.96000000")
+    assert result.ratio == Decimal("0.5")
+
+def test_stock_consolidation_rejects_fractional_result():
+    import pytest
+    from app.domain.corporate_actions import CorporateActionError
+    with pytest.raises(CorporateActionError) as exc:
+        calculate_stock_consolidation(Decimal("3"), Decimal("1.5"), Decimal("10"))
+    assert exc.value.code == "FRACTIONAL_RESULT_UNSUPPORTED"