diff --git a/.proofline/criteria/ac-0039.md b/.proofline/criteria/ac-0039.md new file mode 100644 index 0000000..3b6c215 --- /dev/null +++ b/.proofline/criteria/ac-0039.md @@ -0,0 +1,16 @@ +--- +id: "ac-0039" +status: active +--- + +# Broker ledger venue API and persistence contract + +## Criterion + +Broker ledger write는 optional nullable `execution_venue`와 `order_venue`를 수용해야 한다. `execution_venue`는 broker가 확인한 actual venue인 `KRX`, `NXT` 또는 NULL이고, `order_venue`는 original routing venue인 `KRX`, `NXT`, `SOR` 또는 NULL이다. Portfolio-hub는 두 값을 추정·상호 변환하지 않고 broker ledger row에 저장하며 POST의 new/duplicate 응답과 unified `GET /v1/portfolios/{portfolio_id}/ledger` broker item에서 그대로 반환해야 한다. + +## Verification + +- 성공 조건: `(execution_venue=NULL, order_venue=SOR)`, KRX/NXT actual과 KRX/NXT/SOR order 조합이 POST→repository→실제 PostgreSQL→POST response→unified GET에서 동일하게 round-trip되고 OpenAPI가 두 nullable enum을 노출한다. +- 실패·경계 조건: `execution_venue=SOR`, AUTO 또는 미지원 값과 `order_venue=AUTO` 또는 미지원 값은 `422`나 DB constraint로 거부되며, order venue·시간·default로 actual venue를 채우지 않는다. 필드 생략 request는 계속 유효하고 manual/corporate-action/historical item은 NULL을 반환할 수 있다. +- 검사 방법: schema/OpenAPI·API·repository tests와 실제 PostgreSQL round-trip/check-constraint integration을 실행하고 `LedgerEntry`·request/response·unified mapper의 end-to-end field path를 검토한다. diff --git a/.proofline/criteria/ac-0040.md b/.proofline/criteria/ac-0040.md new file mode 100644 index 0000000..38fd858 --- /dev/null +++ b/.proofline/criteria/ac-0040.md @@ -0,0 +1,16 @@ +--- +id: "ac-0040" +status: active +--- + +# Idempotent venue provenance enrichment + +## Criterion + +동일 `(portfolio_id, order_id, execution_id)`의 broker ledger duplicate는 기존 portfolio row lock을 유지하면서 기존 ledger row를 `FOR UPDATE`로 선택해야 한다. 같은 transaction에서 기존 economic payload(`side`, `quantity`, `price`, `fee_amount`, `tax_amount`)를 먼저 비교하고 불일치면 metadata 변경 없이 `409`로 중단해야 한다. Economic payload가 같을 때 모든 incoming venue를 검사하여 저장값과 서로 다른 non-null이 하나라도 있으면 어느 field도 갱신하지 않고 `409`로 중단하며, 충돌이 없을 때만 저장값 NULL·incoming non-null인 venue들을 하나의 metadata-only update로 보강해야 한다. Incoming NULL은 기존 non-null을 유지해야 한다. 이 경로는 기존 ledger row를 재생성하거나 portfolio cash·asset quantity·snapshot과 WAC/PnL를 다시 변경하지 않아야 한다. + +## Verification + +- 성공 조건: execution/order venue 각각과 두 필드 동시 NULL→non-null duplicate가 같은 ledger row를 갱신하고 `200 duplicate` 최종 저장값을 반환하며 ledger row 수·cash·quantity·snapshot이 최초 `201 new` 이후 변하지 않는다. Incoming NULL replay는 기존 값을 보존한다. +- 실패·경계 조건: economic payload 불일치는 venue 검사·update 전에 `409`이고, execution 또는 order venue의 non-null 불일치는 모든 venue 검사를 완료한 뒤 update 전에 `409`이므로 어느 venue도 부분 갱신되지 않으며 회계가 변하지 않는다. 서로 다른 concurrent backfill은 portfolio·ledger row lock 아래 하나만 성공하거나 같은 값으로 수렴하고 상충 값의 last-write-wins를 허용하지 않는다. +- 검사 방법: API/repository idempotency·conflict tests와 실제 PostgreSQL row-lock/concurrency integration에서 final row·row count·portfolio balances·quantity·snapshot을 함께 검사한다. diff --git a/.proofline/criteria/ac-0041.md b/.proofline/criteria/ac-0041.md new file mode 100644 index 0000000..5907941 --- /dev/null +++ b/.proofline/criteria/ac-0041.md @@ -0,0 +1,16 @@ +--- +id: "ac-0041" +status: active +--- + +# Venue migration and mixed-version compatibility + +## Criterion + +Venue provenance schema 변경은 `portfolio_ledger`에 nullable `execution_venue`와 `order_venue` 및 값 constraint만 additive하게 추가하고 기존 row를 backfill하거나 재해석하지 않아야 한다. 빈 DB bootstrap, 기존 Alembic head에서의 upgrade와 migration replay가 모두 같은 head schema에 수렴해야 하며, Hub가 새 필드를 optional로 먼저 수용한 뒤 core adapter가 전송을 시작하는 혼합 버전 순서에서 구·신 payload와 기존 ledger consumer가 함께 동작해야 한다. + +## Verification + +- 성공 조건: disposable PostgreSQL에서 cold bootstrap, pre-migration legacy fixture upgrade와 replay가 성공하고 기존 row의 두 venue가 NULL인 채 다른 값이 보존된다. 새 core payload와 필드를 생략한 구 payload가 모두 수용되며 manual/corporate-action, authorization, ordering/filter/cursor, joined corporate-action, WAC/PnL 회귀가 통과한다. +- 실패·경계 조건: migration이 historical venue를 KRX/NXT/SOR로 채우거나 기존 row를 삭제·변경하거나 invalid venue를 허용하면 실패한다. Hub보다 core adapter를 먼저 배포하는 순서는 지원된다고 주장하지 않는다. Downgrade는 신규 constraint와 column만 제거하고 기존 schema를 복원한다. +- 검사 방법: Alembic upgrade/downgrade/replay·`app/db/migrations.py` cold/legacy integration, PostgreSQL schema inventory와 기존 ledger API/repository 회귀를 실행하고 interface contract의 Hub-first rollout 문구를 검사한다. diff --git a/.proofline/identities.json b/.proofline/identities.json index 5f851a5..49885af 100644 --- a/.proofline/identities.json +++ b/.proofline/identities.json @@ -1,5 +1,5 @@ { "schema_version": 1, - "next_line_number": 9, - "next_ac_number": 39 + "next_line_number": 10, + "next_ac_number": 42 } diff --git a/.proofline/lines/line-0009/dcy-0009.md b/.proofline/lines/line-0009/dcy-0009.md new file mode 100644 index 0000000..7471314 --- /dev/null +++ b/.proofline/lines/line-0009/dcy-0009.md @@ -0,0 +1,59 @@ +--- +id: "dcy-0009" +status: confirmed +--- + +# Broker ledger venue provenance preservation + +## Problem + +`trex-core`는 broker가 확인한 실제 체결 거래소를 `LedgerTrade.venue`에, 최초 주문 라우팅 거래소를 `LedgerTrade.order_venue`에 구분하여 durable ledger outbox까지 보존한다. 그러나 `PortfolioHubLedger._payload()`가 두 값을 HTTP payload에서 누락하고, `portfolio-hub`의 broker ledger request·저장소·응답 모델에도 대응 필드가 없다. 이 때문에 core 내부에서 보존된 venue provenance가 portfolio-hub 전달 경계에서 소실된다. + +기대 상태는 core의 nullable actual execution venue와 original order venue를 추정·변환 없이 portfolio-hub에 전달하고, 기존 broker ledger 행·manual ledger·corporate action·구버전 client의 동작을 깨뜨리지 않으면서 저장·조회하는 것이다. + +## Evidence + +- `E-001` GitHub Issue #32 `https://github.com/fourblocks2020/portfolio-hub/issues/32`는 `LedgerTrade.venue -> execution_venue`, `LedgerTrade.order_venue -> order_venue` mapping, nullable historical compatibility, duplicate metadata 보강, Hub-first rollout과 venue-only 범위를 구현 계획으로 정한다. +- `E-002` `trex-core` revision `bbdacb121c25a7539f6f9f7fe0559fc03d4542eb`의 `trex_core/application/ledger_port.py`는 `LedgerTrade.venue: Optional[str]`와 `LedgerTrade.order_venue: str`를 별도 필드로 정의한다. +- `E-003` 같은 revision의 `trex_core/adapters/persistence/postgresql.py`와 migration `024_add_ledger_order_venue.sql`은 durable outbox에 `venue`와 `order_venue`를 별도 저장·복원하지만, `trex_core/adapters/ledger/portfolio_hub.py::PortfolioHubLedger._payload()`는 두 값을 HTTP payload에 포함하지 않는다. +- `E-004` `portfolio-hub` revision `567d662913335a90d0de51b8e0e961b6b166d543`의 `app/schemas/v1.py::LedgerTradeRequest`·`LedgerTradeResponse`, `app/domain/models.py::LedgerEntry`, `app/repositories/portfolio_repository.py`와 `app/repositories/postgres.py`에는 execution/order venue 계약이 없다. +- `E-005` 현재 Alembic head의 `portfolio_ledger`에는 두 venue column이 없으며 `app/db/migrations.py`와 Alembic migration이 빈 DB와 legacy DB bootstrap을 함께 소유한다. +- `E-006` portfolio-hub active AC-0032·AC-0034·AC-0035·AC-0037은 unified ledger read, nullable source-specific facts, authorization·manual compatibility와 joined ledger 안정성을 소유하지만 broker venue의 POST validation·persistence·duplicate 보강을 명시하지 않는다. 따라서 새 observable contract는 `create` 대상이고 기존 AC는 의미를 바꾸지 않는 `satisfy` 대상이다. +- `E-007` trex-core active AC-0019는 original `order_venue`와 nullable actual execution venue의 durable 분리, `(order_venue=SOR, venue=NULL)` 보존, 무추론 원칙과 기존 `LedgerWriter` 전달 경로를 이미 소유한다. Core 변경은 기존 AC의 의미를 바꾸지 않는 satisfy 구현이다. + +## Scope + +- `S-001` portfolio-hub에 새 active AC를 `create`하여 broker ledger request가 nullable `execution_venue` (`KRX`, `NXT`)와 nullable `order_venue` (`KRX`, `NXT`, `SOR`)를 수용하고 다른 값과 `execution_venue=SOR`를 거부하는 계약을 소유한다. +- `S-002` `portfolio_ledger`에 nullable venue column과 allowlist constraint를 additive migration으로 추가하고 historical row를 NULL로 유지한다. 빈 DB·legacy DB migration apply/replay와 실제 PostgreSQL round-trip을 검증한다. +- `S-003` 새 active AC를 `create`하여 broker ledger insert, duplicate lookup, domain mapping, POST 응답과 unified GET 응답이 두 venue를 추정·변환 없이 저장·반환하는 계약을 소유한다. +- `S-004` 동일 `(portfolio_id, order_id, execution_id)` duplicate는 portfolio row lock으로 기존 동시성 경계를 유지하면서 기존 ledger row도 `FOR UPDATE`로 선택한다. 같은 transaction에서 기존 economic payload 비교를 먼저 수행해 불일치면 아무 metadata도 갱신하지 않고 `409`로 중단한다. Economic payload가 같을 때만 모든 venue의 non-null 충돌을 먼저 검사하고, 충돌이 없으면 NULL column만 incoming non-null 값으로 한 번에 보강한다. Incoming NULL은 기존 값을 지우지 않으며 이 보강은 ledger row를 재생성하거나 cash·quantity·snapshot 회계를 다시 적용하지 않는다. +- `S-005` 기존 active AC-0032·AC-0034·AC-0035·AC-0037을 `satisfy`하여 source-neutral ordering/filter/cursor, nullable source-specific representation, owner-only authorization, manual/corporate-action 및 WAC/PnL compatibility를 유지한다. +- `S-006` trex-core는 active AC-0019 아래 `PortfolioHubLedger._payload()`가 기존 `LedgerTrade.venue`를 `execution_venue`, 기존 `LedgerTrade.order_venue`를 `order_venue`로 전송하도록 변경한다. producer, `LedgerTrade`, outbox schema와 broker routing은 변경하지 않는다. +- `S-007` 배포 호환 경계는 portfolio-hub 선배포 후 trex-core adapter 배포이며, 구버전 core payload와 historical/manual/corporate-action row는 venue가 NULL인 상태로 계속 유효하다. +- `S-008` 검증은 portfolio-hub schema/API/repository/PostgreSQL migration과 concurrency/idempotency, trex-core exact adapter payload·outbox reload/retry, 양 저장소 교차 contract `201 new -> 200 duplicate -> GET /ledger`를 포함한다. + +## Out of Scope + +- `OS-001` `broker_ordered_at`, `core_observed_filled_at` 또는 다른 timestamp provenance. +- `OS-002` trex-core `LedgerTrade`, producer, **broker execution adapter**, durable outbox schema 또는 주문 routing 변경. `PortfolioHubLedger` HTTP ledger transport adapter의 두 기존 field 직렬화는 `S-006` 범위다. +- `OS-003` historical venue 추정·backfill, 특히 `order_venue`나 시간대로 actual execution venue를 추론하는 동작. +- `OS-004` ledger detail endpoint, idempotency key, broker 주문 실행·라우팅 정책 변경. +- `OS-005` production migration·deployment·runtime restart·실제 broker 주문. + +## Risks and Unknowns + +### Risks + +- `R-001` mixed-version rollout에서 새 core가 구 Hub에 먼저 배포되면 추가 payload field가 거부되거나 유실될 수 있다. Hub를 먼저 배포하고 기존 payload omission을 허용한다. +- `R-002` duplicate metadata 보강이 기존 회계 transaction path를 재실행하면 잔액·수량·snapshot이 중복 반영될 수 있다. row lock과 단일 transaction에서 provenance-only update를 분리하고 회계 불변 회귀를 실행한다. +- `R-003` repository query의 positional mapping과 joined corporate-action projection에 column을 추가할 때 기존 mapping이 어긋날 수 있다. 명시 projection·mapping 검토와 실제 PostgreSQL mixed-row 회귀로 방지한다. +- `R-004` concurrent duplicates가 서로 다른 non-null venue를 보강하면 last-write-wins로 provenance가 변조될 수 있다. 동일 row를 잠그고 NULL-only update와 non-null conflict를 transaction 안에서 판정한다. + +### Assumptions + +- `A-001` Issue #32에 적힌 venue-only 구현 계획과 사용자의 현재 구현 지시가 제품 범위의 근거이며, 추가 timestamp나 broker lifecycle 변경은 승인되지 않았다. +- `A-002` trex-core active AC-0019의 기존 producer/outbox 계약은 변경 없이 유지되고, 이번 core 변경은 transport 누락을 복구하는 satisfy 작업이다. + +### Open Questions + +없음. Issue #32와 현재 source에서 API naming, 값 allowlist, NULL semantics, duplicate policy, migration policy, rollout 순서와 제외 범위가 모두 결정되어 있다. diff --git a/.proofline/lines/line-0009/line-0009.md b/.proofline/lines/line-0009/line-0009.md new file mode 100644 index 0000000..24d6d09 --- /dev/null +++ b/.proofline/lines/line-0009/line-0009.md @@ -0,0 +1,4 @@ +--- +id: "line-0009" +status: discovery +--- diff --git a/.proofline/lines/line-0009/req-0009.md b/.proofline/lines/line-0009/req-0009.md new file mode 100644 index 0000000..905067f --- /dev/null +++ b/.proofline/lines/line-0009/req-0009.md @@ -0,0 +1,53 @@ +--- +id: "req-0009" +status: approved +discovery: "dcy-0009" +criteria: + create: + - "ac-0039" + - "ac-0040" + - "ac-0041" + update: [] + retire: [] + satisfy: + - "ac-0032" + - "ac-0034" + - "ac-0035" + - "ac-0037" +--- + +# Broker ledger venue provenance preservation + +## Objective + +`trex-core`가 이미 durable하게 구분하는 actual execution venue와 original order-routing venue를 portfolio-hub의 broker ledger write·storage·read 경계까지 손실 없이 보존한다. Unknown actual venue와 SOR routing을 정상 조합으로 유지하고, 기존 client·historical/manual/corporate-action row·회계·조회 계약과 혼합 버전 배포의 하위호환성을 보장한다. + +## Scope + +- Broker ledger request에 optional nullable `execution_venue`와 `order_venue`를 추가한다. +- `execution_venue`는 `KRX`, `NXT`, NULL만 허용하고 `order_venue`는 `KRX`, `NXT`, `SOR`, NULL만 허용한다. +- `portfolio_ledger`에 두 nullable column과 같은 allowlist constraint를 additive migration으로 추가한다. 기존 row는 NULL로 유지한다. +- Broker ledger insert·domain mapping·POST response와 unified GET broker item이 저장값을 추정·변환 없이 반환한다. +- 동일 `(portfolio_id, order_id, execution_id)` duplicate는 portfolio row lock에 더해 기존 ledger row를 `FOR UPDATE`로 선택한다. 같은 transaction에서 기존 economic payload를 먼저 비교해 불일치면 metadata 변경 없이 `409`로 중단한다. Economic payload가 같을 때 모든 venue의 non-null 충돌을 먼저 검사하고, 충돌이 없을 때만 NULL→non-null venue를 하나의 metadata-only update로 보강한다. Incoming NULL은 기존 non-null을 지우지 않는다. +- Metadata-only duplicate 보강은 ledger row, portfolio cash·asset quantity·snapshot, WAC/PnL 또는 corporate-action state를 다시 적용하지 않는다. +- 기존 active AC-0032·AC-0034·AC-0035·AC-0037을 의미 변경 없이 satisfy하여 unified ordering/filter/cursor, canonical nullable representation, owner authorization, manual/corporate-action 및 joined ledger compatibility를 유지한다. +- trex-core는 active AC-0019 아래 기존 `LedgerTrade.venue`를 `execution_venue`, `LedgerTrade.order_venue`를 `order_venue`로 HTTP payload에 전달한다. Core producer·domain·outbox schema는 변경하지 않는다. +- 실제 PostgreSQL migration·round-trip·concurrent duplicate와 core adapter payload/reload, 교차 저장소 `201 new -> 200 duplicate -> GET /ledger` contract를 검증한다. + +## Constraints + +- `SOR`는 order routing provenance이며 actual execution venue로 저장하거나 추론하지 않는다. +- Unknown actual venue는 NULL로 보존하고 `order_venue`, 시간 또는 historical default에서 KRX/NXT를 추론하지 않는다. +- 기존 idempotency key와 기존 경제적 payload conflict 판정은 유지한다. +- 추가 request field는 optional이어야 하며 구버전 core/client payload가 계속 유효해야 한다. +- Historical, manual, corporate-action row는 두 venue가 NULL이어도 유효해야 한다. +- Hub-first, core-adapter-second 배포 순서를 문서화하되 production rollout은 수행하지 않는다. +- Normative Design Documents: 없음. 이 REQ와 create AC-0039~AC-0041 및 satisfy 대상 active AC가 normative specification을 소유한다. + +## Non-Goals + +- `broker_ordered_at`, `core_observed_filled_at` 또는 다른 timestamp provenance. +- trex-core `LedgerTrade`, producer, **broker execution adapter**, outbox table/schema 또는 routing policy 변경. `PortfolioHubLedger` HTTP ledger transport adapter의 기존 venue 직렬화는 Scope에 포함한다. +- Historical venue 추정·backfill. +- 새로운 ledger detail endpoint 또는 idempotency key 변경. +- Production migration·deployment·restart·실제 broker 주문. diff --git a/app/api/v1_portfolios.py b/app/api/v1_portfolios.py index 83cd576..6f706a6 100644 --- a/app/api/v1_portfolios.py +++ b/app/api/v1_portfolios.py @@ -27,6 +27,7 @@ from app.domain.models import CorporateAction, Portfolio, PortfolioComment from app.domain.pnl import calculate_current_wac, calculate_realized_pnl from app.domain.services import PORTFOLIO_NOT_FOUND_MESSAGE from app.repositories.factory import get_portfolio_repository_from_settings +from app.repositories.portfolio_repository import LedgerTradeConflictError from app.schemas.corporate_actions import ( CorporateActionRequest, CorporateActionResponse, @@ -604,7 +605,13 @@ def _require_service_token(auth: AuthContext = Depends(require_trade_write)) -> return auth -def _payloads_match(existing, body: LedgerTradeRequest) -> bool: +def _economic_payloads_match(existing, body: LedgerTradeRequest) -> bool: + """Defence in depth for non-Postgres repository implementations. + + The Postgres repository performs this comparison under the portfolio and + ledger row locks; this check preserves the public contract for test or + alternate repository implementations. + """ return ( existing.side == body.side and abs(existing.quantity - float(body.quantity)) < 1e-9 @@ -659,14 +666,21 @@ def record_ledger_trade_v1( executed_at=body.executed_at, fee_amount=float(body.fee_amount), tax_amount=float(body.tax_amount), + execution_venue=body.execution_venue, + order_venue=body.order_venue, ) + except LedgerTradeConflictError: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Duplicate key with different payload", + ) from None except HTTPException: raise finally: _close_repo(repo) if not created: - if not _payloads_match(trade, body): + if not _economic_payloads_match(trade, body): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Duplicate key with different payload", @@ -687,6 +701,8 @@ def record_ledger_trade_v1( fee_amount=trade.fee_amount, tax_amount=trade.tax_amount, executed_at=trade.executed_at, + execution_venue=trade.execution_venue, + order_venue=trade.order_venue, recorded_at=trade.recorded_at, cash_balance_after=trade.cash_balance_after, asset_quantity_after=trade.asset_quantity_after, @@ -827,6 +843,7 @@ def _ledger_entry_to_history_item(entry) -> LedgerHistoryItem | DividendLedgerHi 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, + execution_venue=entry.execution_venue, order_venue=entry.order_venue, ) action = (entry.action or entry.side or "").upper() if action == "DIVIDEND": @@ -853,6 +870,8 @@ def _ledger_entry_to_history_item(entry) -> LedgerHistoryItem | DividendLedgerHi total_value_after=entry.total_value_after, target_ledger_entry_id=entry.target_ledger_entry_id, confirmation_evidence=entry.confirmation_evidence, + execution_venue=entry.execution_venue, + order_venue=entry.order_venue, ) return LedgerHistoryItem( ledger_entry_id=entry.ledger_entry_id, @@ -875,6 +894,8 @@ def _ledger_entry_to_history_item(entry) -> LedgerHistoryItem | DividendLedgerHi total_value_after=entry.total_value_after, target_ledger_entry_id=entry.target_ledger_entry_id, confirmation_evidence=entry.confirmation_evidence, + execution_venue=entry.execution_venue, + order_venue=entry.order_venue, ) diff --git a/app/db/alembic/versions/20260919_0032_ledger_venue_provenance.py b/app/db/alembic/versions/20260919_0032_ledger_venue_provenance.py new file mode 100644 index 0000000..5037154 --- /dev/null +++ b/app/db/alembic/versions/20260919_0032_ledger_venue_provenance.py @@ -0,0 +1,54 @@ +"""Add nullable execution and order venue provenance to portfolio ledger. + +Revision ID: 20260919_0032 +Revises: 20260913_0031 +""" + +from alembic import op + +revision = "20260919_0032" +down_revision = "20260913_0031" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS execution_venue TEXT") + op.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS order_venue TEXT") + op.execute( + """ + DO $$ BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'ck_portfolio_ledger_execution_venue' + AND conrelid = 'portfolio_ledger'::regclass + ) THEN + ALTER TABLE portfolio_ledger + ADD CONSTRAINT ck_portfolio_ledger_execution_venue + CHECK (execution_venue IN ('KRX', 'NXT')); + END IF; + END $$ + """ + ) + op.execute( + """ + DO $$ BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'ck_portfolio_ledger_order_venue' + AND conrelid = 'portfolio_ledger'::regclass + ) THEN + ALTER TABLE portfolio_ledger + ADD CONSTRAINT ck_portfolio_ledger_order_venue + CHECK (order_venue IN ('KRX', 'NXT', 'SOR')); + END IF; + END $$ + """ + ) + + +def downgrade() -> None: + op.drop_constraint("ck_portfolio_ledger_order_venue", "portfolio_ledger", type_="check") + op.drop_constraint("ck_portfolio_ledger_execution_venue", "portfolio_ledger", type_="check") + op.drop_column("portfolio_ledger", "order_venue") + op.drop_column("portfolio_ledger", "execution_venue") diff --git a/app/domain/models.py b/app/domain/models.py index 31f653e..6bb95f6 100644 --- a/app/domain/models.py +++ b/app/domain/models.py @@ -6,7 +6,7 @@ from dataclasses import dataclass, field from datetime import date, datetime from decimal import Decimal from enum import Enum -from typing import Any, Protocol, runtime_checkable +from typing import Any, Literal, Protocol, runtime_checkable class MarketCode(str, Enum): @@ -181,6 +181,8 @@ class LedgerEntry: target_ledger_entry_id: int | None = None confirmation_evidence: str | None = None actor_user_id: str | None = None + execution_venue: Literal["KRX", "NXT"] | None = None + order_venue: Literal["KRX", "NXT", "SOR"] | None = None corporate_action: dict[str, Any] | None = None def __post_init__(self) -> None: diff --git a/app/repositories/portfolio_repository.py b/app/repositories/portfolio_repository.py index 42e262a..1598bad 100644 --- a/app/repositories/portfolio_repository.py +++ b/app/repositories/portfolio_repository.py @@ -9,6 +9,10 @@ from typing import Protocol from app.domain.models import Asset, AuthToken, BrokerCredential, CorporateAction, Daemon, LedgerEntry, Portfolio, PortfolioComment, PortfolioTag, StrategySet, Tag, TokenPrincipal, User +class LedgerTradeConflictError(ValueError): + """A duplicate broker-ledger key disagrees with its persisted payload.""" + + class PortfolioRepository(Protocol): def ping(self) -> None: ... @@ -324,6 +328,8 @@ class PortfolioRepository(Protocol): executed_at: datetime, fee_amount: float = 0.0, tax_amount: float = 0.0, + execution_venue: str | None = None, + order_venue: str | None = None, ) -> tuple[LedgerEntry, bool]: ... def record_manual_ledger( diff --git a/app/repositories/postgres.py b/app/repositories/postgres.py index 6a79bd8..4ea9a6b 100644 --- a/app/repositories/postgres.py +++ b/app/repositories/postgres.py @@ -40,6 +40,7 @@ from app.domain.models import ( TokenPrincipal, User, ) +from app.repositories.portfolio_repository import LedgerTradeConflictError logger = logging.getLogger(__name__) _LEDGER_NUMERIC_SCALE = 8 @@ -382,7 +383,11 @@ class PostgresPortfolioRepository: profit_buffer_after NUMERIC(20,8) NOT NULL DEFAULT 0, strategy_set_id TEXT NOT NULL DEFAULT 'legacy', note TEXT NOT NULL DEFAULT '', - executed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + executed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + execution_venue TEXT NULL CONSTRAINT ck_portfolio_ledger_execution_venue + CHECK (execution_venue IN ('KRX', 'NXT')), + order_venue TEXT NULL CONSTRAINT ck_portfolio_ledger_order_venue + CHECK (order_venue IN ('KRX', 'NXT', 'SOR')) ) """ ) @@ -395,6 +400,33 @@ class PostgresPortfolioRepository: cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS target_ledger_entry_id BIGINT") cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS confirmation_evidence TEXT") cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS actor_user_id TEXT") + cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS execution_venue TEXT") + cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS order_venue TEXT") + cur.execute( + """ + DO $$ + BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'ck_portfolio_ledger_execution_venue' + AND conrelid = 'portfolio_ledger'::regclass + ) THEN + ALTER TABLE portfolio_ledger + ADD CONSTRAINT ck_portfolio_ledger_execution_venue + CHECK (execution_venue IS NULL OR execution_venue IN ('KRX', 'NXT')); + END IF; + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'ck_portfolio_ledger_order_venue' + AND conrelid = 'portfolio_ledger'::regclass + ) THEN + ALTER TABLE portfolio_ledger + ADD CONSTRAINT ck_portfolio_ledger_order_venue + CHECK (order_venue IS NULL OR order_venue IN ('KRX', 'NXT', 'SOR')); + END IF; + END $$; + """ + ) cur.execute( """ DO $$ @@ -2018,6 +2050,8 @@ class PostgresPortfolioRepository: executed_at, fee_amount: float = 0.0, tax_amount: float = 0.0, + execution_venue: str | None = None, + order_venue: str | None = None, ) -> tuple[LedgerEntry, bool]: self.ensure_schema() side_normalized = str(side).lower() @@ -2078,16 +2112,70 @@ 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 + strategy_set_id, note, source, target_ledger_entry_id, + confirmation_evidence, actor_user_id, execution_venue, order_venue 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 + FOR UPDATE """, (portfolio_id, order_id, execution_id), ) existing = cur.fetchone() if existing is not None: + economic_payload_matches = ( + existing[4] == side_normalized + and abs(float(existing[5]) - qty_float) < 1e-9 + and abs(float(existing[6]) - price_float) < 1e-9 + and abs(float(existing[7]) - fee_float) < 1e-9 + and abs(float(existing[8]) - tax_float) < 1e-9 + ) + if not economic_payload_matches: + raise LedgerTradeConflictError("duplicate economic payload conflict") + + existing_execution_venue = existing[21] + existing_order_venue = existing[22] + if ( + existing_execution_venue is not None + and execution_venue is not None + and existing_execution_venue != execution_venue + ) or ( + existing_order_venue is not None + and order_venue is not None + and existing_order_venue != order_venue + ): + raise LedgerTradeConflictError("duplicate venue payload conflict") + + enriched_execution_venue = existing_execution_venue or execution_venue + enriched_order_venue = existing_order_venue or order_venue + if ( + enriched_execution_venue != existing_execution_venue + or enriched_order_venue != existing_order_venue + ): + cur.execute( + """ + UPDATE portfolio_ledger + SET execution_venue = %s, order_venue = %s + WHERE ledger_entry_id = %s AND recorded_at = %s + 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, source, target_ledger_entry_id, + confirmation_evidence, actor_user_id, execution_venue, order_venue + """, + ( + enriched_execution_venue, + enriched_order_venue, + existing[0], + existing[10], + ), + ) + existing = cur.fetchone() # 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. @@ -2102,12 +2190,13 @@ class PostgresPortfolioRepository: 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 + profit_buffer_after, strategy_set_id, note, executed_at, + execution_venue, order_venue ) VALUES ( %s, NOW(), DEFAULT, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, - %s, %s, %s, %s + %s, %s, %s, %s, %s, %s ) RETURNING ledger_entry_id, portfolio_id, order_id, execution_id, side, @@ -2116,7 +2205,8 @@ 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 + strategy_set_id, note, source, target_ledger_entry_id, + confirmation_evidence, actor_user_id, execution_venue, order_venue """, ( portfolio_id, @@ -2134,6 +2224,8 @@ class PostgresPortfolioRepository: strategy_set_id, "", executed_at, + execution_venue, + order_venue, ), ) row = cur.fetchone() @@ -2182,6 +2274,16 @@ class PostgresPortfolioRepository: target_ledger_entry_id=int(row[18]) if len(row) > 18 and row[18] is not None else None, confirmation_evidence=row[19] if len(row) > 19 else None, actor_user_id=row[20] if len(row) > 20 else None, + execution_venue=( + row[21] + if len(row) > 21 and row[21] in {None, "KRX", "NXT"} + else None + ), + order_venue=( + row[22] + if len(row) > 22 and row[22] in {None, "KRX", "NXT", "SOR"} + else None + ), ) def reset(self, portfolio_id: str) -> Portfolio: @@ -2492,7 +2594,8 @@ 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, + execution_venue, order_venue FROM portfolio_ledger WHERE portfolio_id = %s AND ledger_entry_id = %s """, @@ -2591,7 +2694,8 @@ class PostgresPortfolioRepository: RETURNING 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, + execution_venue, order_venue """, ( portfolio_id, @@ -2623,7 +2727,8 @@ 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, + execution_venue, order_venue FROM portfolio_ledger WHERE portfolio_id = %s AND source = 'manual' ORDER BY recorded_at DESC, ledger_entry_id DESC @@ -2661,6 +2766,7 @@ class PostgresPortfolioRepository: ledger.asset_quantity_after, ledger.total_value_after, ledger.profit_buffer_after, ledger.strategy_set_id, ledger.note, ledger.source, ledger.target_ledger_entry_id, ledger.confirmation_evidence, ledger.actor_user_id, + ledger.execution_venue, ledger.order_venue, 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, @@ -2676,8 +2782,8 @@ class PostgresPortfolioRepository: 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]} + if row[23] is not None: + entry.corporate_action = {"corporate_action_id": str(row[23]), "action_type": row[24], "effective_date": row[25], "quantity_before": Decimal(str(row[26])), "quantity_after": Decimal(str(row[27])), "average_price_before": Decimal(str(row[28])), "average_price_after": Decimal(str(row[29])), "total_cost_basis": Decimal(str(row[30])), "cash_balance": Decimal(str(row[31])), "profit_buffer": Decimal(str(row[32])), "evidence_ref": row[33], "actor_user_id": row[34]} result.append(entry) return result diff --git a/app/schemas/portfolio.py b/app/schemas/portfolio.py index 42b3174..572b932 100644 --- a/app/schemas/portfolio.py +++ b/app/schemas/portfolio.py @@ -37,6 +37,8 @@ class LedgerEntryResponse(BaseModel): profit_buffer_after: float = 0.0 strategy_set_id: str = "manual" note: str = "" + execution_venue: Literal["KRX", "NXT"] | None = None + order_venue: Literal["KRX", "NXT", "SOR"] | None = None class PortfolioBase(BaseModel): diff --git a/app/schemas/v1.py b/app/schemas/v1.py index 278a9bc..c786f3a 100644 --- a/app/schemas/v1.py +++ b/app/schemas/v1.py @@ -108,6 +108,8 @@ class UserInfoResponse(BaseModel): # --------------------------------------------------------------------------- class LedgerTradeRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + order_id: str = Field(..., min_length=1) execution_id: str = Field(..., min_length=1) side: Literal["buy", "sell"] @@ -116,6 +118,8 @@ class LedgerTradeRequest(BaseModel): fee_amount: Decimal = Field(default=Decimal(0), ge=0) tax_amount: Decimal = Field(default=Decimal(0), ge=0) executed_at: datetime + execution_venue: Literal["KRX", "NXT"] | None = None + order_venue: Literal["KRX", "NXT", "SOR"] | None = None class LedgerTradeResponse(BaseModel): @@ -129,6 +133,8 @@ class LedgerTradeResponse(BaseModel): fee_amount: float tax_amount: float executed_at: datetime + execution_venue: Literal["KRX", "NXT"] | None = None + order_venue: Literal["KRX", "NXT", "SOR"] | None = None recorded_at: datetime | None = None cash_balance_after: float = 0.0 asset_quantity_after: float = 0.0 @@ -286,6 +292,8 @@ class LedgerHistoryItem(BaseModel): total_value_after: float | None = None target_ledger_entry_id: int | None = None confirmation_evidence: str | None = None + execution_venue: Literal["KRX", "NXT"] | None = None + order_venue: Literal["KRX", "NXT", "SOR"] | None = None class DividendLedgerHistoryItem(BaseModel): @@ -311,6 +319,8 @@ class DividendLedgerHistoryItem(BaseModel): total_value_after: float | None = None target_ledger_entry_id: int | None = None confirmation_evidence: str | None = None + execution_venue: Literal["KRX", "NXT"] | None = None + order_venue: Literal["KRX", "NXT", "SOR"] | None = None class CorporateActionLedgerHistoryItem(BaseModel): @@ -331,6 +341,8 @@ class CorporateActionLedgerHistoryItem(BaseModel): evidence_ref: str actor_user_id: str | None = None recorded_at: datetime + execution_venue: Literal["KRX", "NXT"] | None = None + order_venue: Literal["KRX", "NXT", "SOR"] | None = None class LedgerHistoryResponse(BaseModel): diff --git a/docs/interface-contract.md b/docs/interface-contract.md index e9140b9..ac7ec4c 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -4,8 +4,8 @@ - Baseline Seed: .stagepilot/bootstrap/baseline.yaml - Status: confirmed - Owner: repository-maintainers -- Last Updated (KST): 2026-05-21 00:00 -- Source Discovery / Batch: bootstrap-baseline, bat-001, bat-006, bat-008, bat-010, bat-003, bat-004, bat-005, BAT-006(dcy-004), BAT-007(dcy-004), BAT-008(dcy-004) +- Last Updated (KST): 2026-09-19 +- Source Discovery / Batch: bootstrap-baseline, bat-001, bat-006, bat-008, bat-010, bat-003, bat-004, bat-005, BAT-006(dcy-004), BAT-007(dcy-004), BAT-008(dcy-004), ProofLine req-0009 - Primary Runtime: mixed ## Project Summary @@ -407,11 +407,11 @@ ### Endpoint: `POST /v1/portfolios/{portfolio_id}/ledger/trades` (REQ-029, BAT-011; REQ-033 column rename BAT-015) - Auth: DB service token with `trade-write` permission; `daemon_id IS NULL` → HTTP 403 fail-closed -- Request: `LedgerTradeRequest` — `order_id`, `execution_id`, `side` (buy|sell), `quantity`, `price`, `fee_amount`, `tax_amount`, `executed_at` -- Response: HTTP 201 + `LedgerTradeResponse` with `status="new"`; HTTP 200 + `status="duplicate"` for idempotent replay; HTTP 409 for conflicting payload -- `LedgerTradeResponse` fields: `ledger_entry_id`, `portfolio_id` (renamed from `asset_id` by REQ-033 / dcy-005 — schema and DB column now match in meaning), `order_id`, `execution_id`, `side`, `quantity`, `price`, `fee_amount`, `tax_amount`, `executed_at`, `recorded_at`, `status` -- Error: HTTP 403 (no daemon_id on token, portfolio unassigned, daemon mismatch); HTTP 404 (non-existent portfolio); HTTP 409 (disabled portfolio or payload conflict) -- Idempotency key: `(portfolio_id, order_id, execution_id)` — reuses `uq_ledger_trades_idempotency` constraint +- Request: `LedgerTradeRequest` — `order_id`, `execution_id`, `side` (buy|sell), `quantity`, `price`, `fee_amount`, `tax_amount`, `executed_at`, optional `execution_venue` (`KRX|NXT|null`), optional `order_venue` (`KRX|NXT|SOR|null`). `SOR` is routing provenance only and is rejected as `execution_venue`; the service never infers actual execution venue from routing venue or time. +- Response: HTTP 201 + `LedgerTradeResponse` with `status="new"`; HTTP 200 + `status="duplicate"` for idempotent replay; both responses and broker items from `GET /v1/portfolios/{portfolio_id}/ledger` return persisted `execution_venue` and `order_venue`; HTTP 409 for conflicting payload +- `LedgerTradeResponse` fields: `ledger_entry_id`, `portfolio_id` (renamed from `asset_id` by REQ-033 / dcy-005 — schema and DB column now match in meaning), `order_id`, `execution_id`, `side`, `quantity`, `price`, `fee_amount`, `tax_amount`, `executed_at`, `execution_venue`, `order_venue`, `recorded_at`, `status` +- Error: HTTP 403 (no daemon_id on token, portfolio unassigned, daemon mismatch); HTTP 404 (non-existent portfolio); HTTP 409 (disabled portfolio, economic-payload conflict, or conflicting non-null venue); HTTP 422 (unsupported venue or unknown request field) +- Idempotency key: `(portfolio_id, order_id, execution_id)` — reuses `uq_ledger_trades_idempotency` constraint. Equal economic payloads may atomically enrich only NULL venue metadata; incoming NULL never erases persisted metadata, conflicting non-null values return HTTP 409, and metadata enrichment never reapplies accounting. - **구 path `POST /v1/assets/{asset_id}/ledger/trades`**: BAT-011에서 제거됨. 호출 시 HTTP 404. ### Endpoint: `POST /v1/portfolios/{portfolio_id}/ledger/manual` (REQ-056, BAT-024) diff --git a/scripts/image_release.py b/scripts/image_release.py index 56729a4..9877966 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 = 31 -EXPECTED_ALEMBIC_HEAD = "20260913_0031" +EXPECTED_MIGRATION_COUNT = 32 +EXPECTED_ALEMBIC_HEAD = "20260919_0032" @dataclass(frozen=True) class Result: diff --git a/tests/release/test_image_release.py b/tests/release/test_image_release.py index f07cfc0..ae28afc 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("20260913_0031\n") - if command[:3] == ["docker", "run", "--rm"] and "health-check" in command: return image_release.Result('{"database_revision":"20260913_0031"}\n') + if command[:3] == ["docker", "run", "--rm"] and "alembic_version" in command[-1]: return image_release.Result("20260919_0032\n") + if command[:3] == ["docker", "run", "--rm"] and "health-check" in command: return image_release.Result('{"database_revision":"20260919_0032"}\n') return image_release.Result() def artifacts(tmp_path: Path): @@ -74,12 +74,16 @@ 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", "20260913_0031_corporate_actions.py"] + + [ + "20260820_0030_portfolio_ledger_keyset_index.py", + "20260913_0031_corporate_actions.py", + "20260919_0032_ledger_venue_provenance.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 "20260913_0031" in runs[-1][-1] + assert "migration_inventory" in runs[-1][-1] and "20260919_0032" 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 37aaa63..0989152 100644 --- a/tests/release/test_release_contracts.py +++ b/tests/release/test_release_contracts.py @@ -54,7 +54,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_has_current_head_0031(): +def test_release_version_and_migration_contract_has_current_head_0032(): project = (REPO / "pyproject.toml").read_text() config = (REPO / "app/core/config.py").read_text() migrations = sorted( @@ -64,8 +64,8 @@ def test_release_version_and_migration_contract_has_current_head_0031(): ) assert 'version = "0.3.4"' in project assert 'app_version: str = "0.3.4"' in config - assert len(migrations) == 31 - assert migrations[-1].startswith("20260913_0031_") + assert len(migrations) == 32 + assert migrations[-1].startswith("20260919_0032_") def test_dockerignore_is_deny_by_default_required_inputs_only(): lines=(REPO/".dockerignore").read_text().splitlines(); assert lines[0]=="**" diff --git a/tests/test_bat025_portfolio_ledger_manual_metadata_migration.py b/tests/test_bat025_portfolio_ledger_manual_metadata_migration.py index 889aeb8..8b6bf62 100644 --- a/tests/test_bat025_portfolio_ledger_manual_metadata_migration.py +++ b/tests/test_bat025_portfolio_ledger_manual_metadata_migration.py @@ -34,7 +34,7 @@ from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioReposi _LOCAL_DSN_ENV = "BAT025_LOCAL_POSTGRES_DSN" _LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"} -_HEAD = "20260715_0028" +_HEAD = "20260919_0032" _PRE_BAT025 = "20260628_0027" @@ -138,6 +138,26 @@ def _metadata_columns(settings: Settings) -> dict[str, tuple[str, str, str | Non return {row[0]: (row[1], row[2], row[3]) for row in cur.fetchall()} +def _venue_columns(settings: Settings) -> dict[str, tuple[str, str, str | None]]: + with psycopg.connect( + host=settings.active_db_host, + port=settings.active_db_port, + dbname=settings.active_db_name, + user=settings.active_db_user, + password=settings.active_db_password, + sslmode=settings.active_db_sslmode, + ) as conn, conn.cursor() as cur: + cur.execute( + """ + SELECT column_name, data_type, is_nullable, column_default + FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'portfolio_ledger' + AND column_name IN ('execution_venue', 'order_venue') + """ + ) + return {row[0]: (row[1], row[2], row[3]) for row in cur.fetchall()} + + def _seed_portfolio(repo: PostgresPortfolioRepository, suffix: str) -> str: portfolio_id = f"bat025-{suffix}" user_id = f"bat025-user-{suffix}" @@ -215,14 +235,79 @@ def test_upgrade_from_pre_0028_preserves_business_data_and_backfills_source( "confirmation_evidence": ("text", "YES", None), "actor_user_id": ("text", "YES", None), } + assert _venue_columns(isolated_settings) == { + "execution_venue": ("text", "YES", None), + "order_venue": ("text", "YES", None), + } with repo._connect().cursor() as cur: cur.execute( - "SELECT COUNT(*), side, price::text, quantity::text, note, source " - "FROM portfolio_ledger GROUP BY side, price, quantity, note, source" + "SELECT COUNT(*), side, price::text, quantity::text, note, source, " + "execution_venue, order_venue FROM portfolio_ledger " + "GROUP BY side, price, quantity, note, source, execution_venue, order_venue" ) - count, side, price, quantity, note, source = cur.fetchone() + row = cur.fetchone() + assert row is not None + count, side, price, quantity, note, source, execution_venue, order_venue = row assert count == before_count assert (side, price, quantity, note, source) == ("buy", "123.45000000", "7.00000000", "preserve-me", "broker") + assert (execution_venue, order_venue) == (None, None) + + with pytest.raises(psycopg.errors.CheckViolation), repo._connect().cursor() as cur: + cur.execute( + "UPDATE portfolio_ledger SET execution_venue = 'SOR' WHERE portfolio_id = %s", + (portfolio_id,), + ) + + upgrade_to_head(isolated_settings) + assert _current_revision(isolated_settings) == _HEAD + repo.close() + + +def test_venue_migration_downgrade_removes_only_new_columns( + isolated_settings: Settings, +) -> None: + from alembic import command + + upgrade_to_head(isolated_settings) + repo = _repo(isolated_settings) + portfolio_id = _seed_portfolio(repo, "venue-downgrade") + recorded_at = datetime(2026, 9, 19, tzinfo=UTC) + with repo._connect().cursor() as cur: + cur.execute( + """ + INSERT INTO portfolio_ledger ( + portfolio_id, recorded_at, side, price, quantity, + cash_balance_after, asset_quantity_after, total_value_after, + strategy_set_id, note, executed_at, + execution_venue, order_venue + ) VALUES (%s, %s, 'buy', 100, 1, 900, 1, 1000, + 'downgrade', 'preserve-me', %s, 'NXT', 'SOR') + """, + (portfolio_id, recorded_at, recorded_at), + ) + + command.downgrade(get_alembic_config(isolated_settings), "20260913_0031") + + assert _current_revision(isolated_settings) == "20260913_0031" + assert _venue_columns(isolated_settings) == {} + with repo._connect().cursor() as cur: + cur.execute( + "SELECT COUNT(*), MIN(note) FROM portfolio_ledger WHERE portfolio_id = %s", + (portfolio_id,), + ) + assert cur.fetchone() == (1, "preserve-me") + + upgrade_to_head(isolated_settings) + assert _venue_columns(isolated_settings) == { + "execution_venue": ("text", "YES", None), + "order_venue": ("text", "YES", None), + } + with repo._connect().cursor() as cur: + cur.execute( + "SELECT execution_venue, order_venue FROM portfolio_ledger WHERE portfolio_id = %s", + (portfolio_id,), + ) + assert cur.fetchone() == (None, None) repo.close() diff --git a/tests/test_daemon_assignment.py b/tests/test_daemon_assignment.py index 1723591..a68d918 100644 --- a/tests/test_daemon_assignment.py +++ b/tests/test_daemon_assignment.py @@ -230,6 +230,8 @@ def test_ledger_trade_201_when_daemon_matches(monkeypatch) -> None: trade.price = 50000.0 trade.fee_amount = 0.0 trade.tax_amount = 0.0 + trade.execution_venue = None + trade.order_venue = None trade.executed_at = datetime(2026, 5, 13, 1, 0, 0, tzinfo=timezone.utc) trade.recorded_at = datetime(2026, 5, 13, 1, 0, 0, tzinfo=timezone.utc) mock_repo, client = _setup_ledger_mocks(monkeypatch, principal, portfolio, trade) diff --git a/tests/test_ledger_consolidation_integration.py b/tests/test_ledger_consolidation_integration.py index 441a312..0c535f3 100644 --- a/tests/test_ledger_consolidation_integration.py +++ b/tests/test_ledger_consolidation_integration.py @@ -12,10 +12,13 @@ from __future__ import annotations import os import uuid +from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone +from threading import Barrier import pytest +from app.repositories.portfolio_repository import LedgerTradeConflictError from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioRepository @@ -133,6 +136,182 @@ def test_duplicate_broker_write_is_idempotent(repo, seeded_portfolio): assert cur.fetchone()[0] == 1 +def test_duplicate_venue_enrichment_is_metadata_only(repo, seeded_portfolio): + args = { + "portfolio_id": seeded_portfolio, + "order_id": "ord-venue-enrich", + "execution_id": "exec-venue-enrich", + "side": "buy", + "quantity": 5.0, + "price": 10_000.0, + "executed_at": _now(), + } + _first, created = repo.insert_ledger_trade(**args) + assert created is True + before = repo.get(seeded_portfolio) + + enriched, created = repo.insert_ledger_trade( + **args, + execution_venue="NXT", + order_venue="SOR", + ) + + assert created is False + assert enriched.execution_venue == "NXT" + assert enriched.order_venue == "SOR" + after = repo.get(seeded_portfolio) + assert after.cash_balance == before.cash_balance + assert after.asset_quantity == before.asset_quantity + with repo._connect().cursor() as cur: + cur.execute( + """ + SELECT COUNT(*), MIN(execution_venue), MIN(order_venue) + FROM portfolio_ledger + WHERE portfolio_id = %s AND order_id = %s AND execution_id = %s + """, + (seeded_portfolio, args["order_id"], args["execution_id"]), + ) + assert cur.fetchone() == (1, "NXT", "SOR") + + +def test_duplicate_venue_conflict_is_atomic(repo, seeded_portfolio): + args = { + "portfolio_id": seeded_portfolio, + "order_id": "ord-venue-conflict", + "execution_id": "exec-venue-conflict", + "side": "buy", + "quantity": 1.0, + "price": 10_000.0, + "executed_at": _now(), + } + repo.insert_ledger_trade(**args, execution_venue="KRX") + before = repo.get(seeded_portfolio) + + with pytest.raises(LedgerTradeConflictError): + repo.insert_ledger_trade( + **args, + execution_venue="NXT", + order_venue="SOR", + ) + + after = repo.get(seeded_portfolio) + assert after.cash_balance == before.cash_balance + assert after.asset_quantity == before.asset_quantity + with repo._connect().cursor() as cur: + cur.execute( + """ + SELECT execution_venue, order_venue + FROM portfolio_ledger + WHERE portfolio_id = %s AND order_id = %s AND execution_id = %s + """, + (seeded_portfolio, args["order_id"], args["execution_id"]), + ) + assert cur.fetchone() == ("KRX", None) + + +def test_duplicate_null_venues_do_not_erase_existing_values(repo, seeded_portfolio): + args = { + "portfolio_id": seeded_portfolio, + "order_id": "ord-venue-preserve", + "execution_id": "exec-venue-preserve", + "side": "buy", + "quantity": 1.0, + "price": 10_000.0, + "executed_at": _now(), + } + repo.insert_ledger_trade( + **args, + execution_venue="KRX", + order_venue="SOR", + ) + + duplicate, created = repo.insert_ledger_trade(**args) + + assert created is False + assert duplicate.execution_venue == "KRX" + assert duplicate.order_venue == "SOR" + + +def test_duplicate_economic_conflict_is_rejected(repo, seeded_portfolio): + args = { + "portfolio_id": seeded_portfolio, + "order_id": "ord-economic-conflict", + "execution_id": "exec-economic-conflict", + "side": "buy", + "quantity": 1.0, + "price": 10_000.0, + "executed_at": _now(), + } + repo.insert_ledger_trade(**args) + + with pytest.raises(LedgerTradeConflictError): + repo.insert_ledger_trade(**{**args, "quantity": 2.0}, order_venue="SOR") + + with repo._connect().cursor() as cur: + cur.execute( + """ + SELECT quantity::float8, order_venue + FROM portfolio_ledger + WHERE portfolio_id = %s AND order_id = %s AND execution_id = %s + """, + (seeded_portfolio, args["order_id"], args["execution_id"]), + ) + assert cur.fetchone() == (1.0, None) + + +def test_concurrent_conflicting_venue_enrichment_has_no_last_write_wins( + repo, seeded_portfolio +): + args = { + "portfolio_id": seeded_portfolio, + "order_id": "ord-venue-concurrent", + "execution_id": "exec-venue-concurrent", + "side": "buy", + "quantity": 1.0, + "price": 10_000.0, + "executed_at": _now(), + } + repo.insert_ledger_trade(**args) + barrier = Barrier(2) + + def enrich(execution_venue: str): + worker_repo = PostgresPortfolioRepository(repo._config) + barrier.wait() + try: + entry, created = worker_repo.insert_ledger_trade( + portfolio_id=seeded_portfolio, + order_id="ord-venue-concurrent", + execution_id="exec-venue-concurrent", + side="buy", + quantity=1.0, + price=10_000.0, + executed_at=args["executed_at"], + execution_venue=execution_venue, + ) + return ("ok", created, entry.execution_venue) + except LedgerTradeConflictError: + return ("conflict", False, None) + finally: + worker_repo.close() + + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(enrich, ("KRX", "NXT"))) + + assert sorted(result[0] for result in results) == ["conflict", "ok"] + winner = next(result[2] for result in results if result[0] == "ok") + assert winner in {"KRX", "NXT"} + with repo._connect().cursor() as cur: + cur.execute( + """ + SELECT COUNT(*), MIN(execution_venue), MIN(order_venue) + FROM portfolio_ledger + WHERE portfolio_id = %s AND order_id = %s AND execution_id = %s + """, + (seeded_portfolio, args["order_id"], args["execution_id"]), + ) + assert cur.fetchone() == (1, winner, None) + + def test_negative_balance_allowed(repo, seeded_portfolio): """REQ-037 AC-3 + OQ-3: oversell produces negative asset_quantity without rejection.""" entry, created = repo.insert_ledger_trade( diff --git a/tests/test_v1_portfolios_ledger.py b/tests/test_v1_portfolios_ledger.py index e3c73b2..f5548fa 100644 --- a/tests/test_v1_portfolios_ledger.py +++ b/tests/test_v1_portfolios_ledger.py @@ -147,6 +147,97 @@ def test_ledger_trade_duplicate_same_payload() -> None: assert resp.json()["status"] == "duplicate" +def test_ledger_trade_round_trips_and_forwards_venue_provenance() -> None: + settings = _make_settings() + repo = _make_service_repo( + trade=_make_trade(execution_venue="NXT", order_venue="SOR") + ) + with _with_repo(repo): + client = _make_client(settings) + resp = client.post( + _NEW_PATH, + json=_trade_payload(execution_venue="NXT", order_venue="SOR"), + headers={"Authorization": "Bearer svc-token"}, + ) + + assert resp.status_code == 201 + assert resp.json()["execution_venue"] == "NXT" + assert resp.json()["order_venue"] == "SOR" + assert repo.insert_ledger_trade.call_args.kwargs["execution_venue"] == "NXT" + assert repo.insert_ledger_trade.call_args.kwargs["order_venue"] == "SOR" + + +def test_ledger_trade_allows_unknown_actual_venue_with_sor_route() -> None: + settings = _make_settings() + repo = _make_service_repo(trade=_make_trade(execution_venue=None, order_venue="SOR")) + with _with_repo(repo): + client = _make_client(settings) + resp = client.post( + _NEW_PATH, + json=_trade_payload(execution_venue=None, order_venue="SOR"), + headers={"Authorization": "Bearer svc-token"}, + ) + + assert resp.status_code == 201 + assert resp.json()["execution_venue"] is None + assert resp.json()["order_venue"] == "SOR" + + +def test_ledger_trade_rejects_sor_as_execution_venue() -> None: + settings = _make_settings() + repo = _make_service_repo() + with _with_repo(repo): + client = _make_client(settings) + resp = client.post( + _NEW_PATH, + json=_trade_payload(execution_venue="SOR", order_venue="SOR"), + headers={"Authorization": "Bearer svc-token"}, + ) + + assert resp.status_code == 422 + repo.insert_ledger_trade.assert_not_called() + + +def test_ledger_trade_rejects_unknown_request_fields() -> None: + settings = _make_settings() + repo = _make_service_repo() + with _with_repo(repo): + client = _make_client(settings) + resp = client.post( + _NEW_PATH, + json=_trade_payload(broker_ordered_at="2026-05-01T09:59:59+09:00"), + headers={"Authorization": "Bearer svc-token"}, + ) + + assert resp.status_code == 422 + repo.insert_ledger_trade.assert_not_called() + + +def test_ledger_trade_openapi_exposes_nullable_venue_enums() -> None: + settings = _make_settings() + repo = _make_service_repo() + with _with_repo(repo): + client = _make_client(settings) + response = client.get("/openapi.json") + + assert response.status_code == 200 + properties = response.json()["components"]["schemas"]["LedgerTradeRequest"][ + "properties" + ] + execution_options = properties["execution_venue"]["anyOf"] + order_options = properties["order_venue"]["anyOf"] + assert {tuple(option.get("enum", ())) for option in execution_options} == { + (), + ("KRX", "NXT"), + } + assert {tuple(option.get("enum", ())) for option in order_options} == { + (), + ("KRX", "NXT", "SOR"), + } + assert {option.get("type") for option in execution_options} == {"string", "null"} + assert {option.get("type") for option in order_options} == {"string", "null"} + + # --------------------------------------------------------------------------- # AC-3: same key + different payload → 409 # --------------------------------------------------------------------------- diff --git a/tests/test_v1_portfolios_manual_ledger.py b/tests/test_v1_portfolios_manual_ledger.py index 03af9f0..a38f2f4 100644 --- a/tests/test_v1_portfolios_manual_ledger.py +++ b/tests/test_v1_portfolios_manual_ledger.py @@ -1030,6 +1030,8 @@ def test_unified_ledger_includes_corporate_action_fields() -> None: "evidence_ref": "notice://ca-7", "actor_user_id": _USER_ID, "recorded_at": "2026-09-02T00:00:00Z", + "execution_venue": None, + "order_venue": None, } @@ -1094,6 +1096,7 @@ def test_unified_ledger_paginates_mixed_sources_and_hides_side() -> None: LedgerEntry( ledger_entry_id=99, portfolio_id=_PORTFOLIO_ID, side="buy", quantity=2, price=10, source="broker", recorded_at=datetime.now(UTC), cash_balance_after=900, asset_quantity_after=12, + execution_venue="NXT", order_venue="SOR", ) ) first = client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger?limit=1", headers=_auth()) @@ -1101,6 +1104,8 @@ def test_unified_ledger_paginates_mixed_sources_and_hides_side() -> None: first_item = first.json()["items"][0] assert first_item["source"] == "broker" assert first_item["action"] == "BUY" + assert first_item["execution_venue"] == "NXT" + assert first_item["order_venue"] == "SOR" assert "side" not in first_item assert first.json()["next_cursor"] second = client.get(