fix: make corporate actions atomic and recoverable

.github/workflows/publish-image.yml

@@ -31,7 +31,7 @@ jobs:
         uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4
         with:
           ref: ${{ github.sha }}
-          fetch-depth: 1
+          fetch-depth: 0
       - run: python scripts/image_release.py validate-checkout --actual "$(git rev-parse HEAD)" --expected "${GITHUB_SHA}"
       - run: python scripts/image_release.py validate-event --event "${GITHUB_EVENT_NAME}" --ref "${GITHUB_REF}" --commit "${GITHUB_SHA}" --project pyproject.toml
       - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
@@ -41,7 +41,12 @@ jobs:
       - run: uv sync --extra dev --locked
       - run: uv run pytest -q
       - name: Honest scoped lint
-        run: uv run ruff check --select F,I app/db/migrations.py scripts/image_release.py tests/release
+        run: uv run ruff check --select F,I app/db/migrations.py scripts/image_release.py scripts/ruff_ratchet.py tests/release
+      - name: Reject Ruff findings introduced by v0.3.0 or later patches
+        run: |
+          uv run python scripts/ruff_ratchet.py --base v0.2.1
+          uv run python scripts/ruff_ratchet.py --base v0.3.0
+          uv run python scripts/ruff_ratchet.py --base v0.3.1
       - name: Record exact source migration inventory
         run: uv run python -c 'from app.db.migrations import migration_inventory; print("\n".join(migration_inventory()))' > migration-inventory.txt
       - name: Create excluded synthetic canary

.github/workflows/verify.yml

@@ -11,18 +11,45 @@ permissions:
 jobs:
   verify:
     runs-on: ubuntu-latest
+    services:
+      timescaledb:
+        image: timescale/timescaledb@sha256:252a443e2936039b83dd8da1373d01e59e932d1054fa6adf1bc061f1d56ae60a
+        env:
+          POSTGRES_DB: portfolio_hub
+          POSTGRES_USER: portfolio_hub_user
+          POSTGRES_PASSWORD: change-me
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U portfolio_hub_user -d portfolio_hub"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 12
     permissions:
       contents: read
     steps:
       - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4
+        with:
+          fetch-depth: 0
       - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
         with:
           python-version: "3.11"
       - run: python -m pip install "uv==0.11.14"
       - run: uv sync --extra dev --locked
       - run: uv run pytest -q
+      - name: Verify corporate-action atomicity and guarded recovery in PostgreSQL
+        env:
+          POSTGRES_PASSWORD: change-me
+        run: |
+          export PORTFOLIOHUB_TEST_POSTGRES_DSN="postgresql://portfolio_hub_user:${POSTGRES_PASSWORD}@127.0.0.1:5432/postgres"
+          uv run pytest -q tests/test_corporate_action_atomic_recovery_postgres.py
       - name: Honest scoped lint for release additions
-        run: uv run ruff check --select F,I app/db/migrations.py scripts/image_release.py tests/release
+        run: uv run ruff check --select F,I app/db/migrations.py scripts/image_release.py scripts/ruff_ratchet.py tests/release
+      - name: Reject Ruff findings introduced by v0.3.0 or later patches
+        run: |
+          uv run python scripts/ruff_ratchet.py --base v0.2.1
+          uv run python scripts/ruff_ratchet.py --base v0.3.0
+          uv run python scripts/ruff_ratchet.py --base v0.3.1
       - name: Validate container contracts without registry writes
         run: uv run pytest -q tests/release
       - name: Build and smoke exact pull-request or main source image

app/api/v1_portfolios.py

@@ -24,15 +24,19 @@ from app.core.auth import AuthContext, Permission, require_read, require_trade_w
 from app.core.config import get_settings
 from app.domain.errors import DaemonInactiveError, DaemonNotFoundError
 from app.domain.models import CorporateAction, Portfolio, PortfolioComment
-from app.domain.services import PORTFOLIO_NOT_FOUND_MESSAGE
 from app.domain.pnl import calculate_current_wac, calculate_realized_pnl
+from app.domain.services import PORTFOLIO_NOT_FOUND_MESSAGE
 from app.repositories.factory import get_portfolio_repository_from_settings
+from app.schemas.corporate_actions import (
+    CorporateActionRequest,
+    CorporateActionResponse,
+)
 from app.schemas.v1 import (
+    CorporateActionLedgerHistoryItem,
     DividendLedgerHistoryItem,
     DividendManualLedgerResponse,
     HoldingsCostBasisResponse,
     LedgerHistoryItem,
-    CorporateActionLedgerHistoryItem,
     LedgerHistoryResponse,
     LedgerTradeRequest,
     LedgerTradeResponse,
@@ -57,7 +61,6 @@ from app.schemas.v1 import (
     TradeHistoryItem,
     TradeHistoryResponse,
 )
-from app.schemas.corporate_actions import CorporateActionRequest, CorporateActionResponse
 from app.services.strategy_set_service import (
     StrategyInactiveError,
     StrategyNotFoundError,
@@ -1313,7 +1316,13 @@ 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")):
+def create_corporate_action(
+    portfolio_id: str,
+    payload: CorporateActionRequest,
+    response: Response,
+    auth: AuthContext = Depends(require_trade_write),  # noqa: B008 - FastAPI dependency declaration
+    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()
@@ -1328,8 +1337,14 @@ def create_corporate_action(portfolio_id: str, payload: CorporateActionRequest,
         from app.domain.corporate_actions import canonical_payload_hash
         digest = canonical_payload_hash(payload)
         # 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)
+        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,
+            resume_partial=payload.resume_partial,
+        )
         return CorporateActionResponse(**saved.__dict__)
     except ValueError as exc:
         code = str(exc)

app/cli/commands/portfolio.py

@@ -285,13 +285,22 @@ def corporate_action_portfolio(
     quantity_after: float = typer.Option(..., "--quantity-after", min=0.0000001),
     evidence_ref: str = typer.Option(..., "--evidence-ref"),
     idempotency_key: str = typer.Option(..., "--idempotency-key"),
+    resume_partial: bool = typer.Option(
+        False,
+        "--resume-partial",
+        help="Resume only a verified quantity-only partial corporate action.",
+    ),
     force: bool = typer.Option(False, "--force"),
 ) -> None:
     """Record an auditable stock consolidation without a trade."""
+    import uuid
     from datetime import date
     from decimal import Decimal
-    import uuid
-    from app.domain.corporate_actions import calculate_stock_consolidation, canonical_payload_hash
+
+    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")
@@ -310,11 +319,17 @@ def corporate_action_portfolio(
         portfolio = repository.get(portfolio_id)
         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}
+        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, "resume_partial": resume_partial}
         digest = canonical_payload_hash(payload)
-        calc = calculate_stock_consolidation(Decimal(str(quantity_before)), Decimal(str(quantity_after)), Decimal("0"))
+        calc = calculate_stock_consolidation(Decimal(str(quantity_before)), Decimal(str(quantity_after)), Decimal(0))
         action = CorporateAction(str(uuid.uuid4()), 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)
+        return repository.record_corporate_action(
+            portfolio_id,
+            action=action,
+            idempotency_key=idempotency_key,
+            payload_hash=digest,
+            resume_partial=resume_partial,
+        )
     detail = _with_service_and_repo(ctx, _callback)
     render_output(ctx.obj.console, ctx.obj.output_format, detail)
 

app/core/config.py

@@ -18,7 +18,7 @@ if env_file.exists():
 
 class Settings(BaseSettings):
     app_name: str = "PortfolioHub"
-    app_version: str = "0.3.0"
+    app_version: str = "0.3.3"
     environment: str = "development"
     allowed_origins_raw: str = Field(default="http://localhost:3000", validation_alias="PORTFOLIOHUB_ALLOWED_ORIGINS")
     bootstrap_username: str = Field(default="", validation_alias="PORTFOLIOHUB_BOOTSTRAP_USERNAME")

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

@@ -1,6 +1,7 @@
 """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

app/domain/corporate_actions.py

@@ -1,13 +1,27 @@
 """Pure domain rules for portfolio stock consolidations."""
 from __future__ import annotations
-from dataclasses import dataclass
-from decimal import Decimal, ROUND_HALF_UP
+
 import hashlib
 import json
+from dataclasses import dataclass
+from decimal import ROUND_HALF_UP, Decimal
+
+STORAGE_QUANTUM = Decimal("0.00000001")
+
+
+def quantize_stored_decimal(value: Decimal) -> Decimal:
+    """Round a monetary value to the NUMERIC(*, 8) persistence contract."""
+    return value.quantize(STORAGE_QUANTUM, rounding=ROUND_HALF_UP)
+
 
 def canonical_payload_hash(payload: object) -> str:
     if hasattr(payload, "model_dump"):
-        payload = payload.model_dump(mode="json", exclude_none=True)
+        excluded_fields = (
+            {"resume_partial"} if not getattr(payload, "resume_partial", False) else set()
+        )
+        payload = payload.model_dump(mode="json", exclude_none=True, exclude=excluded_fields)
+    elif isinstance(payload, dict) and payload.get("resume_partial") is False:
+        payload = {key: value for key, value in payload.items() if key != "resume_partial"}
     encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
     return hashlib.sha256(encoded).hexdigest()
 
@@ -34,11 +48,13 @@ def calculate_stock_consolidation(quantity_before: Decimal, quantity_after: Deci
         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)
+    # Preserve the authoritative pre-action total independently. A per-unit WAC
+    # rounded to the storage scale cannot always multiply back to that total
+    # exactly, and that harmless decimal residue is not a fractional share.
+    total = quantize_stored_decimal(q0 * wac)
+    before = quantize_stored_decimal(wac)
+    after = quantize_stored_decimal(total / q1)
+    return StockConsolidation(q0, q1, before, 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)

app/domain/pnl.py

@@ -179,7 +179,11 @@ def calculate_realized_pnl(
             if abs(running_qty - before) > 1e-9:
                 warning_triggered = True
             running_qty = after
-            running_wac = float(row.get("average_price_after", running_wac * before / after))
+            total_cost_basis = row.get("total_cost_basis")
+            if total_cost_basis is not None:
+                running_wac = float(total_cost_basis) / after
+            else:
+                running_wac = float(row.get("average_price_after", running_wac * before / after))
 
     warning: str | None = None
     if warning_triggered:
@@ -272,7 +276,11 @@ def calculate_current_wac(rows: list[dict[str, Any]]) -> CurrentWacResult:
             if abs(running_qty - before) > 1e-9:
                 warning_triggered = True
             running_qty = after
-            running_wac = float(row.get("average_price_after", running_wac * before / after))
+            total_cost_basis = row.get("total_cost_basis")
+            if total_cost_basis is not None:
+                running_wac = float(total_cost_basis) / after
+            else:
+                running_wac = float(row.get("average_price_after", running_wac * before / after))
 
     warning: str | None = None
     if warning_triggered:

app/repositories/portfolio_repository.py

@@ -284,7 +284,15 @@ class PortfolioRepository(Protocol):
         note: str = "",
     ) -> None: ...
 
-    def record_corporate_action(self, portfolio_id: str, *, action: CorporateAction, idempotency_key: str, payload_hash: str) -> CorporateAction: ...
+    def record_corporate_action(
+        self,
+        portfolio_id: str,
+        *,
+        action: CorporateAction,
+        idempotency_key: str,
+        payload_hash: str,
+        resume_partial: bool = False,
+    ) -> CorporateAction: ...
 
     def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]: ...
 

app/repositories/postgres.py

@@ -7,13 +7,17 @@ import json
 import logging
 import uuid
 from dataclasses import dataclass
-from datetime import UTC, date, datetime
+from datetime import UTC, datetime
 from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
 from typing import Any
 
 import psycopg
 from psycopg import sql
 
+from app.domain.corporate_actions import (
+    calculate_stock_consolidation,
+    quantize_stored_decimal,
+)
 from app.domain.errors import (
     DaemonInactiveError,
     DaemonInUseError,
@@ -22,10 +26,10 @@ from app.domain.errors import (
     StrategyNotFoundError,
 )
 from app.domain.models import (
-    CorporateAction,
     Asset,
     AuthToken,
     BrokerCredential,
+    CorporateAction,
     Daemon,
     LedgerEntry,
     Portfolio,
@@ -1628,73 +1632,240 @@ 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."""
+    def record_corporate_action(
+        self,
+        portfolio_id: str,
+        *,
+        action: CorporateAction,
+        idempotency_key: str,
+        payload_hash: str,
+        resume_partial: bool = False,
+    ) -> CorporateAction:
+        """Persist an immutable action and snapshot in one explicit transaction."""
         self.ensure_schema()
-        with self._connect() as conn:
-            with conn.cursor() as cur:
-                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")
-                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[8] != payload_hash:
-                        raise ValueError("IDEMPOTENCY_KEY_REUSE")
-                    (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 row[4] != "enabled":
-                    raise ValueError("CORPORATE_ACTION_PORTFOLIO_DISABLED")
-                if str(row[3]) != str(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, 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[8] != payload_hash:
-                        raise ValueError("IDEMPOTENCY_KEY_REUSE")
-                    (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:
-                    raise ValueError("CORPORATE_ACTION_STALE_QUANTITY")
-                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
-                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))
+        conn = self._connect()
+
+        def hydrate_replay(prior: tuple) -> CorporateAction:
+            if prior[8] != payload_hash:
+                raise ValueError("IDEMPOTENCY_KEY_REUSE")
+            (
+                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
+
+        prior_sql = """
+            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
+        """
+
+        # Repository connections are autocommit connections.  A connection
+        # context alone therefore cannot protect this multi-statement write.
+        with conn.transaction(), conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT asset_quantity, cash_balance, profit_buffer, asset_id,
+                       status, asset_price
+                FROM portfolios
+                WHERE portfolio_id=%s
+                FOR UPDATE
+                """,
+                (portfolio_id,),
+            )
+            row = cur.fetchone()
+            if row is None:
+                raise ValueError("portfolio not found")
+
+            cur.execute(prior_sql, (portfolio_id, idempotency_key))
+            prior = cur.fetchone()
+            if prior:
+                return hydrate_replay(prior)
+
+            snapshot_quantity = Decimal(str(row[0]))
+            cash_balance = Decimal(str(row[1]))
+            profit_buffer = Decimal(str(row[2]))
+            reference_price_before = quantize_stored_decimal(Decimal(str(row[5])))
+            if row[4] != "enabled":
+                raise ValueError("CORPORATE_ACTION_PORTFOLIO_DISABLED")
+            if str(row[3]) != str(action.asset_id):
+                raise ValueError("CORPORATE_ACTION_ASSET_MISMATCH")
+            if cash_balance != action.cash_balance or profit_buffer != 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 trade ledger, never the
+            # portfolio reference price or the already-mutated snapshot.
+            cur.execute(
+                """
+                SELECT side, quantity, price, fee_amount, tax_amount, executed_at
+                FROM portfolio_ledger
+                WHERE portfolio_id=%s AND side IN ('buy','sell')
+                ORDER BY executed_at ASC NULLS FIRST, ledger_entry_id ASC
+                """,
+                (portfolio_id,),
+            )
+            history = [
+                {
+                    "side": ledger_row[0],
+                    "quantity": ledger_row[1],
+                    "price": ledger_row[2],
+                    "fee_amount": ledger_row[3],
+                    "tax_amount": ledger_row[4],
+                    "executed_at": ledger_row[5],
+                }
+                for ledger_row in cur.fetchall()
+            ]
+            from app.domain.pnl import calculate_current_wac
+
+            wac = calculate_current_wac(history)
+            if Decimal(str(wac.current_quantity)) != action.quantity_before:
+                raise ValueError("CORPORATE_ACTION_STALE_QUANTITY")
+
+            if resume_partial:
+                if snapshot_quantity != action.quantity_after:
+                    raise ValueError("CORPORATE_ACTION_RECOVERY_STATE_MISMATCH")
+                cur.execute(
+                    """
+                    SELECT
+                        EXISTS (
+                            SELECT 1 FROM portfolio_corporate_actions
+                            WHERE portfolio_id=%s
+                        ),
+                        EXISTS (
+                            SELECT 1 FROM portfolio_ledger
+                            WHERE portfolio_id=%s AND source='corporate_action'
+                        )
+                    """,
+                    (portfolio_id, portfolio_id),
+                )
+                audit_exists, ledger_exists = cur.fetchone()
+                if audit_exists or ledger_exists:
+                    raise ValueError("CORPORATE_ACTION_RECOVERY_ALREADY_RECORDED")
+            elif snapshot_quantity != action.quantity_before:
+                raise ValueError("CORPORATE_ACTION_STALE_QUANTITY")
+
+            consolidation = calculate_stock_consolidation(
+                action.quantity_before,
+                action.quantity_after,
+                Decimal(str(wac.current_wac or 0)),
+            )
+            action.average_price_before = consolidation.average_price_before
+            action.average_price_after = consolidation.average_price_after
+            action.total_cost_basis = consolidation.total_cost_basis
+            action.cash_balance = cash_balance
+            action.profit_buffer = profit_buffer
+
+            consolidation_ratio = action.quantity_before / action.quantity_after
+            reference_price_after = quantize_stored_decimal(
+                reference_price_before * consolidation_ratio
+            )
+            total_value_after = quantize_stored_decimal(
+                cash_balance + action.quantity_after * reference_price_after
+            )
+
+            cur.execute(
+                """
+                UPDATE portfolios
+                SET asset_quantity=%s, asset_price=%s, updated_at=NOW()
+                WHERE portfolio_id=%s
+                """,
+                (action.quantity_after, reference_price_after, portfolio_id),
+            )
+            cur.execute(
+                """
+                INSERT INTO portfolio_ledger (
+                    portfolio_id, recorded_at, side, source, quantity, price,
+                    cash_balance_after, asset_quantity_after, total_value_after,
+                    profit_buffer_after, executed_at, note
+                ) VALUES (
+                    %s, NOW(), 'stock_consolidation', 'corporate_action', 0, %s,
+                    %s, %s, %s, %s, NOW(), %s
+                )
+                RETURNING ledger_entry_id, recorded_at
+                """,
+                (
+                    portfolio_id,
+                    reference_price_after,
+                    cash_balance,
+                    action.quantity_after,
+                    total_value_after,
+                    profit_buffer,
+                    action.evidence_ref,
+                ),
+            )
+            lid, recorded = cur.fetchone()
+            action.ledger_entry_id = int(lid)
+            action.recorded_at = 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,
+                    cash_balance,
+                    profit_buffer,
+                    lid,
+                ),
+            )
         return action
 
     def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]:

app/schemas/corporate_actions.py

@@ -1,8 +1,10 @@
 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")
@@ -12,6 +14,7 @@ class CorporateActionRequest(BaseModel):
     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
+    resume_partial: bool = False
 
     @model_validator(mode="after")
     def valid_ratio(self):

docs/container-operations.md

@@ -1,11 +1,11 @@
-# PortfolioHub v0.3.0 container contract
+# PortfolioHub v0.3.3 container contract
 
 ## Publication is not deployment
 
-Only a requester-authorized push of the strict tag `v0.3.0` can publish this revision. The workflow checks that the tag equals `[project].version`, checks out `github.sha`, runs the complete Python suite, builds one `linux/amd64` image from `uv.lock`, and tests those exact bytes before publication. It gives the same image exactly two selectors:
+Only a requester-authorized push of the strict tag `v0.3.3` can publish this revision. The workflow checks that the tag equals `[project].version`, checks out `github.sha`, runs the complete Python suite, builds one `linux/amd64` image from `uv.lock`, and tests those exact bytes before publication. It gives the same image exactly two selectors:
 
 - `ghcr.io/fourblocks2020/portfolio-hub:<source-commit-first-12>`
-- `ghcr.io/fourblocks2020/portfolio-hub:v0.3.0`
+- `ghcr.io/fourblocks2020/portfolio-hub:v0.3.3`
 
 It never publishes `latest`. Both selectors must report the same registry digest; deployment and rollback must use the immutable identity:
 
@@ -13,9 +13,9 @@ It never publishes `latest`. Both selectors must report the same registry digest
 ghcr.io/fourblocks2020/portfolio-hub@sha256:<64-lowercase-hex-digest>
 ```
 
-OCI labels are fail-closed build inputs: `org.opencontainers.image.source`, the full 40-character `org.opencontainers.image.revision`, and strict `org.opencontainers.image.version=v0.3.0`. Publication does **not** deploy, replace, start, stop, or restart a production service. This repository provides no production Compose, Kubernetes, Helm, systemd, or Dockge configuration.
+OCI labels are fail-closed build inputs: `org.opencontainers.image.source`, the full 40-character `org.opencontainers.image.revision`, and strict `org.opencontainers.image.version=v0.3.3`. Publication does **not** deploy, replace, start, stop, or restart a production service. This repository provides no production Compose, Kubernetes, Helm, systemd, or Dockge configuration.
 
-The Dockerfile frontend, Python base image, uv image, and GitHub Actions are digest/commit pinned. The workflow intentionally uses `--provenance=false` to preserve the tested `docker save` archive and build-once/no-rebuild identity. Registry-attached provenance, SBOM, signing, and attestations are a nonblocking v0.3.0 supply-chain gap; they must not be added by rebuilding a different image in the publish job.
+The Dockerfile frontend, Python base image, uv image, and GitHub Actions are digest/commit pinned. The workflow intentionally uses `--provenance=false` to preserve the tested `docker save` archive and build-once/no-rebuild identity. Registry-attached provenance, SBOM, signing, and attestations are a nonblocking v0.3.3 supply-chain gap; they must not be added by rebuilding a different image in the publish job.
 
 ## Image runtime
 
@@ -66,19 +66,19 @@ Values are read at process start; changing any value requires a container restar
 | `PORTFOLIOHUB_BOOTSTRAP_PASSWORD` | optional; empty | **yes** | bootstrap password; inject temporarily and remove after first successful start |
 | `PORTFOLIOHUB_ENVIRONMENT` | optional; `development` | no | reported environment metadata |
 | `PORTFOLIOHUB_APP_NAME` | optional; `PortfolioHub` | no | application/OpenAPI display name |
-| `PORTFOLIOHUB_APP_VERSION` | do not override for v0.3.0 | `0.3.0` | no | health/OpenAPI version metadata |
+| `PORTFOLIOHUB_APP_VERSION` | do not override for v0.3.3 | `0.3.3` | no | health/OpenAPI version metadata |
 
-`REDIS_HOST`, `REDIS_PORT`, `REDIS_DB`, `REDIS_PASSWORD`, `LOKI_HOST`, `LOKI_PORT`, and `LOKI_ENABLED` are still parsed for source compatibility but the inspected v0.3.0 application does not open Redis or Loki connections. They are not production dependencies of this image.
+`REDIS_HOST`, `REDIS_PORT`, `REDIS_DB`, `REDIS_PASSWORD`, `LOKI_HOST`, `LOKI_PORT`, and `LOKI_ENABLED` are still parsed for source compatibility but the inspected v0.3.3 application does not open Redis or Loki connections. They are not production dependencies of this image.
 
 ### Legacy and sibling mapping
 
-The existing database variable names remain `TIMESCALEDB_*`; v0.3.0 deliberately does not invent a `DATABASE_URL` alias. Existing `PORTFOLIOHUB_{READ,TRADE_WRITE,ADMIN}_TOKENS` Bearer-token behavior remains compatible. `trex-core`'s `TREX_PORTFOLIO_HUB_BASE_URL` and TOML `portfolio_hub_token`, and `trex-web`'s server-only `PORTFOLIO_HUB_API_BASE_URL`, are **caller-side** settings: point them to `http://portfolio-hub:8000` (or the equivalent service URL) and do not inject them into this container.
+The existing database variable names remain `TIMESCALEDB_*`; v0.3.3 deliberately does not invent a `DATABASE_URL` alias. Existing `PORTFOLIOHUB_{READ,TRADE_WRITE,ADMIN}_TOKENS` Bearer-token behavior remains compatible. `trex-core`'s `TREX_PORTFOLIO_HUB_BASE_URL` and TOML `portfolio_hub_token`, and `trex-web`'s server-only `PORTFOLIO_HUB_API_BASE_URL`, are **caller-side** settings: point them to `http://portfolio-hub:8000` (or the equivalent service URL) and do not inject them into this container.
 
 ## Liveness and readiness
 
 - **Liveness:** unauthenticated `GET /health` must return HTTP 200 with `status=ok`. It is intentionally process-only and does not query PostgreSQL. `db_enabled` only reports configuration; it is not readiness.
 - **Readiness:** run `pfhub --format json health-check` with the same runtime environment. With DB enabled it performs a repository ping, reads `alembic_version` without migration, compares it with the single packaged Alembic head, and exits nonzero on connectivity, missing schema, multiple heads, or revision mismatch.
-- **API identity:** `GET /openapi.json` is stable for smoke verification. Health version is `0.3.0`.
+- **API identity:** `GET /openapi.json` is stable for smoke verification. Health version is `0.3.3`.
 
 Do not route traffic from a static `/health` success alone. Readiness checks never apply, stamp, or repair migrations.
 
@@ -107,7 +107,7 @@ Image rollback is **not** database rollback. Selecting older application bytes d
 
 ## API verification and known caller blockers
 
-v0.3.0 preserves Bearer authentication, retains all-source unified-ledger reads, and adds stock-consolidation corporate-action recording and replay. The release gate runs the complete portfolio-hub source test suite (including auth success/failure, strategy catalog/data, portfolio and ledger APIs), then verifies the built image's health/OpenAPI surface, exact 31-file migration inventory, isolated PostgreSQL migration twice, readiness at `0031`, persistent-volume restart, non-root/read-only operation, secret canary absence, and graceful SIGTERM. Publication consumes the tested archive without rebuilding it.
+v0.3.3 preserves Bearer authentication and all-source unified-ledger reads. Corporate-action quantity/reference-price updates, the complete ledger snapshot (including `total_value_after`), and the audit row execute in one explicit transaction. The guarded `resume_partial` path repairs only a verified quantity-only partial state without raw SQL. The release gate runs the complete portfolio-hub source test suite (including auth success/failure, strategy catalog/data, portfolio and ledger APIs), then verifies the built image's health/OpenAPI surface, exact 31-file migration inventory, isolated PostgreSQL migration twice, readiness at `0031`, persistent-volume restart, non-root/read-only operation, secret canary absence, and graceful SIGTERM. Publication consumes the tested archive without rebuilding it. The changed-scope Ruff gate uses the locked linter and the ACMR Python path selector, compares normalized findings against v0.2.1, v0.3.0, and v0.3.1, and rejects every newly introduced finding.
 
 The strategy catalog API remains additive: strategy responses expose `nick`, admin create/update requests may set it, and portfolio responses expose the current strategy as `strategy_nick`. Existing clients may omit `nick`; supplied values are trimmed and blank-only values are rejected. trex-web v0.2.3 displays `strategy_nick`, but that dependency does not resolve or waive the separate caller blockers below.
 

docs/interface-contract.md

@@ -438,8 +438,12 @@
 - Auth: same owner-scoped user/session token policy as manual write.
 - Query: `limit` 1..200, default 50.
 - `GET /v1/portfolios/{id}/ledger` returns corporate-action events as a distinct `CorporateActionLedgerHistoryItem` with event ID/type, effective date, before/after quantity and WAC, total cost basis, cash/profit buffer, evidence, actor, and recorded time. These events are never represented as BUY/SELL trades.
+- For stock consolidations, WAC fields are rounded half-up to the persisted 8-decimal scale. `total_cost_basis` is preserved independently as the authoritative pre-action acquisition cost; a sub-cent residue when `average_price_after * quantity_after` is recomputed does not make an integral-share consolidation invalid.
+- `POST /v1/portfolios/{id}/corporate-actions` updates quantity and the post-action reference price, appends the `portfolio_ledger` snapshot (including non-null `total_value_after`), and appends the corporate-action audit row in one explicit database transaction. A failure in any statement rolls back every statement.
+- The post-action reference price is `asset_price_before * quantity_before / quantity_after`; ledger `total_value_after` is `cash_balance_after + asset_quantity_after * post_action_reference_price`. Cash and profit buffer are unchanged.
+- `resume_partial=false` is the compatible default. `resume_partial=true` is accepted only when the locked portfolio quantity already equals `quantity_after`, trade-ledger replay still proves `quantity_before`, and both corporate-action audit and corporate-action ledger rows are absent. Same-key replay is idempotent; a different-key recovery after repair is rejected.
 - `GET /v1/portfolios/{id}/trades` remains restricted to BUY/SELL history.
-- `portfolio corporate-action` is the operator command; it requires an idempotency key and confirmation unless `--force` is supplied.
+- `portfolio corporate-action` is the operator command; it requires an idempotency key and confirmation unless `--force` is supplied. `--resume-partial` enables only the guarded recovery mode above.
 
 
 ### Endpoint: `GET /v1/portfolios/{portfolio_id}/returns/summary` (REQ-056, BAT-024)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "portfoliohub"
-version = "0.3.0"
+version = "0.3.3"
 description = "PortfolioHub API for portfolio management and ledger access"
 readme = "README.md"
 requires-python = ">=3.11"

scripts/ruff_ratchet.py

@@ -0,0 +1,128 @@
+"""Reject Ruff findings introduced since a stable release tag."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import shutil
+import subprocess
+import tempfile
+from collections import Counter
+from pathlib import Path
+
+
+def _run(command: list[str], *, cwd: Path, accepted: tuple[int, ...] = (0,)) -> subprocess.CompletedProcess[str]:
+    result = subprocess.run(command, cwd=cwd, text=True, capture_output=True, check=False)
+    if result.returncode not in accepted:
+        raise RuntimeError(
+            f"command failed ({result.returncode}): {' '.join(command)}\n{result.stderr}"
+        )
+    return result
+
+
+def changed_python_paths(repo: Path, base: str) -> list[str]:
+    result = _run(
+        [
+            "git",
+            "diff",
+            "--name-only",
+            "-z",
+            "--diff-filter=ACMR",
+            f"{base}...HEAD",
+            "--",
+            "*.py",
+        ],
+        cwd=repo,
+    )
+    return sorted(path for path in result.stdout.split("\0") if path)
+
+
+def _source_line(root: Path, relative_path: str, row: int) -> str:
+    lines = (root / relative_path).read_text(encoding="utf-8").splitlines()
+    return lines[row - 1].strip() if 0 < row <= len(lines) else ""
+
+
+def normalized_findings(
+    diagnostics: list[dict[str, object]], root: Path
+) -> Counter[tuple[str, str, str, str]]:
+    findings: Counter[tuple[str, str, str, str]] = Counter()
+    for diagnostic in diagnostics:
+        filename = Path(str(diagnostic["filename"])).resolve()
+        relative_path = filename.relative_to(root.resolve()).as_posix()
+        location = diagnostic["location"]
+        if not isinstance(location, dict):
+            raise TypeError("Ruff diagnostic location must be an object")
+        row = int(location["row"])
+        finding = (
+            relative_path,
+            str(diagnostic["code"]),
+            str(diagnostic["message"]),
+            _source_line(root, relative_path, row),
+        )
+        findings[finding] += 1
+    return findings
+
+
+def _ruff_diagnostics(root: Path, paths: list[str]) -> list[dict[str, object]]:
+    if not paths:
+        return []
+    ruff = shutil.which("ruff")
+    if ruff is None:
+        raise RuntimeError("ruff executable is not available")
+    result = _run(
+        [ruff, "check", *paths, "--output-format=json"],
+        cwd=root,
+        accepted=(0, 1),
+    )
+    payload = json.loads(result.stdout)
+    if not isinstance(payload, list):
+        raise TypeError("Ruff JSON output must be a list")
+    return payload
+
+
+def introduced_findings(repo: Path, base: str) -> tuple[list[str], Counter[tuple[str, str, str, str]]]:
+    paths = changed_python_paths(repo, base)
+    if not paths:
+        return paths, Counter()
+
+    current = normalized_findings(_ruff_diagnostics(repo, paths), repo)
+    with tempfile.TemporaryDirectory(prefix="portfoliohub-ruff-base-") as parent:
+        baseline_root = Path(parent) / "tree"
+        _run(["git", "worktree", "add", "--detach", str(baseline_root), base], cwd=repo)
+        try:
+            baseline_paths = [path for path in paths if (baseline_root / path).is_file()]
+            baseline = normalized_findings(
+                _ruff_diagnostics(baseline_root, baseline_paths), baseline_root
+            )
+        finally:
+            _run(
+                ["git", "worktree", "remove", "--force", str(baseline_root)],
+                cwd=repo,
+            )
+    return paths, current - baseline
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--base", required=True, help="stable comparison tag")
+    args = parser.parse_args()
+
+    repo = Path(
+        _run(["git", "rev-parse", "--show-toplevel"], cwd=Path.cwd()).stdout.strip()
+    )
+    ruff_version = _run([shutil.which("ruff") or "ruff", "--version"], cwd=repo).stdout.strip()
+    paths, introduced = introduced_findings(repo, args.base)
+    print(f"base={args.base} head=HEAD selector=ACMR-python files={len(paths)} tool={ruff_version}")
+    for path in paths:
+        print(f"selected={path}")
+    if introduced:
+        for (path, code, message, source), count in sorted(introduced.items()):
+            print(f"NEW {count}x {path}: {code} {message} | {source}")
+        print(f"introduced_findings={sum(introduced.values())}")
+        return 1
+    print("introduced_findings=0")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/release/test_release_contracts.py

@@ -17,6 +17,10 @@ def test_branch_verification_has_no_registry_write_authority_or_path():
     for forbidden in ("packages: write","docker login","docker push","REGISTRY_TOKEN","image_release.py publish"):
         assert forbidden not in text
     assert "uv sync --extra dev --locked" in text and "uv run pytest -q" in text
+    assert "Verify corporate-action atomicity and guarded recovery in PostgreSQL" in text
+    assert "tests/test_corporate_action_atomic_recovery_postgres.py" in text
+    assert "${POSTGRES_PASSWORD}@127.0.0.1:5432/postgres" in text
+    assert "***@" not in text
 
 def test_tag_publication_is_strict_single_build_handoff_without_latest():
     text=PUBLISH.read_text(); verify,publish=text.split("\n  publish:",1)
@@ -58,8 +62,8 @@ def test_release_version_and_migration_contract_has_current_head_0031():
         for path in (REPO / "app/db/alembic/versions").glob("*.py")
         if path.name != "__init__.py"
     )
-    assert 'version = "0.3.0"' in project
-    assert 'app_version: str = "0.3.0"' in config
+    assert 'version = "0.3.3"' in project
+    assert 'app_version: str = "0.3.3"' in config
     assert len(migrations) == 31
     assert migrations[-1].startswith("20260913_0031_")
 
@@ -85,7 +89,7 @@ def test_operations_separate_publication_deployment_migration_and_db_rollback():
         "Image rollback is **not** database rollback",
         "read-only root filesystem",
         "10001:10001",
-        "PortfolioHub v0.3.0 container contract",
+        "PortfolioHub v0.3.3 container contract",
         "v0.2.1 includes migration `0030`",
         "trex-web v0.2.3 displays `strategy_nick`",
     ):

tests/release/test_ruff_ratchet.py

@@ -0,0 +1,33 @@
+from collections import Counter
+
+from scripts.ruff_ratchet import normalized_findings
+
+
+def test_normalized_findings_ignore_line_number_changes(tmp_path):
+    source = tmp_path / "sample.py"
+    source.write_text("\nvalue = missing\n", encoding="utf-8")
+    diagnostics = [
+        {
+            "filename": str(source),
+            "code": "F821",
+            "message": "Undefined name `missing`",
+            "location": {"row": 2, "column": 9},
+        }
+    ]
+
+    assert normalized_findings(diagnostics, tmp_path) == Counter(
+        {("sample.py", "F821", "Undefined name `missing`", "value = missing"): 1}
+    )
+
+
+def test_normalized_findings_preserve_duplicate_count(tmp_path):
+    source = tmp_path / "sample.py"
+    source.write_text("missing\n", encoding="utf-8")
+    diagnostic = {
+        "filename": str(source),
+        "code": "F821",
+        "message": "Undefined name `missing`",
+        "location": {"row": 1, "column": 1},
+    }
+
+    assert normalized_findings([diagnostic, diagnostic], tmp_path).total() == 2

tests/test_corporate_action_atomic_recovery_postgres.py

@@ -0,0 +1,306 @@
+"""Real PostgreSQL regressions for atomic corporate-action writes and recovery."""
+from __future__ import annotations
+
+import os
+from datetime import date
+from decimal import Decimal
+from urllib.parse import unquote, urlparse
+from uuid import uuid4
+
+import psycopg
+import pytest
+from psycopg import sql
+
+from app.core.config import Settings
+from app.db.migrations import upgrade_to_head
+from app.domain.corporate_actions import canonical_payload_hash
+from app.domain.models import CorporateAction
+from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioRepository
+from app.schemas.corporate_actions import CorporateActionRequest
+
+_LOCAL_DSN_ENV = "PORTFOLIOHUB_TEST_POSTGRES_DSN"
+_LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"}
+_PORTFOLIO_ID = "3573f4d6-cc27-4f91-82d9-6330e5250d20"
+_CASH = Decimal("400060.62490000")
+_TOTAL_COST = Decimal("1577211.97990000")
+
+
+def _maintenance_dsn() -> str:
+    dsn = os.environ.get(_LOCAL_DSN_ENV)
+    if not dsn:
+        pytest.skip(f"{_LOCAL_DSN_ENV} is required for real PostgreSQL integration tests")
+    parsed = urlparse(dsn)
+    if parsed.scheme not in {"postgres", "postgresql"}:
+        pytest.fail("test DSN must use postgres/postgresql")
+    if parsed.hostname not in _LOOPBACK_HOSTS or parsed.path.lstrip("/") != "postgres":
+        pytest.fail("tests require a loopback PostgreSQL maintenance database named postgres")
+    return dsn
+
+
+def _settings(dsn: str, database: str) -> Settings:
+    parsed = urlparse(dsn)
+    return Settings(
+        db_enabled=True,
+        timescaledb_host=parsed.hostname,
+        timescaledb_port=parsed.port or 5432,
+        timescaledb_name=database,
+        timescaledb_user=unquote(parsed.username or ""),
+        timescaledb_password=unquote(parsed.password or ""),
+        timescaledb_sslmode="disable",
+        timescaledb_connect_timeout_sec=3,
+        timescaledb_application_name="corporate-action-atomic-test",
+    )
+
+
+def _repo(settings: Settings) -> PostgresPortfolioRepository:
+    return PostgresPortfolioRepository(
+        PortfolioDBConfig(
+            host=settings.active_db_host,
+            port=settings.active_db_port,
+            dbname=settings.active_db_name,
+            user=settings.active_db_user,
+            password=settings.active_db_password,
+            sslmode=settings.active_db_sslmode,
+            connect_timeout_sec=settings.active_db_connect_timeout_sec,
+            application_name=settings.active_db_application_name,
+        )
+    )
+
+
+@pytest.fixture()
+def repository() -> PostgresPortfolioRepository:
+    maintenance_dsn = _maintenance_dsn()
+    database = f"portfolio_hub_ca_{uuid4().hex}"
+    with psycopg.connect(maintenance_dsn, autocommit=True) as conn, conn.cursor() as cur:
+        cur.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(database)))
+    settings = _settings(maintenance_dsn, database)
+    upgrade_to_head(settings)
+    repo = _repo(settings)
+    try:
+        yield repo
+    finally:
+        repo.close()
+        with psycopg.connect(maintenance_dsn, autocommit=True) as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
+                "WHERE datname = %s AND pid <> pg_backend_pid()",
+                (database,),
+            )
+            cur.execute(sql.SQL("DROP DATABASE IF EXISTS {}").format(sql.Identifier(database)))
+
+
+def _seed_daekyo(repo: PostgresPortfolioRepository, *, quantity: Decimal) -> str:
+    user_id = "ca-recovery-user"
+    with repo._connect().cursor() as cur:
+        cur.execute("INSERT INTO users (id, username) VALUES (%s, %s)", (user_id, user_id))
+        cur.execute(
+            "INSERT INTO assets (symbol, markets) VALUES ('019685', ARRAY['KRX']) RETURNING asset_id"
+        )
+        asset_id = str(cur.fetchone()[0])
+        cur.execute(
+            """
+            INSERT INTO portfolios (
+                portfolio_id, name, owner_user_id, cash_balance, asset_quantity,
+                asset_price, profit_buffer, asset_id, broker, status
+            ) VALUES (%s, '대교우B', %s, %s, %s, 643, 0, %s, 'kis', 'enabled')
+            """,
+            (_PORTFOLIO_ID, user_id, _CASH, quantity, asset_id),
+        )
+        # The fee makes the authoritative BUY-ledger cost exactly 1,577,211.9799.
+        cur.execute(
+            """
+            INSERT INTO portfolio_ledger (
+                portfolio_id, recorded_at, side, source, quantity, price, fee_amount, tax_amount,
+                cash_balance_after, asset_quantity_after, total_value_after,
+                profit_buffer_after, strategy_set_id, note, executed_at
+            ) VALUES (%s, '2026-09-09T00:00:00Z', 'buy', 'broker', 1452, 1086, 339.9799, 0,
+                      %s, 1452, %s, 0, 'runtime', 'seed', '2026-09-09T00:00:00Z')
+            """,
+            (_PORTFOLIO_ID, _CASH, _CASH + Decimal(1452) * Decimal(643)),
+        )
+    return asset_id
+
+
+def _request(*, resume_partial: bool = False) -> CorporateActionRequest:
+    return CorporateActionRequest(
+        effective_date=date(2026, 9, 10),
+        quantity_before=Decimal(1452),
+        quantity_after=Decimal(726),
+        evidence_ref="KIS 019685 2026-09-10",
+        resume_partial=resume_partial,
+    )
+
+
+def _action(asset_id: str, request: CorporateActionRequest, key: str) -> CorporateAction:
+    digest = canonical_payload_hash(request)
+    return CorporateAction(
+        str(uuid4()),
+        _PORTFOLIO_ID,
+        asset_id,
+        request.action_type,
+        request.effective_date,
+        request.quantity_before,
+        request.quantity_after,
+        Decimal(0),
+        Decimal(0),
+        Decimal(0),
+        _CASH,
+        Decimal(0),
+        request.evidence_ref,
+        key,
+        digest,
+        0,
+        "ca-recovery-user",
+    )
+
+
+def _record(repo: PostgresPortfolioRepository, asset_id: str, request: CorporateActionRequest, key: str):
+    action = _action(asset_id, request, key)
+    return repo.record_corporate_action(
+        _PORTFOLIO_ID,
+        action=action,
+        idempotency_key=key,
+        payload_hash=action.payload_hash,
+        resume_partial=request.resume_partial,
+    )
+
+
+def _state(repo: PostgresPortfolioRepository) -> tuple:
+    with repo._connect().cursor() as cur:
+        cur.execute(
+            "SELECT asset_quantity, asset_price, cash_balance, profit_buffer "
+            "FROM portfolios WHERE portfolio_id=%s",
+            (_PORTFOLIO_ID,),
+        )
+        portfolio_row = cur.fetchone()
+        assert portfolio_row is not None
+        portfolio = tuple(Decimal(str(value)) for value in portfolio_row)
+        cur.execute(
+            "SELECT COUNT(*) FROM portfolio_ledger "
+            "WHERE portfolio_id=%s AND source='corporate_action'",
+            (_PORTFOLIO_ID,),
+        )
+        ledger_count = cur.fetchone()[0]
+        cur.execute(
+            "SELECT COUNT(*) FROM portfolio_corporate_actions WHERE portfolio_id=%s",
+            (_PORTFOLIO_ID,),
+        )
+        action_count = cur.fetchone()[0]
+    return (*portfolio, ledger_count, action_count)
+
+
+def test_stock_consolidation_persists_reference_price_total_value_and_audit(repository):
+    asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
+
+    saved = _record(repository, asset_id, _request(), "ca-normal")
+
+    assert saved.average_price_after == Decimal("2172.46829187")
+    assert saved.total_cost_basis == _TOTAL_COST
+    assert _state(repository) == (
+        Decimal("726.00000000"),
+        Decimal("1286.00000000"),
+        _CASH,
+        Decimal("0E-8"),
+        1,
+        1,
+    )
+    with repository._connect().cursor() as cur:
+        cur.execute(
+            "SELECT price, total_value_after FROM portfolio_ledger "
+            "WHERE portfolio_id=%s AND source='corporate_action'",
+            (_PORTFOLIO_ID,),
+        )
+        assert cur.fetchone() == (
+            Decimal("1286.00000000"),
+            Decimal("1333696.62490000"),
+        )
+
+
+def test_ledger_failure_rolls_back_portfolio_quantity_and_reference_price(repository):
+    asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
+    with repository._connect().cursor() as cur:
+        cur.execute(
+            """
+            CREATE FUNCTION reject_corporate_action_ledger() RETURNS trigger AS $$
+            BEGIN
+                IF NEW.source = 'corporate_action' THEN
+                    RAISE EXCEPTION 'injected ledger failure';
+                END IF;
+                RETURN NEW;
+            END;
+            $$ LANGUAGE plpgsql
+            """
+        )
+        cur.execute(
+            "CREATE TRIGGER reject_ca_ledger BEFORE INSERT ON portfolio_ledger "
+            "FOR EACH ROW EXECUTE FUNCTION reject_corporate_action_ledger()"
+        )
+
+    with pytest.raises(psycopg.Error, match="injected ledger failure"):
+        _record(repository, asset_id, _request(), "ca-failure")
+
+    assert _state(repository) == (
+        Decimal("1452.00000000"),
+        Decimal("643.00000000"),
+        _CASH,
+        Decimal("0E-8"),
+        0,
+        0,
+    )
+
+
+def test_audit_failure_rolls_back_portfolio_and_inserted_ledger(repository):
+    asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
+    before = _state(repository)
+    with repository._connect().cursor() as cur:
+        cur.execute(
+            """
+            CREATE FUNCTION reject_corporate_action_audit() RETURNS trigger AS $$
+            BEGIN
+                RAISE EXCEPTION 'injected corporate-action audit failure';
+            END;
+            $$ LANGUAGE plpgsql
+            """
+        )
+        cur.execute(
+            "CREATE TRIGGER reject_ca_audit BEFORE INSERT ON portfolio_corporate_actions "
+            "FOR EACH ROW EXECUTE FUNCTION reject_corporate_action_audit()"
+        )
+
+    with pytest.raises(psycopg.Error, match="injected corporate-action audit failure"):
+        _record(repository, asset_id, _request(), "ca-audit-failure")
+
+    assert _state(repository) == before
+
+
+def test_explicit_resume_repairs_only_verified_quantity_only_partial_state(repository):
+    asset_id = _seed_daekyo(repository, quantity=Decimal(726))
+    request = _request(resume_partial=True)
+
+    first = _record(repository, asset_id, request, "ca-recovery")
+    replay = _record(repository, asset_id, request, "ca-recovery")
+
+    assert first.idempotent_replay is False
+    assert replay.idempotent_replay is True
+    assert replay.ledger_entry_id == first.ledger_entry_id
+    assert _state(repository) == (
+        Decimal("726.00000000"),
+        Decimal("1286.00000000"),
+        _CASH,
+        Decimal("0E-8"),
+        1,
+        1,
+    )
+
+    with pytest.raises(ValueError, match="CORPORATE_ACTION_RECOVERY_ALREADY_RECORDED"):
+        _record(repository, asset_id, request, "ca-recovery-different-key")
+    assert _state(repository)[-2:] == (1, 1)
+
+
+def test_resume_rejects_a_normal_unapplied_snapshot(repository):
+    asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
+
+    with pytest.raises(ValueError, match="CORPORATE_ACTION_RECOVERY_STATE_MISMATCH"):
+        _record(repository, asset_id, _request(resume_partial=True), "ca-invalid-recovery")
+
+    assert _state(repository)[-2:] == (0, 0)

tests/test_corporate_actions.py

@@ -1,15 +1,31 @@
 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"))
+    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"))
+        calculate_stock_consolidation(Decimal(3), Decimal("1.5"), Decimal(10))
     assert exc.value.code == "FRACTIONAL_RESULT_UNSUPPORTED"
+
+
+def test_stock_consolidation_allows_stored_wac_rounding_residue():
+    result = calculate_stock_consolidation(
+        Decimal(1452),
+        Decimal(726),
+        Decimal("1086.234145936639256198347107438016528926"),
+    )
+
+    assert result.average_price_before == Decimal("1086.23414594")
+    assert result.average_price_after == Decimal("2172.46829187")
+    assert result.total_cost_basis == Decimal("1577211.97990000")
+    assert result.total_cost_basis - result.average_price_after * result.quantity_after == Decimal("0.00000238")

tests/test_issue34_regressions.py

@@ -1,4 +1,4 @@
-from datetime import date, datetime, timezone
+from datetime import UTC, date, datetime
 from decimal import Decimal
 from pathlib import Path
 
@@ -9,9 +9,9 @@ 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)},
+        {"side": "buy", "quantity": 10, "price": 100, "fee_amount": 0, "tax_amount": 0, "executed_at": datetime(2026, 1, 1, tzinfo=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=UTC)},
+        {"side": "sell", "quantity": 2, "price": 250, "fee_amount": 0, "tax_amount": 0, "executed_at": datetime(2026, 1, 3, tzinfo=UTC)},
     ]
     assert calculate_current_wac(rows).current_wac == 200
     result = calculate_realized_pnl(rows)
@@ -19,13 +19,48 @@ def test_buy_consolidation_sell_replays_wac_and_realized_pnl():
     assert result.total_realized_pnl == 100
 
 
+def test_consolidation_replay_uses_preserved_total_cost_basis():
+    rows = [
+        {"side": "buy", "quantity": 1452, "price": 1086.2341459366393, "fee_amount": 0, "executed_at": datetime(2026, 9, 9, tzinfo=UTC)},
+        {
+            "side": "stock_consolidation",
+            "quantity_before": 1452,
+            "quantity_after": 726,
+            "average_price_after": Decimal("2172.46829187"),
+            "total_cost_basis": Decimal("1577211.97990000"),
+            "executed_at": datetime(2026, 9, 10, tzinfo=UTC),
+        },
+    ]
+
+    result = calculate_current_wac(rows)
+    assert result.current_quantity == 726
+    assert result.current_wac == float(Decimal("1577211.97990000") / Decimal(726))
+
+
 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")
+    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_default_resume_flag_preserves_pre_recovery_idempotency_hash():
+    request = CorporateActionRequest(
+        effective_date=date(2026, 1, 2),
+        quantity_before=Decimal(10),
+        quantity_after=Decimal(5),
+        evidence_ref="ev",
+    )
+    legacy_payload = request.model_dump(mode="json", exclude={"resume_partial"})
+    explicit_default_payload = {**legacy_payload, "resume_partial": False}
+
+    assert canonical_payload_hash(request) == canonical_payload_hash(legacy_payload)
+    assert canonical_payload_hash(explicit_default_payload) == canonical_payload_hash(legacy_payload)
+    assert canonical_payload_hash(
+        request.model_copy(update={"resume_partial": True})
+    ) != canonical_payload_hash(legacy_payload)
+
+
 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

tests/test_issue34_runtime_blockers.py

@@ -1,4 +1,4 @@
-from datetime import date
+from datetime import UTC, date, datetime
 from decimal import Decimal
 from unittest.mock import MagicMock
 from uuid import UUID
@@ -20,6 +20,46 @@ class MissingPortfolioRepo:
         pass
 
 
+class ResumeCaptureRepo:
+    def __init__(self):
+        self.resume_partial = None
+
+    def get(self, portfolio_id):
+        return Portfolio(
+            portfolio_id=portfolio_id,
+            broker="kis",
+            status="enabled",
+            name="대교우B",
+            symbol="019685",
+            asset_id="asset-1",
+            owner_user_id="u1",
+            cash_balance=400060.6249,
+            asset_quantity=726,
+            price=643,
+            profit_buffer=0,
+        )
+
+    def record_corporate_action(
+        self,
+        portfolio_id,
+        *,
+        action,
+        idempotency_key,
+        payload_hash,
+        resume_partial=False,
+    ):
+        self.resume_partial = resume_partial
+        action.average_price_before = Decimal("1086.23414594")
+        action.average_price_after = Decimal("2172.46829187")
+        action.total_cost_basis = Decimal("1577211.97990000")
+        action.ledger_entry_id = 7
+        action.recorded_at = datetime(2026, 9, 13, tzinfo=UTC)
+        return action
+
+    def close(self):
+        pass
+
+
 def test_create_corporate_action_maps_postgres_missing_portfolio_to_404(monkeypatch):
     app = create_app()
     repo = MissingPortfolioRepo()
@@ -35,8 +75,33 @@ def test_create_corporate_action_maps_postgres_missing_portfolio_to_404(monkeypa
     assert response.json()["detail"] == "Portfolio not found"
 
 
+def test_corporate_action_api_passes_explicit_resume_mode(monkeypatch):
+    app = create_app()
+    repo = ResumeCaptureRepo()
+    monkeypatch.setattr(v1_portfolios, "_get_repo", lambda: repo)
+    app.dependency_overrides[v1_portfolios.require_trade_write] = lambda: type(
+        "Auth", (), {"user_id": "u1", "permission": "trade_write"}
+    )()
+
+    response = TestClient(app).post(
+        "/v1/portfolios/pf-1/corporate-actions",
+        headers={"Idempotency-Key": "resume-1"},
+        json={
+            "effective_date": "2026-09-10",
+            "quantity_before": "1452",
+            "quantity_after": "726",
+            "evidence_ref": "KIS 019685 2026-09-10",
+            "resume_partial": True,
+        },
+    )
+
+    assert response.status_code == 201
+    assert response.json()["quantity_after"] == "726"
+    assert repo.resume_partial is True
+
+
 def _action(asset_id="asset-1"):
-    return CorporateAction("ca-1", "pf-1", asset_id, "stock_consolidation", date(2026, 1, 1), Decimal("0"), Decimal("1"), Decimal("0"), Decimal("0"), Decimal("0"), Decimal("10"), Decimal("0"), "ev", "idem", "hash", 0)
+    return CorporateAction("ca-1", "pf-1", asset_id, "stock_consolidation", date(2026, 1, 1), Decimal(2), Decimal(1), Decimal(0), Decimal(0), Decimal(0), Decimal(10), Decimal(0), "ev", "idem", "hash", 0)
 
 
 def _repo_for_row(asset_id):
@@ -47,8 +112,13 @@ def _repo_for_row(asset_id):
     conn.__enter__.return_value = conn
     conn.cursor.return_value.__enter__.return_value = cur
     repo._connect = MagicMock(return_value=conn)
-    cur.fetchone.side_effect = [(Decimal("0"), Decimal("10"), Decimal("0"), asset_id, "enabled"), None, None, None, (7, date(2026, 1, 1))]
-    cur.fetchall.return_value = []
+    cur.fetchone.side_effect = [
+        (Decimal(2), Decimal(10), Decimal(0), asset_id, "enabled", Decimal(643)),
+        None,
+        None,
+        (7, date(2026, 1, 1)),
+    ]
+    cur.fetchall.return_value = [("buy", Decimal(2), Decimal(0), Decimal(0), Decimal(0), datetime(2025, 12, 31, tzinfo=UTC))]
     return repo
 
 
@@ -63,3 +133,59 @@ def test_record_corporate_action_rejects_different_asset_id():
     repo = _repo_for_row(UUID("12345678-1234-5678-1234-567812345678"))
     with pytest.raises(ValueError, match="CORPORATE_ACTION_ASSET_MISMATCH"):
         repo.record_corporate_action("pf-1", action=_action("87654321-4321-8765-4321-876543218765"), idempotency_key="idem", payload_hash="hash")
+
+
+def test_record_corporate_action_accepts_daekyo_rounding_residue():
+    repo = object.__new__(PostgresPortfolioRepository)
+    repo.ensure_schema = MagicMock()
+    conn = MagicMock()
+    cur = MagicMock()
+    conn.__enter__.return_value = conn
+    conn.cursor.return_value.__enter__.return_value = cur
+    repo._connect = MagicMock(return_value=conn)
+    cur.fetchone.side_effect = [
+        (Decimal(1452), Decimal(10), Decimal(0), "asset-1", "enabled", Decimal(643)),
+        None,
+        None,
+        (7, datetime(2026, 9, 13, tzinfo=UTC)),
+    ]
+    cur.fetchall.return_value = [
+        (
+            "buy",
+            Decimal(1452),
+            Decimal("1086.234145936639256198347107438016528926"),
+            Decimal(0),
+            Decimal(0),
+            datetime(2026, 9, 9, tzinfo=UTC),
+        )
+    ]
+    action = CorporateAction(
+        "ca-1",
+        "pf-1",
+        "asset-1",
+        "stock_consolidation",
+        date(2026, 9, 10),
+        Decimal(1452),
+        Decimal(726),
+        Decimal(0),
+        Decimal(0),
+        Decimal(0),
+        Decimal(10),
+        Decimal(0),
+        "KIS 019685 2026-09-10",
+        "idem-daekyo",
+        "hash-daekyo",
+        0,
+    )
+
+    saved = repo.record_corporate_action(
+        "pf-1",
+        action=action,
+        idempotency_key="idem-daekyo",
+        payload_hash="hash-daekyo",
+    )
+
+    assert saved.quantity_after == Decimal(726)
+    assert saved.average_price_after == Decimal("2172.46829187")
+    assert saved.total_cost_basis == Decimal("1577211.97990000")
+    assert saved.ledger_entry_id == 7

uv.lock

@@ -592,7 +592,7 @@ wheels = [
 
 [[package]]
 name = "portfoliohub"
-version = "0.3.0"
+version = "0.3.3"
 source = { editable = "." }
 dependencies = [
     { name = "alembic" },