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

fix: ledger write idempotency for TimescaleDB hypertable (BAT-016 rework)

Browse Source 
BAT-016's initial implementation passed only against mocked repositories and
broke on a real database. Two defects fixed:

1. insert_ledger_trade used `ON CONFLICT (portfolio_id, order_id, execution_id)
   WHERE order_id IS NOT NULL` but portfolio_ledger is a TimescaleDB hypertable,
   which cannot have that partial unique index (must include the partitioning
   column recorded_at). The statement errored against Postgres. Replaced with
   application-level idempotency: SELECT-before-insert inside the existing
   `SELECT ... FOR UPDATE` portfolio row lock, which serializes concurrent
   same-portfolio writes (race-safe). Duplicate returns current authoritative
   portfolio state (dcy-006 OQ-4 b).

2. `with self._connect() as conn:` closes the shared psycopg3 connection on exit,
   leaving the reused repository connection unusable. Switched to
   `with conn.transaction():` (commits without closing).

Added tests/test_ledger_consolidation_integration.py — real-Postgres integration
tests (new / duplicate / negative-balance) that close the mock-only verification
gap. Full suite: 376 passed.

SDLC: REQ-036/037 CHG-20260605-03 reconcile OQ-7 (ii) (DB partial unique →
application-level idempotency) since the hypertable makes the DB constraint
infeasible. REQ-036/037/038 → Implemented; BAT-016 → release-candidate.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
genichin
2026-06-05 22:11:17 +09:00
co-authored by Claude Opus 4.8
parent 44f2102d26
commit 2a19b7f058
11 changed files with 414 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/repositories/postgres.py
+39 -21
View File
@@ -1342,7 +1342,11 @@ class PostgresPortfolioRepository:
if side_normalized not in {"buy", "sell"}:
raise ValueError("side는 buy 또는 sell 이어야 합니다")
with self._connect() as conn:
conn = self._connect()
# Use a transaction block (not `with conn:`) so the shared connection stays open:
# psycopg3 closes the connection when it is used as a context manager, which would
# leave the repository's reused connection unusable for subsequent calls.
with conn.transaction():
with conn.cursor() as cur:
cur.execute(
"""
@@ -1376,6 +1380,40 @@ class PostgresPortfolioRepository:
total_value_after = cash_balance_after + asset_quantity_after * price_float
strategy_set_id = portfolio_row[5] or "manual"
# Application-level idempotency under the portfolios row lock.
# portfolio_ledger is a TimescaleDB hypertable, so a DB-level partial unique
# index on (portfolio_id, order_id, execution_id) is impossible (a unique index
# must include the partitioning column recorded_at). The `SELECT ... FOR UPDATE`
# on the portfolios row above serializes concurrent writes for the same portfolio,
# so this check-then-insert is race-safe for broker writes (order_id present).
if order_id is not None and execution_id is not 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
ORDER BY recorded_at DESC, ledger_entry_id DESC
LIMIT 1
""",
(portfolio_id, order_id, execution_id),
)
existing = cur.fetchone()
if existing is not None:
# Duplicate: do not insert a new row, do not mutate the portfolio.
# OQ-4 (b): return the CURRENT authoritative portfolio state so the
# caller always syncs to the latest balances, not the first-call snapshot.
entry = self._row_to_ledger_entry(existing)
entry.cash_balance_after = cash_balance
entry.asset_quantity_after = asset_quantity
return entry, False
cur.execute(
"""
INSERT INTO portfolio_ledger (
@@ -1389,8 +1427,6 @@ class PostgresPortfolioRepository:
%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,
@@ -1419,24 +1455,6 @@ class PostgresPortfolioRepository:
),
)
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(
"""