3 Commits
Author SHA1 Message Date
genichin f79f3c79b5 chore(release): prepare v0.3.4
Publish verified PortfolioHub image / verify (push) Canceled after 0s
Publish verified PortfolioHub image / publish (push) Canceled after 0s
2026-09-13 23:08:22 +09:00
genichin 8841bdd441 fix: qualify corporate-action ledger reads (#36) 2026-09-13 22:02:43 +09:00
genichin c7d8235bde fix: make corporate actions atomic and recoverable
Publish verified PortfolioHub image / verify (push) Canceled after 0s
Publish verified PortfolioHub image / publish (push) Canceled after 0s
2026-09-13 19:33:07 +09:00
26 changed files with 1005 additions and 120 deletions
+1 -1
View File
@@ -39,7 +39,7 @@ jobs:
python-version: "3.11"
- run: python -m pip install "uv==0.11.14"
- run: uv sync --extra dev --locked
- run: uv run pytest -q
- run: uv run python -m pytest -q
- name: Honest scoped lint
run: uv run ruff check --select F,I app/db/migrations.py scripts/image_release.py scripts/ruff_ratchet.py tests/release
- name: Reject Ruff findings introduced by v0.3.0 or later patches
+22 -2
View File
@@ -11,6 +11,20 @@ permissions:
jobs:
verify:
runs-on: ubuntu-latest
services:
timescaledb:
image: timescale/timescaledb@sha256:252a443e2936039b83dd8da1373d01e59e932d1054fa6adf1bc061f1d56ae60a
env:
POSTGRES_DB: portfolio_hub
POSTGRES_USER: portfolio_hub_user
POSTGRES_PASSWORD: change-me
ports:
- 5432:5432
options: >-
--health-cmd "pg_isready -U portfolio_hub_user -d portfolio_hub"
--health-interval 5s
--health-timeout 5s
--health-retries 12
permissions:
contents: read
steps:
@@ -22,7 +36,13 @@ jobs:
python-version: "3.11"
- run: python -m pip install "uv==0.11.14"
- run: uv sync --extra dev --locked
- run: uv run pytest -q
- run: uv run python -m pytest -q
- name: Verify corporate-action atomicity and guarded recovery in PostgreSQL
env:
POSTGRES_PASSWORD: change-me
run: |
export PORTFOLIOHUB_TEST_POSTGRES_DSN="postgresql://portfolio_hub_user:${POSTGRES_PASSWORD}@127.0.0.1:5432/postgres"
uv run python -m pytest -q tests/test_corporate_action_atomic_recovery_postgres.py
- name: Honest scoped lint for release additions
run: uv run ruff check --select F,I app/db/migrations.py scripts/image_release.py scripts/ruff_ratchet.py tests/release
- name: Reject Ruff findings introduced by v0.3.0 or later patches
@@ -31,7 +51,7 @@ jobs:
uv run python scripts/ruff_ratchet.py --base v0.3.0
uv run python scripts/ruff_ratchet.py --base v0.3.1
- name: Validate container contracts without registry writes
run: uv run pytest -q tests/release
run: uv run python -m pytest -q tests/release
- name: Build and smoke exact pull-request or main source image
run: |
uv run python -c 'from app.db.migrations import migration_inventory; print("\n".join(migration_inventory()))' > migration-inventory.txt
+16
View File
@@ -0,0 +1,16 @@
---
id: "ac-0037"
status: active
---
# 기업행동 원장이 포함된 joined ledger read 안정성
## Criterion
Portfolio owner의 holdings cost-basis와 unified ledger 조회는 같은 portfolio에 기업행동 원장과 일반 거래 원장이 함께 있어도 PostgreSQL column ambiguity 없이 기존 ledger·기업행동 사실을 읽어야 한다. Holdings path는 기존 WAC·실현손익·보유수량·원가 의미와 corporate-action replay mapping을 보존하고, unified ledger path는 기존 latest-first ordering·source filter·keyset cursor·canonical response 의미를 보존해야 한다.
## Verification
- 성공 조건: 실제 PostgreSQL에 일반 BUY/SELL과 `stock_consolidation` 기업행동을 함께 구성했을 때 `get_all_trades_for_pnl()`과 `list_ledger_entries()`가 모두 성공하고 기업행동 metadata를 해당 ledger entry에 결합한다. 같은 fixture의 `GET /v1/portfolios/{portfolio_id}/holdings/cost-basis`와 `GET /v1/portfolios/{portfolio_id}/ledger`가 HTTP 200을 반환하며 unified ledger의 기본 조회·source filter·다음 cursor page가 기존 정렬과 행 구성을 유지한다.
- 실패·경계 조건: JOIN으로 `ledger_entry_id`·`recorded_at` 등 base column이 ambiguous해지거나, SELECT 순서·positional mapping이 바뀌거나, 기업행동이 없는 기존 거래 replay·회계 결과·authorization·API schema·pagination이 달라지면 FAIL이다.
- 검사 방법: 실제 PostgreSQL repository integration과 holdings/unified-ledger API 회귀를 실행하고 `app/repositories/postgres.py`의 joined projection·predicate·ordering qualification 및 반환 mapping을 검토한다.
+2 -2
View File
@@ -1,5 +1,5 @@
{
"schema_version": 1,
"next_line_number": 7,
"next_ac_number": 37
"next_line_number": 8,
"next_ac_number": 38
}
+48
View File
@@ -0,0 +1,48 @@
---
id: "dcy-0007"
status: confirmed
---
# 기업행동 원장이 포함된 보유 원가·전체 원장 조회 복구 (#19)
## Problem
기업행동 원장이 존재하는 포트폴리오에서 `GET /v1/portfolios/{portfolio_id}/holdings/cost-basis`와 `GET /v1/portfolios/{portfolio_id}/ledger?limit=50`가 PostgreSQL ambiguous-column 오류로 HTTP 500을 반환한다. 기업행동 metadata를 읽기 위해 `portfolio_ledger`와 `portfolio_corporate_actions`를 JOIN한 뒤 base ledger 컬럼 일부를 table alias 없이 참조한 것이 원인이다. 그 결과 보유 원가 replay와 전체 원장 조회라는 서로 다른 read path가 함께 중단되고, downstream인 trex-web 포트폴리오 상세에서 보유 정보와 원장 정보를 표시할 수 없다.
## Evidence
- `E-001` Source issue `https://github.com/fourblocks2020/trex-web/issues/19`는 두 API의 HTTP 500과 `psycopg.errors.AmbiguousColumn: column reference "ledger_entry_id" is ambiguous` 운영 관찰을 기록한다. 확인 시각의 issue body SHA-256은 `8e382d87dd3358ac516cdd8e5c683f4fa673253c6eb7fc4d0699ed572378ad5b`, `updated_at`은 `2026-09-13T11:28:56Z`다.
- `E-002` 기준 source는 `portfolio-hub` `c7d8235bdebfa8c2459b0b3300d8ed40a9f465e5`다. `app/repositories/postgres.py`의 `get_all_trades_for_pnl()`은 `portfolio_ledger AS ledger`와 `portfolio_corporate_actions ca`를 JOIN하지만 첫 18개 base projection의 `ledger_entry_id`, `portfolio_id`, `recorded_at` 등을 alias 없이 선택한다.
- `E-003` 같은 파일의 `list_ledger_entries()`는 두 table을 JOIN하면서 base projection, 동적 `WHERE`의 `portfolio_id`·`source`·cursor tuple, `ORDER BY recorded_at, ledger_entry_id`를 alias 없이 참조한다. 현재 source에서 기업행동 table을 JOIN하는 repository query는 이 두 곳뿐이다.
- `E-004` `.proofline/lines/line-0006/req-0006.md`는 approved 상태이며 unified ledger read를 소유한다. Active `AC-0032`는 source-neutral latest-first 조회, `AC-0034`는 source별 canonical representation, `AC-0036`은 PostgreSQL query 회귀를 요구한다. 보유 원가/PnL replay read는 현재 active AC가 직접 소유하지 않는다.
- `E-005` `tests/test_ledger_consolidation_integration.py`는 실제 PostgreSQL repository path를 실행하지만 기업행동 행과 일반 거래 행을 함께 만든 뒤 `get_all_trades_for_pnl()`과 `list_ledger_entries()`를 호출하는 회귀는 없다. Mock 기반 테스트만으로는 PostgreSQL column resolution 오류를 검출할 수 없다.
## Scope
- `S-001` `get_all_trades_for_pnl()`의 모든 base ledger projection을 `ledger.<column>`으로 한정하여 기업행동 원장이 포함된 포트폴리오의 보유 원가 replay가 정상 동작하게 한다. 이 동작은 기존 active AC가 직접 소유하지 않으므로 지속적인 독립 검증이 가능한 AC를 `create`한다.
- `S-002` `list_ledger_entries()`의 base ledger projection, dynamic predicate, keyset cursor tuple과 ordering을 `ledger.<column>`으로 한정하고 `portfolio_ledger AS ledger`를 일관되게 사용한다. Unified ledger의 응답 구조·정렬·cursor·권한 의미를 변경하지 않고 기존 `AC-0032`, `AC-0034`, `AC-0036`을 `satisfy`한다.
- `S-003` 실제 PostgreSQL에서 기업행동 원장과 일반 거래 원장이 공존하는 fixture로 두 repository read를 실행한다. Holdings cost-basis API와 unified ledger API가 HTTP 200을 반환하고 기업행동 metadata 및 일반 거래 행을 기존 schema대로 보존하는 focused API/repository 회귀를 둔다.
- `S-004` 변경은 ambiguous-column을 제거하는 query qualification과 직접 회귀에 한정한다. 회계 수치, WAC·replay 규칙, source/action 의미, authorization, pagination 또는 API schema를 바꾸지 않는다.
## Out of Scope
- `OS-001` 기업행동 원장의 생성·수정·삭제, reversal 또는 recovery 동작 변경.
- `OS-002` broker 주문·취소·체결 처리, producer/outbox 변경 또는 기존 원장 row mutation/backfill.
- `OS-003` holdings cost basis, PnL, corporate-action 회계 산식의 변경이나 임의 보정.
- `OS-004` trex-web UI 변경, 배포·release·production mutation 및 특정 production portfolio 데이터 수정.
## Risks and Unknowns
### Risks
- `R-001` SELECT column 순서가 `_row_to_ledger_entry()`와 corporate-action metadata positional mapping에 결속돼 있다. Qualification 과정에서 순서를 바꾸면 값 mapping이 깨질 수 있으므로 기존 순서를 보존하고 반환 객체를 검증한다.
- `R-002` mock cursor 테스트는 PostgreSQL의 ambiguous-column 판정을 재현하지 못한다. 실제 PostgreSQL 회귀가 없으면 같은 결함이 다시 유입될 수 있으므로 disposable/local integration을 필수 검증 경계로 둔다.
- `R-003` unified ledger의 dynamic keyset predicate qualification이 누락되면 첫 page는 통과해도 source filter 또는 다음 page가 실패할 수 있다. 기본 조회, source filter와 cursor 경계를 함께 검증한다.
### Assumptions
- `A-001` Issue #19에 기록된 운영 오류와 현재 source의 unqualified JOIN query가 같은 결함이라는 판단은 SQL text와 PostgreSQL column-name 충돌에 근거한다. 실제 PostgreSQL pre-fix 회귀로 이를 확인한다.
### Open Questions
없음. Source issue가 query qualification, 유지할 계약, 회귀 범위와 제외 범위를 구체적으로 정했고 현재 source에서 추가 제품·정책 결정을 요구하는 항목은 발견되지 않았다.
@@ -0,0 +1,6 @@
# Activity Log
- 2026-09-13T21:26:26+09:00 — 사용자가 `REQ-0007`, 신규 `AC-0037`, 기존 `AC-0032`·`AC-0034`·`AC-0036` satisfy 집합을 승인했다. `REQ-0007`은 `approved`, `AC-0037`은 `active`로 전환했고 `proofline validate`가 기존 line status warning만 남기고 통과했다.
- 2026-09-13T21:26:26+09:00 — 직접 구현 branch `fix/issue-19-corporate-action-reads`를 생성했다. 계획: 실제 PostgreSQL 실패 회귀 추가 → joined query base column qualification → focused repository/API 회귀 → 전체 회귀·정적 검사 → 독립 QC → commit/PR.
- 2026-09-13T21:40:00+09:00 — `test_joined_ledger_reads_qualify_base_columns` 실제 PostgreSQL 회귀를 추가하고 joined PnL/unified-ledger SQL의 ledger projection·predicate·cursor/order reference를 `ledger.`로 최소 한정했다. API holdings/ledger/PNL focused suite는 50 passed; Ruff는 기존 SIM117 4건만 보고했다.
- 2026-09-13T21:44:31+09:00 — 로컬 TimescaleDB를 기동해 실제 PostgreSQL에서 unqualified JOIN이 SQLSTATE `42702` (`AmbiguousColumn`)를 내는 RED와 qualified JOIN 성공을 확인했다. Mixed BUY+`stock_consolidation` repository 회귀는 source filter·keyset cursor를 포함해 통과했고, 두 API corporate-action 회귀까지 합쳐 `3 passed`였다.
@@ -0,0 +1,35 @@
# AC-0037 Test Report
- 실행 시각: 2026-09-13T21:53:39+09:00
- 기준 Git HEAD: `c7d8235bdebfa8c2459b0b3300d8ed40a9f465e5`
- 검증 대상: 위 HEAD의 `fix/issue-19-corporate-action-reads` working-tree diff
- 실제 backend: `timescale/timescaledb-ha:pg17` (`docker-compose.local-test.yml`, PostgreSQL DSN은 기록하지 않음)
- 결과: PASS
## AC mapping
- Joined projection·metadata mapping·기존 BUY 동작: `tests/test_corporate_action_atomic_recovery_postgres.py::test_joined_ledger_reads_qualify_base_columns`
- Unified ledger source filter·latest-first·keyset cursor: 같은 실제 PostgreSQL test의 `source="corporate_action"` 및 `before_recorded_at`/`before_ledger_entry_id` assertions
- Holdings HTTP 200 및 기존 WAC/total cost basis: `tests/test_v1_portfolios_holdings.py::test_corporate_action_row_keeps_holdings_cost_basis_http_200`
- Unified ledger HTTP 200 및 corporate-action response schema: `tests/test_v1_portfolios_manual_ledger.py::test_unified_ledger_includes_corporate_action_fields`
- Positional mapping drift guard: `tests/test_issue34_regressions.py::test_runtime_schema_has_one_corporate_action_ddl_and_replay_mapping_starts_at_18`
- Authorization·기존 API schema·기업행동 없는 경로: 전체 repository suite
## Executed evidence
1. Unqualified query probe on real PostgreSQL: SQLSTATE `42702`, `AmbiguousColumn` (RED).
2. Same JOIN with `ledger.ledger_entry_id`: success (GREEN).
3. Real PostgreSQL repository regression plus two API regressions: `3 passed, 1 warning`.
4. Full suite with real-PostgreSQL selector enabled: `711 passed, 25 skipped, 17 warnings`.
5. Python compile, `proofline validate`, and `git diff --check`: PASS. ProofLine emitted only pre-existing line-0001–0006 status warnings.
6. Ruff: no finding was introduced by the changed lines; repository has four pre-existing `SIM117` findings in `app/repositories/postgres.py` and pre-existing `UP017` findings in `tests/test_v1_portfolios_holdings.py`.
## Artifact identities
- `app/repositories/postgres.py`: `2fa5ecbdc29a41bbc1379069b15e7cf43c7bf62cf8f00a3fa824ff8c182f410e`
- `tests/test_corporate_action_atomic_recovery_postgres.py`: `0d0c436efe443b53c11b792885ca17e983320afe70d13122c7b89d3d73cf0cdd`
- `tests/test_v1_portfolios_holdings.py`: `daff8ce2bec9f161e1a399d9bb3bea3cbf97707c6d8805cfedb66900b7b036e2`
- `tests/test_v1_portfolios_manual_ledger.py`: `45bb77f66708b0b76e133031b3f35ac06714558c9603ec56c1b642906ca2e681`
- `tests/test_issue34_regressions.py`: `661742e430522af58ebc15a8f97c8385e1f65126bf71a2783716c269a76ac45f`
- `.proofline/lines/line-0007/req-0007.md`: `a3a6ae501f7635cc80a5a96416e61aa3ecce326a732126feba3b9ad1a7c8df73`
- `.proofline/criteria/ac-0037.md`: `99dab6c2d70367c236dc08a65b24d6ddf94e131500e06cd52e9ab89784708d2d`
+4
View File
@@ -0,0 +1,4 @@
---
id: "line-0007"
status: discovery
---
+40
View File
@@ -0,0 +1,40 @@
---
id: "req-0007"
status: approved
discovery: "dcy-0007"
criteria:
create:
- "ac-0037"
update: []
retire: []
satisfy:
- "ac-0032"
- "ac-0034"
- "ac-0036"
---
# 기업행동 원장이 포함된 보유 원가·전체 원장 조회 복구 (#19)
## Objective
기업행동 metadata를 JOIN하는 포트폴리오 read query가 PostgreSQL column ambiguity 없이 기존 보유 원가 replay와 source-neutral 전체 원장 계약을 제공하도록 복구한다.
## Scope
- 기업행동 원장이 있는 포트폴리오의 holdings cost-basis path가 모든 거래와 기업행동 metadata를 기존 순서·의미로 읽어 원가 replay 결과를 반환한다.
- `GET /v1/portfolios/{portfolio_id}/ledger`가 기업행동 행과 일반 거래 행을 함께 기존 latest-first·source filter·keyset pagination 계약으로 반환한다.
- Joined query의 base ledger column reference를 table alias로 한정하고 corporate-action metadata projection은 명시적으로 유지한다.
- 실제 PostgreSQL repository 회귀와 API 회귀로 holdings cost-basis 및 unified ledger HTTP 200, 혼합 행, metadata mapping과 cursor 경계를 검증한다.
## Constraints
- SELECT column 순서와 `_row_to_ledger_entry()` positional mapping을 유지한다.
- API endpoint, response schema, authorization, source/action 의미, ordering 및 cursor 형식을 변경하지 않는다.
- Holdings cost basis, WAC, PnL 또는 corporate-action replay 회계 규칙을 변경하지 않는다.
- 기존 active `AC-0032`, `AC-0034`, `AC-0036`의 의미를 변경하지 않고 satisfy binding으로 적용한다.
## Non-Goals
- 기업행동 또는 일반 ledger row의 생성·수정·삭제·backfill.
- broker 주문·취소·체결, producer/outbox 또는 reversal/recovery 동작 변경.
- trex-web UI 구현, 배포, release, production mutation 또는 특정 production portfolio 데이터 보정.
+7 -1
View File
@@ -1338,7 +1338,13 @@ def create_corporate_action(
digest = canonical_payload_hash(payload)
# The repository derives all WAC and immutable snapshot fields under its row lock.
action = CorporateAction(str(uuid.uuid4()), portfolio_id, portfolio.asset_id, payload.action_type, payload.effective_date, payload.quantity_before, payload.quantity_after, Decimal(0), Decimal(0), Decimal(0), Decimal(str(portfolio.cash_balance)), Decimal(str(portfolio.profit_buffer)), payload.evidence_ref, idempotency_key, digest, 0, auth.user_id)
saved = repo.record_corporate_action(portfolio_id, action=action, idempotency_key=idempotency_key, payload_hash=digest)
saved = repo.record_corporate_action(
portfolio_id,
action=action,
idempotency_key=idempotency_key,
payload_hash=digest,
resume_partial=payload.resume_partial,
)
return CorporateActionResponse(**saved.__dict__)
except ValueError as exc:
code = str(exc)
+13 -2
View File
@@ -285,6 +285,11 @@ def corporate_action_portfolio(
quantity_after: float = typer.Option(..., "--quantity-after", min=0.0000001),
evidence_ref: str = typer.Option(..., "--evidence-ref"),
idempotency_key: str = typer.Option(..., "--idempotency-key"),
resume_partial: bool = typer.Option(
False,
"--resume-partial",
help="Resume only a verified quantity-only partial corporate action.",
),
force: bool = typer.Option(False, "--force"),
) -> None:
"""Record an auditable stock consolidation without a trade."""
@@ -314,11 +319,17 @@ def corporate_action_portfolio(
portfolio = repository.get(portfolio_id)
if not idempotency_key.strip():
raise ValueError("Idempotency-Key must not be empty")
payload = {"action_type": action_type, "effective_date": day.isoformat(), "quantity_before": str(quantity_before), "quantity_after": str(quantity_after), "evidence_ref": evidence_ref, "confirm_daemon_managed": False}
payload = {"action_type": action_type, "effective_date": day.isoformat(), "quantity_before": str(quantity_before), "quantity_after": str(quantity_after), "evidence_ref": evidence_ref, "confirm_daemon_managed": False, "resume_partial": resume_partial}
digest = canonical_payload_hash(payload)
calc = calculate_stock_consolidation(Decimal(str(quantity_before)), Decimal(str(quantity_after)), Decimal(0))
action = CorporateAction(str(uuid.uuid4()), portfolio_id, portfolio.asset_id, action_type, day, calc.quantity_before, calc.quantity_after, calc.average_price_before, calc.average_price_after, calc.total_cost_basis, Decimal(str(portfolio.cash_balance)), Decimal(str(portfolio.profit_buffer)), evidence_ref, idempotency_key, digest, 0, user)
return repository.record_corporate_action(portfolio_id, action=action, idempotency_key=idempotency_key, payload_hash=digest)
return repository.record_corporate_action(
portfolio_id,
action=action,
idempotency_key=idempotency_key,
payload_hash=digest,
resume_partial=resume_partial,
)
detail = _with_service_and_repo(ctx, _callback)
render_output(ctx.obj.console, ctx.obj.output_format, detail)
+1 -1
View File
@@ -18,7 +18,7 @@ if env_file.exists():
class Settings(BaseSettings):
app_name: str = "PortfolioHub"
app_version: str = "0.3.2"
app_version: str = "0.3.4"
environment: str = "development"
allowed_origins_raw: str = Field(default="http://localhost:3000", validation_alias="PORTFOLIOHUB_ALLOWED_ORIGINS")
bootstrap_username: str = Field(default="", validation_alias="PORTFOLIOHUB_BOOTSTRAP_USERNAME")
+6 -1
View File
@@ -16,7 +16,12 @@ def quantize_stored_decimal(value: Decimal) -> Decimal:
def canonical_payload_hash(payload: object) -> str:
if hasattr(payload, "model_dump"):
payload = payload.model_dump(mode="json", exclude_none=True)
excluded_fields = (
{"resume_partial"} if not getattr(payload, "resume_partial", False) else set()
)
payload = payload.model_dump(mode="json", exclude_none=True, exclude=excluded_fields)
elif isinstance(payload, dict) and payload.get("resume_partial") is False:
payload = {key: value for key, value in payload.items() if key != "resume_partial"}
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
return hashlib.sha256(encoded).hexdigest()
+9 -1
View File
@@ -284,7 +284,15 @@ class PortfolioRepository(Protocol):
note: str = "",
) -> None: ...
def record_corporate_action(self, portfolio_id: str, *, action: CorporateAction, idempotency_key: str, payload_hash: str) -> CorporateAction: ...
def record_corporate_action(
self,
portfolio_id: str,
*,
action: CorporateAction,
idempotency_key: str,
payload_hash: str,
resume_partial: bool = False,
) -> CorporateAction: ...
def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]: ...
+231 -62
View File
@@ -14,7 +14,10 @@ from typing import Any
import psycopg
from psycopg import sql
from app.domain.corporate_actions import calculate_stock_consolidation
from app.domain.corporate_actions import (
calculate_stock_consolidation,
quantize_stored_decimal,
)
from app.domain.errors import (
DaemonInactiveError,
DaemonInUseError,
@@ -1629,46 +1632,155 @@ 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."""
def record_corporate_action(
self,
portfolio_id: str,
*,
action: CorporateAction,
idempotency_key: str,
payload_hash: str,
resume_partial: bool = False,
) -> CorporateAction:
"""Persist an immutable action and snapshot in one explicit transaction."""
self.ensure_schema()
with self._connect() as conn: # noqa: SIM117 - keep transaction and cursor lifetimes explicit
with conn.cursor() as cur:
cur.execute("SELECT asset_quantity, cash_balance, profit_buffer, asset_id, status 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, 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, recorded_at FROM portfolio_corporate_actions WHERE portfolio_id=%s AND idempotency_key=%s", (portfolio_id, idempotency_key))
prior = cur.fetchone()
if prior:
conn = self._connect()
def hydrate_replay(prior: tuple) -> CorporateAction:
if prior[8] != payload_hash:
raise ValueError("IDEMPOTENCY_KEY_REUSE")
(action.corporate_action_id, action.portfolio_id, action.asset_id, action.action_type,
action.effective_date, action.actor_user_id, action.evidence_ref,
action.idempotency_key, action.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,
ledger_id, action.recorded_at) = prior
(
action.corporate_action_id,
action.portfolio_id,
action.asset_id,
action.action_type,
action.effective_date,
action.actor_user_id,
action.evidence_ref,
action.idempotency_key,
action.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,
ledger_id,
action.recorded_at,
) = prior
action.corporate_action_id = str(action.corporate_action_id)
action.ledger_entry_id = int(ledger_id)
action.idempotent_replay = True
return action
prior_sql = """
SELECT 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, recorded_at
FROM portfolio_corporate_actions
WHERE portfolio_id=%s AND idempotency_key=%s
"""
# Repository connections are autocommit connections. A connection
# context alone therefore cannot protect this multi-statement write.
with conn.transaction(), conn.cursor() as cur:
cur.execute(
"""
SELECT asset_quantity, cash_balance, profit_buffer, asset_id,
status, asset_price
FROM portfolios
WHERE portfolio_id=%s
FOR UPDATE
""",
(portfolio_id,),
)
row = cur.fetchone()
if row is None:
raise ValueError("portfolio not found")
cur.execute(prior_sql, (portfolio_id, idempotency_key))
prior = cur.fetchone()
if prior:
return hydrate_replay(prior)
snapshot_quantity = Decimal(str(row[0]))
cash_balance = Decimal(str(row[1]))
profit_buffer = Decimal(str(row[2]))
reference_price_before = quantize_stored_decimal(Decimal(str(row[5])))
if row[4] != "enabled":
raise ValueError("CORPORATE_ACTION_PORTFOLIO_DISABLED")
if str(row[3]) != str(action.asset_id):
raise ValueError("CORPORATE_ACTION_ASSET_MISMATCH")
if Decimal(str(row[1])) != action.cash_balance or Decimal(str(row[2])) != action.profit_buffer:
if cash_balance != action.cash_balance or profit_buffer != action.profit_buffer:
raise ValueError("CORPORATE_ACTION_INVARIANT_MISMATCH")
cur.execute("SELECT 1 FROM portfolio_ledger WHERE portfolio_id=%s AND side IN ('buy','sell') AND executed_at::date >= %s LIMIT 1", (portfolio_id, action.effective_date))
cur.execute(
"""
SELECT 1
FROM portfolio_ledger
WHERE portfolio_id=%s
AND side IN ('buy','sell')
AND executed_at::date >= %s
LIMIT 1
""",
(portfolio_id, action.effective_date),
)
if cur.fetchone() is not None:
raise ValueError("CORPORATE_ACTION_HISTORICAL_CONFLICT")
# Derive WAC from the authoritative locked ledger, never portfolios.price.
cur.execute("SELECT side, quantity, price, fee_amount, tax_amount, executed_at FROM portfolio_ledger WHERE portfolio_id=%s ORDER BY executed_at ASC NULLS FIRST, ledger_entry_id ASC", (portfolio_id,))
history = [{"side": r[0], "quantity": r[1], "price": r[2], "fee_amount": r[3], "tax_amount": r[4], "executed_at": r[5]} for r in cur.fetchall()]
# Derive WAC from the authoritative trade ledger, never the
# portfolio reference price or the already-mutated snapshot.
cur.execute(
"""
SELECT side, quantity, price, fee_amount, tax_amount, executed_at
FROM portfolio_ledger
WHERE portfolio_id=%s AND side IN ('buy','sell')
ORDER BY executed_at ASC NULLS FIRST, ledger_entry_id ASC
""",
(portfolio_id,),
)
history = [
{
"side": ledger_row[0],
"quantity": ledger_row[1],
"price": ledger_row[2],
"fee_amount": ledger_row[3],
"tax_amount": ledger_row[4],
"executed_at": ledger_row[5],
}
for ledger_row in cur.fetchall()
]
from app.domain.pnl import calculate_current_wac
wac = calculate_current_wac(history)
if Decimal(str(row[0])) != action.quantity_before or Decimal(str(wac.current_quantity)) != action.quantity_before:
if Decimal(str(wac.current_quantity)) != action.quantity_before:
raise ValueError("CORPORATE_ACTION_STALE_QUANTITY")
if resume_partial:
if snapshot_quantity != action.quantity_after:
raise ValueError("CORPORATE_ACTION_RECOVERY_STATE_MISMATCH")
cur.execute(
"""
SELECT
EXISTS (
SELECT 1 FROM portfolio_corporate_actions
WHERE portfolio_id=%s
),
EXISTS (
SELECT 1 FROM portfolio_ledger
WHERE portfolio_id=%s AND source='corporate_action'
)
""",
(portfolio_id, portfolio_id),
)
audit_exists, ledger_exists = cur.fetchone()
if audit_exists or ledger_exists:
raise ValueError("CORPORATE_ACTION_RECOVERY_ALREADY_RECORDED")
elif snapshot_quantity != action.quantity_before:
raise ValueError("CORPORATE_ACTION_STALE_QUANTITY")
consolidation = calculate_stock_consolidation(
action.quantity_before,
action.quantity_after,
@@ -1677,28 +1789,83 @@ class PostgresPortfolioRepository:
action.average_price_before = consolidation.average_price_before
action.average_price_after = consolidation.average_price_after
action.total_cost_basis = consolidation.total_cost_basis
cur.execute("SELECT 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, 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[8] != payload_hash:
raise ValueError("IDEMPOTENCY_KEY_REUSE")
(action.corporate_action_id, action.portfolio_id, action.asset_id, action.action_type,
action.effective_date, action.actor_user_id, action.evidence_ref,
action.idempotency_key, action.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,
ledger_id, action.recorded_at) = prior
action.corporate_action_id = str(action.corporate_action_id)
action.ledger_entry_id = int(ledger_id)
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, updated_at=NOW() 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))
action.cash_balance = cash_balance
action.profit_buffer = profit_buffer
consolidation_ratio = action.quantity_before / action.quantity_after
reference_price_after = quantize_stored_decimal(
reference_price_before * consolidation_ratio
)
total_value_after = quantize_stored_decimal(
cash_balance + action.quantity_after * reference_price_after
)
cur.execute(
"""
UPDATE portfolios
SET asset_quantity=%s, asset_price=%s, updated_at=NOW()
WHERE portfolio_id=%s
""",
(action.quantity_after, reference_price_after, portfolio_id),
)
cur.execute(
"""
INSERT INTO portfolio_ledger (
portfolio_id, recorded_at, side, source, quantity, price,
cash_balance_after, asset_quantity_after, total_value_after,
profit_buffer_after, executed_at, note
) VALUES (
%s, NOW(), 'stock_consolidation', 'corporate_action', 0, %s,
%s, %s, %s, %s, NOW(), %s
)
RETURNING ledger_entry_id, recorded_at
""",
(
portfolio_id,
reference_price_after,
cash_balance,
action.quantity_after,
total_value_after,
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))
action.ledger_entry_id = int(lid)
action.recorded_at = 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,
cash_balance,
profit_buffer,
lid,
),
)
return action
def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]:
@@ -1803,13 +1970,13 @@ class PostgresPortfolioRepository:
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, source,
ledger.ledger_entry_id, ledger.portfolio_id, ledger.order_id, ledger.execution_id, ledger.side,
ledger.quantity::float8, ledger.price::float8,
ledger.fee_amount::float8, ledger.tax_amount::float8,
ledger.executed_at, ledger.recorded_at,
ledger.cash_balance_after::float8, ledger.asset_quantity_after::float8,
ledger.total_value_after::float8, ledger.profit_buffer_after::float8,
ledger.strategy_set_id, ledger.note, ledger.source,
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,
@@ -2476,30 +2643,32 @@ class PostgresPortfolioRepository:
limit: int = 50,
) -> list[LedgerEntry]:
self.ensure_schema()
clauses = ["portfolio_id = %s"]
clauses = ["ledger.portfolio_id = %s"]
params: list[Any] = [portfolio_id]
if source is not None:
clauses.append("source = %s")
clauses.append("ledger.source = %s")
params.append(source)
if before_recorded_at is not None and before_ledger_entry_id is not None:
clauses.append("(recorded_at, ledger_entry_id) < (%s, %s)")
clauses.append("(ledger.recorded_at, ledger.ledger_entry_id) < (%s, %s)")
params.extend([before_recorded_at, before_ledger_entry_id])
params.append(int(limit))
with self._connect().cursor() as cur:
cur.execute(
"""
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,
SELECT ledger.ledger_entry_id, ledger.portfolio_id, ledger.order_id, ledger.execution_id, ledger.side,
ledger.quantity, ledger.price, ledger.fee_amount, ledger.tax_amount,
ledger.executed_at, ledger.recorded_at, ledger.cash_balance_after,
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,
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,
ca.profit_buffer, ca.evidence_ref, ca.actor_user_id
FROM portfolio_ledger
LEFT JOIN portfolio_corporate_actions ca ON ca.ledger_entry_id = portfolio_ledger.ledger_entry_id
FROM portfolio_ledger AS ledger
LEFT JOIN portfolio_corporate_actions AS ca ON ca.ledger_entry_id = ledger.ledger_entry_id
WHERE """ + " AND ".join(clauses) + """
ORDER BY recorded_at DESC, ledger_entry_id DESC
ORDER BY ledger.recorded_at DESC, ledger.ledger_entry_id DESC
LIMIT %s
""",
params,
+1
View File
@@ -14,6 +14,7 @@ class CorporateActionRequest(BaseModel):
quantity_after: Decimal = Field(..., gt=0, max_digits=20, decimal_places=8)
evidence_ref: str = Field(..., min_length=1, max_length=500)
confirm_daemon_managed: bool = False
resume_partial: bool = False
@model_validator(mode="after")
def valid_ratio(self):
+10 -10
View File
@@ -1,11 +1,11 @@
# PortfolioHub v0.3.2 container contract
# PortfolioHub v0.3.4 container contract
## Publication is not deployment
Only a requester-authorized push of the strict tag `v0.3.2` can publish this revision. The workflow checks that the tag equals `[project].version`, checks out `github.sha`, runs the complete Python suite, builds one `linux/amd64` image from `uv.lock`, and tests those exact bytes before publication. It gives the same image exactly two selectors:
Only a requester-authorized push of the strict tag `v0.3.4` can publish this revision. The workflow checks that the tag equals `[project].version`, checks out `github.sha`, runs the complete Python suite, builds one `linux/amd64` image from `uv.lock`, and tests those exact bytes before publication. It gives the same image exactly two selectors:
- `ghcr.io/fourblocks2020/portfolio-hub:<source-commit-first-12>`
- `ghcr.io/fourblocks2020/portfolio-hub:v0.3.2`
- `ghcr.io/fourblocks2020/portfolio-hub:v0.3.4`
It never publishes `latest`. Both selectors must report the same registry digest; deployment and rollback must use the immutable identity:
@@ -13,9 +13,9 @@ It never publishes `latest`. Both selectors must report the same registry digest
ghcr.io/fourblocks2020/portfolio-hub@sha256:<64-lowercase-hex-digest>
```
OCI labels are fail-closed build inputs: `org.opencontainers.image.source`, the full 40-character `org.opencontainers.image.revision`, and strict `org.opencontainers.image.version=v0.3.2`. Publication does **not** deploy, replace, start, stop, or restart a production service. This repository provides no production Compose, Kubernetes, Helm, systemd, or Dockge configuration.
OCI labels are fail-closed build inputs: `org.opencontainers.image.source`, the full 40-character `org.opencontainers.image.revision`, and strict `org.opencontainers.image.version=v0.3.4`. Publication does **not** deploy, replace, start, stop, or restart a production service. This repository provides no production Compose, Kubernetes, Helm, systemd, or Dockge configuration.
The Dockerfile frontend, Python base image, uv image, and GitHub Actions are digest/commit pinned. The workflow intentionally uses `--provenance=false` to preserve the tested `docker save` archive and build-once/no-rebuild identity. Registry-attached provenance, SBOM, signing, and attestations are a nonblocking v0.3.2 supply-chain gap; they must not be added by rebuilding a different image in the publish job.
The Dockerfile frontend, Python base image, uv image, and GitHub Actions are digest/commit pinned. The workflow intentionally uses `--provenance=false` to preserve the tested `docker save` archive and build-once/no-rebuild identity. Registry-attached provenance, SBOM, signing, and attestations are a nonblocking v0.3.4 supply-chain gap; they must not be added by rebuilding a different image in the publish job.
## Image runtime
@@ -66,19 +66,19 @@ Values are read at process start; changing any value requires a container restar
| `PORTFOLIOHUB_BOOTSTRAP_PASSWORD` | optional; empty | **yes** | bootstrap password; inject temporarily and remove after first successful start |
| `PORTFOLIOHUB_ENVIRONMENT` | optional; `development` | no | reported environment metadata |
| `PORTFOLIOHUB_APP_NAME` | optional; `PortfolioHub` | no | application/OpenAPI display name |
| `PORTFOLIOHUB_APP_VERSION` | do not override for v0.3.2 | `0.3.2` | no | health/OpenAPI version metadata |
| `PORTFOLIOHUB_APP_VERSION` | do not override for v0.3.4 | `0.3.4` | no | health/OpenAPI version metadata |
`REDIS_HOST`, `REDIS_PORT`, `REDIS_DB`, `REDIS_PASSWORD`, `LOKI_HOST`, `LOKI_PORT`, and `LOKI_ENABLED` are still parsed for source compatibility but the inspected v0.3.2 application does not open Redis or Loki connections. They are not production dependencies of this image.
`REDIS_HOST`, `REDIS_PORT`, `REDIS_DB`, `REDIS_PASSWORD`, `LOKI_HOST`, `LOKI_PORT`, and `LOKI_ENABLED` are still parsed for source compatibility but the inspected v0.3.4 application does not open Redis or Loki connections. They are not production dependencies of this image.
### Legacy and sibling mapping
The existing database variable names remain `TIMESCALEDB_*`; v0.3.2 deliberately does not invent a `DATABASE_URL` alias. Existing `PORTFOLIOHUB_{READ,TRADE_WRITE,ADMIN}_TOKENS` Bearer-token behavior remains compatible. `trex-core`'s `TREX_PORTFOLIO_HUB_BASE_URL` and TOML `portfolio_hub_token`, and `trex-web`'s server-only `PORTFOLIO_HUB_API_BASE_URL`, are **caller-side** settings: point them to `http://portfolio-hub:8000` (or the equivalent service URL) and do not inject them into this container.
The existing database variable names remain `TIMESCALEDB_*`; v0.3.4 deliberately does not invent a `DATABASE_URL` alias. Existing `PORTFOLIOHUB_{READ,TRADE_WRITE,ADMIN}_TOKENS` Bearer-token behavior remains compatible. `trex-core`'s `TREX_PORTFOLIO_HUB_BASE_URL` and TOML `portfolio_hub_token`, and `trex-web`'s server-only `PORTFOLIO_HUB_API_BASE_URL`, are **caller-side** settings: point them to `http://portfolio-hub:8000` (or the equivalent service URL) and do not inject them into this container.
## Liveness and readiness
- **Liveness:** unauthenticated `GET /health` must return HTTP 200 with `status=ok`. It is intentionally process-only and does not query PostgreSQL. `db_enabled` only reports configuration; it is not readiness.
- **Readiness:** run `pfhub --format json health-check` with the same runtime environment. With DB enabled it performs a repository ping, reads `alembic_version` without migration, compares it with the single packaged Alembic head, and exits nonzero on connectivity, missing schema, multiple heads, or revision mismatch.
- **API identity:** `GET /openapi.json` is stable for smoke verification. Health version is `0.3.2`.
- **API identity:** `GET /openapi.json` is stable for smoke verification. Health version is `0.3.4`.
Do not route traffic from a static `/health` success alone. Readiness checks never apply, stamp, or repair migrations.
@@ -107,7 +107,7 @@ Image rollback is **not** database rollback. Selecting older application bytes d
## API verification and known caller blockers
v0.3.2 preserves Bearer authentication and all-source unified-ledger reads. It records integral-share stock consolidations even when 8-decimal WAC storage creates a harmless recomputation residue: WAC is rounded half-up to 8 decimals, while the authoritative total cost basis remains independently stored and drives replay. The release gate runs the complete portfolio-hub source test suite (including auth success/failure, strategy catalog/data, portfolio and ledger APIs), then verifies the built image's health/OpenAPI surface, exact 31-file migration inventory, isolated PostgreSQL migration twice, readiness at `0031`, persistent-volume restart, non-root/read-only operation, secret canary absence, and graceful SIGTERM. Publication consumes the tested archive without rebuilding it. The changed-scope Ruff gate uses the locked linter and the ACMR Python path selector, compares normalized findings against v0.2.1, v0.3.0, and v0.3.1, and rejects every newly introduced finding.
v0.3.4 preserves Bearer authentication and all-source unified-ledger reads. Corporate-action quantity/reference-price updates, the complete ledger snapshot (including `total_value_after`), and the audit row execute in one explicit transaction. The guarded `resume_partial` path repairs only a verified quantity-only partial state without raw SQL. The release gate runs the complete portfolio-hub source test suite (including auth success/failure, strategy catalog/data, portfolio and ledger APIs), then verifies the built image's health/OpenAPI surface, exact 31-file migration inventory, isolated PostgreSQL migration twice, readiness at `0031`, persistent-volume restart, non-root/read-only operation, secret canary absence, and graceful SIGTERM. Publication consumes the tested archive without rebuilding it. The changed-scope Ruff gate uses the locked linter and the ACMR Python path selector, compares normalized findings against v0.2.1, v0.3.0, and v0.3.1, and rejects every newly introduced finding.
The strategy catalog API remains additive: strategy responses expose `nick`, admin create/update requests may set it, and portfolio responses expose the current strategy as `strategy_nick`. Existing clients may omit `nick`; supplied values are trimmed and blank-only values are rejected. trex-web v0.2.3 displays `strategy_nick`, but that dependency does not resolve or waive the separate caller blockers below.
+4 -1
View File
@@ -439,8 +439,11 @@
- Query: `limit` 1..200, default 50.
- `GET /v1/portfolios/{id}/ledger` returns corporate-action events as a distinct `CorporateActionLedgerHistoryItem` with event ID/type, effective date, before/after quantity and WAC, total cost basis, cash/profit buffer, evidence, actor, and recorded time. These events are never represented as BUY/SELL trades.
- For stock consolidations, WAC fields are rounded half-up to the persisted 8-decimal scale. `total_cost_basis` is preserved independently as the authoritative pre-action acquisition cost; a sub-cent residue when `average_price_after * quantity_after` is recomputed does not make an integral-share consolidation invalid.
- `POST /v1/portfolios/{id}/corporate-actions` updates quantity and the post-action reference price, appends the `portfolio_ledger` snapshot (including non-null `total_value_after`), and appends the corporate-action audit row in one explicit database transaction. A failure in any statement rolls back every statement.
- The post-action reference price is `asset_price_before * quantity_before / quantity_after`; ledger `total_value_after` is `cash_balance_after + asset_quantity_after * post_action_reference_price`. Cash and profit buffer are unchanged.
- `resume_partial=false` is the compatible default. `resume_partial=true` is accepted only when the locked portfolio quantity already equals `quantity_after`, trade-ledger replay still proves `quantity_before`, and both corporate-action audit and corporate-action ledger rows are absent. Same-key replay is idempotent; a different-key recovery after repair is rejected.
- `GET /v1/portfolios/{id}/trades` remains restricted to BUY/SELL history.
- `portfolio corporate-action` is the operator command; it requires an idempotency key and confirmation unless `--force` is supplied.
- `portfolio corporate-action` is the operator command; it requires an idempotency key and confirmation unless `--force` is supplied. `--resume-partial` enables only the guarded recovery mode above.
### Endpoint: `GET /v1/portfolios/{portfolio_id}/returns/summary` (REQ-056, BAT-024)
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "portfoliohub"
version = "0.3.2"
version = "0.3.4"
description = "PortfolioHub API for portfolio management and ledger access"
readme = "README.md"
requires-python = ">=3.11"
+8 -4
View File
@@ -16,7 +16,11 @@ def test_branch_verification_has_no_registry_write_authority_or_path():
text=VERIFY.read_text(); assert "branches: [main]" in text and "pull_request:" in text
for forbidden in ("packages: write","docker login","docker push","REGISTRY_TOKEN","image_release.py publish"):
assert forbidden not in text
assert "uv sync --extra dev --locked" in text and "uv run pytest -q" in text
assert "uv sync --extra dev --locked" in text and "uv run python -m pytest -q" in text
assert "Verify corporate-action atomicity and guarded recovery in PostgreSQL" in text
assert "tests/test_corporate_action_atomic_recovery_postgres.py" in text
assert "${POSTGRES_PASSWORD}@127.0.0.1:5432/postgres" in text
assert "***@" not in text
def test_tag_publication_is_strict_single_build_handoff_without_latest():
text=PUBLISH.read_text(); verify,publish=text.split("\n publish:",1)
@@ -58,8 +62,8 @@ def test_release_version_and_migration_contract_has_current_head_0031():
for path in (REPO / "app/db/alembic/versions").glob("*.py")
if path.name != "__init__.py"
)
assert 'version = "0.3.2"' in project
assert 'app_version: str = "0.3.2"' in config
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_")
@@ -85,7 +89,7 @@ def test_operations_separate_publication_deployment_migration_and_db_rollback():
"Image rollback is **not** database rollback",
"read-only root filesystem",
"10001:10001",
"PortfolioHub v0.3.2 container contract",
"PortfolioHub v0.3.4 container contract",
"v0.2.1 includes migration `0030`",
"trex-web v0.2.3 displays `strategy_nick`",
):
@@ -0,0 +1,333 @@
"""Real PostgreSQL regressions for atomic corporate-action writes and recovery."""
from __future__ import annotations
import os
from datetime import date
from decimal import Decimal
from urllib.parse import unquote, urlparse
from uuid import uuid4
import psycopg
import pytest
from psycopg import sql
from app.core.config import Settings
from app.db.migrations import upgrade_to_head
from app.domain.corporate_actions import canonical_payload_hash
from app.domain.models import CorporateAction
from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioRepository
from app.schemas.corporate_actions import CorporateActionRequest
_LOCAL_DSN_ENV = "PORTFOLIOHUB_TEST_POSTGRES_DSN"
_LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"}
_PORTFOLIO_ID = "3573f4d6-cc27-4f91-82d9-6330e5250d20"
_CASH = Decimal("400060.62490000")
_TOTAL_COST = Decimal("1577211.97990000")
def _maintenance_dsn() -> str:
dsn = os.environ.get(_LOCAL_DSN_ENV)
if not dsn:
pytest.skip(f"{_LOCAL_DSN_ENV} is required for real PostgreSQL integration tests")
parsed = urlparse(dsn)
if parsed.scheme not in {"postgres", "postgresql"}:
pytest.fail("test DSN must use postgres/postgresql")
if parsed.hostname not in _LOOPBACK_HOSTS or parsed.path.lstrip("/") != "postgres":
pytest.fail("tests require a loopback PostgreSQL maintenance database named postgres")
return dsn
def _settings(dsn: str, database: str) -> Settings:
parsed = urlparse(dsn)
return Settings(
db_enabled=True,
timescaledb_host=parsed.hostname,
timescaledb_port=parsed.port or 5432,
timescaledb_name=database,
timescaledb_user=unquote(parsed.username or ""),
timescaledb_password=unquote(parsed.password or ""),
timescaledb_sslmode="disable",
timescaledb_connect_timeout_sec=3,
timescaledb_application_name="corporate-action-atomic-test",
)
def _repo(settings: Settings) -> PostgresPortfolioRepository:
return PostgresPortfolioRepository(
PortfolioDBConfig(
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,
connect_timeout_sec=settings.active_db_connect_timeout_sec,
application_name=settings.active_db_application_name,
)
)
@pytest.fixture()
def repository() -> PostgresPortfolioRepository:
maintenance_dsn = _maintenance_dsn()
database = f"portfolio_hub_ca_{uuid4().hex}"
with psycopg.connect(maintenance_dsn, autocommit=True) as conn, conn.cursor() as cur:
cur.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(database)))
settings = _settings(maintenance_dsn, database)
upgrade_to_head(settings)
repo = _repo(settings)
try:
yield repo
finally:
repo.close()
with psycopg.connect(maintenance_dsn, autocommit=True) as conn, conn.cursor() as cur:
cur.execute(
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
"WHERE datname = %s AND pid <> pg_backend_pid()",
(database,),
)
cur.execute(sql.SQL("DROP DATABASE IF EXISTS {}").format(sql.Identifier(database)))
def _seed_daekyo(repo: PostgresPortfolioRepository, *, quantity: Decimal) -> str:
user_id = "ca-recovery-user"
with repo._connect().cursor() as cur:
cur.execute("INSERT INTO users (id, username) VALUES (%s, %s)", (user_id, user_id))
cur.execute(
"INSERT INTO assets (symbol, markets) VALUES ('019685', ARRAY['KRX']) RETURNING asset_id"
)
asset_id = str(cur.fetchone()[0])
cur.execute(
"""
INSERT INTO portfolios (
portfolio_id, name, owner_user_id, cash_balance, asset_quantity,
asset_price, profit_buffer, asset_id, broker, status
) VALUES (%s, '대교우B', %s, %s, %s, 643, 0, %s, 'kis', 'enabled')
""",
(_PORTFOLIO_ID, user_id, _CASH, quantity, asset_id),
)
# The fee makes the authoritative BUY-ledger cost exactly 1,577,211.9799.
cur.execute(
"""
INSERT INTO portfolio_ledger (
portfolio_id, recorded_at, side, source, quantity, price, fee_amount, tax_amount,
cash_balance_after, asset_quantity_after, total_value_after,
profit_buffer_after, strategy_set_id, note, executed_at
) VALUES (%s, '2026-09-09T00:00:00Z', 'buy', 'broker', 1452, 1086, 339.9799, 0,
%s, 1452, %s, 0, 'runtime', 'seed', '2026-09-09T00:00:00Z')
""",
(_PORTFOLIO_ID, _CASH, _CASH + Decimal(1452) * Decimal(643)),
)
return asset_id
def _request(*, resume_partial: bool = False) -> CorporateActionRequest:
return CorporateActionRequest(
effective_date=date(2026, 9, 10),
quantity_before=Decimal(1452),
quantity_after=Decimal(726),
evidence_ref="KIS 019685 2026-09-10",
resume_partial=resume_partial,
)
def _action(asset_id: str, request: CorporateActionRequest, key: str) -> CorporateAction:
digest = canonical_payload_hash(request)
return CorporateAction(
str(uuid4()),
_PORTFOLIO_ID,
asset_id,
request.action_type,
request.effective_date,
request.quantity_before,
request.quantity_after,
Decimal(0),
Decimal(0),
Decimal(0),
_CASH,
Decimal(0),
request.evidence_ref,
key,
digest,
0,
"ca-recovery-user",
)
def _record(repo: PostgresPortfolioRepository, asset_id: str, request: CorporateActionRequest, key: str):
action = _action(asset_id, request, key)
return repo.record_corporate_action(
_PORTFOLIO_ID,
action=action,
idempotency_key=key,
payload_hash=action.payload_hash,
resume_partial=request.resume_partial,
)
def test_joined_ledger_reads_qualify_base_columns(repository):
"""Mixed trade/corporate-action rows must be readable through both joins."""
asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
_record(repository, asset_id, _request(), "ca-joined-read")
pnl_entries = repository.get_all_trades_for_pnl(_PORTFOLIO_ID)
ledger_entries = repository.list_ledger_entries(_PORTFOLIO_ID, limit=50)
assert len(pnl_entries) == 2
assert len(ledger_entries) == 2
assert any(entry.corporate_action for entry in pnl_entries)
assert any(entry.corporate_action for entry in ledger_entries)
assert ledger_entries[0].recorded_at >= ledger_entries[1].recorded_at
corporate_entries = repository.list_ledger_entries(_PORTFOLIO_ID, source="corporate_action", limit=50)
assert len(corporate_entries) == 1
assert corporate_entries[0].corporate_action is not None
next_page = repository.list_ledger_entries(
_PORTFOLIO_ID,
before_recorded_at=ledger_entries[0].recorded_at,
before_ledger_entry_id=ledger_entries[0].ledger_entry_id,
limit=50,
)
assert [entry.source for entry in next_page] == ["broker"]
def _state(repo: PostgresPortfolioRepository) -> tuple:
with repo._connect().cursor() as cur:
cur.execute(
"SELECT asset_quantity, asset_price, cash_balance, profit_buffer "
"FROM portfolios WHERE portfolio_id=%s",
(_PORTFOLIO_ID,),
)
portfolio_row = cur.fetchone()
assert portfolio_row is not None
portfolio = tuple(Decimal(str(value)) for value in portfolio_row)
cur.execute(
"SELECT COUNT(*) FROM portfolio_ledger "
"WHERE portfolio_id=%s AND source='corporate_action'",
(_PORTFOLIO_ID,),
)
ledger_count = cur.fetchone()[0]
cur.execute(
"SELECT COUNT(*) FROM portfolio_corporate_actions WHERE portfolio_id=%s",
(_PORTFOLIO_ID,),
)
action_count = cur.fetchone()[0]
return (*portfolio, ledger_count, action_count)
def test_stock_consolidation_persists_reference_price_total_value_and_audit(repository):
asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
saved = _record(repository, asset_id, _request(), "ca-normal")
assert saved.average_price_after == Decimal("2172.46829187")
assert saved.total_cost_basis == _TOTAL_COST
assert _state(repository) == (
Decimal("726.00000000"),
Decimal("1286.00000000"),
_CASH,
Decimal("0E-8"),
1,
1,
)
with repository._connect().cursor() as cur:
cur.execute(
"SELECT price, total_value_after FROM portfolio_ledger "
"WHERE portfolio_id=%s AND source='corporate_action'",
(_PORTFOLIO_ID,),
)
assert cur.fetchone() == (
Decimal("1286.00000000"),
Decimal("1333696.62490000"),
)
def test_ledger_failure_rolls_back_portfolio_quantity_and_reference_price(repository):
asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
with repository._connect().cursor() as cur:
cur.execute(
"""
CREATE FUNCTION reject_corporate_action_ledger() RETURNS trigger AS $$
BEGIN
IF NEW.source = 'corporate_action' THEN
RAISE EXCEPTION 'injected ledger failure';
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql
"""
)
cur.execute(
"CREATE TRIGGER reject_ca_ledger BEFORE INSERT ON portfolio_ledger "
"FOR EACH ROW EXECUTE FUNCTION reject_corporate_action_ledger()"
)
with pytest.raises(psycopg.Error, match="injected ledger failure"):
_record(repository, asset_id, _request(), "ca-failure")
assert _state(repository) == (
Decimal("1452.00000000"),
Decimal("643.00000000"),
_CASH,
Decimal("0E-8"),
0,
0,
)
def test_audit_failure_rolls_back_portfolio_and_inserted_ledger(repository):
asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
before = _state(repository)
with repository._connect().cursor() as cur:
cur.execute(
"""
CREATE FUNCTION reject_corporate_action_audit() RETURNS trigger AS $$
BEGIN
RAISE EXCEPTION 'injected corporate-action audit failure';
END;
$$ LANGUAGE plpgsql
"""
)
cur.execute(
"CREATE TRIGGER reject_ca_audit BEFORE INSERT ON portfolio_corporate_actions "
"FOR EACH ROW EXECUTE FUNCTION reject_corporate_action_audit()"
)
with pytest.raises(psycopg.Error, match="injected corporate-action audit failure"):
_record(repository, asset_id, _request(), "ca-audit-failure")
assert _state(repository) == before
def test_explicit_resume_repairs_only_verified_quantity_only_partial_state(repository):
asset_id = _seed_daekyo(repository, quantity=Decimal(726))
request = _request(resume_partial=True)
first = _record(repository, asset_id, request, "ca-recovery")
replay = _record(repository, asset_id, request, "ca-recovery")
assert first.idempotent_replay is False
assert replay.idempotent_replay is True
assert replay.ledger_entry_id == first.ledger_entry_id
assert _state(repository) == (
Decimal("726.00000000"),
Decimal("1286.00000000"),
_CASH,
Decimal("0E-8"),
1,
1,
)
with pytest.raises(ValueError, match="CORPORATE_ACTION_RECOVERY_ALREADY_RECORDED"):
_record(repository, asset_id, request, "ca-recovery-different-key")
assert _state(repository)[-2:] == (1, 1)
def test_resume_rejects_a_normal_unapplied_snapshot(repository):
asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
with pytest.raises(ValueError, match="CORPORATE_ACTION_RECOVERY_STATE_MISMATCH"):
_record(repository, asset_id, _request(resume_partial=True), "ca-invalid-recovery")
assert _state(repository)[-2:] == (0, 0)
+18 -1
View File
@@ -44,10 +44,27 @@ def test_canonical_hash_same_payload_and_changed_payload_differ():
assert canonical_payload_hash(a) != canonical_payload_hash(b)
def test_default_resume_flag_preserves_pre_recovery_idempotency_hash():
request = CorporateActionRequest(
effective_date=date(2026, 1, 2),
quantity_before=Decimal(10),
quantity_after=Decimal(5),
evidence_ref="ev",
)
legacy_payload = request.model_dump(mode="json", exclude={"resume_partial"})
explicit_default_payload = {**legacy_payload, "resume_partial": False}
assert canonical_payload_hash(request) == canonical_payload_hash(legacy_payload)
assert canonical_payload_hash(explicit_default_payload) == canonical_payload_hash(legacy_payload)
assert canonical_payload_hash(
request.model_copy(update={"resume_partial": True})
) != canonical_payload_hash(legacy_payload)
def test_runtime_schema_has_one_corporate_action_ddl_and_replay_mapping_starts_at_18():
source = Path("app/repositories/postgres.py").read_text()
assert source.count("CREATE TABLE IF NOT EXISTS portfolio_corporate_actions") == 1
assert "strategy_set_id, note, source," in source
assert "ledger.strategy_set_id, ledger.note, ledger.source," in source
assert "ca.corporate_action_id, ca.action_type" in source
assert 'row[18] is not None' in source
+73 -4
View File
@@ -7,7 +7,7 @@ import pytest
from fastapi.testclient import TestClient
from app.api import v1_portfolios
from app.domain.models import CorporateAction
from app.domain.models import CorporateAction, Portfolio
from app.main import create_app
from app.repositories.postgres import PostgresPortfolioRepository
@@ -20,6 +20,46 @@ class MissingPortfolioRepo:
pass
class ResumeCaptureRepo:
def __init__(self):
self.resume_partial = None
def get(self, portfolio_id):
return Portfolio(
portfolio_id=portfolio_id,
broker="kis",
status="enabled",
name="대교우B",
symbol="019685",
asset_id="asset-1",
owner_user_id="u1",
cash_balance=400060.6249,
asset_quantity=726,
price=643,
profit_buffer=0,
)
def record_corporate_action(
self,
portfolio_id,
*,
action,
idempotency_key,
payload_hash,
resume_partial=False,
):
self.resume_partial = resume_partial
action.average_price_before = Decimal("1086.23414594")
action.average_price_after = Decimal("2172.46829187")
action.total_cost_basis = Decimal("1577211.97990000")
action.ledger_entry_id = 7
action.recorded_at = datetime(2026, 9, 13, tzinfo=UTC)
return action
def close(self):
pass
def test_create_corporate_action_maps_postgres_missing_portfolio_to_404(monkeypatch):
app = create_app()
repo = MissingPortfolioRepo()
@@ -35,6 +75,31 @@ def test_create_corporate_action_maps_postgres_missing_portfolio_to_404(monkeypa
assert response.json()["detail"] == "Portfolio not found"
def test_corporate_action_api_passes_explicit_resume_mode(monkeypatch):
app = create_app()
repo = ResumeCaptureRepo()
monkeypatch.setattr(v1_portfolios, "_get_repo", lambda: repo)
app.dependency_overrides[v1_portfolios.require_trade_write] = lambda: type(
"Auth", (), {"user_id": "u1", "permission": "trade_write"}
)()
response = TestClient(app).post(
"/v1/portfolios/pf-1/corporate-actions",
headers={"Idempotency-Key": "resume-1"},
json={
"effective_date": "2026-09-10",
"quantity_before": "1452",
"quantity_after": "726",
"evidence_ref": "KIS 019685 2026-09-10",
"resume_partial": True,
},
)
assert response.status_code == 201
assert response.json()["quantity_after"] == "726"
assert repo.resume_partial is True
def _action(asset_id="asset-1"):
return CorporateAction("ca-1", "pf-1", asset_id, "stock_consolidation", date(2026, 1, 1), Decimal(2), Decimal(1), Decimal(0), Decimal(0), Decimal(0), Decimal(10), Decimal(0), "ev", "idem", "hash", 0)
@@ -47,7 +112,12 @@ def _repo_for_row(asset_id):
conn.__enter__.return_value = conn
conn.cursor.return_value.__enter__.return_value = cur
repo._connect = MagicMock(return_value=conn)
cur.fetchone.side_effect = [(Decimal(2), Decimal(10), Decimal(0), asset_id, "enabled"), None, None, None, (7, date(2026, 1, 1))]
cur.fetchone.side_effect = [
(Decimal(2), Decimal(10), Decimal(0), asset_id, "enabled", Decimal(643)),
None,
None,
(7, date(2026, 1, 1)),
]
cur.fetchall.return_value = [("buy", Decimal(2), Decimal(0), Decimal(0), Decimal(0), datetime(2025, 12, 31, tzinfo=UTC))]
return repo
@@ -74,8 +144,7 @@ def test_record_corporate_action_accepts_daekyo_rounding_residue():
conn.cursor.return_value.__enter__.return_value = cur
repo._connect = MagicMock(return_value=conn)
cur.fetchone.side_effect = [
(Decimal(1452), Decimal(10), Decimal(0), "asset-1", "enabled"),
None,
(Decimal(1452), Decimal(10), Decimal(0), "asset-1", "enabled", Decimal(643)),
None,
None,
(7, datetime(2026, 9, 13, tzinfo=UTC)),
+35 -1
View File
@@ -7,7 +7,7 @@ Covers AC-1~11 from REQ-043.
from __future__ import annotations
from contextlib import ExitStack
from datetime import datetime, timezone
from datetime import UTC, datetime, timezone
from unittest.mock import MagicMock, patch
import pytest
@@ -286,6 +286,40 @@ class TestAC2PartialSellWacUnchanged:
assert resp.json()["total_cost_basis"] == pytest.approx(500_000.0)
def test_corporate_action_row_keeps_holdings_cost_basis_http_200() -> None:
buy = _make_entry(
ledger_entry_id=1,
side="buy",
quantity=20,
price=100,
executed_at=datetime(2026, 9, 1, tzinfo=UTC),
)
corporate_action = LedgerEntry(
ledger_entry_id=2,
portfolio_id=_PORTFOLIO_ID,
source="corporate_action",
side="stock_consolidation",
quantity=10,
price=200,
recorded_at=datetime(2026, 9, 2, tzinfo=UTC),
corporate_action={"action_type": "stock_consolidation"},
)
repo = _make_service_repo(
portfolio=_make_portfolio(asset_quantity=10),
trades=[buy, corporate_action],
)
with _with_repo(repo):
response = _make_client(_make_settings()).get(
_PATH,
headers={"Authorization": "Bearer tok"},
)
assert response.status_code == 200
assert response.json()["cost_basis_per_unit"] == pytest.approx(100)
assert response.json()["total_cost_basis"] == pytest.approx(1000)
# ---------------------------------------------------------------------------
# AC-3: user read_token (owner) → 200 + normal response
# ---------------------------------------------------------------------------
+57 -1
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import os
import time
from contextlib import ExitStack
from datetime import UTC, datetime
from datetime import UTC, date, datetime
from decimal import Decimal
from typing import Any
from unittest.mock import patch
@@ -977,6 +977,62 @@ def test_unified_ledger_includes_dividend_fields() -> None:
assert item["profit_buffer_after"] == created.json()["profit_buffer_after"]
def test_unified_ledger_includes_corporate_action_fields() -> None:
repo = _FakeRepo(_portfolio())
repo.entries.append(
LedgerEntry(
ledger_entry_id=7,
portfolio_id=_PORTFOLIO_ID,
source="corporate_action",
side="stock_consolidation",
quantity=10,
price=200,
recorded_at=datetime(2026, 9, 2, tzinfo=UTC),
corporate_action={
"corporate_action_id": "ca-7",
"action_type": "stock_consolidation",
"effective_date": date(2026, 9, 2),
"quantity_before": Decimal(20),
"quantity_after": Decimal(10),
"average_price_before": Decimal(100),
"average_price_after": Decimal(200),
"total_cost_basis": Decimal(2000),
"cash_balance": Decimal(1000),
"profit_buffer": Decimal(500),
"evidence_ref": "notice://ca-7",
"actor_user_id": _USER_ID,
},
)
)
client = _make_client(repo)
response = client.get(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger?source=corporate_action",
headers=_auth(),
)
assert response.status_code == 200
item = response.json()["items"][0]
assert item == {
"ledger_entry_id": 7,
"portfolio_id": _PORTFOLIO_ID,
"source": "corporate_action",
"event_id": "ca-7",
"event_type": "stock_consolidation",
"effective_date": "2026-09-02",
"quantity_before": "20",
"quantity_after": "10",
"average_price_before": "100",
"average_price_after": "200",
"total_cost_basis": "2000",
"cash_balance": "1000",
"profit_buffer": "500",
"evidence_ref": "notice://ca-7",
"actor_user_id": _USER_ID,
"recorded_at": "2026-09-02T00:00:00Z",
}
def test_unified_ledger_openapi_contains_nullable_amount() -> None:
repo = _FakeRepo(_portfolio())
client = _make_client(repo)
Generated
+1 -1
View File
@@ -592,7 +592,7 @@ wheels = [
[[package]]
name = "portfoliohub"
version = "0.3.2"
version = "0.3.4"
source = { editable = "." }
dependencies = [
{ name = "alembic" },