Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c7d8235bde |
@@ -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:
|
||||
@@ -23,6 +37,12 @@ jobs:
|
||||
- run: python -m pip install "uv==0.11.14"
|
||||
- run: uv sync --extra dev --locked
|
||||
- run: uv run 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 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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
@@ -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.3"
|
||||
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")
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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]: ...
|
||||
|
||||
|
||||
+212
-45
@@ -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]:
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
# PortfolioHub v0.3.2 container contract
|
||||
# PortfolioHub v0.3.3 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.3` 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.3`
|
||||
|
||||
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.3`. 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.3 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.3 | `0.3.3` | 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.3 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.3 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.3`.
|
||||
|
||||
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.3 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.
|
||||
|
||||
|
||||
@@ -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
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "portfoliohub"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
description = "PortfolioHub API for portfolio management and ledger access"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.11"
|
||||
|
||||
@@ -17,6 +17,10 @@ def test_branch_verification_has_no_registry_write_authority_or_path():
|
||||
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 "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.3"' in project
|
||||
assert 'app_version: str = "0.3.3"' 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.3 container contract",
|
||||
"v0.2.1 includes migration `0030`",
|
||||
"trex-web v0.2.3 displays `strategy_nick`",
|
||||
):
|
||||
|
||||
@@ -0,0 +1,306 @@
|
||||
"""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 _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)
|
||||
@@ -44,6 +44,23 @@ 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
|
||||
|
||||
@@ -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)),
|
||||
|
||||
Reference in New Issue
Block a user