diff --git a/app/api/v1_portfolios.py b/app/api/v1_portfolios.py index 0262dc2..e59318e 100644 --- a/app/api/v1_portfolios.py +++ b/app/api/v1_portfolios.py @@ -18,12 +18,13 @@ import uuid from datetime import date, datetime, timedelta, timezone from decimal import Decimal -from fastapi import APIRouter, Depends, HTTPException, Query, Response, status +from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response, status from app.core.auth import AuthContext, Permission, require_read, require_trade_write from app.core.config import get_settings from app.domain.errors import DaemonInactiveError, DaemonNotFoundError -from app.domain.models import Portfolio, PortfolioComment +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.repositories.factory import get_portfolio_repository_from_settings from app.schemas.v1 import ( @@ -31,6 +32,7 @@ from app.schemas.v1 import ( DividendManualLedgerResponse, HoldingsCostBasisResponse, LedgerHistoryItem, + CorporateActionLedgerHistoryItem, LedgerHistoryResponse, LedgerTradeRequest, LedgerTradeResponse, @@ -55,6 +57,7 @@ from app.schemas.v1 import ( TradeHistoryItem, TradeHistoryResponse, ) +from app.schemas.corporate_actions import CorporateActionRequest, CorporateActionResponse from app.services.strategy_set_service import ( StrategyInactiveError, StrategyNotFoundError, @@ -809,7 +812,19 @@ def _decode_ledger_cursor(cursor: str, *, portfolio_id: str, source: str | None) raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="invalid ledger cursor") from None -def _ledger_entry_to_history_item(entry) -> LedgerHistoryItem | DividendLedgerHistoryItem: +def _ledger_entry_to_history_item(entry) -> LedgerHistoryItem | DividendLedgerHistoryItem | CorporateActionLedgerHistoryItem: + if getattr(entry, "corporate_action", None): + event = entry.corporate_action + return CorporateActionLedgerHistoryItem( + ledger_entry_id=entry.ledger_entry_id, portfolio_id=entry.portfolio_id, + source="corporate_action", event_id=str(event["corporate_action_id"]), + event_type=event["action_type"], effective_date=event["effective_date"], + quantity_before=event["quantity_before"], quantity_after=event["quantity_after"], + average_price_before=event["average_price_before"], average_price_after=event["average_price_after"], + total_cost_basis=event["total_cost_basis"], cash_balance=event["cash_balance"], + profit_buffer=event["profit_buffer"], evidence_ref=event["evidence_ref"], + actor_user_id=event.get("actor_user_id"), recorded_at=entry.recorded_at, + ) action = (entry.action or entry.side or "").upper() if action == "DIVIDEND": amount_decimal = entry.price_decimal if entry.price_decimal is not None else Decimal(str(entry.price)) @@ -1295,3 +1310,31 @@ def get_holdings_cost_basis( cost_basis_from=wac_result.cost_basis_from, data_completeness_warning=wac_result.data_completeness_warning, ) + + +@router.post("/{portfolio_id}/corporate-actions", response_model=CorporateActionResponse, status_code=status.HTTP_201_CREATED) +def create_corporate_action(portfolio_id: str, payload: CorporateActionRequest, response: Response, auth: AuthContext = Depends(require_trade_write), idempotency_key: str = Header(..., alias="Idempotency-Key")): + if not idempotency_key or not idempotency_key.strip(): + raise HTTPException(status_code=422, detail="Idempotency-Key must not be empty") + repo = _get_repo() + try: + portfolio = repo.get(portfolio_id) + if portfolio is None: + raise HTTPException(status_code=404, detail="Portfolio not found") + if portfolio.owner_user_id != auth.user_id and auth.permission != Permission.ADMIN: + 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 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) + return CorporateActionResponse(**saved.__dict__) + except ValueError as exc: + code = str(exc) + if code == PORTFOLIO_NOT_FOUND_MESSAGE: + raise HTTPException(status_code=404, detail="Portfolio not found") from exc + raise HTTPException(status_code=409 if code.startswith(("IDEMPOTENCY", "CORPORATE_ACTION_")) else 400, detail=code) from exc + finally: + _close_repo(repo) diff --git a/app/cli/commands/portfolio.py b/app/cli/commands/portfolio.py index 83e157a..0fd227c 100644 --- a/app/cli/commands/portfolio.py +++ b/app/cli/commands/portfolio.py @@ -274,6 +274,51 @@ def sell_portfolio( render_output(ctx.obj.console, ctx.obj.output_format, detail) +@portfolio_app.command("corporate-action") +def corporate_action_portfolio( + ctx: typer.Context, + name: str = typer.Argument(...), + user: str = typer.Option(..., "--user"), + action_type: str = typer.Option("stock_consolidation", "--type"), + effective_date: str = typer.Option(..., "--effective-date"), + quantity_before: float = typer.Option(..., "--quantity-before", min=0.0000001), + quantity_after: float = typer.Option(..., "--quantity-after", min=0.0000001), + evidence_ref: str = typer.Option(..., "--evidence-ref"), + idempotency_key: str = typer.Option(..., "--idempotency-key"), + force: bool = typer.Option(False, "--force"), +) -> None: + """Record an auditable stock consolidation without a trade.""" + 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.models import CorporateAction + if action_type != "stock_consolidation": + ctx.obj.console.print("Error: INVALID_CORPORATE_ACTION_TYPE") + raise typer.Exit(code=1) + try: + day = date.fromisoformat(effective_date) + except ValueError: + ctx.obj.console.print("Error: INVALID_EFFECTIVE_DATE") + raise typer.Exit(code=1) + if quantity_after >= quantity_before: + ctx.obj.console.print("Error: INVALID_CORPORATE_ACTION_RATIO") + raise typer.Exit(code=1) + _confirm(force, f"Record stock consolidation {quantity_before:g} → {quantity_after:g}; cash/profit buffer unchanged?") + def _callback(service, repository, app_context): + portfolio_id = _resolve_portfolio_id(repository, name, user) + 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} + digest = canonical_payload_hash(payload) + 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) + detail = _with_service_and_repo(ctx, _callback) + render_output(ctx.obj.console, ctx.obj.output_format, detail) + + @portfolio_app.command("invest") def invest_portfolio( ctx: typer.Context, diff --git a/app/db/alembic/versions/20260913_0031_corporate_actions.py b/app/db/alembic/versions/20260913_0031_corporate_actions.py new file mode 100644 index 0000000..538d3ba --- /dev/null +++ b/app/db/alembic/versions/20260913_0031_corporate_actions.py @@ -0,0 +1,31 @@ +"""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) + )""")) + op.get_bind().execute(text( + "CREATE INDEX IF NOT EXISTS idx_corporate_actions_portfolio " + "ON portfolio_corporate_actions (portfolio_id, effective_date)" + )) + +def downgrade() -> None: + op.get_bind().execute(text("DROP TABLE IF EXISTS portfolio_corporate_actions")) diff --git a/app/domain/corporate_actions.py b/app/domain/corporate_actions.py new file mode 100644 index 0000000..82b4302 --- /dev/null +++ b/app/domain/corporate_actions.py @@ -0,0 +1,45 @@ +"""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 + +def canonical_payload_hash(payload: object) -> str: + if hasattr(payload, "model_dump"): + payload = payload.model_dump(mode="json", exclude_none=True) + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + return hashlib.sha256(encoded).hexdigest() + +class CorporateActionError(ValueError): + code = "INVALID_CORPORATE_ACTION_RATIO" + 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 diff --git a/app/domain/models.py b/app/domain/models.py index d0bd324..31f653e 100644 --- a/app/domain/models.py +++ b/app/domain/models.py @@ -3,7 +3,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from datetime import datetime +from datetime import date, datetime from decimal import Decimal from enum import Enum from typing import Any, Protocol, runtime_checkable @@ -132,6 +132,29 @@ class Portfolio: managed_by_daemon_id: str | None = None +@dataclass +class CorporateAction: + corporate_action_id: str + portfolio_id: str + asset_id: str | None + action_type: str + 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 + idempotency_key: str + payload_hash: str + ledger_entry_id: int + actor_user_id: str | None = None + recorded_at: datetime | None = None + idempotent_replay: bool = False + + @dataclass class LedgerEntry: portfolio_id: str @@ -158,6 +181,7 @@ class LedgerEntry: target_ledger_entry_id: int | None = None confirmation_evidence: str | None = None actor_user_id: str | None = None + corporate_action: dict[str, Any] | None = None def __post_init__(self) -> None: if self.side is None and self.action: diff --git a/app/domain/pnl.py b/app/domain/pnl.py index 2a0663b..85d65a4 100644 --- a/app/domain/pnl.py +++ b/app/domain/pnl.py @@ -170,6 +170,17 @@ def calculate_realized_pnl( if running_qty < 0: running_qty = 0.0 + elif side == "stock_consolidation": + before = float(row.get("quantity_before", running_qty)) + after = float(row.get("quantity_after", row.get("asset_quantity_after", 0.0))) + if before <= 0 or after <= 0 or running_qty <= 0: + warning_triggered = True + continue + 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)) + warning: str | None = None if warning_triggered: warning = ( @@ -252,6 +263,17 @@ def calculate_current_wac(rows: list[dict[str, Any]]) -> CurrentWacResult: if running_qty < 0: running_qty = 0.0 + elif side == "stock_consolidation": + before = float(row.get("quantity_before", running_qty)) + after = float(row.get("quantity_after", row.get("asset_quantity_after", 0.0))) + if before <= 0 or after <= 0 or running_qty <= 0: + warning_triggered = True + continue + 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)) + warning: str | None = None if warning_triggered: warning = ( diff --git a/app/domain/services.py b/app/domain/services.py index e6d6055..6211433 100644 --- a/app/domain/services.py +++ b/app/domain/services.py @@ -7,7 +7,7 @@ from typing import Callable, Protocol from app.domain.models import LedgerEntry, LiveAssetSource, Portfolio, StrategySet -_NOT_FOUND_MSG = "포트폴리오를 찾을 수 없습니다" +PORTFOLIO_NOT_FOUND_MESSAGE = "포트폴리오를 찾을 수 없습니다" _INVEST_ACTION_MAP = { "deposit": "INVEST", "profit": "PROFIT", @@ -250,7 +250,7 @@ def apply_reset( portfolio = repo.reset(portfolio_id) return portfolio, False except ValueError as exc: - if str(exc) != _NOT_FOUND_MSG: + if str(exc) != PORTFOLIO_NOT_FOUND_MESSAGE: raise symbol = infer_symbol(portfolio_id) portfolio = repo.get_or_create(portfolio_id, symbol, initial_cash=0.0) diff --git a/app/repositories/portfolio_repository.py b/app/repositories/portfolio_repository.py index e550cec..496e475 100644 --- a/app/repositories/portfolio_repository.py +++ b/app/repositories/portfolio_repository.py @@ -6,7 +6,7 @@ from datetime import datetime from decimal import Decimal from typing import Protocol -from app.domain.models import Asset, AuthToken, BrokerCredential, Daemon, LedgerEntry, Portfolio, PortfolioComment, PortfolioTag, StrategySet, Tag, TokenPrincipal, User +from app.domain.models import Asset, AuthToken, BrokerCredential, CorporateAction, Daemon, LedgerEntry, Portfolio, PortfolioComment, PortfolioTag, StrategySet, Tag, TokenPrincipal, User class PortfolioRepository(Protocol): @@ -284,6 +284,8 @@ class PortfolioRepository(Protocol): note: str = "", ) -> None: ... + def record_corporate_action(self, portfolio_id: str, *, action: CorporateAction, idempotency_key: str, payload_hash: str) -> CorporateAction: ... + def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]: ... def list_trades_by_date( diff --git a/app/repositories/postgres.py b/app/repositories/postgres.py index df1b726..c2af3ba 100644 --- a/app/repositories/postgres.py +++ b/app/repositories/postgres.py @@ -7,7 +7,7 @@ import json import logging import uuid from dataclasses import dataclass -from datetime import UTC, datetime +from datetime import UTC, date, datetime from decimal import ROUND_HALF_UP, Decimal, InvalidOperation from typing import Any @@ -22,6 +22,7 @@ from app.domain.errors import ( StrategyNotFoundError, ) from app.domain.models import ( + CorporateAction, Asset, AuthToken, BrokerCredential, @@ -332,6 +333,32 @@ class PostgresPortfolioRepository: ) cur.execute("UPDATE portfolios SET name = portfolio_id WHERE name IS NULL OR name = ''") cur.execute("ALTER TABLE portfolios ALTER COLUMN name SET NOT NULL") + cur.execute( + """ + CREATE TABLE IF NOT EXISTS portfolio_corporate_actions ( + corporate_action_id UUID PRIMARY KEY, + portfolio_id TEXT NOT NULL 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) + ) + """ + ) + cur.execute("CREATE INDEX IF NOT EXISTS idx_corporate_actions_portfolio ON portfolio_corporate_actions (portfolio_id, effective_date)") cur.execute( """ CREATE TABLE IF NOT EXISTS portfolio_ledger ( @@ -418,6 +445,7 @@ class PostgresPortfolioRepository: ON portfolios (broker, status, updated_at DESC) """ ) + cur.execute( """ CREATE TABLE IF NOT EXISTS portfolio_comments ( @@ -1095,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 @@ -1598,6 +1628,75 @@ class PostgresPortfolioRepository: ), ) + def record_corporate_action(self, portfolio_id: str, *, action: CorporateAction, idempotency_key: str, payload_hash: str) -> CorporateAction: + """Persist an immutable action and snapshot atomically.""" + self.ensure_schema() + with self._connect() as conn: + with conn.cursor() as cur: + cur.execute("SELECT asset_quantity, cash_balance, profit_buffer, asset_id, 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)) + return action + def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]: self.ensure_schema() with self._connect().cursor() as cur: @@ -1612,7 +1711,7 @@ class PostgresPortfolioRepository: total_value_after::float8, profit_buffer_after::float8, strategy_set_id, note FROM portfolio_ledger - WHERE portfolio_id = %s + WHERE portfolio_id = %s AND side IN ('buy', 'sell') ORDER BY recorded_at DESC, ledger_entry_id DESC LIMIT %s """, @@ -1636,7 +1735,7 @@ class PostgresPortfolioRepository: boundaries (start of day / start of next day). """ self.ensure_schema() - clauses = ["portfolio_id = %s"] + clauses = ["portfolio_id = %s", "side IN ('buy','sell')"] params: list[object] = [portfolio_id] if date is not None: @@ -1706,16 +1805,35 @@ class PostgresPortfolioRepository: executed_at, recorded_at, cash_balance_after::float8, asset_quantity_after::float8, total_value_after::float8, profit_buffer_after::float8, - strategy_set_id, note - FROM portfolio_ledger - WHERE portfolio_id = %s - AND side IN ('buy', 'sell') - ORDER BY executed_at ASC, ledger_entry_id ASC + strategy_set_id, note, source, + ca.corporate_action_id, ca.action_type, ca.effective_date, + 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 + ORDER BY ledger.executed_at ASC, ledger.ledger_entry_id ASC """, (portfolio_id,), ) rows = cur.fetchall() - return [self._row_to_ledger_entry(row) for row in rows] + entries = [] + for row in rows: + entry = self._row_to_ledger_entry(row) + if len(row) > 18 and row[18] is not None: + 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 def insert_ledger_trade( self, @@ -2369,15 +2487,26 @@ class PostgresPortfolioRepository: SELECT ledger_entry_id, portfolio_id, order_id, execution_id, side, quantity, price, fee_amount, tax_amount, executed_at, recorded_at, cash_balance_after, asset_quantity_after, total_value_after, profit_buffer_after, strategy_set_id, - note, source, target_ledger_entry_id, confirmation_evidence, actor_user_id + note, source, target_ledger_entry_id, confirmation_evidence, actor_user_id, + ca.corporate_action_id, ca.action_type, ca.effective_date, + 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 FROM portfolio_ledger + LEFT JOIN portfolio_corporate_actions ca ON ca.ledger_entry_id = portfolio_ledger.ledger_entry_id WHERE """ + " AND ".join(clauses) + """ ORDER BY recorded_at DESC, ledger_entry_id DESC LIMIT %s """, params, ) - return [self._row_to_ledger_entry(row) for row in cur.fetchall()] + result = [] + for row in cur.fetchall(): + entry = self._row_to_ledger_entry(row) + if row[21] is not None: + entry.corporate_action = {"corporate_action_id": str(row[21]), "action_type": row[22], "effective_date": row[23], "quantity_before": Decimal(str(row[24])), "quantity_after": Decimal(str(row[25])), "average_price_before": Decimal(str(row[26])), "average_price_after": Decimal(str(row[27])), "total_cost_basis": Decimal(str(row[28])), "cash_balance": Decimal(str(row[29])), "profit_buffer": Decimal(str(row[30])), "evidence_ref": row[31], "actor_user_id": row[32]} + result.append(entry) + return result def get_return_summaries(self, portfolio_ids: list[str]) -> dict[str, dict[str, float]]: if not portfolio_ids: diff --git a/app/schemas/corporate_actions.py b/app/schemas/corporate_actions.py new file mode 100644 index 0000000..ff48237 --- /dev/null +++ b/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 diff --git a/app/schemas/v1.py b/app/schemas/v1.py index d0796a5..278a9bc 100644 --- a/app/schemas/v1.py +++ b/app/schemas/v1.py @@ -313,8 +313,28 @@ class DividendLedgerHistoryItem(BaseModel): confirmation_evidence: str | None = None +class CorporateActionLedgerHistoryItem(BaseModel): + """Auditable corporate-action event in the unified ledger read model.""" + ledger_entry_id: int + portfolio_id: str + source: Literal["corporate_action"] + event_id: str + event_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 + + class LedgerHistoryResponse(BaseModel): - items: list[LedgerHistoryItem | DividendLedgerHistoryItem] + items: list[LedgerHistoryItem | DividendLedgerHistoryItem | CorporateActionLedgerHistoryItem] next_cursor: str | None = None diff --git a/docs/container-operations.md b/docs/container-operations.md index 1750e5c..179be43 100644 --- a/docs/container-operations.md +++ b/docs/container-operations.md @@ -82,30 +82,32 @@ The existing database variable names remain `TIMESCALEDB_*`; v0.2.1 deliberately Do not route traffic from a static `/health` success alone. Readiness checks never apply, stamp, or repair migrations. -## One-shot migration and 0029 -> 0030 policy +## One-shot migration and 0029 -> 0031 policy -The installed image contains exactly **30** migration files and one Alembic head, `20260820_0030`. The application default command does not invoke Alembic. Migration is a separate pre-start operation using the exact same digest and database environment: +The installed image contains exactly **31** migration files and one Alembic head, `20260913_0031`. The application default command does not invoke Alembic. Migration is a separate pre-start operation using the exact same digest and database environment: 1. back up the database and record the current immutable image digest and `alembic_version`; 2. require the normal upgrade origin `20260814_0029`; for an older revision, review and apply the full ordered chain first; 3. stop writes or otherwise establish an operator-approved maintenance boundary; 4. execute `pfhub migrate-db` once with restart disabled; 5. stop rollout on any nonzero result and preserve sanitized logs; -6. run `pfhub --format json health-check` and require both `database_revision` and `expected_revision` to equal `20260820_0030`; +6. run `pfhub --format json health-check` and require both `database_revision` and `expected_revision` to equal `20260913_0031`; 7. start application replicas and verify `/health`, `/openapi.json`, authenticated caller paths, and strategy data; -8. a repeated `pfhub migrate-db` is expected to be idempotent and remain at `0030`, but it is verification—not an application-start hook. +8. a repeated `pfhub migrate-db` is expected to be idempotent and remain at `0031`, but it is verification—not an application-start hook. Migration `0029` is additive/idempotent around the strategy catalog: it preserves existing strategy IDs and portfolio references, reuses the canonical `target_value`/`{}` strategy, and backfills catalog data. Before migration, inspect duplicate or invalid strategy references and take a backup. After migration, verify strategy listing and portfolio active-strategy fields before resuming writes. v0.2.1 includes migration `0030`, an idempotent index on `(portfolio_id, recorded_at DESC, ledger_entry_id DESC)` for unified-ledger keyset reads. Operators at `0029` must run the same-image one-shot migration before rollout; it advances to `20260820_0030` without mutating ledger rows. -A pre-Alembic legacy schema is detected only when all three tables `strategy_sets`, `portfolios`, and `portfolio_ledger` exist; the CLI stamps `0027` and then runs real upgrades through `0030`. This compatibility path must be used only after backup and schema review. A partially matching schema is not auto-stamped and must fail for operator investigation. +Migration `0031` is additive corporate-action storage. Operators must complete the same-image migration through `20260913_0031` before using the corporate-action endpoint; the endpoint is not available for traffic until readiness reports that head. + +A pre-Alembic legacy schema is detected only when all three tables `strategy_sets`, `portfolios`, and `portfolio_ledger` exist; the CLI stamps `0027` and then runs real upgrades through `0031`. This compatibility path must be used only after backup and schema review. A partially matching schema is not auto-stamped and must fail for operator investigation. Image rollback is **not** database rollback. Selecting older application bytes does not reverse schema/data changes, tokens, or side effects. Database restore, downgrade, or forward-fix requires separate review and authorization; no automatic downgrade is provided. ## API verification and known caller blockers -v0.2.1 preserves Bearer authentication and adds all-source unified-ledger reads. 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 30-file migration inventory, isolated PostgreSQL migration twice, readiness at `0030`, persistent-volume restart, non-root/read-only operation, secret canary absence, and graceful SIGTERM. Publication consumes the tested archive without rebuilding it. +v0.2.1 preserves Bearer authentication and adds all-source unified-ledger reads. 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 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. diff --git a/docs/data-model.md b/docs/data-model.md index 18b04c1..7134eec 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -372,7 +372,7 @@ - Alembic revisions in migrations/versions/ are the ordered schema history for persistent environments. - app/repositories/postgres.py includes ensure_schema() for bootstrap safety, but long-term schema evolution should run through Alembic. -- Current Alembic head is `20260820_0030`; the preceding `0029` strategy columns, default seed, and constraints and the `0030` unified-ledger keyset index are guarded for ensure_schema-first and repeatable execution. +- Current Alembic head is `20260913_0031`; migration `0031` additively stores corporate actions for the corporate-action endpoint. Operators must migrate an environment to `0031` before using that endpoint. The preceding `0029` strategy columns, `0030` unified-ledger keyset index, and `0031` corporate-action storage are guarded for ensure_schema-first and repeatable execution. - Comments, tags, auth tokens, and ledger rows are normalized into dedicated tables keyed to users or portfolios. ## Current Gaps / Planned Changes diff --git a/docs/interface-contract.md b/docs/interface-contract.md index 4819b82..346b2dc 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -437,7 +437,10 @@ - Auth: same owner-scoped user/session token policy as manual write. - Query: `limit` 1..200, default 50. -- Response: `ManualLedgerHistoryResponse {"items": [ManualLedgerResponse], "total": }` ordered newest first. +- `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. +- `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. + ### Endpoint: `GET /v1/portfolios/{portfolio_id}/returns/summary` (REQ-056, BAT-024) diff --git a/docs/project-structure.md b/docs/project-structure.md index 3acf709..52f083c 100644 --- a/docs/project-structure.md +++ b/docs/project-structure.md @@ -138,13 +138,13 @@ |-----------|-----------|------| | ensure_schema() 전용 | `idx_auth_tokens_service_name_unique` (partial index, BAT-006/REQ-017: `AND revoked_at IS NULL` 조건 추가) | Alembic revision 20260520_0022에도 CONCURRENTLY 버전 존재 | | Alembic 전용 | `portfolios.market`, `auth_tokens.runtime_id`, `idx_auth_tokens_runtime_id_unique`, `broker_credentials`, `ledger_trades` | ensure_schema()에 없음 — v1 API 전제 조건 | -| 양쪽 공유 (IF NOT EXISTS 보호) | users, auth_tokens, strategy_sets (`status`, `revision`, `updated_at` 포함), portfolios, portfolio_ledger, portfolio_comments, portfolio_tags + 대부분의 인덱스 | ensure_schema() 우선 실행 후 Alembic `20260820_0030`까지 실행 시 충돌 없음; 기본 전략 seed/constraints와 unified-ledger keyset index도 존재 여부 guard 적용 | +| 양쪽 공유 (IF NOT EXISTS 보호) | users, auth_tokens, strategy_sets (`status`, `revision`, `updated_at` 포함), portfolios, portfolio_ledger, portfolio_comments, portfolio_tags + 대부분의 인덱스 | ensure_schema() 우선 실행 후 Alembic `20260913_0031`까지 실행 시 충돌 없음; 기본 전략 seed/constraints, unified-ledger keyset index, additive corporate-action storage도 존재 여부 guard 적용 | ### 허용 실행 순서 1. `init-db` 선행 → App 시작 → 정상 운영 (권장) 2. App 시작 (ensure_schema() 선행) → `migrate-db` 후속 실행 (legacy stamp 경로) -3. 이미 Alembic 관리 DB → `migrate-db` 증분 upgrade +3. 이미 Alembic 관리 DB → `migrate-db` 증분 upgrade through `20260913_0031` before using the corporate-action endpoint ### 운영 제약 diff --git a/scripts/image_release.py b/scripts/image_release.py index 497a82d..56729a4 100755 --- a/scripts/image_release.py +++ b/scripts/image_release.py @@ -25,8 +25,8 @@ COMMIT_RE = re.compile(r"^[0-9a-f]{40}$") VERSION_REF_RE = re.compile(r"^refs/tags/v(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)$") IMAGE_ID_RE = re.compile(r"^sha256:[0-9a-f]{64}$") DIGEST_RE = re.compile(r"(?:^|\s)digest:\s*(sha256:[0-9a-f]{64})(?:\s|$)") -EXPECTED_MIGRATION_COUNT = 30 -EXPECTED_ALEMBIC_HEAD = "20260820_0030" +EXPECTED_MIGRATION_COUNT = 31 +EXPECTED_ALEMBIC_HEAD = "20260913_0031" @dataclass(frozen=True) class Result: diff --git a/tests/release/test_image_release.py b/tests/release/test_image_release.py index aa8625a..f07cfc0 100644 --- a/tests/release/test_image_release.py +++ b/tests/release/test_image_release.py @@ -33,8 +33,8 @@ class RecordingRunner: if command[:3] == ["docker", "run", "--rm"] and "urllib.request" in command[-1]: enabled = "TARGET:8000" not in command[-1] and "plain" not in command[-1] return image_release.Result('{"status":"ok","db_enabled":' + str(enabled).lower() + '}') - if command[:3] == ["docker", "run", "--rm"] and "alembic_version" in command[-1]: return image_release.Result("20260820_0030\n") - if command[:3] == ["docker", "run", "--rm"] and "health-check" in command: return image_release.Result('{"database_revision":"20260820_0030"}\n') + if command[:3] == ["docker", "run", "--rm"] and "alembic_version" in command[-1]: return image_release.Result("20260913_0031\n") + if command[:3] == ["docker", "run", "--rm"] and "health-check" in command: return image_release.Result('{"database_revision":"20260913_0031"}\n') return image_release.Result() def artifacts(tmp_path: Path): @@ -74,12 +74,12 @@ def test_project_version_must_match_release_tag(tmp_path): def test_installed_smoke_has_import_cli_exact_inventory_and_cleanup(): inventory = tuple( [f"202601{i:02d}_{i:04d}_migration.py" for i in range(1, 30)] - + ["20260820_0030_portfolio_ledger_keyset_index.py"] + + ["20260820_0030_portfolio_ledger_keyset_index.py", "20260913_0031_corporate_actions.py"] ) runner = RecordingRunner(); image_release.smoke_installed_image("candidate", inventory, runner, 17) runs = [x[0] for x in runner.commands if x[0][:2] == ["docker","run"]] assert len(runs) == 3 and all("--read-only" in x and "none" in x for x in runs) - assert "migration_inventory" in runs[-1][-1] and "20260820_0030" in runs[-1][-1] + assert "migration_inventory" in runs[-1][-1] and "20260913_0031" in runs[-1][-1] assert len([x for x in runner.commands if x[0][:3] == ["docker","rm","--force"]]) == 3 def _assert_exact_startup_cleanup(commands: list[list[str]]) -> None: diff --git a/tests/release/test_release_contracts.py b/tests/release/test_release_contracts.py index e062a9b..f3e7149 100644 --- a/tests/release/test_release_contracts.py +++ b/tests/release/test_release_contracts.py @@ -50,7 +50,7 @@ def test_workflow_actions_are_commit_pinned_and_verify_version_is_project_derive assert 'RELEASE_VERSION=v0.1.3' not in verify -def test_release_version_and_migration_contract_are_v021_head_0030(): +def test_release_version_and_migration_contract_has_current_head_0031(): project = (REPO / "pyproject.toml").read_text() config = (REPO / "app/core/config.py").read_text() migrations = sorted( @@ -60,8 +60,8 @@ def test_release_version_and_migration_contract_are_v021_head_0030(): ) assert 'version = "0.2.1"' in project assert 'app_version: str = "0.2.1"' in config - assert len(migrations) == 30 - assert migrations[-1].startswith("20260820_0030") + assert len(migrations) == 31 + assert migrations[-1].startswith("20260913_0031_") def test_dockerignore_is_deny_by_default_required_inputs_only(): lines=(REPO/".dockerignore").read_text().splitlines(); assert lines[0]=="**" diff --git a/tests/test_corporate_actions.py b/tests/test_corporate_actions.py new file mode 100644 index 0000000..e866151 --- /dev/null +++ b/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" diff --git a/tests/test_issue34_regressions.py b/tests/test_issue34_regressions.py new file mode 100644 index 0000000..2253187 --- /dev/null +++ b/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 diff --git a/tests/test_issue34_runtime_blockers.py b/tests/test_issue34_runtime_blockers.py new file mode 100644 index 0000000..16987ca --- /dev/null +++ b/tests/test_issue34_runtime_blockers.py @@ -0,0 +1,65 @@ +from datetime import date +from decimal import Decimal +from unittest.mock import MagicMock +from uuid import UUID + +import pytest +from fastapi.testclient import TestClient + +from app.api import v1_portfolios +from app.domain.models import CorporateAction, Portfolio +from app.main import create_app +from app.repositories.postgres import PostgresPortfolioRepository + + +class MissingPortfolioRepo: + def get(self, portfolio_id): + raise ValueError("포트폴리오를 찾을 수 없습니다") + + def close(self): + pass + + +def test_create_corporate_action_maps_postgres_missing_portfolio_to_404(monkeypatch): + app = create_app() + repo = MissingPortfolioRepo() + 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"})() + client = TestClient(app) + response = client.post( + "/v1/portfolios/missing/corporate-actions", + headers={"Idempotency-Key": "idem-1"}, + json={"effective_date": "2026-01-01", "quantity_before": "10", "quantity_after": "5", "evidence_ref": "ev"}, + ) + assert response.status_code == 404 + assert response.json()["detail"] == "Portfolio not found" + + +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) + + +def _repo_for_row(asset_id): + 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("0"), Decimal("10"), Decimal("0"), asset_id, "enabled"), None, None, None, (7, date(2026, 1, 1))] + cur.fetchall.return_value = [] + return repo + + +def test_record_corporate_action_accepts_uuid_row_against_string_domain_id(): + asset = UUID("12345678-1234-5678-1234-567812345678") + action = _action(str(asset)) + saved = _repo_for_row(asset).record_corporate_action("pf-1", action=action, idempotency_key="idem", payload_hash="hash") + assert saved.ledger_entry_id == 7 + + +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")