"""Postgres repository implementation for portfolios and ledger.""" from __future__ import annotations import hashlib import uuid from dataclasses import dataclass from datetime import datetime, timezone from typing import Any 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 FROM portfolios p JOIN assets a ON p.asset_id = a.asset_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( """ 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)] 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, ) @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 "", ) 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 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 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 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