diff --git a/app/domain/models.py b/app/domain/models.py index d0bd324..b563498 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 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 8dbf013..075936f 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,23 @@ class PostgresPortfolioRepository: ) cur.execute("UPDATE portfolios SET name = portfolio_id WHERE name IS NULL OR name = ''") cur.execute("ALTER TABLE portfolios ALTER COLUMN name SET NOT NULL") + cur.execute( + """ + CREATE TABLE IF NOT EXISTS portfolio_corporate_actions ( + corporate_action_id UUID PRIMARY KEY, portfolio_id TEXT NOT NULL, + asset_id UUID, action_type TEXT NOT NULL, effective_date DATE NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), actor_user_id TEXT, + evidence_ref TEXT NOT NULL, idempotency_key TEXT NOT NULL, + payload_hash TEXT NOT NULL, quantity_before NUMERIC(20,8) NOT NULL, + quantity_after NUMERIC(20,8) NOT NULL, average_price_before NUMERIC(20,8) NOT NULL, + average_price_after NUMERIC(20,8) NOT NULL, total_cost_basis NUMERIC(20,8) NOT NULL, + cash_balance NUMERIC(20,8) NOT NULL, profit_buffer NUMERIC(20,8) NOT NULL, + ledger_entry_id BIGINT NOT NULL, + UNIQUE(portfolio_id, idempotency_key) + ) + """ + ) + cur.execute("CREATE INDEX IF NOT EXISTS idx_corporate_actions_portfolio ON portfolio_corporate_actions (portfolio_id, effective_date)") cur.execute( """ CREATE TABLE IF NOT EXISTS portfolio_ledger ( @@ -1616,6 +1634,32 @@ class PostgresPortfolioRepository: ), ) + def record_corporate_action(self, portfolio_id: str, *, action: CorporateAction, idempotency_key: str, payload_hash: str) -> CorporateAction: + """Persist an immutable action and snapshot atomically.""" + self.ensure_schema() + with self._connect() as conn: + with conn.cursor() as cur: + cur.execute("SELECT asset_quantity, cash_balance, profit_buffer, asset_id FROM portfolios WHERE portfolio_id=%s FOR UPDATE", (portfolio_id,)) + row = cur.fetchone() + if row is None: + raise ValueError("portfolio not found") + cur.execute("SELECT corporate_action_id, payload_hash, ledger_entry_id, recorded_at FROM portfolio_corporate_actions WHERE portfolio_id=%s AND idempotency_key=%s", (portfolio_id, idempotency_key)) + prior = cur.fetchone() + if prior: + if prior[1] != payload_hash: + raise ValueError("IDEMPOTENCY_KEY_REUSE") + action.corporate_action_id, action.ledger_entry_id, action.recorded_at = str(prior[0]), int(prior[2]), prior[3] + action.idempotent_replay = True + return action + if Decimal(str(row[0])) != action.quantity_before: + raise ValueError("CORPORATE_ACTION_STALE_QUANTITY") + cur.execute("UPDATE portfolios SET asset_quantity=%s WHERE portfolio_id=%s", (float(action.quantity_after), portfolio_id)) + cur.execute("INSERT INTO portfolio_ledger (portfolio_id, recorded_at, side, source, quantity, price, cash_balance_after, asset_quantity_after, profit_buffer_after, executed_at, note) VALUES (%s,NOW(),'stock_consolidation','corporate_action',0,0,%s,%s,%s,NOW(),%s) RETURNING ledger_entry_id, recorded_at", (portfolio_id, float(action.cash_balance), float(action.quantity_after), float(action.profit_buffer), action.evidence_ref)) + lid, recorded = cur.fetchone() + action.ledger_entry_id, action.recorded_at = int(lid), recorded + cur.execute("INSERT INTO portfolio_corporate_actions (corporate_action_id,portfolio_id,asset_id,action_type,effective_date,actor_user_id,evidence_ref,idempotency_key,payload_hash,quantity_before,quantity_after,average_price_before,average_price_after,total_cost_basis,cash_balance,profit_buffer,ledger_entry_id) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", (action.corporate_action_id,portfolio_id,action.asset_id,action.action_type,action.effective_date,action.actor_user_id,action.evidence_ref,idempotency_key,payload_hash,action.quantity_before,action.quantity_after,action.average_price_before,action.average_price_after,action.total_cost_basis,action.cash_balance,action.profit_buffer,lid)) + return action + def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]: self.ensure_schema() with self._connect().cursor() as cur: