"""Postgres repository implementation for portfolios and ledger."""

from __future__ import annotations

import hashlib
import json
import logging
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any

logger = logging.getLogger(__name__)

import psycopg
from psycopg import sql

from app.domain.models import Asset, AuthToken, BrokerCredential, Daemon, LedgerEntry, Portfolio, PortfolioComment, PortfolioTag, StrategySet, Tag, TokenPrincipal, User

_SELECT_USER = """
    SELECT
        id,
        username,
        email,
        permission,
        is_active,
        password_hash,
        created_at,
        updated_at,
        last_login_at
    FROM users
"""

_SELECT_AUTH_TOKEN = """
    SELECT
        id,
        user_id,
        name,
        token_prefix,
        permission,
        token_type,
        expires_at,
        last_used_at,
        revoked_at,
        created_at,
        runtime_id,
        daemon_id
    FROM auth_tokens
"""

_SELECT_PORTFOLIO = """
    SELECT
        p.portfolio_id,
        p.broker,
        p.status,
        p.name,
        a.symbol,
        p.cash_balance,
        p.asset_quantity,
        p.asset_price,
        p.profit_buffer,
        p.active_strategy_id,
        p.avabile_strategies,
        p.owner_user_id,
        p.created_at,
        p.updated_at,
        p.last_trade_at,
        p.managed_by_daemon_id,
        p.asset_id,
        COALESCE(p.tag_ids, '{}'::uuid[]),
        a.display_name,
        ss.name,
        ss.config,
        a.markets
    FROM portfolios p
    JOIN assets a ON p.asset_id = a.asset_id
    LEFT JOIN strategy_sets ss ON p.active_strategy_id = ss.id
"""

_SELECT_TAG = """
    SELECT
        tag_id,
        name,
        created_by_user_id,
        is_admin_tag,
        is_active,
        created_at
    FROM tags
"""

_SELECT_ASSET = """
    SELECT
        asset_id,
        symbol,
        markets,
        display_name,
        status,
        created_at,
        updated_at,
        COALESCE(tag_ids, '{}'::uuid[])
    FROM assets
"""


@dataclass(frozen=True)
class PortfolioDBConfig:
    host: str
    port: int
    dbname: str
    user: str
    password: str
    sslmode: str
    connect_timeout_sec: int = 10
    application_name: str = "portfoliohub"


class PostgresPortfolioRepository:
    def __init__(self, config: PortfolioDBConfig) -> None:
        self._config = config
        self._connection: psycopg.Connection | None = None

    def _connect(self) -> psycopg.Connection:
        if self._connection is None:
            self._connection = psycopg.connect(
                host=self._config.host,
                port=self._config.port,
                dbname=self._config.dbname,
                user=self._config.user,
                password=self._config.password,
                sslmode=self._config.sslmode,
                connect_timeout=self._config.connect_timeout_sec,
                application_name=self._config.application_name,
                autocommit=True,
            )
        return self._connection

    def close(self) -> None:
        if self._connection is not None:
            self._connection.close()
            self._connection = None

    def ping(self) -> None:
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1")
            cur.fetchone()

    def ensure_schema(self) -> None:
        conn = self._connect()
        with conn.cursor() as cur:
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS users (
                    id TEXT PRIMARY KEY,
                    username TEXT NOT NULL UNIQUE,
                    email TEXT UNIQUE,
                    permission TEXT NOT NULL DEFAULT 'read',
                    is_active BOOLEAN NOT NULL DEFAULT TRUE,
                    password_hash TEXT,
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    last_login_at TIMESTAMPTZ
                )
                """
            )
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS auth_tokens (
                    id TEXT PRIMARY KEY,
                    user_id TEXT NOT NULL,
                    name TEXT NOT NULL,
                    token_prefix TEXT NOT NULL,
                    token_hash TEXT NOT NULL UNIQUE,
                    permission TEXT NOT NULL,
                    token_type TEXT NOT NULL,
                    runtime_id TEXT,
                    expires_at TIMESTAMPTZ,
                    last_used_at TIMESTAMPTZ,
                    revoked_at TIMESTAMPTZ,
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
                )
                """
            )
            cur.execute("ALTER TABLE auth_tokens ADD COLUMN IF NOT EXISTS runtime_id TEXT")
            cur.execute("ALTER TABLE auth_tokens ADD COLUMN IF NOT EXISTS daemon_id TEXT")
            cur.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_auth_tokens_user_created
                ON auth_tokens (user_id, created_at DESC)
                """
            )
            cur.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_auth_tokens_hash_active
                ON auth_tokens (token_hash)
                """
            )
            cur.execute(
                """
                CREATE UNIQUE INDEX IF NOT EXISTS idx_auth_tokens_service_name_unique
                ON auth_tokens (name)
                WHERE token_type = 'service' AND revoked_at IS NULL
                """
            )
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS strategy_sets (
                    id TEXT PRIMARY KEY,
                    name TEXT NOT NULL,
                    config TEXT NOT NULL DEFAULT '',
                    nick TEXT NOT NULL UNIQUE,
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    UNIQUE (name, config)
                )
                """
            )
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS assets (
                    asset_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                    symbol TEXT NOT NULL UNIQUE,
                    markets TEXT[] NOT NULL DEFAULT '{}',
                    display_name TEXT,
                    status TEXT NOT NULL DEFAULT 'active',
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
                )
                """
            )
            cur.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_assets_markets_gin
                ON assets USING GIN (markets)
                """
            )
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS portfolios (
                    portfolio_id TEXT PRIMARY KEY,
                    broker TEXT NOT NULL DEFAULT 'kis',
                    status TEXT NOT NULL DEFAULT 'enabled',
                    name TEXT NOT NULL,
                    asset_id UUID NOT NULL REFERENCES assets(asset_id) ON DELETE RESTRICT,
                    owner_user_id TEXT,
                    cash_balance DOUBLE PRECISION NOT NULL,
                    asset_quantity DOUBLE PRECISION NOT NULL,
                    asset_price DOUBLE PRECISION NOT NULL DEFAULT 0,
                    profit_buffer DOUBLE PRECISION NOT NULL DEFAULT 0,
                    active_strategy_id TEXT DEFAULT NULL,
                    avabile_strategies TEXT[] DEFAULT NULL,
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    last_trade_at TIMESTAMPTZ
                )
                """
            )
            cur.execute("ALTER TABLE portfolios ADD COLUMN IF NOT EXISTS name TEXT")
            cur.execute("ALTER TABLE portfolios ADD COLUMN IF NOT EXISTS owner_user_id TEXT")
            cur.execute("ALTER TABLE portfolios ADD COLUMN IF NOT EXISTS managed_by_daemon_id TEXT")
            cur.execute("ALTER TABLE portfolios ADD COLUMN IF NOT EXISTS asset_id UUID")
            cur.execute("ALTER TABLE portfolios ADD COLUMN IF NOT EXISTS tag_ids UUID[] NOT NULL DEFAULT '{}'")
            cur.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_portfolios_tag_ids_gin
                ON portfolios USING GIN (tag_ids)
                """
            )
            cur.execute("UPDATE portfolios SET name = portfolio_id WHERE name IS NULL OR name = ''")
            cur.execute("ALTER TABLE portfolios ALTER COLUMN name SET NOT NULL")
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS portfolio_ledger (
                    ledger_entry_id BIGINT GENERATED BY DEFAULT AS IDENTITY,
                    portfolio_id TEXT NOT NULL,
                    recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    order_id TEXT NULL,
                    execution_id TEXT NULL,
                    side TEXT NOT NULL,
                    price NUMERIC(20,8) NOT NULL,
                    quantity NUMERIC(20,8) NOT NULL,
                    fee_amount NUMERIC(20,8) NOT NULL DEFAULT 0,
                    tax_amount NUMERIC(20,8) NOT NULL DEFAULT 0,
                    cash_balance_after NUMERIC(20,8) NOT NULL,
                    asset_quantity_after NUMERIC(20,8) NOT NULL,
                    total_value_after NUMERIC(20,8) NOT NULL,
                    profit_buffer_after NUMERIC(20,8) NOT NULL DEFAULT 0,
                    strategy_set_id TEXT NOT NULL DEFAULT 'legacy',
                    note TEXT NOT NULL DEFAULT '',
                    executed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
                )
                """
            )
            cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS ledger_entry_id BIGINT GENERATED BY DEFAULT AS IDENTITY")
            cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS order_id TEXT")
            cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS execution_id TEXT")
            cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS side TEXT")
            cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS executed_at TIMESTAMPTZ")
            cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS source TEXT NOT NULL DEFAULT 'broker'")
            cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS target_ledger_entry_id BIGINT")
            cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS confirmation_evidence TEXT")
            cur.execute("ALTER TABLE portfolio_ledger ADD COLUMN IF NOT EXISTS actor_user_id TEXT")
            cur.execute(
                """
                DO $$
                BEGIN
                    IF EXISTS (
                        SELECT 1
                        FROM information_schema.columns
                        WHERE table_name = 'portfolio_ledger'
                          AND column_name = 'action'
                    ) THEN
                        UPDATE portfolio_ledger SET side = lower(action) WHERE side IS NULL;
                    END IF;
                END $$;
                """
            )
            cur.execute("UPDATE portfolio_ledger SET executed_at = recorded_at WHERE executed_at IS NULL")
            cur.execute("UPDATE portfolio_ledger SET side = 'manual' WHERE side IS NULL")
            cur.execute("ALTER TABLE portfolio_ledger ALTER COLUMN side SET NOT NULL")
            cur.execute("ALTER TABLE portfolio_ledger ALTER COLUMN executed_at SET NOT NULL")
            for column in (
                "price",
                "quantity",
                "fee_amount",
                "tax_amount",
                "cash_balance_after",
                "asset_quantity_after",
                "total_value_after",
                "profit_buffer_after",
            ):
                cur.execute(
                    f"ALTER TABLE portfolio_ledger ALTER COLUMN {column} TYPE NUMERIC(20,8) USING {column}::numeric(20,8)"
                )
            cur.execute("ALTER TABLE portfolio_ledger DROP COLUMN IF EXISTS action")
            cur.execute("DROP TABLE IF EXISTS ledger_trades")

            cur.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_portfolio_ledger_portfolio_time
                ON portfolio_ledger (portfolio_id, recorded_at DESC)
                """
            )
            cur.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_portfolio_ledger_portfolio_executed
                ON portfolio_ledger (portfolio_id, executed_at DESC)
                """
            )

            cur.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_portfolios_broker_status_updated
                ON portfolios (broker, status, updated_at DESC)
                """
            )
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS portfolio_comments (
                    id TEXT PRIMARY KEY,
                    portfolio_id TEXT NOT NULL,
                    content TEXT NOT NULL,
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
                )
                """
            )
            cur.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_portfolio_comments_portfolio_created
                ON portfolio_comments (portfolio_id, created_at DESC)
                """
            )
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS portfolio_tags (
                    portfolio_id TEXT NOT NULL,
                    tag_value TEXT NOT NULL,
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    PRIMARY KEY (portfolio_id, tag_value)
                )
                """
            )
            cur.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_portfolio_tags_tag_portfolio
                ON portfolio_tags (tag_value, portfolio_id)
                """
            )
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS tags (
                    tag_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                    name TEXT NOT NULL,
                    created_by_user_id TEXT NOT NULL,
                    is_admin_tag BOOLEAN NOT NULL DEFAULT FALSE,
                    is_active BOOLEAN NOT NULL DEFAULT TRUE,
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    UNIQUE(name)
                )
                """
            )
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS daemons (
                    id TEXT PRIMARY KEY,
                    owner_user_id TEXT NOT NULL,
                    daemon_slug TEXT NOT NULL,
                    display_name TEXT,
                    is_active BOOLEAN NOT NULL DEFAULT TRUE,
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    UNIQUE (owner_user_id, daemon_slug)
                )
                """
            )
            cur.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_daemons_owner
                ON daemons (owner_user_id)
                """
            )
            # NOTE(REQ-047 Step2 / BAT-021): daemon-scoped PK 전환 이후 최종 스키마.
            # 이 CREATE TABLE IF NOT EXISTS는 Step2 적용 환경에서 테이블이 이미 존재해 스킵된다.
            # Step2 미적용 환경에서는 alembic revision(Step1→Step2)을 먼저 적용해야 한다.
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS broker_credentials (
                    daemon_id TEXT NOT NULL,
                    broker TEXT NOT NULL,
                    user_id TEXT NOT NULL,
                    account_ref TEXT NOT NULL,
                    encrypted_payload TEXT NOT NULL,
                    credential_version INTEGER NOT NULL DEFAULT 1,
                    last_rotated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    PRIMARY KEY (daemon_id, broker)
                )
                """
            )
            cur.execute("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'timescaledb')")
            if cur.fetchone()[0]:
                cur.execute(
                    """
                    SELECT create_hypertable('portfolio_ledger', 'recorded_at', if_not_exists => TRUE)
                    """
                )

    @staticmethod
    def _row_to_user(row: Any) -> User:
        if row is None:
            raise ValueError("사용자를 찾을 수 없습니다")

        return User(
            id=row[0],
            username=row[1],
            email=row[2],
            permission=row[3] if len(row) > 3 else "read",
            is_active=bool(row[4]) if len(row) > 4 else True,
            password_hash=row[5] if len(row) > 5 else None,
            created_at=row[6] if len(row) > 6 else None,
            updated_at=row[7] if len(row) > 7 else None,
            last_login_at=row[8] if len(row) > 8 else None,
        )

    @staticmethod
    def _row_to_auth_token(row: Any) -> AuthToken:
        if row is None:
            raise ValueError("토큰을 찾을 수 없습니다")
        return AuthToken(
            id=row[0],
            user_id=row[1],
            name=row[2],
            token_prefix=row[3],
            permission=row[4],
            token_type=row[5],
            expires_at=row[6] if len(row) > 6 else None,
            last_used_at=row[7] if len(row) > 7 else None,
            revoked_at=row[8] if len(row) > 8 else None,
            created_at=row[9] if len(row) > 9 else None,
            runtime_id=row[10] if len(row) > 10 else None,
            daemon_id=row[11] if len(row) > 11 else None,
        )

    @staticmethod
    def _row_to_broker_credential(row: Any) -> BrokerCredential:
        """Step2(REQ-047) 이후 컬럼 순서: daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at."""
        if row is None:
            raise ValueError("브로커 인증 정보를 찾을 수 없습니다")
        return BrokerCredential(
            daemon_id=row[0],
            user_id=row[1],
            broker=row[2],
            account_ref=row[3],
            encrypted_payload=row[4],
            credential_version=int(row[5]) if len(row) > 5 else 1,
            last_rotated_at=row[6] if len(row) > 6 else None,
        )

    @staticmethod
    def _row_to_portfolio(row: Any) -> Portfolio:
        if row is None:
            raise ValueError("포트폴리오를 찾을 수 없습니다")

        raw_available = row[10] if len(row) > 10 else None
        if raw_available is None:
            avabile_strategies = None
        elif isinstance(raw_available, list):
            avabile_strategies = [str(item) for item in raw_available]
        else:
            avabile_strategies = [str(raw_available)]

        # REQ-055: strategy_name (row[19] = ss.name from LEFT JOIN strategy_sets)
        strategy_name: str | None = row[19] if len(row) > 19 else None

        # REQ-055: strategy_params — JSON 파싱 (파싱 불가 시 {} + WARNING, 500 금지)
        _config_raw = row[20] if len(row) > 20 else None
        if not _config_raw:
            strategy_params: dict[str, Any] = {}
        else:
            try:
                _parsed = json.loads(_config_raw)
                strategy_params = _parsed if isinstance(_parsed, dict) else {}
            except (json.JSONDecodeError, ValueError):
                logger.warning(
                    "strategy_sets.config 파싱 실패 — portfolio_id=%s, active_strategy_id=%s, config=%r",
                    row[0],
                    row[9] if len(row) > 9 else None,
                    _config_raw[:200] if isinstance(_config_raw, str) else _config_raw,
                )
                strategy_params = {}

        # REQ-055: markets (row[21] = a.markets from assets JOIN)
        _markets_raw = row[21] if len(row) > 21 else None
        markets: list[str] = list(_markets_raw) if _markets_raw else []

        return Portfolio(
            portfolio_id=row[0],
            broker=row[1] or "kis",
            status=row[2] or "enabled",
            name=row[3],
            symbol=row[4],
            tags=None,
            cash_balance=float(row[5]),
            asset_quantity=float(row[6]),
            price=float(row[7]),
            profit_buffer=float(row[8]),
            active_strategy_id=row[9] if len(row) > 9 else None,
            avabile_strategies=avabile_strategies,
            owner_user_id=row[11] if len(row) > 11 else None,
            created_at=row[12] if len(row) > 12 else None,
            updated_at=row[13] if len(row) > 13 else None,
            last_trade_at=row[14] if len(row) > 14 else None,
            managed_by_daemon_id=row[15] if len(row) > 15 else None,
            asset_id=str(row[16]) if len(row) > 16 and row[16] is not None else None,
            tag_ids=[str(t) for t in row[17]] if len(row) > 17 and row[17] is not None else [],
            display_name=row[18] if len(row) > 18 else None,
            strategy_name=strategy_name,
            strategy_params=strategy_params,
            markets=markets,
        )

    @staticmethod
    def _row_to_portfolio_tag(row: Any) -> PortfolioTag:
        if row is None:
            raise ValueError("태그를 찾을 수 없습니다")
        return PortfolioTag(portfolio_id=row[0], tag=row[1], created_at=row[2] if len(row) > 2 else None)

    @staticmethod
    def _row_to_asset(row: Any) -> Asset:
        if row is None:
            raise ValueError("asset을 찾을 수 없습니다")
        return Asset(
            asset_id=str(row[0]),
            symbol=row[1],
            markets=list(row[2]) if row[2] else [],
            display_name=row[3],
            status=row[4] or "active",
            created_at=row[5],
            updated_at=row[6],
            tag_ids=[str(t) for t in row[7]] if len(row) > 7 and row[7] is not None else [],
        )

    @staticmethod
    def _row_to_tag(row: Any) -> Tag:
        if row is None:
            raise ValueError("태그를 찾을 수 없습니다")
        return Tag(
            tag_id=str(row[0]),
            name=row[1],
            created_by_user_id=row[2],
            is_admin_tag=bool(row[3]),
            is_active=bool(row[4]),
            created_at=row[5] if len(row) > 5 else None,
        )

    def _load_tags_for_portfolios(self, cur: psycopg.Cursor, portfolio_ids: list[str]) -> dict[str, list[str]]:
        if not portfolio_ids:
            return {}
        cur.execute(
            """
            SELECT portfolio_id, tag_value
            FROM portfolio_tags
            WHERE portfolio_id = ANY(%s)
            ORDER BY tag_value ASC
            """,
            (portfolio_ids,),
        )
        tags_by_portfolio: dict[str, list[str]] = {portfolio_id: [] for portfolio_id in portfolio_ids}
        for portfolio_id, tag_value in cur.fetchall():
            tags_by_portfolio.setdefault(str(portfolio_id), []).append(str(tag_value))
        return tags_by_portfolio

    def _attach_tags(self, cur: psycopg.Cursor, portfolios: list[Portfolio]) -> list[Portfolio]:
        tags_by_portfolio = self._load_tags_for_portfolios(cur, [portfolio.portfolio_id for portfolio in portfolios])
        for portfolio in portfolios:
            portfolio.tags = tags_by_portfolio.get(portfolio.portfolio_id, [])
        return portfolios

    @staticmethod
    def _row_to_ledger_entry_legacy(row: Any) -> LedgerEntry:
        return LedgerEntry(
            ledger_entry_id=0,
            portfolio_id=row[0],
            order_id=None,
            execution_id=None,
            side=row[2],
            quantity=float(row[4]),
            price=float(row[3]),
            fee_amount=float(row[5]),
            tax_amount=float(row[6]),
            executed_at=row[1],
            recorded_at=row[1],
            cash_balance_after=float(row[7]),
            asset_quantity_after=float(row[8]),
            total_value_after=float(row[9]),
            profit_buffer_after=float(row[10]),
            strategy_set_id=row[11] or "legacy",
            note=row[12] or "",
        )

    def _ensure_user_uniqueness(
        self,
        cur: psycopg.Cursor,
        *,
        username: str | None = None,
        email: str | None = None,
        exclude_user_id: str | None = None,
    ) -> None:
        if username is not None:
            if exclude_user_id is None:
                cur.execute("SELECT 1 FROM users WHERE username = %s", (username,))
            else:
                cur.execute("SELECT 1 FROM users WHERE username = %s AND id <> %s", (username, exclude_user_id))
            if cur.fetchone() is not None:
                raise ValueError(f"이미 존재하는 username입니다: {username}")

        if email is not None:
            if exclude_user_id is None:
                cur.execute("SELECT 1 FROM users WHERE email = %s", (email,))
            else:
                cur.execute("SELECT 1 FROM users WHERE email = %s AND id <> %s", (email, exclude_user_id))
            if cur.fetchone() is not None:
                raise ValueError(f"이미 존재하는 email입니다: {email}")

    def get_user(self, user_id: str) -> User:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_USER + "WHERE id = %s", (user_id,))
            row = cur.fetchone()
        return self._row_to_user(row)

    def get_user_by_username(self, username: str) -> User | None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_USER + "WHERE username = %s", (username,))
            row = cur.fetchone()
        if row is None:
            return None
        return self._row_to_user(row)

    def list_users(self) -> list[User]:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_USER + "ORDER BY created_at DESC, username ASC")
            rows = cur.fetchall()
        return [self._row_to_user(row) for row in rows]

    def create_user(
        self,
        username: str,
        *,
        email: str | None = None,
        password_hash: str | None = None,
        permission: str = "read",
        is_active: bool = True,
    ) -> User:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            self._ensure_user_uniqueness(cur, username=username, email=email)
            user_id = str(uuid.uuid4())
            cur.execute(
                """
                INSERT INTO users (id, username, email, permission, is_active, password_hash)
                VALUES (%s, %s, %s, %s, %s, %s)
                RETURNING id, username, email, permission, is_active, password_hash, created_at, updated_at, last_login_at
                """,
                (user_id, username, email, permission, is_active, password_hash),
            )
            row = cur.fetchone()
        return self._row_to_user(row)

    def update_user(
        self,
        user_id: str,
        *,
        username: str | None = None,
        email: str | None = None,
        password_hash: str | None = None,
        permission: str | None = None,
        is_active: bool | None = None,
    ) -> User:
        self.ensure_schema()
        assignments: list[str] = []
        params: list[object] = []
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_USER + "WHERE id = %s", (user_id,))
            existing_row = cur.fetchone()
            if existing_row is None:
                raise ValueError("사용자를 찾을 수 없습니다")

            self._ensure_user_uniqueness(cur, username=username, email=email, exclude_user_id=user_id)

            if username is not None:
                assignments.append("username = %s")
                params.append(username)
            if email is not None:
                assignments.append("email = %s")
                params.append(email)
            if password_hash is not None:
                assignments.append("password_hash = %s")
                params.append(password_hash)
            if permission is not None:
                assignments.append("permission = %s")
                params.append(permission)
            if is_active is not None:
                assignments.append("is_active = %s")
                params.append(is_active)

            if not assignments:
                return self._row_to_user(existing_row)

            assignments.append("updated_at = NOW()")
            cur.execute(
                f"""
                UPDATE users
                SET {', '.join(assignments)}
                WHERE id = %s
                RETURNING id, username, email, permission, is_active, password_hash, created_at, updated_at, last_login_at
                """,
                (*params, user_id),
            )
            row = cur.fetchone()
        return self._row_to_user(row)

    def delete_user(self, user_id: str) -> None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM users WHERE id = %s", (user_id,))
            if cur.fetchone() is None:
                raise ValueError("사용자를 찾을 수 없습니다")
            cur.execute("DELETE FROM auth_tokens WHERE user_id = %s", (user_id,))
            cur.execute("DELETE FROM users WHERE id = %s", (user_id,))

    def update_user_last_login(self, user_id: str) -> None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                "UPDATE users SET last_login_at = NOW(), updated_at = NOW() WHERE id = %s",
                (user_id,),
            )

    def upsert_broker_credential(
        self,
        user_id: str,
        broker: str,
        *,
        account_ref: str,
        encrypted_payload: str,
        daemon_id: str | None = None,
    ) -> BrokerCredential:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM users WHERE id = %s", (user_id,))
            if cur.fetchone() is None:
                raise ValueError("사용자를 찾을 수 없습니다")
            # NOTE(REQ-047 Step2 / BAT-021): PK 전환 이후 ON CONFLICT (daemon_id, broker).
            # daemon_id는 NOT NULL(Step2 제약). Step2 미적용 환경에서는 이 코드를 배포하지 마라.
            # 배포 순서: Step1 alembic upgrade → backfill CLI → Step2 alembic upgrade → 코드 배포.
            cur.execute(
                """
                INSERT INTO broker_credentials (daemon_id, user_id, broker, account_ref, encrypted_payload)
                VALUES (%s, %s, %s, %s, %s)
                ON CONFLICT (daemon_id, broker) DO UPDATE
                SET account_ref = EXCLUDED.account_ref,
                    encrypted_payload = EXCLUDED.encrypted_payload,
                    user_id = EXCLUDED.user_id,
                    credential_version = broker_credentials.credential_version + 1,
                    last_rotated_at = NOW()
                RETURNING daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
                """,
                (daemon_id, user_id, broker, account_ref, encrypted_payload),
            )
            row = cur.fetchone()
        return self._row_to_broker_credential(row)

    def list_broker_credentials(
        self,
        user_id: str,
        daemon_id: str | None = None,
    ) -> list[BrokerCredential]:
        self.ensure_schema()
        # NOTE(REQ-047 Step2 / BAT-021): daemon_id 컬럼은 Step2 이후 항상 존재(NOT NULL).
        # Step2 미적용 환경에서는 이 코드를 배포하지 마라.
        with self._connect().cursor() as cur:
            if daemon_id is not None:
                cur.execute(
                    """
                    SELECT daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
                    FROM broker_credentials
                    WHERE user_id = %s AND daemon_id = %s
                    ORDER BY broker ASC
                    """,
                    (user_id, daemon_id),
                )
            else:
                cur.execute(
                    """
                    SELECT daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
                    FROM broker_credentials
                    WHERE user_id = %s
                    ORDER BY broker ASC
                    """,
                    (user_id,),
                )
            rows = cur.fetchall()
        return [self._row_to_broker_credential(row) for row in rows]

    def update_broker_credential_daemon_id(
        self,
        user_id: str,
        broker: str,
        daemon_id: str,
    ) -> bool:
        """백필용: 특정 (user_id, broker) 인증정보에 daemon_id를 설정한다 (REQ-048)."""
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                UPDATE broker_credentials
                SET daemon_id = %s
                WHERE user_id = %s AND broker = %s AND daemon_id IS NULL
                RETURNING user_id
                """,
                (daemon_id, user_id, broker),
            )
            row = cur.fetchone()
        return row is not None

    def delete_broker_credential(
        self,
        user_id: str,
        broker: str,
        daemon_id: str | None = None,
    ) -> bool:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            if daemon_id is not None:
                cur.execute(
                    "DELETE FROM broker_credentials WHERE daemon_id = %s AND broker = %s RETURNING user_id",
                    (daemon_id, broker),
                )
            else:
                cur.execute(
                    "DELETE FROM broker_credentials WHERE user_id = %s AND broker = %s RETURNING user_id",
                    (user_id, broker),
                )
            row = cur.fetchone()
        return row is not None

    def create_auth_token(
        self,
        user_id: str,
        *,
        name: str,
        token_hash: str,
        token_prefix: str,
        permission: str,
        token_type: str,
        expires_at: datetime | None = None,
        runtime_id: str | None = None,
        daemon_id: str | None = None,
    ) -> AuthToken:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM users WHERE id = %s", (user_id,))
            if cur.fetchone() is None:
                raise ValueError("사용자를 찾을 수 없습니다")
            token_id = str(uuid.uuid4())
            cur.execute(
                """
                INSERT INTO auth_tokens (
                    id, user_id, name, token_prefix, token_hash, permission, token_type,
                    expires_at, runtime_id, daemon_id
                )
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                RETURNING
                    id, user_id, name, token_prefix, permission, token_type,
                    expires_at, last_used_at, revoked_at, created_at, runtime_id, daemon_id
                """,
                (token_id, user_id, name, token_prefix, token_hash, permission, token_type,
                 expires_at, runtime_id, daemon_id),
            )
            row = cur.fetchone()
        return self._row_to_auth_token(row)

    def list_auth_tokens(self, user_id: str) -> list[AuthToken]:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                _SELECT_AUTH_TOKEN + "WHERE user_id = %s ORDER BY created_at DESC, id ASC",
                (user_id,),
            )
            rows = cur.fetchall()
        return [self._row_to_auth_token(row) for row in rows]

    def list_service_tokens(
        self,
        user_id: str | None = None,
        *,
        include_revoked: bool = False,
    ) -> list[AuthToken]:
        self.ensure_schema()
        conditions = ["token_type = 'service'"]
        params: list = []
        if user_id is not None:
            conditions.append("user_id = %s")
            params.append(user_id)
        if not include_revoked:
            conditions.append("revoked_at IS NULL")
        where = "WHERE " + " AND ".join(conditions) + " "
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_AUTH_TOKEN + where + "ORDER BY created_at DESC, id ASC", params)
            rows = cur.fetchall()
        return [self._row_to_auth_token(row) for row in rows]

    def revoke_auth_token(self, user_id: str, token_id: str) -> None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                "UPDATE auth_tokens SET revoked_at = NOW() WHERE id = %s AND user_id = %s RETURNING id",
                (token_id, user_id),
            )
            row = cur.fetchone()
            if row is None:
                raise ValueError("토큰을 찾을 수 없습니다")

    def get_service_token_by_name(self, name: str) -> AuthToken | None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                _SELECT_AUTH_TOKEN + "WHERE name = %s AND token_type = 'service' AND revoked_at IS NULL ORDER BY created_at DESC LIMIT 1",
                (name,),
            )
            row = cur.fetchone()
        if row is None:
            return None
        return self._row_to_auth_token(row)

    def get_token_principal(self, token_hash: str) -> TokenPrincipal | None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                SELECT t.id, t.user_id, t.permission, t.token_type, t.runtime_id, t.daemon_id
                FROM auth_tokens t
                WHERE token_hash = %s
                  AND t.revoked_at IS NULL
                  AND (t.expires_at IS NULL OR t.expires_at > NOW())
                """,
                (token_hash,),
            )
            row = cur.fetchone()
            if row is None:
                return None
            cur.execute("UPDATE auth_tokens SET last_used_at = NOW() WHERE id = %s", (row[0],))
        return TokenPrincipal(
            token_id=row[0],
            user_id=row[1],
            permission=row[2],
            token_type=row[3],
            runtime_id=row[4] if len(row) > 4 else None,
            daemon_id=row[5] if len(row) > 5 else None,
        )

    def get(self, portfolio_id: str) -> Portfolio:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_PORTFOLIO + "WHERE p.portfolio_id = %s", (portfolio_id,))
            row = cur.fetchone()
            portfolio = self._row_to_portfolio(row)
            self._attach_tags(cur, [portfolio])
        return portfolio

    def get_portfolio_by_name_and_user(self, owner_user_id: str, name: str) -> Portfolio | None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                _SELECT_PORTFOLIO + "WHERE p.owner_user_id = %s AND p.name = %s",
                (owner_user_id, name),
            )
            row = cur.fetchone()
            if row is None:
                return None
            portfolio = self._row_to_portfolio(row)
            self._attach_tags(cur, [portfolio])
        return portfolio

    def get_or_create(
        self,
        portfolio_id: str,
        symbol: str,
        *,
        name: str | None = None,
        broker: str = "kis",
        status: str = "enabled",
        active_strategy_id: str | None = None,
        avabile_strategies: list[str] | None = None,
        owner_user_id: str | None = None,
        initial_cash: float = 0.0,
    ) -> Portfolio:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT asset_id FROM assets WHERE symbol = %s", (symbol,))
            row = cur.fetchone()
            if row is None:
                raise ValueError(f"asset not found for symbol: {symbol}")
            asset_id = row[0]
            cur.execute(
                """
                INSERT INTO portfolios (
                    portfolio_id, broker, status, name, asset_id,
                    owner_user_id, cash_balance, asset_quantity, asset_price, profit_buffer,
                    active_strategy_id, avabile_strategies
                )
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                ON CONFLICT (portfolio_id) DO NOTHING
                """,
                (
                    portfolio_id,
                    broker,
                    status,
                    name or symbol,
                    asset_id,
                    owner_user_id,
                    float(initial_cash),
                    0.0,
                    0.0,
                    0.0,
                    active_strategy_id,
                    avabile_strategies,
                ),
            )
            cur.execute(_SELECT_PORTFOLIO + "WHERE p.portfolio_id = %s", (portfolio_id,))
            row = cur.fetchone()
        return self._row_to_portfolio(row)

    def create_portfolio(
        self,
        portfolio_id: str,
        *,
        owner_user_id: str,
        broker: str,
        status: str,
        name: str,
        asset_id: str,
        cash_balance: float,
        asset_quantity: float,
        asset_price: float,
        profit_buffer: float,
        active_strategy_id: str | None,
        avabile_strategies: list[str] | None,
    ) -> Portfolio:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            if cur.fetchone() is not None:
                raise ValueError(f"이미 존재하는 포트폴리오입니다: {portfolio_id}")
            cur.execute(
                """
                INSERT INTO portfolios (
                    portfolio_id, broker, status, name, asset_id,
                    owner_user_id, cash_balance, asset_quantity, asset_price, profit_buffer,
                    active_strategy_id, avabile_strategies
                ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                """,
                (
                    portfolio_id,
                    broker,
                    status,
                    name,
                    asset_id,
                    owner_user_id,
                    float(cash_balance),
                    float(asset_quantity),
                    float(asset_price),
                    float(profit_buffer),
                    active_strategy_id,
                    avabile_strategies,
                ),
            )
        portfolio = self.get(portfolio_id)
        portfolio.tags = []
        return portfolio

    def update_portfolio(self, portfolio_id: str, **fields: object) -> Portfolio:
        self.ensure_schema()
        allowed = {
            "broker": "broker",
            "status": "status",
            "name": "name",
            "asset_price": "asset_price",
            "active_strategy_id": "active_strategy_id",
            "avabile_strategies": "avabile_strategies",
            "managed_by_daemon_id": "managed_by_daemon_id",
        }
        assignments: list[str] = []
        params: list[object] = []
        for key, value in fields.items():
            if value is None or key not in allowed:
                continue
            if key == "avabile_strategies":
                assignments.append("avabile_strategies = %s")
                params.append(value)
            else:
                assignments.append(f"{allowed[key]} = %s")
                params.append(value)
        if not assignments:
            return self.get(portfolio_id)

        assignments.append("updated_at = NOW()")
        with self._connect().cursor() as cur:
            cur.execute(
                f"""
                UPDATE portfolios
                SET {', '.join(assignments)}
                WHERE portfolio_id = %s
                """,
                (*params, portfolio_id),
            )
        return self.get(portfolio_id)

    def list_portfolios(
        self,
        *,
        broker: str | None = None,
        status: str | None = None,
        owner_user_id: str | None = None,
        include_tags: list[str] | None = None,
        exclude_tags: list[str] | None = None,
        match_mode: str | None = None,
        managed_by_daemon_id: str | None = None,
        market: str | None = None,
    ) -> list[Portfolio]:
        self.ensure_schema()
        query = _SELECT_PORTFOLIO
        clauses: list[str] = []
        params: list[object] = []
        if broker is not None:
            clauses.append("p.broker = %s")
            params.append(broker)
        if status is not None:
            clauses.append("p.status = %s")
            params.append(status)
        if owner_user_id is not None:
            clauses.append("p.owner_user_id = %s")
            params.append(owner_user_id)
        if managed_by_daemon_id is not None:
            clauses.append("p.managed_by_daemon_id = %s")
            params.append(managed_by_daemon_id)
        normalized_mode = (match_mode or "OR").upper()
        if include_tags:
            if normalized_mode == "AND":
                clauses.append(
                    """
                    p.portfolio_id IN (
                        SELECT portfolio_id
                        FROM portfolio_tags
                        WHERE tag_value = ANY(%s)
                        GROUP BY portfolio_id
                        HAVING COUNT(DISTINCT tag_value) = %s
                    )
                    """
                )
                params.append(include_tags)
                params.append(len(set(include_tags)))
            else:
                clauses.append(
                    "p.portfolio_id IN (SELECT DISTINCT portfolio_id FROM portfolio_tags WHERE tag_value = ANY(%s))"
                )
                params.append(include_tags)
        if exclude_tags:
            clauses.append(
                "p.portfolio_id NOT IN (SELECT DISTINCT portfolio_id FROM portfolio_tags WHERE tag_value = ANY(%s))"
            )
            params.append(exclude_tags)
        if market is not None:
            clauses.append("%s = ANY(a.markets)")
            params.append(market)
        if clauses:
            query += "WHERE " + " AND ".join(clauses) + " "
        query += "ORDER BY p.updated_at DESC, p.portfolio_id ASC"
        with self._connect().cursor() as cur:
            cur.execute(query, tuple(params))
            rows = cur.fetchall()
            portfolios = [self._row_to_portfolio(row) for row in rows]
            self._attach_tags(cur, portfolios)
        return portfolios

    def add_tag(self, portfolio_id: str, *, tag: str) -> PortfolioTag:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            if cur.fetchone() is None:
                raise ValueError("포트폴리오를 찾을 수 없습니다")
            try:
                cur.execute(
                    """
                    INSERT INTO portfolio_tags (portfolio_id, tag_value)
                    VALUES (%s, %s)
                    RETURNING portfolio_id, tag_value, created_at
                    """,
                    (portfolio_id, tag),
                )
            except psycopg.errors.UniqueViolation as exc:
                raise ValueError(f"이미 연결된 태그입니다: {tag}") from exc
            row = cur.fetchone()
        return self._row_to_portfolio_tag(row)

    def list_tags(self, portfolio_id: str) -> list[PortfolioTag]:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            if cur.fetchone() is None:
                raise ValueError("포트폴리오를 찾을 수 없습니다")
            cur.execute(
                """
                SELECT portfolio_id, tag_value, created_at
                FROM portfolio_tags
                WHERE portfolio_id = %s
                ORDER BY tag_value ASC
                """,
                (portfolio_id,),
            )
            rows = cur.fetchall()
        return [self._row_to_portfolio_tag(row) for row in rows]

    def remove_tag(self, portfolio_id: str, *, tag: str) -> None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            if cur.fetchone() is None:
                raise ValueError("포트폴리오를 찾을 수 없습니다")
            cur.execute(
                "DELETE FROM portfolio_tags WHERE portfolio_id = %s AND tag_value = %s RETURNING portfolio_id",
                (portfolio_id, tag),
            )
            if cur.fetchone() is None:
                raise ValueError("태그를 찾을 수 없습니다")

    def get_or_create_strategy_set(self, name: str, config: str) -> StrategySet:
        self.ensure_schema()
        nick = f"{name}:{hashlib.sha1(config.encode('utf-8')).hexdigest()[:12]}"
        with self._connect().cursor() as cur:
            new_id = str(uuid.uuid4())
            cur.execute(
                """
                INSERT INTO strategy_sets (id, name, config, nick)
                VALUES (%s, %s, %s, %s)
                ON CONFLICT (name, config) DO NOTHING
                """,
                (new_id, name, config, nick),
            )
            cur.execute(
                """
                SELECT id, name, config, nick, created_at
                FROM strategy_sets
                WHERE name = %s AND config = %s
                """,
                (name, config),
            )
            row = cur.fetchone()
        return StrategySet(id=row[0], name=row[1], config=row[2], nick=row[3], created_at=row[4])

    def get_strategy_set(self, id: str) -> StrategySet | None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                "SELECT id, name, config, nick, created_at FROM strategy_sets WHERE id = %s",
                (id,),
            )
            row = cur.fetchone()
        if row is None:
            return None
        return StrategySet(id=row[0], name=row[1], config=row[2], nick=row[3], created_at=row[4])

    def list_strategy_sets(self) -> list[StrategySet]:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                "SELECT id, name, config, nick, created_at FROM strategy_sets ORDER BY created_at DESC"
            )
            rows = cur.fetchall()
        return [StrategySet(id=row[0], name=row[1], config=row[2], nick=row[3], created_at=row[4]) for row in rows]

    def delete_strategy_set(self, id: str) -> None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            # 1. Check if this strategy set is referenced by any portfolio
            cur.execute(
                """
                SELECT COUNT(*) FROM portfolios
                WHERE active_strategy_id = %s
                   OR avabile_strategies @> ARRAY[%s]
                """,
                (id, id),
            )
            count = cur.fetchone()[0]
            if count > 0:
                raise ValueError(f"Strategy set is in use by {count} portfolio(s) and cannot be deleted")
            
            # 2. If not in use, proceed with deletion
            cur.execute("DELETE FROM strategy_sets WHERE id = %s", (id,))

    def save(
        self,
        portfolio: Portfolio,
        *,
        current_price: float,
        action: str,
        trade_price: float,
        quantity: float,
        fee_amount: float,
        tax_amount: float,
        total_value_after: float,
        strategy_set_id: str = "manual",
        note: str = "",
    ) -> None:
        self.ensure_schema()
        trade_time = datetime.now(timezone.utc)
        with self._connect().cursor() as cur:
            cur.execute(
                """
                UPDATE portfolios
                SET broker = %s,
                    status = %s,
                    cash_balance = %s,
                    asset_quantity = %s,
                    asset_price = %s,
                    profit_buffer = %s,
                    active_strategy_id = %s,
                    avabile_strategies = %s,
                    updated_at = %s,
                    last_trade_at = CASE WHEN %s IN ('BUY', 'SELL') THEN %s ELSE last_trade_at END
                WHERE portfolio_id = %s
                """,
                (
                    portfolio.broker,
                    portfolio.status,
                    float(portfolio.cash_balance),
                    float(portfolio.asset_quantity),
                    float(current_price),
                    float(portfolio.profit_buffer),
                    portfolio.active_strategy_id,
                    portfolio.avabile_strategies,
                    trade_time,
                    action,
                    trade_time,
                    portfolio.portfolio_id,
                ),
            )
            if action in {"BUY", "SELL"}:
                cur.execute(
                    """
                    INSERT INTO portfolio_ledger (
                        portfolio_id, recorded_at, side, price, quantity, fee_amount,
                        tax_amount, cash_balance_after, asset_quantity_after,
                        total_value_after, profit_buffer_after, strategy_set_id, note, executed_at
                    ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                    """,
                    (
                        portfolio.portfolio_id,
                        trade_time,
                        str(action).lower(),
                        float(trade_price),
                        float(quantity),
                        float(fee_amount),
                        float(tax_amount),
                        float(portfolio.cash_balance),
                        float(portfolio.asset_quantity),
                        float(total_value_after),
                        float(portfolio.profit_buffer),
                        strategy_set_id,
                        note,
                        trade_time,
                    ),
                )

    def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                SELECT
                    ledger_entry_id, portfolio_id, order_id, execution_id, side,
                    quantity::float8, price::float8,
                    fee_amount::float8, tax_amount::float8,
                    executed_at, recorded_at,
                    cash_balance_after::float8, asset_quantity_after::float8,
                    total_value_after::float8, profit_buffer_after::float8,
                    strategy_set_id, note
                FROM portfolio_ledger
                WHERE portfolio_id = %s
                ORDER BY recorded_at DESC, ledger_entry_id DESC
                LIMIT %s
                """,
                (portfolio_id, int(limit)),
            )
            rows = cur.fetchall()
        return [self._row_to_ledger_entry(row) for row in rows]

    def list_trades_by_date(
        self,
        portfolio_id: str,
        *,
        date: "datetime | None" = None,
        from_: "datetime | None" = None,
        to_: "datetime | None" = None,
    ) -> list[LedgerEntry]:
        """Return ledger entries for portfolio_id filtered by date range (REQ-039).

        Either `date` (single day) or `from_`/`to_` (range, both inclusive) must be
        provided. The caller is responsible for converting date strings to datetime
        boundaries (start of day / start of next day).
        """
        self.ensure_schema()
        clauses = ["portfolio_id = %s"]
        params: list[object] = [portfolio_id]

        if date is not None:
            # Single day: [date 00:00:00, date+1day 00:00:00)
            clauses.append("executed_at >= %s")
            params.append(date)
            import datetime as _dt
            next_day = date + _dt.timedelta(days=1)
            clauses.append("executed_at < %s")
            params.append(next_day)
        else:
            if from_ is not None:
                clauses.append("executed_at >= %s")
                params.append(from_)
            if to_ is not None:
                import datetime as _dt
                # to_ is inclusive — extend to end of that day
                next_day = to_ + _dt.timedelta(days=1)
                clauses.append("executed_at < %s")
                params.append(next_day)

        where = "WHERE " + " AND ".join(clauses)
        with self._connect().cursor() as cur:
            cur.execute(
                f"""
                SELECT
                    ledger_entry_id, portfolio_id, order_id, execution_id, side,
                    quantity::float8, price::float8,
                    fee_amount::float8, tax_amount::float8,
                    executed_at, recorded_at,
                    cash_balance_after::float8, asset_quantity_after::float8,
                    total_value_after::float8, profit_buffer_after::float8,
                    strategy_set_id, note
                FROM portfolio_ledger
                {where}
                ORDER BY executed_at ASC, ledger_entry_id ASC
                """,
                tuple(params),
            )
            rows = cur.fetchall()
        return [self._row_to_ledger_entry(row) for row in rows]

    def get_all_trades_for_pnl(
        self,
        portfolio_id: str,
        *,
        symbol: str | None = None,
    ) -> list[LedgerEntry]:
        """Return all buy/sell ledger entries for portfolio_id for WAC PnL calculation (REQ-040/041).

        Returns ALL buy/sell history (not bounded by from_/to_) so that the WAC
        calculation is accurate. The caller (domain layer) applies the date filter
        to sell events when collecting realized PnL items.

        portfolio_ledger does not store symbol per row — symbol filtering is handled
        at the portfolio level (one portfolio = one symbol). The `symbol` parameter
        is forwarded to the domain layer via the returned rows; it is not used in SQL.
        """
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                SELECT
                    ledger_entry_id, portfolio_id, order_id, execution_id, side,
                    quantity::float8, price::float8,
                    fee_amount::float8, tax_amount::float8,
                    executed_at, recorded_at,
                    cash_balance_after::float8, asset_quantity_after::float8,
                    total_value_after::float8, profit_buffer_after::float8,
                    strategy_set_id, note
                FROM portfolio_ledger
                WHERE portfolio_id = %s
                  AND side IN ('buy', 'sell')
                ORDER BY executed_at ASC, ledger_entry_id ASC
                """,
                (portfolio_id,),
            )
            rows = cur.fetchall()
        return [self._row_to_ledger_entry(row) for row in rows]

    def insert_ledger_trade(
        self,
        *,
        portfolio_id: str,
        order_id: str,
        execution_id: str,
        side: str,
        quantity: float,
        price: float,
        executed_at,
        fee_amount: float = 0.0,
        tax_amount: float = 0.0,
    ) -> tuple[LedgerEntry, bool]:
        self.ensure_schema()
        side_normalized = str(side).lower()
        if side_normalized not in {"buy", "sell"}:
            raise ValueError("side는 buy 또는 sell 이어야 합니다")

        conn = self._connect()
        # Use a transaction block (not `with conn:`) so the shared connection stays open:
        # psycopg3 closes the connection when it is used as a context manager, which would
        # leave the repository's reused connection unusable for subsequent calls.
        with conn.transaction():
            with conn.cursor() as cur:
                cur.execute(
                    """
                    SELECT
                        portfolio_id, cash_balance, asset_quantity, asset_price, profit_buffer,
                        active_strategy_id
                    FROM portfolios
                    WHERE portfolio_id = %s
                    FOR UPDATE
                    """,
                    (portfolio_id,),
                )
                portfolio_row = cur.fetchone()
                if portfolio_row is None:
                    raise ValueError("포트폴리오를 찾을 수 없습니다")

                cash_balance = float(portfolio_row[1])
                asset_quantity = float(portfolio_row[2])
                price_float = float(price)
                qty_float = float(quantity)
                fee_float = float(fee_amount)
                tax_float = float(tax_amount)

                if side_normalized == "buy":
                    cash_balance_after = cash_balance - (qty_float * price_float + fee_float + tax_float)
                    asset_quantity_after = asset_quantity + qty_float
                else:
                    cash_balance_after = cash_balance + (qty_float * price_float - fee_float - tax_float)
                    asset_quantity_after = asset_quantity - qty_float

                total_value_after = cash_balance_after + asset_quantity_after * price_float
                strategy_set_id = portfolio_row[5] or "manual"

                # Application-level idempotency under the portfolios row lock.
                # portfolio_ledger is a TimescaleDB hypertable, so a DB-level partial unique
                # index on (portfolio_id, order_id, execution_id) is impossible (a unique index
                # must include the partitioning column recorded_at). The `SELECT ... FOR UPDATE`
                # on the portfolios row above serializes concurrent writes for the same portfolio,
                # so this check-then-insert is race-safe for broker writes (order_id present).
                if order_id is not None and execution_id is not None:
                    cur.execute(
                        """
                        SELECT
                            ledger_entry_id, portfolio_id, order_id, execution_id, side,
                            quantity::float8, price::float8,
                            fee_amount::float8, tax_amount::float8,
                            executed_at, recorded_at,
                            cash_balance_after::float8, asset_quantity_after::float8,
                            total_value_after::float8, profit_buffer_after::float8,
                            strategy_set_id, note
                        FROM portfolio_ledger
                        WHERE portfolio_id = %s AND order_id = %s AND execution_id = %s
                        ORDER BY recorded_at DESC, ledger_entry_id DESC
                        LIMIT 1
                        """,
                        (portfolio_id, order_id, execution_id),
                    )
                    existing = cur.fetchone()
                    if existing is not None:
                        # Duplicate: do not insert a new row, do not mutate the portfolio.
                        # OQ-4 (b): return the CURRENT authoritative portfolio state so the
                        # caller always syncs to the latest balances, not the first-call snapshot.
                        entry = self._row_to_ledger_entry(existing)
                        entry.cash_balance_after = cash_balance
                        entry.asset_quantity_after = asset_quantity
                        return entry, False

                cur.execute(
                    """
                    INSERT INTO portfolio_ledger (
                        portfolio_id, recorded_at, ledger_entry_id, order_id, execution_id,
                        side, price, quantity, fee_amount, tax_amount,
                        cash_balance_after, asset_quantity_after, total_value_after,
                        profit_buffer_after, strategy_set_id, note, executed_at
                    ) VALUES (
                        %s, NOW(), DEFAULT, %s, %s,
                        %s, %s, %s, %s, %s,
                        %s, %s, %s,
                        %s, %s, %s, %s
                    )
                    RETURNING
                        ledger_entry_id, portfolio_id, order_id, execution_id, side,
                        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
                    """,
                    (
                        portfolio_id,
                        order_id,
                        execution_id,
                        side_normalized,
                        price_float,
                        qty_float,
                        fee_float,
                        tax_float,
                        cash_balance_after,
                        asset_quantity_after,
                        total_value_after,
                        float(portfolio_row[4]),
                        strategy_set_id,
                        "",
                        executed_at,
                    ),
                )
                row = cur.fetchone()

                cur.execute(
                    """
                    UPDATE portfolios
                    SET cash_balance = %s,
                        asset_quantity = %s,
                        asset_price = %s,
                        updated_at = NOW(),
                        last_trade_at = %s
                    WHERE portfolio_id = %s
                    """,
                    (cash_balance_after, asset_quantity_after, price_float, executed_at, portfolio_id),
                )
                return self._row_to_ledger_entry(row), True

    @staticmethod
    def _row_to_ledger_entry(row: Any) -> LedgerEntry:
        if row is None:
            raise ValueError("원장 항목을 찾을 수 없습니다")
        return LedgerEntry(
            ledger_entry_id=int(row[0]),
            portfolio_id=row[1],
            order_id=row[2],
            execution_id=row[3],
            side=row[4],
            quantity=float(row[5]),
            price=float(row[6]),
            fee_amount=float(row[7]),
            tax_amount=float(row[8]),
            executed_at=row[9],
            recorded_at=row[10],
            cash_balance_after=float(row[11]),
            asset_quantity_after=float(row[12]),
            total_value_after=float(row[13]),
            profit_buffer_after=float(row[14]),
            strategy_set_id=row[15] or "legacy",
            note=row[16] or "",
            source=row[17] if len(row) > 17 and row[17] else "broker",
            target_ledger_entry_id=int(row[18]) if len(row) > 18 and row[18] is not None else None,
            confirmation_evidence=row[19] if len(row) > 19 else None,
            actor_user_id=row[20] if len(row) > 20 else None,
        )

    def reset(self, portfolio_id: str) -> Portfolio:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                UPDATE portfolios
                SET cash_balance = 0,
                    asset_quantity = 0,
                    asset_price = 0,
                    profit_buffer = 0,
                    active_strategy_id = NULL,
                    updated_at = NOW(),
                    last_trade_at = NULL
                WHERE portfolio_id = %s
                """,
                (portfolio_id,),
            )
            if cur.rowcount == 0:
                raise ValueError("포트폴리오를 찾을 수 없습니다")
            cur.execute("DELETE FROM portfolio_ledger WHERE portfolio_id = %s", (portfolio_id,))
        return self.get(portfolio_id)

    def adjust_buffer(self, portfolio_id: str, amount: float, action: str, note: str = "") -> Portfolio:
        self.ensure_schema()
        normalized = str(action).upper()
        if normalized not in {"INVEST", "PROFIT", "WITHDRAW"}:
            raise ValueError("action은 INVEST, PROFIT, WITHDRAW 중 하나여야 합니다")
        if amount <= 0:
            raise ValueError("amount는 0보다 커야 합니다")
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_PORTFOLIO + "WHERE p.portfolio_id = %s", (portfolio_id,))
            row = cur.fetchone()
            if row is None:
                raise ValueError("포트폴리오를 찾을 수 없습니다")
            portfolio = self._row_to_portfolio(row)
            updated_buffer = float(portfolio.profit_buffer)
            if normalized == "INVEST":
                updated_buffer += float(amount)
            else:
                updated_buffer -= float(amount)
                if updated_buffer < 0:
                    raise ValueError("수익버퍼 잔액이 부족합니다")
            cur.execute(
                """
                UPDATE portfolios
                SET profit_buffer = %s,
                    updated_at = NOW()
                WHERE portfolio_id = %s
                """,
                (updated_buffer, portfolio_id),
            )
            total_value = float(portfolio.cash_balance) + float(portfolio.asset_quantity) * float(portfolio.price)
            cur.execute(
                """
                INSERT INTO portfolio_ledger (
                    portfolio_id, recorded_at, side, price, quantity, fee_amount,
                    tax_amount, cash_balance_after, asset_quantity_after,
                    total_value_after, profit_buffer_after, strategy_set_id, note, executed_at
                ) VALUES (%s, NOW(), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())
                """,
                (
                    portfolio_id,
                    normalized.lower(),
                    0.0,
                    0.0,
                    0.0,
                    0.0,
                    float(portfolio.cash_balance),
                    float(portfolio.asset_quantity),
                    float(total_value),
                    float(updated_buffer),
                    "manual",
                    note,
                ),
            )
        return self.get(portfolio_id)

    def transfer_buffer(self, portfolio_id: str, amount: float, direction: str, note: str = "") -> Portfolio:
        self.ensure_schema()
        normalized = str(direction).lower()
        if normalized not in {"to-buffer", "from-buffer"}:
            raise ValueError("direction은 to-buffer 또는 from-buffer 여야 합니다")
        if amount <= 0:
            raise ValueError("amount는 0보다 커야 합니다")
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_PORTFOLIO + "WHERE p.portfolio_id = %s", (portfolio_id,))
            row = cur.fetchone()
            if row is None:
                raise ValueError("포트폴리오를 찾을 수 없습니다")
            portfolio = self._row_to_portfolio(row)

            cash = float(portfolio.cash_balance)
            buf = float(portfolio.profit_buffer)
            if normalized == "to-buffer":
                if cash < float(amount):
                    raise ValueError("이체 금액이 현금 잔고를 초과합니다")
                cash -= float(amount)
                buf += float(amount)
                ledger_action = "BUFFER_IN"
            else:
                if buf < float(amount):
                    raise ValueError("이체 금액이 수익버퍼 잔고를 초과합니다")
                buf -= float(amount)
                cash += float(amount)
                ledger_action = "BUFFER_OUT"
            cur.execute(
                """
                UPDATE portfolios
                SET cash_balance = %s,
                    profit_buffer = %s,
                    updated_at = NOW()
                WHERE portfolio_id = %s
                """,
                (cash, buf, portfolio_id),
            )
            total_value = float(cash) + float(portfolio.asset_quantity) * float(portfolio.price)
            cur.execute(
                """
                INSERT INTO portfolio_ledger (
                    portfolio_id, recorded_at, side, price, quantity, fee_amount,
                    tax_amount, cash_balance_after, asset_quantity_after,
                    total_value_after, profit_buffer_after, strategy_set_id, note, executed_at
                ) VALUES (%s, NOW(), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())
                """,
                (
                    portfolio_id,
                    ledger_action.lower(),
                    0.0,
                    0.0,
                    0.0,
                    0.0,
                    float(cash),
                    float(portfolio.asset_quantity),
                    float(total_value),
                    float(buf),
                    "manual",
                    note,
                ),
            )
        return self.get(portfolio_id)

    def _manual_total_value(self, cash: float, quantity: float, price: float, buffer: float) -> float:
        return float(cash) + float(quantity) * float(price) + float(buffer)

    def record_manual_ledger(
        self,
        portfolio_id: str,
        *,
        action: str,
        amount: float | None = None,
        quantity: float | None = None,
        price: float | None = None,
        fee: float = 0.0,
        tax: float = 0.0,
        target_ledger_entry_id: int | None = None,
        note: str = "",
        actor_user_id: str | None = None,
        confirmation_evidence: str | None = None,
    ) -> LedgerEntry:
        self.ensure_schema()
        normalized = str(action).upper()
        if normalized not in {"INVEST", "BUFFER_OUT", "BUFFER_IN", "BUY", "SELL", "WITHDRAW", "PROFIT", "REVERSAL"}:
            raise ValueError("unsupported manual ledger action")
        conn = self._connect()
        with conn.transaction():
            with conn.cursor() as cur:
                cur.execute(_SELECT_PORTFOLIO + "WHERE p.portfolio_id = %s FOR UPDATE OF p", (portfolio_id,))
                row = cur.fetchone()
                if row is None:
                    raise ValueError("포트폴리오를 찾을 수 없습니다")
                portfolio = self._row_to_portfolio(row)
                cash = float(portfolio.cash_balance)
                qty = float(portfolio.asset_quantity)
                buffer = float(portfolio.profit_buffer)
                asset_price = float(portfolio.price)
                ledger_price = float(price or 0.0)
                ledger_quantity = float(quantity or 0.0)
                fee_f = float(fee or 0.0)
                tax_f = float(tax or 0.0)
                amount_f = float(amount or 0.0)
                reverse_target: LedgerEntry | None = None

                if normalized == "INVEST":
                    buffer += amount_f
                    ledger_price = amount_f
                elif normalized == "BUFFER_OUT":
                    if buffer < amount_f:
                        raise ValueError("INSUFFICIENT_BUFFER")
                    buffer -= amount_f
                    cash += amount_f
                    ledger_price = amount_f
                elif normalized == "BUFFER_IN":
                    if cash < amount_f:
                        raise ValueError("INSUFFICIENT_CASH")
                    cash -= amount_f
                    buffer += amount_f
                    ledger_price = amount_f
                elif normalized == "WITHDRAW":
                    if buffer < amount_f:
                        raise ValueError("INSUFFICIENT_BUFFER")
                    buffer -= amount_f
                    ledger_price = amount_f
                elif normalized == "PROFIT":
                    if buffer < amount_f:
                        raise ValueError("INSUFFICIENT_BUFFER")
                    buffer -= amount_f
                    ledger_price = amount_f
                elif normalized == "BUY":
                    required = ledger_quantity * ledger_price + fee_f + tax_f
                    if cash < required:
                        raise ValueError("INSUFFICIENT_CASH")
                    cash -= required
                    qty += ledger_quantity
                    asset_price = ledger_price
                elif normalized == "SELL":
                    if qty < ledger_quantity:
                        raise ValueError("INSUFFICIENT_QUANTITY")
                    cash += ledger_quantity * ledger_price - fee_f - tax_f
                    qty -= ledger_quantity
                    asset_price = ledger_price
                elif normalized == "REVERSAL":
                    if target_ledger_entry_id is None:
                        raise ValueError("target_ledger_entry_id is required")
                    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 portfolio_id = %s AND ledger_entry_id = %s
                        """,
                        (portfolio_id, target_ledger_entry_id),
                    )
                    reverse_target = self._row_to_ledger_entry(cur.fetchone())
                    cur.execute(
                        "SELECT 1 FROM portfolio_ledger WHERE portfolio_id = %s AND side = 'reversal' AND target_ledger_entry_id = %s LIMIT 1",
                        (portfolio_id, target_ledger_entry_id),
                    )
                    if cur.fetchone() is not None:
                        raise ValueError("LEDGER_ENTRY_ALREADY_REVERSED")
                    target_action = (reverse_target.action or "").upper()
                    amount_f = float(reverse_target.price)
                    ledger_price = float(reverse_target.price)
                    ledger_quantity = float(reverse_target.quantity)
                    fee_f = float(reverse_target.fee_amount)
                    tax_f = float(reverse_target.tax_amount)
                    if target_action == "BUY":
                        cash += ledger_quantity * ledger_price + fee_f + tax_f
                        qty -= ledger_quantity
                    elif target_action == "SELL":
                        if cash < ledger_quantity * ledger_price - fee_f - tax_f:
                            raise ValueError("INSUFFICIENT_CASH")
                        cash -= ledger_quantity * ledger_price - fee_f - tax_f
                        qty += ledger_quantity
                    elif target_action == "BUFFER_OUT":
                        if cash < amount_f:
                            raise ValueError("INSUFFICIENT_CASH")
                        cash -= amount_f
                        buffer += amount_f
                    elif target_action == "BUFFER_IN":
                        if buffer < amount_f:
                            raise ValueError("INSUFFICIENT_BUFFER")
                        cash += amount_f
                        buffer -= amount_f
                    elif target_action == "INVEST":
                        if buffer < amount_f:
                            raise ValueError("INSUFFICIENT_BUFFER")
                        buffer -= amount_f
                    elif target_action in {"WITHDRAW", "PROFIT"}:
                        buffer += amount_f
                    else:
                        raise ValueError("unsupported reversal target action")

                if qty < 0:
                    raise ValueError("INSUFFICIENT_QUANTITY")
                if cash < 0:
                    raise ValueError("INSUFFICIENT_CASH")
                if buffer < 0:
                    raise ValueError("INSUFFICIENT_BUFFER")

                total_value = self._manual_total_value(cash, qty, asset_price, buffer)
                cur.execute(
                    """
                    UPDATE portfolios
                    SET cash_balance = %s,
                        asset_quantity = %s,
                        asset_price = %s,
                        profit_buffer = %s,
                        updated_at = NOW(),
                        last_trade_at = NOW()
                    WHERE portfolio_id = %s
                    """,
                    (cash, qty, asset_price, buffer, portfolio_id),
                )
                cur.execute(
                    """
                    INSERT INTO portfolio_ledger (
                        portfolio_id, recorded_at, side, price, quantity, fee_amount,
                        tax_amount, cash_balance_after, asset_quantity_after,
                        total_value_after, profit_buffer_after, strategy_set_id, note, executed_at,
                        source, target_ledger_entry_id, confirmation_evidence, actor_user_id
                    ) VALUES (%s, NOW(), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), %s, %s, %s, %s)
                    RETURNING ledger_entry_id, portfolio_id, order_id, execution_id, side, quantity, price,
                              fee_amount, tax_amount, executed_at, recorded_at, cash_balance_after,
                              asset_quantity_after, total_value_after, profit_buffer_after, strategy_set_id,
                              note, source, target_ledger_entry_id, confirmation_evidence, actor_user_id
                    """,
                    (
                        portfolio_id,
                        normalized.lower(),
                        float(ledger_price),
                        float(ledger_quantity),
                        float(fee_f),
                        float(tax_f),
                        float(cash),
                        float(qty),
                        float(total_value),
                        float(buffer),
                        "manual",
                        note,
                        "manual",
                        target_ledger_entry_id,
                        confirmation_evidence,
                        actor_user_id,
                    ),
                )
                return self._row_to_ledger_entry(cur.fetchone())

    def list_manual_ledger_entries(self, portfolio_id: str, limit: int = 50) -> list[LedgerEntry]:
        self.ensure_schema()
        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 portfolio_id = %s AND source = 'manual'
                ORDER BY recorded_at DESC, ledger_entry_id DESC
                LIMIT %s
                """,
                (portfolio_id, int(limit)),
            )
            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 {}
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                SELECT ledger.portfolio_id,
                       COALESCE(SUM(ledger.price) FILTER (WHERE ledger.side = 'invest'), 0),
                       COALESCE(SUM(ledger.price) FILTER (WHERE ledger.side = 'withdraw'), 0),
                       COALESCE(SUM(ledger.price) FILTER (WHERE ledger.side = 'profit'), 0)
                FROM portfolio_ledger AS ledger
                WHERE ledger.portfolio_id = ANY(%s)
                  AND ledger.source = 'manual'
                  AND ledger.side IN ('invest', 'withdraw', 'profit')
                  AND NOT EXISTS (
                      SELECT 1
                      FROM portfolio_ledger AS reversal
                      WHERE reversal.portfolio_id = ledger.portfolio_id
                        AND reversal.source = 'manual'
                        AND reversal.side = 'reversal'
                        AND reversal.target_ledger_entry_id = ledger.ledger_entry_id
                  )
                GROUP BY ledger.portfolio_id
                """,
                (portfolio_ids,),
            )
            rows = cur.fetchall()
        return {
            str(portfolio_id): {
                "total_invested": float(invested),
                "total_withdrawn": float(withdrawn),
                "net_contributed_capital": float(invested) - float(withdrawn),
                "cumulative_profit_withdrawn": float(profit),
            }
            for portfolio_id, invested, withdrawn, profit in rows
        }

    def get_return_summary(self, portfolio_id: str) -> dict[str, float]:
        summaries = self.get_return_summaries([portfolio_id])
        return summaries.get(
            portfolio_id,
            {
                "total_invested": 0.0,
                "total_withdrawn": 0.0,
                "net_contributed_capital": 0.0,
                "cumulative_profit_withdrawn": 0.0,
            },
        )

    def get_asset_by_id(self, asset_id: str) -> Asset | None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_ASSET + "WHERE asset_id = %s", (asset_id,))
            row = cur.fetchone()
        return None if row is None else self._row_to_asset(row)

    def get_asset_by_symbol(self, symbol: str) -> Asset | None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_ASSET + "WHERE symbol = %s", (symbol,))
            row = cur.fetchone()
        return None if row is None else self._row_to_asset(row)

    def list_assets(self) -> list[Asset]:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_ASSET + "ORDER BY symbol ASC")
            rows = cur.fetchall()
        return [self._row_to_asset(row) for row in rows]

    def insert_asset(
        self,
        *,
        symbol: str,
        markets: list[str],
        display_name: str | None = None,
        status: str = "active",
    ) -> Asset:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                INSERT INTO assets (symbol, markets, display_name, status)
                VALUES (%s, %s, %s, %s)
                RETURNING asset_id, symbol, markets, display_name, status, created_at, updated_at,
                          COALESCE(tag_ids, '{}'::uuid[])
                """,
                (symbol, markets, display_name, status),
            )
            row = cur.fetchone()
        return self._row_to_asset(row)

    def update_asset(
        self,
        asset_id: str,
        *,
        markets: list[str] | None = None,
        display_name: str | None = None,
        status: str | None = None,
    ) -> Asset:
        self.ensure_schema()
        assignments: list[str] = ["updated_at = NOW()"]
        params: list[object] = []
        if markets is not None:
            assignments.append("markets = %s")
            params.append(markets)
        if display_name is not None:
            assignments.append("display_name = %s")
            params.append(display_name)
        if status is not None:
            assignments.append("status = %s")
            params.append(status)
        params.append(asset_id)
        with self._connect().cursor() as cur:
            cur.execute(
                f"""
                UPDATE assets SET {', '.join(assignments)}
                WHERE asset_id = %s
                """,
                tuple(params),
            )
            if cur.rowcount == 0:
                raise ValueError(f"asset을 찾을 수 없습니다: {asset_id}")
        return self.get_asset_by_id(asset_id)  # type: ignore[return-value]

    def delete(self, portfolio_id: str, *, clear_ledger: bool = True) -> None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT portfolio_id FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            if cur.fetchone() is None:
                raise ValueError("포트폴리오를 찾을 수 없습니다")
            if clear_ledger:
                cur.execute("DELETE FROM portfolio_ledger WHERE portfolio_id = %s", (portfolio_id,))
            # Comments are always removed with the portfolio (no FK constraint, handled in application layer)
            cur.execute("DELETE FROM portfolio_comments WHERE portfolio_id = %s", (portfolio_id,))
            cur.execute("DELETE FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))

    @staticmethod
    def _row_to_comment(row: Any) -> PortfolioComment:
        if row is None:
            raise ValueError("댓글을 찾을 수 없습니다")
        return PortfolioComment(
            id=row[0],
            portfolio_id=row[1],
            content=row[2],
            created_at=row[3],
            updated_at=row[4],
        )

    def create_comment(self, portfolio_id: str, *, content: str) -> PortfolioComment:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            if cur.fetchone() is None:
                raise ValueError("포트폴리오를 찾을 수 없습니다")
            comment_id = str(uuid.uuid4())
            cur.execute(
                """
                INSERT INTO portfolio_comments (id, portfolio_id, content)
                VALUES (%s, %s, %s)
                RETURNING id, portfolio_id, content, created_at, updated_at
                """,
                (comment_id, portfolio_id, content),
            )
            row = cur.fetchone()
        return self._row_to_comment(row)

    def list_comments(self, portfolio_id: str) -> list[PortfolioComment]:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            if cur.fetchone() is None:
                raise ValueError("포트폴리오를 찾을 수 없습니다")
            cur.execute(
                """
                SELECT id, portfolio_id, content, created_at, updated_at
                FROM portfolio_comments
                WHERE portfolio_id = %s
                ORDER BY created_at DESC
                """,
                (portfolio_id,),
            )
            rows = cur.fetchall()
        return [self._row_to_comment(row) for row in rows]

    def update_comment(self, portfolio_id: str, comment_id: str, *, content: str) -> PortfolioComment:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            if cur.fetchone() is None:
                raise ValueError("포트폴리오를 찾을 수 없습니다")
            cur.execute(
                """
                UPDATE portfolio_comments
                SET content = %s, updated_at = NOW()
                WHERE id = %s AND portfolio_id = %s
                RETURNING id, portfolio_id, content, created_at, updated_at
                """,
                (content, comment_id, portfolio_id),
            )
            row = cur.fetchone()
            if row is None:
                raise ValueError("댓글을 찾을 수 없습니다")
        return self._row_to_comment(row)

    def delete_comment(self, portfolio_id: str, comment_id: str) -> None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            if cur.fetchone() is None:
                raise ValueError("포트폴리오를 찾을 수 없습니다")
            cur.execute(
                "DELETE FROM portfolio_comments WHERE id = %s AND portfolio_id = %s RETURNING id",
                (comment_id, portfolio_id),
            )
            if cur.fetchone() is None:
                raise ValueError("댓글을 찾을 수 없습니다")

    def soft_delete_portfolio(self, portfolio_id: str) -> Portfolio:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                UPDATE portfolios
                SET status = 'disabled', managed_by_daemon_id = NULL, updated_at = NOW()
                WHERE portfolio_id = %s
                """,
                (portfolio_id,),
            )
            if cur.rowcount == 0:
                raise ValueError(f"포트폴리오를 찾을 수 없습니다: {portfolio_id}")
        return self.get(portfolio_id)

    def list_portfolios_with_strategies(
        self,
        owner_user_id: str,
        *,
        include_disabled: bool = True,
        managed_by_daemon_id: str | None = None,
        market: str | None = None,
    ) -> list[Portfolio]:
        status_filter = None if include_disabled else "enabled"
        return self.list_portfolios(
            owner_user_id=owner_user_id,
            status=status_filter,
            managed_by_daemon_id=managed_by_daemon_id,
            market=market,
        )

    def create_tag(
        self,
        *,
        name: str,
        created_by_user_id: str,
        is_admin_tag: bool = False,
    ) -> Tag:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            try:
                cur.execute(
                    """
                    INSERT INTO tags (name, created_by_user_id, is_admin_tag)
                    VALUES (%s, %s, %s)
                    RETURNING tag_id, name, created_by_user_id, is_admin_tag, is_active, created_at
                    """,
                    (name, created_by_user_id, is_admin_tag),
                )
            except psycopg.errors.UniqueViolation as exc:
                raise ValueError(f"이미 존재하는 태그 이름입니다: {name}") from exc
            row = cur.fetchone()
        return self._row_to_tag(row)

    def get_tag_by_id(self, tag_id: str) -> Tag | None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_TAG + "WHERE tag_id = %s", (tag_id,))
            row = cur.fetchone()
        return self._row_to_tag(row) if row is not None else None

    def get_tags_by_ids(self, tag_ids: list[str]) -> dict[str, Tag]:
        if not tag_ids:
            return {}
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(_SELECT_TAG + "WHERE tag_id = ANY(%s)", (tag_ids,))
            rows = cur.fetchall()
        return {str(row[0]): self._row_to_tag(row) for row in rows}

    def list_tags_master(self, *, include_inactive: bool = False) -> list[Tag]:
        self.ensure_schema()
        query = _SELECT_TAG
        if not include_inactive:
            query += "WHERE is_active = TRUE "
        query += "ORDER BY name ASC"
        with self._connect().cursor() as cur:
            cur.execute(query)
            rows = cur.fetchall()
        return [self._row_to_tag(row) for row in rows]

    def deactivate_tag(self, tag_id: str) -> Tag:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                UPDATE tags SET is_active = FALSE
                WHERE tag_id = %s
                RETURNING tag_id, name, created_by_user_id, is_admin_tag, is_active, created_at
                """,
                (tag_id,),
            )
            row = cur.fetchone()
            if row is None:
                raise ValueError(f"태그를 찾을 수 없습니다: {tag_id}")
        return self._row_to_tag(row)

    def append_portfolio_tag_id(self, portfolio_id: str, tag_id: str) -> None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                UPDATE portfolios
                SET tag_ids = array_append(tag_ids, %s::uuid)
                WHERE portfolio_id = %s AND NOT (%s::uuid = ANY(tag_ids))
                """,
                (tag_id, portfolio_id, tag_id),
            )
            if cur.rowcount == 0:
                cur.execute("SELECT portfolio_id FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
                if cur.fetchone() is None:
                    raise ValueError(f"portfolio를 찾을 수 없습니다: {portfolio_id}")

    def remove_portfolio_tag_id(self, portfolio_id: str, tag_id: str) -> None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                UPDATE portfolios
                SET tag_ids = array_remove(tag_ids, %s::uuid)
                WHERE portfolio_id = %s
                """,
                (tag_id, portfolio_id),
            )
            if cur.rowcount == 0:
                raise ValueError(f"portfolio를 찾을 수 없습니다: {portfolio_id}")

    def append_asset_tag_id(self, asset_id: str, tag_id: str) -> None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                UPDATE assets
                SET tag_ids = array_append(tag_ids, %s::uuid)
                WHERE asset_id = %s AND NOT (%s::uuid = ANY(tag_ids))
                """,
                (tag_id, asset_id, tag_id),
            )
            if cur.rowcount == 0:
                cur.execute("SELECT asset_id FROM assets WHERE asset_id = %s", (asset_id,))
                if cur.fetchone() is None:
                    raise ValueError(f"asset을 찾을 수 없습니다: {asset_id}")

    def remove_asset_tag_id(self, asset_id: str, tag_id: str) -> None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                UPDATE assets
                SET tag_ids = array_remove(tag_ids, %s::uuid)
                WHERE asset_id = %s
                """,
                (tag_id, asset_id),
            )
            if cur.rowcount == 0:
                raise ValueError(f"asset을 찾을 수 없습니다: {asset_id}")

    def rotate_service_token(self, name: str) -> tuple[str, AuthToken]:
        from app.core.security import generate_token, hash_token, token_prefix as _token_prefix

        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                _SELECT_AUTH_TOKEN + "WHERE name = %s AND token_type = 'service' AND revoked_at IS NULL",
                (name,),
            )
            existing_row = cur.fetchone()
            if existing_row is None:
                raise ValueError(f"서비스 토큰을 찾을 수 없습니다: {name}")
            existing = self._row_to_auth_token(existing_row)

            new_raw = generate_token()
            new_runtime_id = str(uuid.uuid4())
            new_hash = hash_token(new_raw)
            new_prefix = _token_prefix(new_raw)

            cur.execute("UPDATE auth_tokens SET revoked_at = NOW() WHERE id = %s", (existing.id,))

            token_id = str(uuid.uuid4())
            cur.execute(
                """
                INSERT INTO auth_tokens (
                    id, user_id, name, token_prefix, token_hash, permission, token_type,
                    expires_at, runtime_id, daemon_id
                )
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                RETURNING
                    id, user_id, name, token_prefix, permission, token_type,
                    expires_at, last_used_at, revoked_at, created_at, runtime_id, daemon_id
                """,
                (
                    token_id,
                    existing.user_id,
                    name,
                    new_prefix,
                    new_hash,
                    existing.permission,
                    "service",
                    existing.expires_at,
                    new_runtime_id,
                    existing.daemon_id,
                ),
            )
            row = cur.fetchone()
        return new_raw, self._row_to_auth_token(row)

    def rebind_service_token(self, name: str, daemon_id: str) -> AuthToken:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                _SELECT_AUTH_TOKEN + "WHERE name = %s AND token_type = 'service' AND revoked_at IS NULL ORDER BY created_at DESC LIMIT 1",
                (name,),
            )
            existing_row = cur.fetchone()
            if existing_row is None:
                raise ValueError(f"서비스 토큰을 찾을 수 없습니다: {name}")
            existing = self._row_to_auth_token(existing_row)

            cur.execute(
                """
                UPDATE auth_tokens
                SET daemon_id = %s
                WHERE id = %s
                RETURNING
                    id, user_id, name, token_prefix, permission, token_type,
                    expires_at, last_used_at, revoked_at, created_at, runtime_id, daemon_id
                """,
                (daemon_id, existing.id),
            )
            row = cur.fetchone()
        return self._row_to_auth_token(row)

    @staticmethod
    def _row_to_daemon(row: Any) -> Daemon:
        if row is None:
            raise ValueError("데몬을 찾을 수 없습니다")
        return Daemon(
            id=row[0],
            owner_user_id=row[1],
            daemon_slug=row[2],
            display_name=row[3],
            is_active=bool(row[4]),
            created_at=row[5] if len(row) > 5 else None,
            updated_at=row[6] if len(row) > 6 else None,
        )

    def create_daemon(
        self,
        owner_user_id: str,
        *,
        daemon_slug: str,
        display_name: str | None = None,
    ) -> Daemon:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute("SELECT 1 FROM users WHERE id = %s", (owner_user_id,))
            if cur.fetchone() is None:
                raise ValueError("사용자를 찾을 수 없습니다")
            daemon_id = str(uuid.uuid4())
            try:
                cur.execute(
                    """
                    INSERT INTO daemons (id, owner_user_id, daemon_slug, display_name)
                    VALUES (%s, %s, %s, %s)
                    RETURNING id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
                    """,
                    (daemon_id, owner_user_id, daemon_slug, display_name),
                )
            except psycopg.errors.UniqueViolation as exc:
                raise ValueError(f"이미 존재하는 daemon slug입니다: {daemon_slug}") from exc
            row = cur.fetchone()
        return self._row_to_daemon(row)

    def get_daemon_by_id(self, daemon_id: str) -> Daemon | None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                SELECT id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
                FROM daemons WHERE id = %s
                """,
                (daemon_id,),
            )
            row = cur.fetchone()
        if row is None:
            return None
        return self._row_to_daemon(row)

    def get_daemon_by_slug(self, owner_user_id: str, daemon_slug: str) -> Daemon | None:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                SELECT id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
                FROM daemons WHERE owner_user_id = %s AND daemon_slug = %s
                """,
                (owner_user_id, daemon_slug),
            )
            row = cur.fetchone()
        if row is None:
            return None
        return self._row_to_daemon(row)

    def list_daemons_for_user(self, owner_user_id: str) -> list[Daemon]:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                SELECT id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
                FROM daemons WHERE owner_user_id = %s ORDER BY created_at ASC
                """,
                (owner_user_id,),
            )
            rows = cur.fetchall()
        return [self._row_to_daemon(row) for row in rows]

    def list_daemons_all(self) -> list[Daemon]:
        """Return all daemons across all owners (admin use only)."""
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                SELECT id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
                FROM daemons ORDER BY created_at ASC
                """,
            )
            rows = cur.fetchall()
        return [self._row_to_daemon(row) for row in rows]

    def update_daemon(
        self,
        daemon_id: str,
        *,
        display_name: str | None = None,
        is_active: bool | None = None,
    ) -> Daemon:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                SELECT id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
                FROM daemons WHERE id = %s
                """,
                (daemon_id,),
            )
            existing_row = cur.fetchone()
            if existing_row is None:
                raise ValueError("데몬을 찾을 수 없습니다")

            assignments: list[str] = []
            params: list[object] = []
            if display_name is not None:
                assignments.append("display_name = %s")
                params.append(display_name)
            if is_active is not None:
                assignments.append("is_active = %s")
                params.append(is_active)

            if not assignments:
                return self._row_to_daemon(existing_row)

            assignments.append("updated_at = NOW()")
            cur.execute(
                f"""
                UPDATE daemons
                SET {', '.join(assignments)}
                WHERE id = %s
                RETURNING id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
                """,
                (*params, daemon_id),
            )
            row = cur.fetchone()
        return self._row_to_daemon(row)

    def revoke_tokens_by_daemon_id(self, daemon_id: str) -> int:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                "UPDATE auth_tokens SET revoked_at = NOW() WHERE daemon_id = %s AND revoked_at IS NULL",
                (daemon_id,),
            )
            return cur.rowcount

    def get_broker_credential_by_daemon(
        self,
        daemon_id: str,
        broker: str,
    ) -> BrokerCredential | None:
        """Return a single BrokerCredential by (daemon_id, broker) PK, or None if absent (REQ-052)."""
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                """
                SELECT daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
                FROM broker_credentials
                WHERE daemon_id = %s AND broker = %s
                """,
                (daemon_id, broker),
            )
            row = cur.fetchone()
        if row is None:
            return None
        return self._row_to_broker_credential(row)

    def delete_credentials_by_daemon_id(self, daemon_id: str) -> int:
        """Delete ALL broker credentials for daemon_id (owner re-assignment invalidation — REQ-053 AC-4).

        Returns number of rows deleted.
        """
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                "DELETE FROM broker_credentials WHERE daemon_id = %s",
                (daemon_id,),
            )
            return cur.rowcount

    def update_daemon_owner(
        self,
        daemon_id: str,
        *,
        owner_user_id: str,
    ) -> Daemon:
        """Change daemon owner and invalidate all existing BrokerCredentials (REQ-053 AC-4).

        Owner change and credential invalidation are executed atomically in a single
        transaction (BAT-022 QC Major fix).  If either statement fails the entire
        transaction is rolled back, preventing the state where the owner has changed but
        the previous owner's credentials remain (credential leak — dcy-011 OQ-3).

        Execution order inside the transaction:
          1. DELETE FROM broker_credentials WHERE daemon_id = %s   (credential first)
          2. UPDATE daemons SET owner_user_id = %s ...             (owner second)
        This order is also safe if the transaction machinery is somehow bypassed: if
        only step 1 completes the owner is unchanged and credentials are gone (no leak);
        if only step 2 completes the old credentials are still present but the caller
        still owns them (no cross-owner leak).
        """
        self.ensure_schema()
        conn = self._connect()
        # Use conn.transaction() (not `with conn:`) so the shared connection stays open
        # after the block (psycopg3 closes the connection when used as a context manager).
        with conn.transaction():
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT id FROM daemons WHERE id = %s",
                    (daemon_id,),
                )
                if cur.fetchone() is None:
                    raise ValueError("데몬을 찾을 수 없습니다")
                # Step 1: delete credentials first (safe-order)
                cur.execute(
                    "DELETE FROM broker_credentials WHERE daemon_id = %s",
                    (daemon_id,),
                )
                # Step 2: update owner
                cur.execute(
                    """
                    UPDATE daemons
                    SET owner_user_id = %s, updated_at = NOW()
                    WHERE id = %s
                    RETURNING id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
                    """,
                    (owner_user_id, daemon_id),
                )
                row = cur.fetchone()
        return self._row_to_daemon(row)

    def list_service_tokens_with_null_daemon(self) -> list[AuthToken]:
        self.ensure_schema()
        with self._connect().cursor() as cur:
            cur.execute(
                _SELECT_AUTH_TOKEN
                + "WHERE token_type = 'service' AND daemon_id IS NULL AND revoked_at IS NULL "
                + "ORDER BY created_at ASC",
            )
            rows = cur.fetchall()
        return [self._row_to_auth_token(row) for row in rows]


def database_exists(config: PortfolioDBConfig, *, maintenance_db: str = "postgres") -> bool:
    with psycopg.connect(
        host=config.host,
        port=config.port,
        dbname=maintenance_db,
        user=config.user,
        password=config.password,
        sslmode=config.sslmode,
        connect_timeout=config.connect_timeout_sec,
        application_name=config.application_name,
        autocommit=True,
    ) as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT 1 FROM pg_database WHERE datname = %s", (config.dbname,))
            return cur.fetchone() is not None


def create_database_if_missing(config: PortfolioDBConfig, *, maintenance_db: str = "postgres") -> bool:
    with psycopg.connect(
        host=config.host,
        port=config.port,
        dbname=maintenance_db,
        user=config.user,
        password=config.password,
        sslmode=config.sslmode,
        connect_timeout=config.connect_timeout_sec,
        application_name=config.application_name,
        autocommit=True,
    ) as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT 1 FROM pg_database WHERE datname = %s", (config.dbname,))
            if cur.fetchone() is not None:
                return False
            cur.execute(sql.SQL("CREATE DATABASE {}") .format(sql.Identifier(config.dbname)))
            return True