feat(portfolios): add unified ledger history API

app/api/v1_portfolios.py

@@ -8,6 +8,11 @@ REQ-043: GET /v1/portfolios/{id}/holdings/cost-basis — current holding WAC cos
 
 from __future__ import annotations
 
+import base64
+import binascii
+import hashlib
+import hmac
+import json
 import logging
 import uuid
 from datetime import date, datetime, timedelta, timezone
@@ -15,11 +20,15 @@ from datetime import date, datetime, timedelta, timezone
 from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
 
 from app.core.auth import AuthContext, Permission, require_read, require_trade_write
+from app.core.config import get_settings
 from app.domain.errors import DaemonInactiveError, DaemonNotFoundError
 from app.domain.models import Portfolio, PortfolioComment
 from app.domain.pnl import calculate_current_wac, calculate_realized_pnl
 from app.repositories.factory import get_portfolio_repository_from_settings
 from app.schemas.v1 import (
+    HoldingsCostBasisResponse,
+    LedgerHistoryItem,
+    LedgerHistoryResponse,
     LedgerTradeRequest,
     LedgerTradeResponse,
     ManualLedgerHistoryResponse,
@@ -35,7 +44,6 @@ from app.schemas.v1 import (
     PortfolioReturnSummaryResponse,
     PortfolioTagListResponse,
     PortfolioUpdateRequest,
-    HoldingsCostBasisResponse,
     RealizedPnlItem,
     RealizedPnlResponse,
     TagApplyRequest,
@@ -44,7 +52,10 @@ from app.schemas.v1 import (
     TradeHistoryItem,
     TradeHistoryResponse,
 )
-from app.services.strategy_set_service import StrategyInactiveError, StrategyNotFoundError
+from app.services.strategy_set_service import (
+    StrategyInactiveError,
+    StrategyNotFoundError,
+)
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/v1/portfolios", tags=["v1"])
@@ -710,6 +721,62 @@ def _manual_error_code(message: str) -> str:
     return "MANUAL_LEDGER_CONFLICT"
 
 
+def _encode_ledger_cursor(*, portfolio_id: str, source: str | None, recorded_at: datetime, ledger_entry_id: int) -> str:
+    payload = {
+        "portfolio_id": portfolio_id,
+        "source": source,
+        "recorded_at": recorded_at.isoformat(),
+        "ledger_entry_id": ledger_entry_id,
+    }
+    raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
+    signature = hmac.new(get_settings().ledger_cursor_secret.encode(), raw, hashlib.sha256).digest()
+    return base64.urlsafe_b64encode(raw + signature).decode().rstrip("=")
+
+
+def _decode_ledger_cursor(cursor: str, *, portfolio_id: str, source: str | None) -> tuple[datetime, int]:
+    try:
+        encoded = cursor.encode()
+        raw_and_signature = base64.urlsafe_b64decode(encoded + b"=" * (-len(encoded) % 4))
+        if base64.urlsafe_b64encode(raw_and_signature).rstrip(b"=") != encoded:
+            raise ValueError
+        raw, signature = raw_and_signature[:-32], raw_and_signature[-32:]
+        expected = hmac.new(get_settings().ledger_cursor_secret.encode(), raw, hashlib.sha256).digest()
+        payload = json.loads(raw)
+        recorded_at = datetime.fromisoformat(payload["recorded_at"])
+        ledger_entry_id = int(payload["ledger_entry_id"])
+        if not hmac.compare_digest(signature, expected) or payload["portfolio_id"] != portfolio_id or payload["source"] != source:
+            raise ValueError
+        if recorded_at.tzinfo is None:
+            raise ValueError
+        return recorded_at, ledger_entry_id
+    except (KeyError, TypeError, ValueError, json.JSONDecodeError, UnicodeDecodeError, binascii.Error):
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="invalid ledger cursor") from None
+
+
+def _ledger_entry_to_history_item(entry) -> LedgerHistoryItem:
+    return LedgerHistoryItem(
+        ledger_entry_id=entry.ledger_entry_id,
+        portfolio_id=entry.portfolio_id,
+        source=entry.source or "broker",
+        action=(entry.action or entry.side or "").upper(),
+        recorded_at=entry.recorded_at,
+        executed_at=entry.executed_at,
+        order_id=entry.order_id,
+        execution_id=entry.execution_id,
+        quantity=entry.quantity,
+        price=entry.price,
+        fee_amount=entry.fee_amount,
+        tax_amount=entry.tax_amount,
+        note=entry.note,
+        cash_balance_after=entry.cash_balance_after,
+        asset_quantity_after=entry.asset_quantity_after,
+        profit_buffer_after=entry.profit_buffer_after,
+        total_value_after=entry.total_value_after,
+        target_ledger_entry_id=entry.target_ledger_entry_id,
+        confirmation_evidence=entry.confirmation_evidence,
+    )
+
+
 @router.post(
     "/{portfolio_id}/ledger/manual",
     response_model=ManualLedgerResponse,
@@ -788,6 +855,43 @@ def list_manual_ledger_v1(
     return ManualLedgerHistoryResponse(items=items, total=len(items))
 
 
+@router.get("/{portfolio_id}/ledger", response_model=LedgerHistoryResponse)
+def list_ledger_v1(
+    portfolio_id: str,
+    source: str | None = Query(default=None, min_length=1),
+    cursor: str | None = Query(default=None),
+    limit: int = Query(default=50, ge=1, le=200),
+    auth: AuthContext = Depends(require_read),
+) -> LedgerHistoryResponse:
+    if auth.token_type in {"service", "client"} or auth.user_id is None:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="ledger requires user/session token")
+    before_recorded_at = before_ledger_entry_id = None
+    if cursor is not None:
+        before_recorded_at, before_ledger_entry_id = _decode_ledger_cursor(cursor, portfolio_id=portfolio_id, source=source)
+    repo = _get_repo()
+    try:
+        _get_owner_portfolio_or_404(auth, portfolio_id, repo)
+        entries = repo.list_ledger_entries(
+            portfolio_id,
+            source=source,
+            before_recorded_at=before_recorded_at,
+            before_ledger_entry_id=before_ledger_entry_id,
+            limit=limit + 1,
+        )
+    finally:
+        _close_repo(repo)
+    page, extra = entries[:limit], entries[limit:]
+    next_cursor = None
+    if extra and page:
+        last = page[-1]
+        if last.recorded_at is None:
+            raise RuntimeError("ledger entry is missing recorded_at")
+        next_cursor = _encode_ledger_cursor(
+            portfolio_id=portfolio_id, source=source, recorded_at=last.recorded_at, ledger_entry_id=last.ledger_entry_id
+        )
+    return LedgerHistoryResponse(items=[_ledger_entry_to_history_item(entry) for entry in page], next_cursor=next_cursor)
+
+
 @router.get("/{portfolio_id}/returns/summary", response_model=PortfolioReturnSummaryResponse)
 def get_portfolio_return_summary_v1(
     portfolio_id: str,

app/core/config.py

@@ -27,6 +27,7 @@ class Settings(BaseSettings):
     read_tokens_raw: str = Field(default="", validation_alias="PORTFOLIOHUB_READ_TOKENS")
     trade_write_tokens_raw: str = Field(default="", validation_alias="PORTFOLIOHUB_TRADE_WRITE_TOKENS")
     admin_tokens_raw: str = Field(default="", validation_alias="PORTFOLIOHUB_ADMIN_TOKENS")
+    ledger_cursor_secret: str = Field(default="development-ledger-cursor-secret", validation_alias="PORTFOLIOHUB_LEDGER_CURSOR_SECRET")
     db_enabled: bool = Field(default=False, validation_alias="PORTFOLIOHUB_DB_ENABLED")
     session_token_ttl_hours: int = Field(default=12, validation_alias="PORTFOLIOHUB_SESSION_TOKEN_TTL_HOURS")
     client_token_default_ttl_days: int = Field(default=365, validation_alias="PORTFOLIOHUB_CLIENT_TOKEN_DEFAULT_TTL_DAYS")

app/db/alembic/versions/20260820_0030_portfolio_ledger_keyset_index.py

@@ -0,0 +1,31 @@
+"""portfolio_ledger: keyset pagination index for unified ledger reads
+
+Revision ID: 20260820_0030
+Revises: 20260814_0029
+Create Date: 2026-08-20 10:50:00
+"""
+
+from __future__ import annotations
+
+from alembic import op
+from sqlalchemy import text
+
+revision = "20260820_0030"
+down_revision = "20260814_0029"
+branch_labels = None
+depends_on = None
+
+_INDEX = "idx_portfolio_ledger_portfolio_recorded_entry_desc"
+
+
+def upgrade() -> None:
+    op.get_bind().execute(
+        text(
+            f"CREATE INDEX IF NOT EXISTS {_INDEX} "
+            "ON portfolio_ledger (portfolio_id, recorded_at DESC, ledger_entry_id DESC)"
+        )
+    )
+
+
+def downgrade() -> None:
+    op.get_bind().execute(text(f"DROP INDEX IF EXISTS {_INDEX}"))

app/repositories/portfolio_repository.py

@@ -333,6 +333,16 @@ class PortfolioRepository(Protocol):
 
     def list_manual_ledger_entries(self, portfolio_id: str, limit: int = 50) -> list[LedgerEntry]: ...
 
+    def list_ledger_entries(
+        self,
+        portfolio_id: str,
+        *,
+        source: str | None = None,
+        before_recorded_at: datetime | None = None,
+        before_ledger_entry_id: int | None = None,
+        limit: int = 50,
+    ) -> list[LedgerEntry]: ...
+
     def get_return_summary(self, portfolio_id: str) -> dict[str, float]: ...
 
     def get_return_summaries(self, portfolio_ids: list[str]) -> dict[str, dict[str, float]]: ...

app/repositories/postgres.py

@@ -2210,6 +2210,41 @@ class PostgresPortfolioRepository:
             )
             return [self._row_to_ledger_entry(row) for row in cur.fetchall()]
 
+    def list_ledger_entries(
+        self,
+        portfolio_id: str,
+        *,
+        source: str | None = None,
+        before_recorded_at: datetime | None = None,
+        before_ledger_entry_id: int | None = None,
+        limit: int = 50,
+    ) -> list[LedgerEntry]:
+        self.ensure_schema()
+        clauses = ["portfolio_id = %s"]
+        params: list[Any] = [portfolio_id]
+        if source is not None:
+            clauses.append("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)")
+            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
+                FROM portfolio_ledger
+                WHERE """ + " AND ".join(clauses) + """
+                ORDER BY recorded_at DESC, ledger_entry_id DESC
+                LIMIT %s
+                """,
+                params,
+            )
+            return [self._row_to_ledger_entry(row) for row in cur.fetchall()]
+
     def get_return_summaries(self, portfolio_ids: list[str]) -> dict[str, dict[str, float]]:
         if not portfolio_ids:
             return {}

app/schemas/v1.py

@@ -220,6 +220,35 @@ class ManualLedgerHistoryResponse(BaseModel):
     total: int
 
 
+class LedgerHistoryItem(BaseModel):
+    """Source-neutral, owner-visible ledger fact for GET /ledger."""
+
+    ledger_entry_id: int
+    portfolio_id: str
+    source: str
+    action: str
+    recorded_at: datetime
+    executed_at: datetime | None = None
+    order_id: str | None = None
+    execution_id: str | None = None
+    quantity: float | None = None
+    price: float | None = None
+    fee_amount: float | None = None
+    tax_amount: float | None = None
+    note: str | None = None
+    cash_balance_after: float | None = None
+    asset_quantity_after: float | None = None
+    profit_buffer_after: float | None = None
+    total_value_after: float | None = None
+    target_ledger_entry_id: int | None = None
+    confirmation_evidence: str | None = None
+
+
+class LedgerHistoryResponse(BaseModel):
+    items: list[LedgerHistoryItem]
+    next_cursor: str | None = None
+
+
 class PortfolioReturnSummaryResponse(BaseModel):
     portfolio_id: str
     total_invested: float = Field(