diff --git a/app/repositories/portfolio_repository.py b/app/repositories/portfolio_repository.py index e49eba9..e4d8e63 100644 --- a/app/repositories/portfolio_repository.py +++ b/app/repositories/portfolio_repository.py @@ -5,7 +5,7 @@ from __future__ import annotations from datetime import datetime from typing import Protocol -from app.domain.models import AuthToken, BrokerCredential, Daemon, LedgerEntry, Portfolio, PortfolioComment, PortfolioTag, StrategySet, TokenPrincipal, User +from app.domain.models import AuthToken, BrokerCredential, Daemon, LedgerEntry, LedgerTrade, Portfolio, PortfolioComment, PortfolioTag, StrategySet, TokenPrincipal, User class PortfolioRepository(Protocol): @@ -198,6 +198,20 @@ class PortfolioRepository(Protocol): def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]: ... + def insert_ledger_trade( + self, + *, + asset_id: str, + order_id: str, + execution_id: str, + side: str, + quantity: float, + price: float, + executed_at: datetime, + fee_amount: float = 0.0, + tax_amount: float = 0.0, + ) -> tuple[LedgerTrade, bool]: ... + def reset(self, portfolio_id: str) -> Portfolio: ... def adjust_buffer(self, portfolio_id: str, amount: float, action: str, note: str = "") -> Portfolio: ... diff --git a/app/repositories/postgres.py b/app/repositories/postgres.py index eae614f..1429b1a 100644 --- a/app/repositories/postgres.py +++ b/app/repositories/postgres.py @@ -1169,6 +1169,73 @@ class PostgresPortfolioRepository: rows = cur.fetchall() return [self._row_to_ledger_entry(row) for row in rows] + def insert_ledger_trade( + self, + *, + asset_id: str, + order_id: str, + execution_id: str, + side: str, + quantity: float, + price: float, + executed_at, + fee_amount: float = 0.0, + tax_amount: float = 0.0, + ) -> tuple: + self.ensure_schema() + with self._connect().cursor() as cur: + cur.execute( + """ + INSERT INTO ledger_trades + (asset_id, order_id, execution_id, side, quantity, price, + fee_amount, tax_amount, executed_at) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT ON CONSTRAINT uq_ledger_trades_idempotency DO NOTHING + RETURNING + ledger_entry_id, asset_id, order_id, execution_id, side, + quantity::float8, price::float8, + fee_amount::float8, tax_amount::float8, + executed_at, recorded_at + """, + (asset_id, order_id, execution_id, side, quantity, price, + fee_amount, tax_amount, executed_at), + ) + row = cur.fetchone() + if row is not None: + return self._row_to_ledger_trade(row), True + # Conflict: fetch existing record + cur.execute( + """ + SELECT + ledger_entry_id, asset_id, order_id, execution_id, side, + quantity::float8, price::float8, + fee_amount::float8, tax_amount::float8, + executed_at, recorded_at + FROM ledger_trades + WHERE asset_id = %s AND order_id = %s AND execution_id = %s + """, + (asset_id, order_id, execution_id), + ) + row = cur.fetchone() + return self._row_to_ledger_trade(row), False + + @staticmethod + def _row_to_ledger_trade(row) -> "LedgerTrade": + from app.domain.models import LedgerTrade + return LedgerTrade( + ledger_entry_id=row[0], + asset_id=row[1], + order_id=row[2], + execution_id=row[3], + side=row[4], + quantity=float(row[5]), + price=float(row[6]), + fee_amount=float(row[7]), + tax_amount=float(row[8]), + executed_at=row[9], + recorded_at=row[10], + ) + def reset(self, portfolio_id: str) -> Portfolio: self.ensure_schema() with self._connect().cursor() as cur: