trex/portfolio-hub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: consolidate ledger trade writes into portfolio_ledger

Browse Source 
- add ledger consolidation migration and remove ledger_trades usage
- return portfolio state after v1 ledger trade writes
- deprecate legacy buy/sell endpoints with deprecation headers
- add BAT-016, DCY-006, and REQ-036~038 documentation
- update repository models and tests for unified ledger entries
This commit is contained in:
genichin
2026-06-05 20:55:24 +09:00
parent 2470c0a00c
commit 44f2102d26
23 changed files with 1254 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/repositories/postgres.py
+216 -71
View File
@@ -261,28 +261,79 @@ class PostgresPortfolioRepository:
cur.execute(
"""
CREATE TABLE IF NOT EXISTS portfolio_ledger (
ledger_entry_id BIGINT GENERATED BY DEFAULT AS IDENTITY,
portfolio_id TEXT NOT NULL,
recorded_at TIMESTAMPTZ NOT NULL,
action TEXT NOT NULL,
price DOUBLE PRECISION NOT NULL,
quantity DOUBLE PRECISION NOT NULL,
fee_amount DOUBLE PRECISION NOT NULL DEFAULT 0,
tax_amount DOUBLE PRECISION NOT NULL DEFAULT 0,
cash_balance_after DOUBLE PRECISION NOT NULL,
asset_quantity_after DOUBLE PRECISION NOT NULL,
total_value_after DOUBLE PRECISION NOT NULL,
profit_buffer_after DOUBLE PRECISION NOT NULL DEFAULT 0,
recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
order_id TEXT NULL,
execution_id TEXT NULL,
side TEXT NOT NULL,
price NUMERIC(20,8) NOT NULL,
quantity NUMERIC(20,8) NOT NULL,
fee_amount NUMERIC(20,8) NOT NULL DEFAULT 0,
tax_amount NUMERIC(20,8) NOT NULL DEFAULT 0,
cash_balance_after NUMERIC(20,8) NOT NULL,
asset_quantity_after NUMERIC(20,8) NOT NULL,
total_value_after NUMERIC(20,8) NOT NULL,
profit_buffer_after NUMERIC(20,8) NOT NULL DEFAULT 0,
strategy_set_id TEXT NOT NULL DEFAULT 'legacy',
note TEXT NOT NULL DEFAULT ''
note TEXT NOT NULL DEFAULT '',
executed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
)
"""
)
cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS ledger_entry_id BIGINT GENERATED BY DEFAULT AS IDENTITY")
cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS order_id TEXT")
cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS execution_id TEXT")
cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS side TEXT")
cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS executed_at TIMESTAMPTZ")
cur.execute(
"""
DO $$
BEGIN
IF EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_name = 'portfolio_ledger'
AND column_name = 'action'
) THEN
UPDATE portfolio_ledger SET side = lower(action) WHERE side IS NULL;
END IF;
END $$;
"""
)
cur.execute("UPDATE portfolio_ledger SET executed_at = recorded_at WHERE executed_at IS NULL")
cur.execute("UPDATE portfolio_ledger SET side = 'manual' WHERE side IS NULL")
cur.execute("ALTER TABLE portfolio_ledger ALTER COLUMN side SET NOT NULL")
cur.execute("ALTER TABLE portfolio_ledger ALTER COLUMN executed_at SET NOT NULL")
for column in (
"price",
"quantity",
"fee_amount",
"tax_amount",
"cash_balance_after",
"asset_quantity_after",
"total_value_after",
"profit_buffer_after",
):
cur.execute(
f"ALTER TABLE portfolio_ledger ALTER COLUMN {column} TYPE NUMERIC(20,8) USING {column}::numeric(20,8)"
)
cur.execute("ALTER TABLE portfolio_ledger DROP COLUMN IF EXISTS action")
cur.execute("DROP TABLE IF EXISTS ledger_trades")
cur.execute(
"""
CREATE INDEX IF NOT EXISTS idx_portfolio_ledger_portfolio_time
ON portfolio_ledger (portfolio_id, recorded_at DESC)
"""
)
cur.execute(
"""
CREATE INDEX IF NOT EXISTS idx_portfolio_ledger_portfolio_executed
ON portfolio_ledger (portfolio_id, executed_at DESC)
"""
)
cur.execute(
"""
CREATE INDEX IF NOT EXISTS idx_portfolios_broker_status_updated
@@ -506,15 +557,19 @@ class PostgresPortfolioRepository:
return portfolios
@staticmethod
def _row_to_ledger_entry(row: Any) -> LedgerEntry:
def _row_to_ledger_entry_legacy(row: Any) -> LedgerEntry:
return LedgerEntry(
ledger_entry_id=0,
portfolio_id=row[0],
recorded_at=row[1],
action=row[2],
price=float(row[3]),
order_id=None,
execution_id=None,
side=row[2],
quantity=float(row[4]),
price=float(row[3]),
fee_amount=float(row[5]),
tax_amount=float(row[6]),
executed_at=row[1],
recorded_at=row[1],
cash_balance_after=float(row[7]),
asset_quantity_after=float(row[8]),
total_value_after=float(row[9]),
@@ -1223,15 +1278,15 @@ class PostgresPortfolioRepository:
cur.execute(
"""
INSERT INTO portfolio_ledger (
portfolio_id, recorded_at, action, price, quantity, fee_amount,
portfolio_id, recorded_at, side, price, quantity, fee_amount,
tax_amount, cash_balance_after, asset_quantity_after,
total_value_after, profit_buffer_after, strategy_set_id, note
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
total_value_after, profit_buffer_after, strategy_set_id, note, executed_at
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""",
(
portfolio.portfolio_id,
trade_time,
action,
str(action).lower(),
float(trade_price),
float(quantity),
float(fee_amount),
@@ -1242,6 +1297,7 @@ class PostgresPortfolioRepository:
float(portfolio.profit_buffer),
strategy_set_id,
note,
trade_time,
),
)
@@ -1251,12 +1307,16 @@ class PostgresPortfolioRepository:
cur.execute(
"""
SELECT
portfolio_id, recorded_at, action, price, quantity, fee_amount,
tax_amount, cash_balance_after, asset_quantity_after,
total_value_after, profit_buffer_after, strategy_set_id, note
ledger_entry_id, portfolio_id, order_id, execution_id, side,
quantity::float8, price::float8,
fee_amount::float8, tax_amount::float8,
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
ORDER BY recorded_at DESC
ORDER BY recorded_at DESC, ledger_entry_id DESC
LIMIT %s
""",
(portfolio_id, int(limit)),
@@ -1276,49 +1336,128 @@ class PostgresPortfolioRepository:
executed_at,
fee_amount: float = 0.0,
tax_amount: float = 0.0,
) -> tuple:
) -> tuple[LedgerEntry, bool]:
self.ensure_schema()
with self._connect().cursor() as cur:
cur.execute(
"""
INSERT INTO ledger_trades
(portfolio_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, portfolio_id, order_id, execution_id, side,
quantity::float8, price::float8,
fee_amount::float8, tax_amount::float8,
executed_at, recorded_at
""",
(portfolio_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, portfolio_id, order_id, execution_id, side,
quantity::float8, price::float8,
fee_amount::float8, tax_amount::float8,
executed_at, recorded_at
FROM ledger_trades
WHERE portfolio_id = %s AND order_id = %s AND execution_id = %s
""",
(portfolio_id, order_id, execution_id),
)
row = cur.fetchone()
return self._row_to_ledger_trade(row), False
side_normalized = str(side).lower()
if side_normalized not in {"buy", "sell"}:
raise ValueError("side는 buy 또는 sell 이어야 합니다")
with self._connect() as conn:
with conn.cursor() as cur:
cur.execute(
"""
SELECT
portfolio_id, cash_balance, asset_quantity, asset_price, profit_buffer,
active_strategy_id
FROM portfolios
WHERE portfolio_id = %s
FOR UPDATE
""",
(portfolio_id,),
)
portfolio_row = cur.fetchone()
if portfolio_row is None:
raise ValueError("포트폴리오를 찾을 수 없습니다")
cash_balance = float(portfolio_row[1])
asset_quantity = float(portfolio_row[2])
price_float = float(price)
qty_float = float(quantity)
fee_float = float(fee_amount)
tax_float = float(tax_amount)
if side_normalized == "buy":
cash_balance_after = cash_balance - (qty_float * price_float + fee_float + tax_float)
asset_quantity_after = asset_quantity + qty_float
else:
cash_balance_after = cash_balance + (qty_float * price_float - fee_float - tax_float)
asset_quantity_after = asset_quantity - qty_float
total_value_after = cash_balance_after + asset_quantity_after * price_float
strategy_set_id = portfolio_row[5] or "manual"
cur.execute(
"""
INSERT INTO portfolio_ledger (
portfolio_id, recorded_at, ledger_entry_id, order_id, execution_id,
side, price, quantity, fee_amount, tax_amount,
cash_balance_after, asset_quantity_after, total_value_after,
profit_buffer_after, strategy_set_id, note, executed_at
) VALUES (
%s, NOW(), DEFAULT, %s, %s,
%s, %s, %s, %s, %s,
%s, %s, %s,
%s, %s, %s, %s
)
ON CONFLICT (portfolio_id, order_id, execution_id) WHERE order_id IS NOT NULL
DO NOTHING
RETURNING
ledger_entry_id, portfolio_id, order_id, execution_id, side,
quantity::float8, price::float8,
fee_amount::float8, tax_amount::float8,
executed_at, recorded_at,
cash_balance_after::float8, asset_quantity_after::float8,
total_value_after::float8, profit_buffer_after::float8,
strategy_set_id, note
""",
(
portfolio_id,
order_id,
execution_id,
side_normalized,
price_float,
qty_float,
fee_float,
tax_float,
cash_balance_after,
asset_quantity_after,
total_value_after,
float(portfolio_row[4]),
strategy_set_id,
"",
executed_at,
),
)
row = cur.fetchone()
if row is None:
cur.execute(
"""
SELECT
ledger_entry_id, portfolio_id, order_id, execution_id, side,
quantity::float8, price::float8,
fee_amount::float8, tax_amount::float8,
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 order_id = %s AND execution_id = %s
""",
(portfolio_id, order_id, execution_id),
)
row = cur.fetchone()
return self._row_to_ledger_entry(row), False
cur.execute(
"""
UPDATE portfolios
SET cash_balance = %s,
asset_quantity = %s,
asset_price = %s,
updated_at = NOW(),
last_trade_at = %s
WHERE portfolio_id = %s
""",
(cash_balance_after, asset_quantity_after, price_float, executed_at, portfolio_id),
)
return self._row_to_ledger_entry(row), True
@staticmethod
def _row_to_ledger_trade(row) -> "LedgerTrade":
from app.domain.models import LedgerTrade
return LedgerTrade(
ledger_entry_id=row[0],
def _row_to_ledger_entry(row: Any) -> LedgerEntry:
if row is None:
raise ValueError("원장 항목을 찾을 수 없습니다")
return LedgerEntry(
ledger_entry_id=int(row[0]),
portfolio_id=row[1],
order_id=row[2],
execution_id=row[3],
@@ -1329,6 +1468,12 @@ class PostgresPortfolioRepository:
tax_amount=float(row[8]),
executed_at=row[9],
recorded_at=row[10],
cash_balance_after=float(row[11]),
asset_quantity_after=float(row[12]),
total_value_after=float(row[13]),
profit_buffer_after=float(row[14]),
strategy_set_id=row[15] or "legacy",
note=row[16] or "",
)
def reset(self, portfolio_id: str) -> Portfolio:
@@ -1386,14 +1531,14 @@ class PostgresPortfolioRepository:
cur.execute(
"""
INSERT INTO portfolio_ledger (
portfolio_id, recorded_at, action, price, quantity, fee_amount,
portfolio_id, recorded_at, side, price, quantity, fee_amount,
tax_amount, cash_balance_after, asset_quantity_after,
total_value_after, profit_buffer_after, strategy_set_id, note
) VALUES (%s, NOW(), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
total_value_after, profit_buffer_after, strategy_set_id, note, executed_at
) VALUES (%s, NOW(), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())
""",
(
portfolio_id,
normalized,
normalized.lower(),
0.0,
0.0,
0.0,
@@ -1450,14 +1595,14 @@ class PostgresPortfolioRepository:
cur.execute(
"""
INSERT INTO portfolio_ledger (
portfolio_id, recorded_at, action, price, quantity, fee_amount,
portfolio_id, recorded_at, side, price, quantity, fee_amount,
tax_amount, cash_balance_after, asset_quantity_after,
total_value_after, profit_buffer_after, strategy_set_id, note
) VALUES (%s, NOW(), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
total_value_after, profit_buffer_after, strategy_set_id, note, executed_at
) VALUES (%s, NOW(), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())
""",
(
portfolio_id,
ledger_action,
ledger_action.lower(),
0.0,
0.0,
0.0,