commit 3e3acbd13d126c4bc53b0b5a50c310c2b0c220f9 Author: Justin Ko Date: Fri Apr 3 07:33:32 2026 +0000 chore: initial project scaffold and gitignore diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..24aca58 --- /dev/null +++ b/.env.example @@ -0,0 +1,30 @@ +# Application Configuration +PORTFOLIOHUB_ALLOWED_ORIGINS=http://localhost:3000 +PORTFOLIOHUB_READ_TOKENS=read-token-123 +PORTFOLIOHUB_TRADE_WRITE_TOKENS=trade-token-456 +PORTFOLIOHUB_ADMIN_TOKENS=admin-token-789 +PORTFOLIOHUB_ENVIRONMENT=development + +# Enable DB-backed repository (uses TIMESCALEDB_* below) +PORTFOLIOHUB_DB_ENABLED=false + +# TimescaleDB / PostgreSQL connection settings +TIMESCALEDB_HOST=localhost +TIMESCALEDB_PORT=5432 +TIMESCALEDB_NAME=portfoliohub +TIMESCALEDB_USER=portfoliohub +TIMESCALEDB_PASSWORD= +TIMESCALEDB_SSLMODE=prefer +TIMESCALEDB_CONNECT_TIMEOUT_SEC=10 +TIMESCALEDB_APPLICATION_NAME=portfoliohub + +# Redis (for caching, sessions, etc.) +REDIS_HOST=localhost +REDIS_PORT=6379 +REDIS_DB=0 +REDIS_PASSWORD= + +# Loki (for logging) +LOKI_HOST=localhost +LOKI_PORT=3100 +LOKI_ENABLED=false diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3d36bfe --- /dev/null +++ b/.gitignore @@ -0,0 +1,46 @@ +# Environment +.env +.env.* +!.env.example + +# Project local artifacts +temp/ + +# Python virtual environments +.venv/ +venv/ +env/ + +# Python cache and bytecode +__pycache__/ +*.py[cod] +*$py.class + +# Packaging and build artifacts +*.egg-info/ +.eggs/ +dist/ +build/ +pip-wheel-metadata/ + +# Test and coverage artifacts +.pytest_cache/ +.coverage +.coverage.* +htmlcov/ + +# Type checker / linter / formatter caches +.mypy_cache/ +.ruff_cache/ +.pyre/ + +# Notebook checkpoints +.ipynb_checkpoints/ + +# Editor / IDE +.vscode/ +.idea/ + +# OS junk +.DS_Store +Thumbs.db \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..6c2012a --- /dev/null +++ b/README.md @@ -0,0 +1,130 @@ +# PortfolioHub + +PortfolioHub is a FastAPI service that extracts portfolio management from `jsbot-kis` and exposes it through an authenticated HTTP API. + +## Current status + +This repository is in bootstrap phase. + +- FastAPI application entrypoint is available. +- Token-based authentication and role checks are implemented. +- Portfolio API contracts are defined. +- Core portfolio domain services are ported from `jsbot-kis`. +- Postgres repository scaffolding is implemented and gated behind `PORTFOLIOHUB_DB_ENABLED=true`. + +## Configuration + +Copy `.env.example` to `.env` and edit values as needed: + +```bash +cp .env.example .env +``` + +The application loads configuration in this order (higher priority first): +1. Environment variables +2. Variables from `.env` file +3. Default values in code + +When deploying to production, set environment variables directly instead of using `.env`. + +### Key configuration options + +- **DB repository switch**: `PORTFOLIOHUB_DB_ENABLED` (enables DB-backed persistence) +- **TimescaleDB / PostgreSQL**: `TIMESCALEDB_*` (single source of DB connection settings) +- **Redis**: `REDIS_*` (for caching and sessions) +- **Loki**: `LOKI_*` (for centralized logging) +- **Authentication**: `PORTFOLIOHUB_*_TOKENS` (bearer tokens with role-based access) + +For complete configuration options, see `.env.example`. + +## Run locally + +```bash +python -m venv .venv +source .venv/bin/activate +pip install -e .[dev] + +# Configure .env +cp .env.example .env + +# Start server +uvicorn app.main:app --reload + +# In another terminal, run tests +pytest +``` + +## CLI + +The repository now includes a `portfoliohub` CLI that mirrors the main portfolio flows from `jsbot-kis`. + +```bash +portfoliohub --help +portfoliohub portfolio list +portfoliohub portfolio status 005930:tv --recent 10 +portfoliohub portfolio create 005930:tv --symbol 005930 --cash-balance 10000000 --force +portfoliohub portfolio buy 005930:tv --quantity 10 --price 71000 --fee 5000 --force +portfoliohub portfolio sell 005930:tv --quantity 5 --price 72000 --tax 3000 --force +portfoliohub portfolio invest 005930:tv --type deposit --amount 50000 --force +portfoliohub portfolio transfer 005930:tv --direction to-buffer --amount 100000 --force +portfoliohub portfolio reset 005930:tv --force +portfoliohub portfolio delete 005930:tv --keep-ledger --force +``` + +Supported commands: + +- `init-db` +- `migrate-db` +- `portfolio list` +- `portfolio status` +- `portfolio create` +- `portfolio update` +- `portfolio buy` +- `portfolio sell` +- `portfolio invest` +- `portfolio transfer` +- `portfolio reset` +- `portfolio delete` +- `health-check` + +Write commands ask for confirmation by default. Use `--force` to skip the prompt for scripts and automation. + +The CLI uses the same repository and settings layer as the HTTP API. To target the external TimescaleDB, set `PORTFOLIOHUB_DB_ENABLED=true` and provide `TIMESCALEDB_*` values in the environment or `.env`. + +Initialize a new database explicitly before the first use: + +```bash +PORTFOLIOHUB_DB_ENABLED=true portfoliohub init-db +PORTFOLIOHUB_DB_ENABLED=true portfoliohub health-check +PORTFOLIOHUB_DB_ENABLED=true portfoliohub portfolio list +``` + +`init-db` will: + +- connect to the maintenance database, defaulting to `postgres` +- create the target database if it does not exist +- run Alembic `upgrade head` + +`migrate-db` will: + +- connect to the target database +- run Alembic `upgrade head` +- apply any future schema revisions in order + +This repository now uses Alembic for versioned schema migrations. Future schema changes should be added under `migrations/versions`. + +## Authentication model + +PortfolioHub currently supports bearer tokens with three permission levels. + +- `read`: list and status endpoints +- `trade-write`: mutation endpoints such as buy, sell, invest, transfer, reset +- `admin`: delete and future operational actions + +The default `.env.example` includes example tokens for development. + +## Planned next steps + +1. Implement Postgres/Timescale repository. +2. Port domain logic from `jsbot-kis` services. +3. Add integration tests for full portfolio flows. diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..8559dec --- /dev/null +++ b/alembic.ini @@ -0,0 +1,36 @@ +[alembic] +script_location = migrations +sqlalchemy.url = postgresql+psycopg://postgres:postgres@localhost:5432/postgres + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S \ No newline at end of file diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..f065969 --- /dev/null +++ b/app/__init__.py @@ -0,0 +1 @@ +"""PortfolioHub application package.""" diff --git a/app/api/__init__.py b/app/api/__init__.py new file mode 100644 index 0000000..d45717f --- /dev/null +++ b/app/api/__init__.py @@ -0,0 +1 @@ +"""API routers for PortfolioHub.""" diff --git a/app/api/health.py b/app/api/health.py new file mode 100644 index 0000000..592029a --- /dev/null +++ b/app/api/health.py @@ -0,0 +1,10 @@ +"""Health and readiness endpoints.""" + +from fastapi import APIRouter + +router = APIRouter(tags=["health"]) + + +@router.get("/health") +def healthcheck() -> dict[str, str]: + return {"status": "ok"} diff --git a/app/api/portfolios.py b/app/api/portfolios.py new file mode 100644 index 0000000..6488fce --- /dev/null +++ b/app/api/portfolios.py @@ -0,0 +1,114 @@ +"""Portfolio API routes.""" + +from fastapi import APIRouter, Depends, Response, status + +from app.core.auth import require_admin, require_read, require_trade_write +from app.schemas.portfolio import ( + BuyRequest, + CreatePortfolioRequest, + InvestRequest, + PortfolioDetailResponse, + PortfolioListItem, + ResetPortfolioRequest, + SellRequest, + TransferRequest, + UpdatePortfolioRequest, +) +from app.services.portfolio_service import PortfolioService, get_portfolio_service + +router = APIRouter(prefix="/portfolios", tags=["portfolios"]) + + +@router.get("", response_model=list[PortfolioListItem], dependencies=[Depends(require_read)]) +def list_portfolios( + broker: str | None = None, + status_filter: str | None = None, + service: PortfolioService = Depends(get_portfolio_service), +) -> list[PortfolioListItem]: + return service.list_portfolios(broker=broker, status=status_filter) + + +@router.get("/{portfolio_id}", response_model=PortfolioDetailResponse, dependencies=[Depends(require_read)]) +def get_portfolio( + portfolio_id: str, + recent_trades: int = 5, + service: PortfolioService = Depends(get_portfolio_service), +) -> PortfolioDetailResponse: + return service.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=recent_trades) + + +@router.post( + "", + response_model=PortfolioDetailResponse, + status_code=status.HTTP_201_CREATED, + dependencies=[Depends(require_trade_write)], +) +def create_portfolio( + payload: CreatePortfolioRequest, + service: PortfolioService = Depends(get_portfolio_service), +) -> PortfolioDetailResponse: + return service.create_portfolio(payload) + + +@router.patch("/{portfolio_id}", response_model=PortfolioDetailResponse, dependencies=[Depends(require_trade_write)]) +def update_portfolio( + portfolio_id: str, + payload: UpdatePortfolioRequest, + service: PortfolioService = Depends(get_portfolio_service), +) -> PortfolioDetailResponse: + return service.update_portfolio(portfolio_id, payload) + + +@router.post("/{portfolio_id}/buy", response_model=PortfolioDetailResponse, dependencies=[Depends(require_trade_write)]) +def buy_portfolio_asset( + portfolio_id: str, + payload: BuyRequest, + service: PortfolioService = Depends(get_portfolio_service), +) -> PortfolioDetailResponse: + return service.buy(portfolio_id, payload) + + +@router.post("/{portfolio_id}/sell", response_model=PortfolioDetailResponse, dependencies=[Depends(require_trade_write)]) +def sell_portfolio_asset( + portfolio_id: str, + payload: SellRequest, + service: PortfolioService = Depends(get_portfolio_service), +) -> PortfolioDetailResponse: + return service.sell(portfolio_id, payload) + + +@router.post("/{portfolio_id}/invest", response_model=PortfolioDetailResponse, dependencies=[Depends(require_trade_write)]) +def invest_buffer( + portfolio_id: str, + payload: InvestRequest, + service: PortfolioService = Depends(get_portfolio_service), +) -> PortfolioDetailResponse: + return service.invest(portfolio_id, payload) + + +@router.post("/{portfolio_id}/transfer", response_model=PortfolioDetailResponse, dependencies=[Depends(require_trade_write)]) +def transfer_buffer( + portfolio_id: str, + payload: TransferRequest, + service: PortfolioService = Depends(get_portfolio_service), +) -> PortfolioDetailResponse: + return service.transfer(portfolio_id, payload) + + +@router.post("/{portfolio_id}/reset", response_model=PortfolioDetailResponse, dependencies=[Depends(require_trade_write)]) +def reset_portfolio( + portfolio_id: str, + payload: ResetPortfolioRequest, + service: PortfolioService = Depends(get_portfolio_service), +) -> PortfolioDetailResponse: + return service.reset(portfolio_id, payload) + + +@router.delete("/{portfolio_id}", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(require_admin)]) +def delete_portfolio( + portfolio_id: str, + keep_ledger: bool = False, + service: PortfolioService = Depends(get_portfolio_service), +) -> Response: + service.delete(portfolio_id, keep_ledger=keep_ledger) + return Response(status_code=status.HTTP_204_NO_CONTENT) diff --git a/app/cli/__init__.py b/app/cli/__init__.py new file mode 100644 index 0000000..048fd43 --- /dev/null +++ b/app/cli/__init__.py @@ -0,0 +1 @@ +"""CLI package for PortfolioHub.""" diff --git a/app/cli/commands/__init__.py b/app/cli/commands/__init__.py new file mode 100644 index 0000000..cb5739a --- /dev/null +++ b/app/cli/commands/__init__.py @@ -0,0 +1 @@ +"""CLI command groups.""" diff --git a/app/cli/commands/health.py b/app/cli/commands/health.py new file mode 100644 index 0000000..f7ad306 --- /dev/null +++ b/app/cli/commands/health.py @@ -0,0 +1,104 @@ +"""Health-related CLI commands.""" + +from __future__ import annotations + +import typer + +from app.cli.core.utils import print_json +from app.db.migrations import bootstrap_or_upgrade_to_head +from app.repositories.factory import build_portfolio_db_config +from app.repositories.postgres import PostgresPortfolioRepository, create_database_if_missing, database_exists + + +def initialize_database(*, settings, maintenance_db: str = "postgres") -> dict[str, object]: + config = build_portfolio_db_config(settings) + existed_before = database_exists(config, maintenance_db=maintenance_db) + created = False + if not existed_before: + created = create_database_if_missing(config, maintenance_db=maintenance_db) + + bootstrap_or_upgrade_to_head(settings) + + return { + "database": config.dbname, + "host": config.host, + "port": config.port, + "created": created, + "schema_initialized": True, + "migration_applied": "head", + } + + +def register_health_commands(app: typer.Typer) -> None: + @app.command("init-db") + def init_db( + ctx: typer.Context, + maintenance_db: str = typer.Option("postgres", help="Database to connect to for CREATE DATABASE."), + ) -> None: + app_context = ctx.obj + try: + result = initialize_database(settings=app_context.settings, maintenance_db=maintenance_db) + except Exception as exc: + app_context.console.print(f"init-db failed: {exc}") + raise typer.Exit(code=1) from exc + else: + if app_context.output_format == "json": + print_json(app_context.console, result) + else: + app_context.console.print(f"database: {result['database']}") + app_context.console.print(f"host: {result['host']}") + app_context.console.print(f"port: {result['port']}") + app_context.console.print(f"created: {result['created']}") + app_context.console.print("schema_initialized: True") + app_context.console.print("migration_applied: head") + + @app.command("migrate-db") + def migrate_db(ctx: typer.Context) -> None: + app_context = ctx.obj + try: + bootstrap_or_upgrade_to_head(app_context.settings) + result = { + "database": app_context.settings.active_db_name, + "host": app_context.settings.active_db_host, + "port": app_context.settings.active_db_port, + "migration_applied": "head", + } + except Exception as exc: + app_context.console.print(f"migrate-db failed: {exc}") + raise typer.Exit(code=1) from exc + else: + if app_context.output_format == "json": + print_json(app_context.console, result) + else: + app_context.console.print(f"database: {result['database']}") + app_context.console.print(f"host: {result['host']}") + app_context.console.print(f"port: {result['port']}") + app_context.console.print("migration_applied: head") + + @app.command("health-check") + def health_check(ctx: typer.Context) -> None: + app_context = ctx.obj + repository = app_context.repository_factory(app_context.settings) + try: + health = { + "app_name": app_context.settings.app_name, + "environment": app_context.settings.environment, + "db_enabled": app_context.settings.db_enabled, + "active_db_host": app_context.settings.active_db_host, + "active_db_port": app_context.settings.active_db_port, + "repository": repository.__class__.__name__, + } + if hasattr(repository, "ping"): + repository.ping() + except Exception as exc: + app_context.console.print(f"health-check failed: {exc}") + raise typer.Exit(code=1) from exc + else: + if app_context.output_format == "json": + print_json(app_context.console, health) + else: + for key, value in health.items(): + app_context.console.print(f"{key}: {value}") + finally: + if hasattr(repository, "close"): + repository.close() \ No newline at end of file diff --git a/app/cli/commands/portfolio.py b/app/cli/commands/portfolio.py new file mode 100644 index 0000000..d6b9857 --- /dev/null +++ b/app/cli/commands/portfolio.py @@ -0,0 +1,262 @@ +"""Portfolio CLI commands.""" + +from __future__ import annotations + +import json +from typing import Any + +import typer + +from app.cli.core.utils import execute_command, render_output +from app.schemas.portfolio import ( + BuyRequest, + CreatePortfolioRequest, + InvestRequest, + ResetPortfolioRequest, + SellRequest, + TransferRequest, + UpdatePortfolioRequest, +) +from app.services.portfolio_service import PortfolioService + + +portfolio_app = typer.Typer(help="Portfolio lifecycle and trade commands.") + + +def _parse_strategy_params(value: str) -> dict[str, Any]: + if not value.strip(): + return {} + parsed = json.loads(value) + if not isinstance(parsed, dict): + raise ValueError("strategy params must be a JSON object") + return parsed + + +def _with_service(ctx: typer.Context, callback): + app_context = ctx.obj + repository = app_context.repository_factory(app_context.settings) + service = PortfolioService(repository) + try: + result = execute_command(app_context.console, callback=lambda: callback(service, app_context)) + return result + finally: + if hasattr(repository, "close"): + repository.close() + + +def _confirm(force: bool, message: str) -> None: + if force: + return + confirmed = typer.confirm(message) + if not confirmed: + raise typer.Exit(code=0) + + +@portfolio_app.command("list") +def list_portfolios( + ctx: typer.Context, + broker: str | None = typer.Option(None, help="Filter by broker."), + status: str | None = typer.Option(None, help="Filter by status."), +) -> None: + detail = _with_service(ctx, lambda service, app_context: service.list_portfolios(broker=broker, status=status)) + render_output(ctx.obj.console, ctx.obj.output_format, detail, empty_message="No portfolios found.") + + +@portfolio_app.command("status") +def portfolio_status( + ctx: typer.Context, + portfolio_id: str = typer.Argument(..., help="Portfolio identifier."), + recent: int = typer.Option(5, "--recent", min=1, help="Number of recent ledger entries."), +) -> None: + detail = _with_service( + ctx, + lambda service, app_context: service.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=recent), + ) + render_output(ctx.obj.console, ctx.obj.output_format, detail) + + +@portfolio_app.command("create") +def create_portfolio( + ctx: typer.Context, + portfolio_id: str = typer.Argument(..., help="Portfolio identifier."), + symbol: str = typer.Option(..., help="Trading symbol."), + broker: str = typer.Option("kis", help="Broker code."), + status: str = typer.Option("enabled", help="Portfolio status."), + cash_balance: float = typer.Option(0.0, min=0.0, help="Starting cash balance."), + asset_quantity: float = typer.Option(0.0, min=0.0, help="Starting asset quantity."), + asset_price: float = typer.Option(0.0, min=0.0, help="Starting asset price."), + profit_buffer: float = typer.Option(0.0, min=0.0, help="Initial profit buffer."), + strategy_name: str = typer.Option("target_value", help="Strategy name."), + strategy_params: str = typer.Option("{}", help="Strategy params as JSON object."), + force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."), +) -> None: + _confirm(force, f"Create portfolio {portfolio_id} for symbol {symbol}?") + detail = _with_service( + ctx, + lambda service, app_context: service.create_portfolio( + CreatePortfolioRequest( + portfolio_id=portfolio_id, + symbol=symbol, + broker=broker, + status=status, + cash_balance=cash_balance, + asset_quantity=asset_quantity, + asset_price=asset_price, + profit_buffer=profit_buffer, + strategy_name=strategy_name, + strategy_params=_parse_strategy_params(strategy_params), + ) + ), + ) + render_output(ctx.obj.console, ctx.obj.output_format, detail) + + +@portfolio_app.command("update") +def update_portfolio( + ctx: typer.Context, + portfolio_id: str = typer.Argument(..., help="Portfolio identifier."), + broker: str | None = typer.Option(None, help="Broker code."), + status: str | None = typer.Option(None, help="Portfolio status."), + symbol: str | None = typer.Option(None, help="Trading symbol."), + strategy_name: str | None = typer.Option(None, help="Strategy name."), + strategy_params: str | None = typer.Option(None, help="Strategy params as JSON object."), + asset_price: float | None = typer.Option(None, min=0.0, help="Current asset price."), + force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."), +) -> None: + _confirm(force, f"Update portfolio {portfolio_id}?") + parsed_strategy_params = None if strategy_params is None else _parse_strategy_params(strategy_params) + detail = _with_service( + ctx, + lambda service, app_context: service.update_portfolio( + portfolio_id, + UpdatePortfolioRequest( + broker=broker, + status=status, + symbol=symbol, + strategy_name=strategy_name, + strategy_params=parsed_strategy_params, + asset_price=asset_price, + ), + ), + ) + render_output(ctx.obj.console, ctx.obj.output_format, detail) + + +@portfolio_app.command("buy") +def buy_portfolio( + ctx: typer.Context, + portfolio_id: str = typer.Argument(..., help="Portfolio identifier."), + quantity: float = typer.Option(..., min=0.0000001, help="Quantity to buy."), + price: float = typer.Option(..., min=0.0000001, help="Execution price."), + fee: float = typer.Option(0.0, min=0.0, help="Fee amount."), + tax: float = typer.Option(0.0, min=0.0, help="Tax amount."), + note: str = typer.Option("", help="Ledger note."), + force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."), +) -> None: + total_cost = quantity * price + fee + tax + _confirm(force, f"Buy {quantity:g} units of {portfolio_id} at {price:,.0f} (total {total_cost:,.0f})?") + detail = _with_service( + ctx, + lambda service, app_context: service.buy( + portfolio_id, + BuyRequest(quantity=quantity, price=price, fee=fee, tax=tax, note=note), + ), + ) + render_output(ctx.obj.console, ctx.obj.output_format, detail) + + +@portfolio_app.command("sell") +def sell_portfolio( + ctx: typer.Context, + portfolio_id: str = typer.Argument(..., help="Portfolio identifier."), + quantity: float = typer.Option(..., min=0.0000001, help="Quantity to sell."), + price: float = typer.Option(..., min=0.0000001, help="Execution price."), + fee: float = typer.Option(0.0, min=0.0, help="Fee amount."), + tax: float = typer.Option(0.0, min=0.0, help="Tax amount."), + note: str = typer.Option("", help="Ledger note."), + force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."), +) -> None: + net_amount = quantity * price - fee - tax + _confirm(force, f"Sell {quantity:g} units of {portfolio_id} at {price:,.0f} (net {net_amount:,.0f})?") + detail = _with_service( + ctx, + lambda service, app_context: service.sell( + portfolio_id, + SellRequest(quantity=quantity, price=price, fee=fee, tax=tax, note=note), + ), + ) + render_output(ctx.obj.console, ctx.obj.output_format, detail) + + +@portfolio_app.command("invest") +def invest_portfolio( + ctx: typer.Context, + portfolio_id: str = typer.Argument(..., help="Portfolio identifier."), + movement_type: str = typer.Option(..., "--type", help="deposit, profit, or withdraw."), + amount: float = typer.Option(..., min=0.0000001, help="Buffer adjustment amount."), + note: str = typer.Option("", help="Ledger note."), + force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."), +) -> None: + _confirm(force, f"Apply {movement_type} of {amount:,.0f} to portfolio {portfolio_id}?") + detail = _with_service( + ctx, + lambda service, app_context: service.invest( + portfolio_id, + InvestRequest(movement_type=movement_type, amount=amount, note=note), + ), + ) + render_output(ctx.obj.console, ctx.obj.output_format, detail) + + +@portfolio_app.command("transfer") +def transfer_portfolio( + ctx: typer.Context, + portfolio_id: str = typer.Argument(..., help="Portfolio identifier."), + direction: str = typer.Option(..., help="to-buffer or from-buffer."), + amount: float = typer.Option(..., min=0.0000001, help="Transfer amount."), + note: str = typer.Option("", help="Ledger note."), + force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."), +) -> None: + _confirm(force, f"Transfer {amount:,.0f} {direction} for portfolio {portfolio_id}?") + detail = _with_service( + ctx, + lambda service, app_context: service.transfer( + portfolio_id, + TransferRequest(direction=direction, amount=amount, note=note), + ), + ) + render_output(ctx.obj.console, ctx.obj.output_format, detail) + + +@portfolio_app.command("reset") +def reset_portfolio( + ctx: typer.Context, + portfolio_id: str = typer.Argument(..., help="Portfolio identifier."), + infer_symbol: str | None = typer.Option(None, help="Fallback symbol when portfolio does not exist."), + force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."), +) -> None: + _confirm(force, f"Reset portfolio {portfolio_id}?") + detail = _with_service( + ctx, + lambda service, app_context: service.reset(portfolio_id, ResetPortfolioRequest(infer_symbol=infer_symbol)), + ) + render_output(ctx.obj.console, ctx.obj.output_format, detail) + + +@portfolio_app.command("delete") +def delete_portfolio( + ctx: typer.Context, + portfolio_id: str = typer.Argument(..., help="Portfolio identifier."), + keep_ledger: bool = typer.Option(False, help="Preserve ledger history."), + force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."), +) -> None: + suffix = " and keep ledger" if keep_ledger else "" + _confirm(force, f"Delete portfolio {portfolio_id}{suffix}?") + _with_service( + ctx, + lambda service, app_context: service.delete(portfolio_id, keep_ledger=keep_ledger), + ) + if ctx.obj.output_format == "json": + render_output(ctx.obj.console, ctx.obj.output_format, {"deleted": portfolio_id, "keep_ledger": keep_ledger}) + else: + ctx.obj.console.print(f"Deleted portfolio: {portfolio_id}") \ No newline at end of file diff --git a/app/cli/core/__init__.py b/app/cli/core/__init__.py new file mode 100644 index 0000000..03acd69 --- /dev/null +++ b/app/cli/core/__init__.py @@ -0,0 +1 @@ +"""CLI helpers.""" diff --git a/app/cli/core/utils.py b/app/cli/core/utils.py new file mode 100644 index 0000000..7415112 --- /dev/null +++ b/app/cli/core/utils.py @@ -0,0 +1,161 @@ +"""Formatting and execution helpers for the CLI.""" + +from __future__ import annotations + +import json +from dataclasses import asdict, is_dataclass +from typing import Any, Callable, TypeVar + +import typer +from pydantic import BaseModel, ValidationError +from rich.console import Console +from rich.panel import Panel +from rich.table import Table + +from app.core.errors import FeatureNotReadyError +from app.schemas.portfolio import PortfolioDetailResponse, PortfolioListItem + +T = TypeVar("T") + + +def format_currency(value: float | None) -> str: + if value is None: + return "N/A" + return f"{value:,.0f}" + + +def to_plain_data(value: Any) -> Any: + if isinstance(value, BaseModel): + return value.model_dump(mode="json") + if is_dataclass(value): + return asdict(value) + if isinstance(value, list): + return [to_plain_data(item) for item in value] + if isinstance(value, dict): + return {key: to_plain_data(item) for key, item in value.items()} + return value + + +def print_portfolio_table(console: Console, portfolios: list[PortfolioListItem]) -> None: + table = Table(title="Portfolios", expand=True) + table.add_column("ID", style="cyan", no_wrap=True) + table.add_column("Broker") + table.add_column("Status", no_wrap=True) + table.add_column("Symbol", no_wrap=True) + table.add_column("Cash", justify="right") + table.add_column("Qty", justify="right") + table.add_column("Price", justify="right") + table.add_column("Buffer", justify="right") + table.add_column("Total", justify="right") + for item in portfolios: + table.add_row( + item.portfolio_id, + item.broker, + item.status, + item.symbol, + format_currency(item.cash_balance), + f"{item.asset_quantity:g}", + format_currency(item.price), + format_currency(item.profit_buffer), + format_currency(item.total_value), + ) + console.print(table) + + +def print_portfolio_detail(console: Console, detail: PortfolioDetailResponse) -> None: + portfolio = detail.portfolio + position_value = float(portfolio.asset_quantity) * float(portfolio.price) + bot_total = float(portfolio.cash_balance) + position_value + user_total = bot_total + float(portfolio.profit_buffer) + panel = Panel.fit( + "\n".join( + [ + f"id: {portfolio.portfolio_id}", + f"broker: {portfolio.broker}", + f"status: {portfolio.status}", + f"symbol: {portfolio.symbol}", + f"strategy: {portfolio.strategy_name}", + f"cash_balance: {format_currency(portfolio.cash_balance)}", + f"asset_quantity: {portfolio.asset_quantity:g}", + f"price: {format_currency(portfolio.price)}", + f"profit_buffer: {format_currency(portfolio.profit_buffer)}", + f"position_value: {format_currency(position_value)}", + f"bot_total: {format_currency(bot_total)}", + f"user_total: {format_currency(user_total)}", + f"live_quantity: {detail.live_quantity if detail.live_quantity is not None else 'N/A'}", + ] + ), + title="Portfolio Status", + ) + console.print(panel) + + if detail.strategy_set is not None: + console.print( + Panel.fit( + "\n".join( + [ + f"id: {detail.strategy_set.id}", + f"name: {detail.strategy_set.strategy_name}", + f"config: {detail.strategy_set.strategy_config}", + ] + ), + title="Strategy Set", + ) + ) + + ledger_table = Table(title="Recent Ledger Entries", expand=True) + ledger_table.add_column("Recorded At", style="cyan", no_wrap=True) + ledger_table.add_column("Action") + ledger_table.add_column("Qty", justify="right") + ledger_table.add_column("Price", justify="right") + ledger_table.add_column("Cash After", justify="right") + ledger_table.add_column("Qty After", justify="right") + ledger_table.add_column("Buffer After", justify="right") + for entry in detail.recent_entries: + ledger_table.add_row( + entry.recorded_at.isoformat(timespec="seconds"), + entry.action, + f"{entry.quantity:g}", + format_currency(entry.price), + format_currency(entry.cash_balance_after), + f"{entry.asset_quantity_after:g}", + format_currency(entry.profit_buffer_after), + ) + console.print(ledger_table) + + +def print_json(console: Console, payload: Any) -> None: + console.print_json(data=to_plain_data(payload)) + + +def render_output(console: Console, output_format: str, payload: Any, *, empty_message: str | None = None) -> None: + if output_format == "json": + print_json(console, payload) + return + + if isinstance(payload, list): + if not payload and empty_message: + console.print(empty_message) + return + print_portfolio_table(console, payload) + return + + if isinstance(payload, PortfolioDetailResponse): + print_portfolio_detail(console, payload) + return + + console.print(payload) + + +def execute_command(console: Console, callback: Callable[[], T]) -> T: + try: + return callback() + except ValidationError as exc: + console.print(f"Input validation failed: {exc}") + raise typer.Exit(code=1) from exc + except FeatureNotReadyError as exc: + console.print(str(exc)) + raise typer.Exit(code=1) from exc + except ValueError as exc: + console.print(str(exc)) + raise typer.Exit(code=1) from exc diff --git a/app/cli/main.py b/app/cli/main.py new file mode 100644 index 0000000..780a038 --- /dev/null +++ b/app/cli/main.py @@ -0,0 +1,56 @@ +"""Typer entrypoint for PortfolioHub.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Callable + +import typer +from rich.console import Console + +from app.cli.commands.health import register_health_commands +from app.cli.commands.portfolio import portfolio_app +from app.core.config import Settings, get_settings +from app.repositories.factory import get_portfolio_repository_from_settings +from app.repositories.portfolio_repository import PortfolioRepository + + +@dataclass +class AppContext: + settings: Settings + output_format: str + console: Console + repository_factory: Callable[[Settings], PortfolioRepository] + + +def create_app( + *, + settings_factory: Callable[[], Settings] = get_settings, + repository_factory: Callable[[Settings], PortfolioRepository] = get_portfolio_repository_from_settings, +) -> typer.Typer: + app = typer.Typer(help="PortfolioHub command line interface.", no_args_is_help=True) + + @app.callback() + def main_callback( + ctx: typer.Context, + output_format: str = typer.Option("table", "--format", help="Output format: table or json."), + ) -> None: + settings = settings_factory() + console = Console(no_color=True, width=160) + ctx.obj = AppContext( + settings=settings, + output_format=output_format, + console=console, + repository_factory=repository_factory, + ) + + app.add_typer(portfolio_app, name="portfolio") + register_health_commands(app) + return app + + +app = create_app() + + +def main() -> None: + app() \ No newline at end of file diff --git a/app/core/__init__.py b/app/core/__init__.py new file mode 100644 index 0000000..6ee6af5 --- /dev/null +++ b/app/core/__init__.py @@ -0,0 +1 @@ +"""Core application modules.""" diff --git a/app/core/auth.py b/app/core/auth.py new file mode 100644 index 0000000..37de50a --- /dev/null +++ b/app/core/auth.py @@ -0,0 +1,79 @@ +"""Bearer token authentication and authorization.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum + +from fastapi import Depends, HTTPException, status +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer + +from app.core.config import Settings, get_settings + + +class Permission(StrEnum): + READ = "read" + TRADE_WRITE = "trade-write" + ADMIN = "admin" + + +@dataclass(frozen=True) +class AuthContext: + token: str + permission: Permission + + +security = HTTPBearer(auto_error=False) + + +def _resolve_permission(token: str, settings: Settings) -> Permission | None: + if token in settings.admin_tokens: + return Permission.ADMIN + if token in settings.trade_write_tokens: + return Permission.TRADE_WRITE + if token in settings.read_tokens: + return Permission.READ + return None + + +def _has_permission(current: Permission, required: Permission) -> bool: + if current == Permission.ADMIN: + return True + if current == required: + return True + if current == Permission.TRADE_WRITE and required == Permission.READ: + return True + return False + + +def require_permission( + required: Permission, + credentials: HTTPAuthorizationCredentials | None = Depends(security), + settings: Settings = Depends(get_settings), +) -> AuthContext: + if credentials is None or credentials.scheme.lower() != "bearer": + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token") + + permission = _resolve_permission(credentials.credentials, settings) + if permission is None: + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") + + if not _has_permission(permission, required): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permission") + + return AuthContext(token=credentials.credentials, permission=permission) + + +def permission_dependency(required: Permission): + def dependency( + credentials: HTTPAuthorizationCredentials | None = Depends(security), + settings: Settings = Depends(get_settings), + ) -> AuthContext: + return require_permission(required, credentials=credentials, settings=settings) + + return dependency + + +require_read = permission_dependency(Permission.READ) +require_trade_write = permission_dependency(Permission.TRADE_WRITE) +require_admin = permission_dependency(Permission.ADMIN) diff --git a/app/core/config.py b/app/core/config.py new file mode 100644 index 0000000..82a73c4 --- /dev/null +++ b/app/core/config.py @@ -0,0 +1,130 @@ +"""Application configuration.""" + +from __future__ import annotations + +from functools import lru_cache +from pathlib import Path +from urllib.parse import quote_plus + +from dotenv import load_dotenv +from pydantic import Field, computed_field +from pydantic_settings import BaseSettings, SettingsConfigDict + +# Load .env file from project root +env_file = Path(__file__).parent.parent.parent / ".env" +if env_file.exists(): + load_dotenv(env_file) + + +class Settings(BaseSettings): + app_name: str = "PortfolioHub" + app_version: str = "0.1.0" + environment: str = "development" + allowed_origins_raw: str = Field(default="http://localhost:3000", validation_alias="PORTFOLIOHUB_ALLOWED_ORIGINS") + read_tokens_raw: str = Field(default="", validation_alias="PORTFOLIOHUB_READ_TOKENS") + trade_write_tokens_raw: str = Field(default="", validation_alias="PORTFOLIOHUB_TRADE_WRITE_TOKENS") + admin_tokens_raw: str = Field(default="", validation_alias="PORTFOLIOHUB_ADMIN_TOKENS") + db_enabled: bool = Field(default=False, validation_alias="PORTFOLIOHUB_DB_ENABLED") + + # External services configuration + timescaledb_host: str = Field(default="localhost", validation_alias="TIMESCALEDB_HOST") + timescaledb_port: int = Field(default=5432, validation_alias="TIMESCALEDB_PORT") + timescaledb_name: str = Field(default="portfoliohub", validation_alias="TIMESCALEDB_NAME") + timescaledb_user: str = Field(default="portfoliohub", validation_alias="TIMESCALEDB_USER") + timescaledb_password: str = Field(default="", validation_alias="TIMESCALEDB_PASSWORD") + timescaledb_sslmode: str = Field(default="prefer", validation_alias="TIMESCALEDB_SSLMODE") + timescaledb_connect_timeout_sec: int = Field(default=10, validation_alias="TIMESCALEDB_CONNECT_TIMEOUT_SEC") + timescaledb_application_name: str = Field(default="portfoliohub", validation_alias="TIMESCALEDB_APPLICATION_NAME") + + redis_host: str = Field(default="localhost", validation_alias="REDIS_HOST") + redis_port: int = Field(default=6379, validation_alias="REDIS_PORT") + redis_db: int = Field(default=0, validation_alias="REDIS_DB") + redis_password: str = Field(default="", validation_alias="REDIS_PASSWORD") + + loki_host: str = Field(default="localhost", validation_alias="LOKI_HOST") + loki_port: int = Field(default=3100, validation_alias="LOKI_PORT") + loki_enabled: bool = Field(default=False, validation_alias="LOKI_ENABLED") + + model_config = SettingsConfigDict( + env_prefix="PORTFOLIOHUB_", + case_sensitive=False, + extra="ignore", + ) + + @computed_field + @property + def allowed_origins(self) -> list[str]: + return [item.strip() for item in self.allowed_origins_raw.split(",") if item.strip()] + + @computed_field + @property + def read_tokens(self) -> set[str]: + return {item.strip() for item in self.read_tokens_raw.split(",") if item.strip()} + + @computed_field + @property + def trade_write_tokens(self) -> set[str]: + return {item.strip() for item in self.trade_write_tokens_raw.split(",") if item.strip()} + + @computed_field + @property + def admin_tokens(self) -> set[str]: + return {item.strip() for item in self.admin_tokens_raw.split(",") if item.strip()} + + @computed_field + @property + def active_db_host(self) -> str: + return self.timescaledb_host + + @computed_field + @property + def active_db_port(self) -> int: + return self.timescaledb_port + + @computed_field + @property + def active_db_name(self) -> str: + return self.timescaledb_name + + @computed_field + @property + def active_db_user(self) -> str: + return self.timescaledb_user + + @computed_field + @property + def active_db_password(self) -> str: + return self.timescaledb_password + + @computed_field + @property + def active_db_sslmode(self) -> str: + return self.timescaledb_sslmode + + @computed_field + @property + def active_db_connect_timeout_sec(self) -> int: + return self.timescaledb_connect_timeout_sec + + @computed_field + @property + def active_db_application_name(self) -> str: + return self.timescaledb_application_name + + @computed_field + @property + def active_database_url(self) -> str: + user = quote_plus(self.active_db_user) + password = quote_plus(self.active_db_password) + database = quote_plus(self.active_db_name) + return ( + f"postgresql+psycopg://{user}:{password}@{self.active_db_host}:{self.active_db_port}/{database}" + f"?sslmode={quote_plus(self.active_db_sslmode)}" + f"&connect_timeout={self.active_db_connect_timeout_sec}" + f"&application_name={quote_plus(self.active_db_application_name)}" + ) + + +@lru_cache +def get_settings() -> Settings: + return Settings() diff --git a/app/core/errors.py b/app/core/errors.py new file mode 100644 index 0000000..dbd4cdc --- /dev/null +++ b/app/core/errors.py @@ -0,0 +1,21 @@ +"""Exception types and FastAPI handlers.""" + +from fastapi import FastAPI, Request, status +from fastapi.responses import JSONResponse + + +class PortfolioHubError(Exception): + """Base application exception.""" + + +class FeatureNotReadyError(PortfolioHubError): + """Raised when an endpoint contract exists but the implementation is pending.""" + + +def register_exception_handlers(app: FastAPI) -> None: + @app.exception_handler(FeatureNotReadyError) + async def handle_not_ready(_: Request, exc: FeatureNotReadyError) -> JSONResponse: + return JSONResponse( + status_code=status.HTTP_501_NOT_IMPLEMENTED, + content={"detail": str(exc) or "Feature not implemented yet"}, + ) diff --git a/app/db/__init__.py b/app/db/__init__.py new file mode 100644 index 0000000..142b929 --- /dev/null +++ b/app/db/__init__.py @@ -0,0 +1 @@ +"""Database tooling for migrations and bootstrap.""" diff --git a/app/db/migrations.py b/app/db/migrations.py new file mode 100644 index 0000000..cd8114d --- /dev/null +++ b/app/db/migrations.py @@ -0,0 +1,81 @@ +"""Alembic helpers for PortfolioHub database migrations.""" + +from __future__ import annotations + +from pathlib import Path + +from alembic import command +from alembic.config import Config +import psycopg + +from app.core.config import Settings + + +def get_alembic_config(settings: Settings) -> Config: + project_root = Path(__file__).resolve().parents[2] + config = Config(str(project_root / "alembic.ini")) + config.set_main_option("script_location", str(project_root / "migrations")) + config.set_main_option("sqlalchemy.url", settings.active_database_url.replace("%", "%%")) + return config + + +def upgrade_to_head(settings: Settings) -> None: + command.upgrade(get_alembic_config(settings), "head") + + +def bootstrap_or_upgrade_to_head(settings: Settings) -> str: + if _has_alembic_version_table(settings): + upgrade_to_head(settings) + return "head" + + if _has_legacy_portfolio_schema(settings): + command.stamp(get_alembic_config(settings), "head") + return "head" + + upgrade_to_head(settings) + return "head" + + +def current_revision(settings: Settings) -> str | None: + output: list[str] = [] + + def capture(message: str, *_args, **_kwargs) -> None: + if message.strip(): + output.append(message.strip()) + + config = get_alembic_config(settings) + command.current(config, verbose=False) + return output[-1] if output else None + + +def _connect(settings: Settings) -> psycopg.Connection: + return psycopg.connect( + host=settings.active_db_host, + port=settings.active_db_port, + dbname=settings.active_db_name, + user=settings.active_db_user, + password=settings.active_db_password, + sslmode=settings.active_db_sslmode, + connect_timeout=settings.active_db_connect_timeout_sec, + application_name=settings.active_db_application_name, + autocommit=True, + ) + + +def _has_alembic_version_table(settings: Settings) -> bool: + with _connect(settings) as conn: + with conn.cursor() as cur: + cur.execute("SELECT to_regclass('public.alembic_version')") + return cur.fetchone()[0] is not None + + +def _has_legacy_portfolio_schema(settings: Settings) -> bool: + required_tables = {"strategy_sets", "portfolios", "portfolio_ledger"} + with _connect(settings) as conn: + with conn.cursor() as cur: + cur.execute( + "SELECT tablename FROM pg_tables WHERE schemaname = 'public' AND tablename = ANY(%s)", + (list(required_tables),), + ) + found = {row[0] for row in cur.fetchall()} + return found == required_tables \ No newline at end of file diff --git a/app/domain/__init__.py b/app/domain/__init__.py new file mode 100644 index 0000000..d1bebd4 --- /dev/null +++ b/app/domain/__init__.py @@ -0,0 +1 @@ +"""Domain models and services for PortfolioHub.""" diff --git a/app/domain/models.py b/app/domain/models.py new file mode 100644 index 0000000..439fa19 --- /dev/null +++ b/app/domain/models.py @@ -0,0 +1,55 @@ +"""Portfolio domain models.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Protocol, runtime_checkable + + +@dataclass +class StrategySet: + id: str + strategy_name: str + strategy_config: str + created_at: datetime | None = None + + +@dataclass +class Portfolio: + portfolio_id: str + broker: str + status: str + symbol: str + strategy_name: str + strategy_params: dict[str, Any] = field(default_factory=dict) + cash_balance: float = 0.0 + asset_quantity: float = 0.0 + price: float = 0.0 + profit_buffer: float = 0.0 + current_strategy_set_id: str | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + last_trade_at: datetime | None = None + + +@dataclass +class LedgerEntry: + portfolio_id: str + recorded_at: datetime + action: str + price: float + quantity: float + cash_balance_after: float + asset_quantity_after: float + total_value_after: float + fee_amount: float = 0.0 + tax_amount: float = 0.0 + profit_buffer_after: float = 0.0 + strategy_set_id: str = "legacy" + note: str = "" + + +@runtime_checkable +class LiveAssetSource(Protocol): + def get_asset_quantity(self, symbol: str) -> float: ... diff --git a/app/domain/services.py b/app/domain/services.py new file mode 100644 index 0000000..012539e --- /dev/null +++ b/app/domain/services.py @@ -0,0 +1,280 @@ +"""Portfolio domain services ported from jsbot-kis.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Callable, Protocol + +from app.domain.models import LedgerEntry, LiveAssetSource, Portfolio, StrategySet + +_NOT_FOUND_MSG = "포트폴리오를 찾을 수 없습니다" +_INVEST_ACTION_MAP = { + "deposit": "INVEST", + "profit": "PROFIT", + "withdraw": "WITHDRAW", +} + + +class PortfolioDomainRepository(Protocol): + def get(self, portfolio_id: str) -> Portfolio: ... + def get_or_create( + self, + portfolio_id: str, + symbol: str, + *, + broker: str = "kis", + status: str = "enabled", + strategy_name: str = "", + strategy_params: dict | None = None, + initial_cash: float = 0.0, + ) -> Portfolio: ... + def create_portfolio( + self, + portfolio_id: str, + *, + broker: str, + status: str, + symbol: str, + strategy_name: str, + strategy_params: dict, + cash_balance: float, + asset_quantity: float, + asset_price: float, + profit_buffer: float, + ) -> Portfolio: ... + def update_portfolio(self, portfolio_id: str, **fields: object) -> Portfolio: ... + def list_portfolios(self, *, broker: str | None = None, status: str | None = None) -> list[Portfolio]: ... + def get_strategy_set(self, id: str) -> StrategySet | None: ... + def get_or_create_strategy_set(self, strategy_name: str, strategy_config: str) -> StrategySet: ... + 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: ... + def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]: ... + def reset(self, portfolio_id: str) -> Portfolio: ... + def adjust_buffer(self, portfolio_id: str, amount: float, action: str, note: str = "") -> Portfolio: ... + def transfer_buffer(self, portfolio_id: str, amount: float, direction: str, note: str = "") -> Portfolio: ... + def delete(self, portfolio_id: str, clear_ledger: bool = True) -> None: ... + + +def apply_buy( + repo: PortfolioDomainRepository, + portfolio_id: str, + *, + quantity: float, + price: float, + fee: float = 0.0, + tax: float = 0.0, + note: str = "", +) -> Portfolio: + quantity = float(quantity) + price = float(price) + fee = float(fee) + tax = float(tax) + + if quantity <= 0: + raise ValueError("quantity는 0보다 커야 합니다") + if price <= 0: + raise ValueError("price는 0보다 커야 합니다") + if fee < 0 or tax < 0: + raise ValueError("fee/tax는 음수가 될 수 없습니다") + + portfolio = repo.get(portfolio_id) + gross = quantity * price + total_cost = gross + fee + tax + if float(portfolio.cash_balance) < total_cost: + raise ValueError("현금 잔고가 부족합니다") + + updated = Portfolio( + portfolio_id=portfolio.portfolio_id, + broker=portfolio.broker, + status=portfolio.status, + symbol=portfolio.symbol, + strategy_name=portfolio.strategy_name, + strategy_params=portfolio.strategy_params, + cash_balance=float(portfolio.cash_balance) - total_cost, + asset_quantity=float(portfolio.asset_quantity) + quantity, + price=price, + profit_buffer=float(portfolio.profit_buffer), + current_strategy_set_id=portfolio.current_strategy_set_id, + created_at=portfolio.created_at, + updated_at=portfolio.updated_at, + last_trade_at=portfolio.last_trade_at, + ) + total_value_after = float(updated.cash_balance) + float(updated.asset_quantity) * price + repo.save( + updated, + current_price=price, + action="BUY", + trade_price=price, + quantity=quantity, + fee_amount=fee, + tax_amount=tax, + total_value_after=total_value_after, + strategy_set_id=portfolio.current_strategy_set_id or "manual", + note=note, + ) + return repo.get(portfolio_id) + + +def apply_sell( + repo: PortfolioDomainRepository, + portfolio_id: str, + *, + quantity: float, + price: float, + fee: float = 0.0, + tax: float = 0.0, + note: str = "", +) -> Portfolio: + quantity = float(quantity) + price = float(price) + fee = float(fee) + tax = float(tax) + + if quantity <= 0: + raise ValueError("quantity는 0보다 커야 합니다") + if price <= 0: + raise ValueError("price는 0보다 커야 합니다") + if fee < 0 or tax < 0: + raise ValueError("fee/tax는 음수가 될 수 없습니다") + + portfolio = repo.get(portfolio_id) + if float(portfolio.asset_quantity) < quantity: + raise ValueError("보유 수량이 부족합니다") + + gross = quantity * price + net_proceeds = gross - fee - tax + + updated = Portfolio( + portfolio_id=portfolio.portfolio_id, + broker=portfolio.broker, + status=portfolio.status, + symbol=portfolio.symbol, + strategy_name=portfolio.strategy_name, + strategy_params=portfolio.strategy_params, + cash_balance=float(portfolio.cash_balance) + net_proceeds, + asset_quantity=float(portfolio.asset_quantity) - quantity, + price=price, + profit_buffer=float(portfolio.profit_buffer), + current_strategy_set_id=portfolio.current_strategy_set_id, + created_at=portfolio.created_at, + updated_at=portfolio.updated_at, + last_trade_at=portfolio.last_trade_at, + ) + total_value_after = float(updated.cash_balance) + float(updated.asset_quantity) * price + repo.save( + updated, + current_price=price, + action="SELL", + trade_price=price, + quantity=quantity, + fee_amount=fee, + tax_amount=tax, + total_value_after=total_value_after, + strategy_set_id=portfolio.current_strategy_set_id or "manual", + note=note, + ) + return repo.get(portfolio_id) + + +def apply_invest( + repo: PortfolioDomainRepository, + portfolio_id: str, + *, + movement_type: str, + amount: float, + note: str = "", +) -> Portfolio: + amount = float(amount) + if amount <= 0: + raise ValueError("--amount는 0보다 커야 합니다") + + key = str(movement_type).lower() + if key not in _INVEST_ACTION_MAP: + raise ValueError("movement_type은 deposit, profit, withdraw 중 하나여야 합니다") + + return repo.adjust_buffer(portfolio_id, amount, _INVEST_ACTION_MAP[key], note) + + +def apply_transfer( + repo: PortfolioDomainRepository, + portfolio_id: str, + *, + direction: str, + amount: float, + note: str = "", +) -> Portfolio: + amount = float(amount) + if amount <= 0: + raise ValueError("--amount는 0보다 커야 합니다") + return repo.transfer_buffer(portfolio_id, amount, direction, note) + + +def apply_reset( + repo: PortfolioDomainRepository, + portfolio_id: str, + *, + infer_symbol: Callable[[str], str], +) -> tuple[Portfolio, bool]: + try: + portfolio = repo.reset(portfolio_id) + return portfolio, False + except ValueError as exc: + if str(exc) != _NOT_FOUND_MSG: + raise + symbol = infer_symbol(portfolio_id) + portfolio = repo.get_or_create(portfolio_id, symbol, initial_cash=0.0) + return portfolio, True + + +def apply_delete( + repo: PortfolioDomainRepository, + portfolio_id: str, + *, + clear_ledger: bool = True, +) -> None: + repo.delete(portfolio_id, clear_ledger=clear_ledger) + + +@dataclass +class PortfolioStatusResult: + portfolio: Portfolio + recent_entries: list[LedgerEntry] + strategy_set: StrategySet | None + live_quantity: float | None + + +def get_portfolio_status( + repo: PortfolioDomainRepository, + portfolio_id: str, + *, + recent_trades: int = 5, + live_source: LiveAssetSource | None = None, +) -> PortfolioStatusResult: + portfolio = repo.get(portfolio_id) + recent_entries = repo.list_recent_trades(portfolio_id, limit=recent_trades) + strategy_set = None + if portfolio.current_strategy_set_id: + strategy_set = repo.get_strategy_set(portfolio.current_strategy_set_id) + + live_quantity = None + if live_source is not None: + live_quantity = live_source.get_asset_quantity(portfolio.symbol) + + return PortfolioStatusResult( + portfolio=portfolio, + recent_entries=recent_entries, + strategy_set=strategy_set, + live_quantity=live_quantity, + ) diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..a66e0d7 --- /dev/null +++ b/app/main.py @@ -0,0 +1,35 @@ +"""FastAPI application entrypoint.""" + +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware + +from app.api.health import router as health_router +from app.api.portfolios import router as portfolio_router +from app.core.config import get_settings +from app.core.errors import register_exception_handlers + + +def create_app() -> FastAPI: + settings = get_settings() + app = FastAPI( + title=settings.app_name, + version=settings.app_version, + docs_url="/docs", + redoc_url="/redoc", + ) + + app.add_middleware( + CORSMiddleware, + allow_origins=settings.allowed_origins, + allow_credentials=True, + allow_methods=["GET", "POST", "PATCH", "DELETE", "OPTIONS"], + allow_headers=["Authorization", "Content-Type"], + ) + + app.include_router(health_router) + app.include_router(portfolio_router) + register_exception_handlers(app) + return app + + +app = create_app() diff --git a/app/repositories/__init__.py b/app/repositories/__init__.py new file mode 100644 index 0000000..666095f --- /dev/null +++ b/app/repositories/__init__.py @@ -0,0 +1 @@ +"""Repository layer for PortfolioHub.""" diff --git a/app/repositories/factory.py b/app/repositories/factory.py new file mode 100644 index 0000000..133bac8 --- /dev/null +++ b/app/repositories/factory.py @@ -0,0 +1,32 @@ +"""Repository factory helpers shared by API and CLI.""" + +from __future__ import annotations + +from app.core.config import Settings +from app.core.errors import FeatureNotReadyError +from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioRepository +from app.repositories.portfolio_repository import PortfolioRepository + + +class UnimplementedPortfolioRepository: + def __getattr__(self, _: str) -> object: + raise FeatureNotReadyError("Portfolio persistence is not implemented yet") + + +def build_portfolio_db_config(settings: Settings) -> PortfolioDBConfig: + return PortfolioDBConfig( + host=settings.active_db_host, + port=settings.active_db_port, + dbname=settings.active_db_name, + user=settings.active_db_user, + password=settings.active_db_password, + sslmode=settings.active_db_sslmode, + connect_timeout_sec=settings.active_db_connect_timeout_sec, + application_name=settings.active_db_application_name, + ) + + +def get_portfolio_repository_from_settings(settings: Settings) -> PortfolioRepository: + if not settings.db_enabled: + return UnimplementedPortfolioRepository() + return PostgresPortfolioRepository(build_portfolio_db_config(settings)) \ No newline at end of file diff --git a/app/repositories/portfolio_repository.py b/app/repositories/portfolio_repository.py new file mode 100644 index 0000000..b61bcdc --- /dev/null +++ b/app/repositories/portfolio_repository.py @@ -0,0 +1,74 @@ +"""Portfolio repository contract.""" + +from __future__ import annotations + +from typing import Protocol + +from app.domain.models import LedgerEntry, Portfolio, StrategySet + + +class PortfolioRepository(Protocol): + def ping(self) -> None: ... + + def get(self, portfolio_id: str) -> Portfolio: ... + + def get_or_create( + self, + portfolio_id: str, + symbol: str, + *, + broker: str = "kis", + status: str = "enabled", + strategy_name: str = "", + strategy_params: dict | None = None, + initial_cash: float = 0.0, + ) -> Portfolio: ... + + def create_portfolio( + self, + portfolio_id: str, + *, + broker: str, + status: str, + symbol: str, + strategy_name: str, + strategy_params: dict, + cash_balance: float, + asset_quantity: float, + asset_price: float, + profit_buffer: float, + ) -> Portfolio: ... + + def update_portfolio(self, portfolio_id: str, **fields: object) -> Portfolio: ... + + def list_portfolios(self, *, broker: str | None = None, status: str | None = None) -> list[Portfolio]: ... + + def get_or_create_strategy_set(self, strategy_name: str, strategy_config: str) -> StrategySet: ... + + def get_strategy_set(self, id: str) -> StrategySet | None: ... + + 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: ... + + def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]: ... + + def reset(self, portfolio_id: str) -> Portfolio: ... + + def adjust_buffer(self, portfolio_id: str, amount: float, action: str, note: str = "") -> Portfolio: ... + + def transfer_buffer(self, portfolio_id: str, amount: float, direction: str, note: str = "") -> Portfolio: ... + + def delete(self, portfolio_id: str, *, clear_ledger: bool = True) -> None: ... + diff --git a/app/repositories/postgres.py b/app/repositories/postgres.py new file mode 100644 index 0000000..b5151f7 --- /dev/null +++ b/app/repositories/postgres.py @@ -0,0 +1,687 @@ +"""Postgres repository implementation for portfolios and ledger.""" + +from __future__ import annotations + +import json +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 LedgerEntry, Portfolio, StrategySet + +_SELECT_PORTFOLIO = """ + SELECT + portfolio_id, + broker, + status, + symbol, + strategy_name, + strategy_params, + cash_balance, + asset_quantity, + asset_price, + profit_buffer, + current_strategy_set_id, + created_at, + updated_at, + last_trade_at + FROM portfolios +""" + + +@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 strategy_sets ( + id TEXT PRIMARY KEY, + strategy_name TEXT NOT NULL, + strategy_config TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (strategy_name, strategy_config) + ) + """ + ) + 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', + symbol TEXT NOT NULL, + strategy_name TEXT NOT NULL DEFAULT '', + strategy_params JSONB NOT NULL DEFAULT '{}'::jsonb, + 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, + current_strategy_set_id TEXT DEFAULT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + last_trade_at TIMESTAMPTZ + ) + """ + ) + cur.execute( + """ + CREATE TABLE IF NOT EXISTS portfolio_ledger ( + portfolio_id TEXT NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL, + action TEXT NOT NULL, + price DOUBLE PRECISION NOT NULL, + quantity DOUBLE PRECISION NOT NULL, + fee_amount DOUBLE PRECISION NOT NULL DEFAULT 0, + tax_amount DOUBLE PRECISION NOT NULL DEFAULT 0, + cash_balance_after DOUBLE PRECISION NOT NULL, + asset_quantity_after DOUBLE PRECISION NOT NULL, + total_value_after DOUBLE PRECISION NOT NULL, + profit_buffer_after DOUBLE PRECISION NOT NULL DEFAULT 0, + strategy_set_id TEXT NOT NULL DEFAULT 'legacy', + note TEXT NOT NULL DEFAULT '' + ) + """ + ) + 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_portfolios_broker_status_updated + ON portfolios (broker, status, updated_at DESC) + """ + ) + 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_portfolio(row: Any) -> Portfolio: + if row is None: + raise ValueError("포트폴리오를 찾을 수 없습니다") + + raw_params = row[5] if len(row) > 5 else {} + if isinstance(raw_params, str): + try: + strategy_params = json.loads(raw_params) if raw_params else {} + except json.JSONDecodeError: + strategy_params = {} + elif isinstance(raw_params, dict): + strategy_params = raw_params + else: + strategy_params = {} + + return Portfolio( + portfolio_id=row[0], + broker=row[1] or "kis", + status=row[2] or "enabled", + symbol=row[3], + strategy_name=row[4] or "", + strategy_params=strategy_params, + cash_balance=float(row[6]), + asset_quantity=float(row[7]), + price=float(row[8]), + profit_buffer=float(row[9]), + current_strategy_set_id=row[10] if len(row) > 10 else None, + created_at=row[11] if len(row) > 11 else None, + updated_at=row[12] if len(row) > 12 else None, + last_trade_at=row[13] if len(row) > 13 else None, + ) + + @staticmethod + def _row_to_ledger_entry(row: Any) -> LedgerEntry: + return LedgerEntry( + portfolio_id=row[0], + recorded_at=row[1], + action=row[2], + price=float(row[3]), + quantity=float(row[4]), + fee_amount=float(row[5]), + tax_amount=float(row[6]), + 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 get(self, portfolio_id: str) -> Portfolio: + self.ensure_schema() + with self._connect().cursor() as cur: + cur.execute(_SELECT_PORTFOLIO + "WHERE portfolio_id = %s", (portfolio_id,)) + row = cur.fetchone() + return self._row_to_portfolio(row) + + def get_or_create( + self, + portfolio_id: str, + symbol: str, + *, + broker: str = "kis", + status: str = "enabled", + strategy_name: str = "", + strategy_params: dict[str, Any] | None = None, + initial_cash: float = 0.0, + ) -> Portfolio: + self.ensure_schema() + with self._connect().cursor() as cur: + cur.execute( + """ + INSERT INTO portfolios ( + portfolio_id, broker, status, symbol, strategy_name, strategy_params, + cash_balance, asset_quantity, asset_price, profit_buffer + ) + VALUES (%s, %s, %s, %s, %s, %s::jsonb, %s, %s, %s, %s) + ON CONFLICT (portfolio_id) DO NOTHING + """, + ( + portfolio_id, + broker, + status, + symbol, + strategy_name, + json.dumps(strategy_params or {}), + float(initial_cash), + 0.0, + 0.0, + 0.0, + ), + ) + cur.execute(_SELECT_PORTFOLIO + "WHERE portfolio_id = %s", (portfolio_id,)) + row = cur.fetchone() + return self._row_to_portfolio(row) + + def create_portfolio( + self, + portfolio_id: str, + *, + broker: str, + status: str, + symbol: str, + strategy_name: str, + strategy_params: dict, + cash_balance: float, + asset_quantity: float, + asset_price: float, + profit_buffer: float, + ) -> 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, symbol, strategy_name, strategy_params, + cash_balance, asset_quantity, asset_price, profit_buffer + ) VALUES (%s, %s, %s, %s, %s, %s::jsonb, %s, %s, %s, %s) + RETURNING + portfolio_id, broker, status, symbol, strategy_name, strategy_params, + cash_balance, asset_quantity, asset_price, profit_buffer, + current_strategy_set_id, created_at, updated_at, last_trade_at + """, + ( + portfolio_id, + broker, + status, + symbol, + strategy_name, + json.dumps(strategy_params), + float(cash_balance), + float(asset_quantity), + float(asset_price), + float(profit_buffer), + ), + ) + row = cur.fetchone() + return self._row_to_portfolio(row) + + def update_portfolio(self, portfolio_id: str, **fields: object) -> Portfolio: + self.ensure_schema() + allowed = { + "broker": "broker", + "status": "status", + "symbol": "symbol", + "strategy_name": "strategy_name", + "strategy_params": "strategy_params", + "asset_price": "asset_price", + "current_strategy_set_id": "current_strategy_set_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 == "strategy_params": + assignments.append("strategy_params = %s::jsonb") + params.append(json.dumps(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 + RETURNING + portfolio_id, broker, status, symbol, strategy_name, strategy_params, + cash_balance, asset_quantity, asset_price, profit_buffer, + current_strategy_set_id, created_at, updated_at, last_trade_at + """, + (*params, portfolio_id), + ) + row = cur.fetchone() + return self._row_to_portfolio(row) + + def list_portfolios(self, *, broker: str | None = None, status: str | None = None) -> list[Portfolio]: + self.ensure_schema() + query = _SELECT_PORTFOLIO + clauses: list[str] = [] + params: list[object] = [] + if broker is not None: + clauses.append("broker = %s") + params.append(broker) + if status is not None: + clauses.append("status = %s") + params.append(status) + if clauses: + query += "WHERE " + " AND ".join(clauses) + " " + query += "ORDER BY updated_at DESC, portfolio_id ASC" + with self._connect().cursor() as cur: + cur.execute(query, tuple(params)) + rows = cur.fetchall() + return [self._row_to_portfolio(row) for row in rows] + + def get_or_create_strategy_set(self, strategy_name: str, strategy_config: str) -> StrategySet: + self.ensure_schema() + with self._connect().cursor() as cur: + new_id = str(uuid.uuid4()) + cur.execute( + """ + INSERT INTO strategy_sets (id, strategy_name, strategy_config) + VALUES (%s, %s, %s) + ON CONFLICT (strategy_name, strategy_config) DO NOTHING + """, + (new_id, strategy_name, strategy_config), + ) + cur.execute( + """ + SELECT id, strategy_name, strategy_config, created_at + FROM strategy_sets + WHERE strategy_name = %s AND strategy_config = %s + """, + (strategy_name, strategy_config), + ) + row = cur.fetchone() + return StrategySet(id=row[0], strategy_name=row[1], strategy_config=row[2], created_at=row[3]) + + def get_strategy_set(self, id: str) -> StrategySet | None: + self.ensure_schema() + with self._connect().cursor() as cur: + cur.execute( + "SELECT id, strategy_name, strategy_config, created_at FROM strategy_sets WHERE id = %s", + (id,), + ) + row = cur.fetchone() + if row is None: + return None + return StrategySet(id=row[0], strategy_name=row[1], strategy_config=row[2], created_at=row[3]) + + 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, + symbol = %s, + strategy_name = %s, + strategy_params = %s::jsonb, + cash_balance = %s, + asset_quantity = %s, + asset_price = %s, + profit_buffer = %s, + current_strategy_set_id = %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, + portfolio.symbol, + portfolio.strategy_name, + json.dumps(portfolio.strategy_params), + float(portfolio.cash_balance), + float(portfolio.asset_quantity), + float(current_price), + float(portfolio.profit_buffer), + portfolio.current_strategy_set_id, + trade_time, + action, + trade_time, + portfolio.portfolio_id, + ), + ) + if action in {"BUY", "SELL"}: + cur.execute( + """ + INSERT INTO portfolio_ledger ( + portfolio_id, recorded_at, action, price, quantity, fee_amount, + tax_amount, cash_balance_after, asset_quantity_after, + total_value_after, profit_buffer_after, strategy_set_id, note + ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + """, + ( + portfolio.portfolio_id, + trade_time, + action, + 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, + ), + ) + + 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 + portfolio_id, recorded_at, action, price, quantity, fee_amount, + tax_amount, cash_balance_after, asset_quantity_after, + total_value_after, profit_buffer_after, strategy_set_id, note + FROM portfolio_ledger + WHERE portfolio_id = %s + ORDER BY recorded_at DESC + LIMIT %s + """, + (portfolio_id, int(limit)), + ) + rows = cur.fetchall() + return [self._row_to_ledger_entry(row) for row in rows] + + 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, + current_strategy_set_id = NULL, + updated_at = NOW(), + last_trade_at = NULL + WHERE portfolio_id = %s + RETURNING + portfolio_id, broker, status, symbol, strategy_name, strategy_params, + cash_balance, asset_quantity, asset_price, profit_buffer, + current_strategy_set_id, created_at, updated_at, last_trade_at + """, + (portfolio_id,), + ) + row = cur.fetchone() + if row is None: + raise ValueError("포트폴리오를 찾을 수 없습니다") + cur.execute("DELETE FROM portfolio_ledger WHERE portfolio_id = %s", (portfolio_id,)) + return self._row_to_portfolio(row) + + 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 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 + RETURNING + portfolio_id, broker, status, symbol, strategy_name, strategy_params, + cash_balance, asset_quantity, asset_price, profit_buffer, + current_strategy_set_id, created_at, updated_at, last_trade_at + """, + (updated_buffer, portfolio_id), + ) + updated_row = cur.fetchone() + total_value = float(portfolio.cash_balance) + float(portfolio.asset_quantity) * float(portfolio.price) + cur.execute( + """ + INSERT INTO portfolio_ledger ( + portfolio_id, recorded_at, action, price, quantity, fee_amount, + tax_amount, cash_balance_after, asset_quantity_after, + total_value_after, profit_buffer_after, strategy_set_id, note + ) VALUES (%s, NOW(), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + """, + ( + portfolio_id, + normalized, + 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._row_to_portfolio(updated_row) + + 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 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 + RETURNING + portfolio_id, broker, status, symbol, strategy_name, strategy_params, + cash_balance, asset_quantity, asset_price, profit_buffer, + current_strategy_set_id, created_at, updated_at, last_trade_at + """, + (cash, buf, portfolio_id), + ) + updated_row = cur.fetchone() + total_value = float(cash) + float(portfolio.asset_quantity) * float(portfolio.price) + cur.execute( + """ + INSERT INTO portfolio_ledger ( + portfolio_id, recorded_at, action, price, quantity, fee_amount, + tax_amount, cash_balance_after, asset_quantity_after, + total_value_after, profit_buffer_after, strategy_set_id, note + ) VALUES (%s, NOW(), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + """, + ( + portfolio_id, + ledger_action, + 0.0, + 0.0, + 0.0, + 0.0, + float(cash), + float(portfolio.asset_quantity), + float(total_value), + float(buf), + "manual", + note, + ), + ) + return self._row_to_portfolio(updated_row) + + 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,)) + cur.execute("DELETE FROM portfolios WHERE portfolio_id = %s", (portfolio_id,)) + + +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 diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py new file mode 100644 index 0000000..a1e48e6 --- /dev/null +++ b/app/schemas/__init__.py @@ -0,0 +1 @@ +"""Pydantic schemas for PortfolioHub.""" diff --git a/app/schemas/portfolio.py b/app/schemas/portfolio.py new file mode 100644 index 0000000..81369f3 --- /dev/null +++ b/app/schemas/portfolio.py @@ -0,0 +1,121 @@ +"""Portfolio request and response schemas.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field + + +class StrategySetResponse(BaseModel): + id: str + strategy_name: str + strategy_config: str + created_at: datetime | None = None + + +class LedgerEntryResponse(BaseModel): + portfolio_id: str + recorded_at: datetime + action: str + price: float + quantity: float + fee_amount: float = 0.0 + tax_amount: float = 0.0 + cash_balance_after: float + asset_quantity_after: float + total_value_after: float + profit_buffer_after: float = 0.0 + strategy_set_id: str = "manual" + note: str = "" + + +class PortfolioBase(BaseModel): + broker: str = "kis" + status: str = "enabled" + symbol: str + strategy_name: str = "target_value" + strategy_params: dict[str, Any] = Field(default_factory=dict) + + +class PortfolioSummary(BaseModel): + portfolio_id: str + broker: str + status: str + symbol: str + strategy_name: str + strategy_params: dict[str, Any] = Field(default_factory=dict) + cash_balance: float = 0.0 + asset_quantity: float = 0.0 + price: float = 0.0 + profit_buffer: float = 0.0 + current_strategy_set_id: str | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + last_trade_at: datetime | None = None + + +class PortfolioListItem(PortfolioSummary): + total_value: float = 0.0 + + +class PortfolioDetailResponse(BaseModel): + portfolio: PortfolioSummary + recent_entries: list[LedgerEntryResponse] = Field(default_factory=list) + strategy_set: StrategySetResponse | None = None + live_quantity: float | None = None + + +class CreatePortfolioRequest(PortfolioBase): + portfolio_id: str + cash_balance: float = Field(default=0.0, ge=0.0) + asset_quantity: float = Field(default=0.0, ge=0.0) + asset_price: float = Field(default=0.0, ge=0.0) + profit_buffer: float = Field(default=0.0, ge=0.0) + + +class UpdatePortfolioRequest(BaseModel): + broker: str | None = None + status: str | None = None + symbol: str | None = None + strategy_name: str | None = None + strategy_params: dict[str, Any] | None = None + asset_price: float | None = Field(default=None, ge=0.0) + + +class BuyRequest(BaseModel): + quantity: float = Field(gt=0.0) + price: float = Field(gt=0.0) + fee: float = Field(default=0.0, ge=0.0) + tax: float = Field(default=0.0, ge=0.0) + note: str = "" + + +class SellRequest(BaseModel): + quantity: float = Field(gt=0.0) + price: float = Field(gt=0.0) + fee: float = Field(default=0.0, ge=0.0) + tax: float = Field(default=0.0, ge=0.0) + note: str = "" + + +class InvestRequest(BaseModel): + movement_type: Literal["deposit", "profit", "withdraw"] + amount: float = Field(gt=0.0) + note: str = "" + + +class TransferRequest(BaseModel): + direction: Literal["to-buffer", "from-buffer"] + amount: float = Field(gt=0.0) + note: str = "" + + +class ResetPortfolioRequest(BaseModel): + infer_symbol: str | None = None + + +class ErrorResponse(BaseModel): + detail: str + model_config = ConfigDict(extra="ignore") diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..85b9661 --- /dev/null +++ b/app/services/__init__.py @@ -0,0 +1 @@ +"""Service layer for PortfolioHub.""" diff --git a/app/services/portfolio_service.py b/app/services/portfolio_service.py new file mode 100644 index 0000000..fc31713 --- /dev/null +++ b/app/services/portfolio_service.py @@ -0,0 +1,218 @@ +"""Portfolio service orchestration.""" + +from __future__ import annotations + +import json + +from fastapi import Depends + +from app.core.config import Settings, get_settings +from app.domain.models import LedgerEntry, Portfolio, StrategySet +from app.domain.services import ( + apply_buy, + apply_delete, + apply_invest, + apply_reset, + apply_sell, + apply_transfer, + get_portfolio_status, +) +from app.repositories.factory import get_portfolio_repository_from_settings +from app.repositories.portfolio_repository import PortfolioRepository +from app.schemas.portfolio import ( + BuyRequest, + CreatePortfolioRequest, + InvestRequest, + PortfolioDetailResponse, + PortfolioListItem, + ResetPortfolioRequest, + SellRequest, + TransferRequest, + UpdatePortfolioRequest, +) + + +def _infer_symbol(portfolio_id: str) -> str: + parts = [part.strip() for part in portfolio_id.split(":") if part.strip()] + return parts[0] if parts else "" + + +def _to_portfolio_summary(portfolio: Portfolio): + from app.schemas.portfolio import PortfolioSummary + + return PortfolioSummary( + portfolio_id=portfolio.portfolio_id, + broker=portfolio.broker, + status=portfolio.status, + symbol=portfolio.symbol, + strategy_name=portfolio.strategy_name, + strategy_params=portfolio.strategy_params, + cash_balance=portfolio.cash_balance, + asset_quantity=portfolio.asset_quantity, + price=portfolio.price, + profit_buffer=portfolio.profit_buffer, + current_strategy_set_id=portfolio.current_strategy_set_id, + created_at=portfolio.created_at, + updated_at=portfolio.updated_at, + last_trade_at=portfolio.last_trade_at, + ) + + +def _to_strategy_set_response(strategy_set: StrategySet | None): + from app.schemas.portfolio import StrategySetResponse + + if strategy_set is None: + return None + return StrategySetResponse( + id=strategy_set.id, + strategy_name=strategy_set.strategy_name, + strategy_config=strategy_set.strategy_config, + created_at=strategy_set.created_at, + ) + + +def _to_ledger_entry_response(entry: LedgerEntry): + from app.schemas.portfolio import LedgerEntryResponse + + return LedgerEntryResponse( + portfolio_id=entry.portfolio_id, + recorded_at=entry.recorded_at, + action=entry.action, + price=entry.price, + quantity=entry.quantity, + fee_amount=entry.fee_amount, + tax_amount=entry.tax_amount, + cash_balance_after=entry.cash_balance_after, + asset_quantity_after=entry.asset_quantity_after, + total_value_after=entry.total_value_after, + profit_buffer_after=entry.profit_buffer_after, + strategy_set_id=entry.strategy_set_id, + note=entry.note, + ) + + +def _to_detail_response(status_result) -> PortfolioDetailResponse: + return PortfolioDetailResponse( + portfolio=_to_portfolio_summary(status_result.portfolio), + recent_entries=[_to_ledger_entry_response(entry) for entry in status_result.recent_entries], + strategy_set=_to_strategy_set_response(status_result.strategy_set), + live_quantity=status_result.live_quantity, + ) + + +def get_portfolio_repository(settings: Settings = Depends(get_settings)) -> PortfolioRepository: + return get_portfolio_repository_from_settings(settings) + + +class PortfolioService: + def __init__(self, repository: PortfolioRepository) -> None: + self._repository = repository + + def list_portfolios(self, *, broker: str | None = None, status: str | None = None) -> list[PortfolioListItem]: + portfolios = self._repository.list_portfolios(broker=broker, status=status) + return [ + PortfolioListItem( + **_to_portfolio_summary(portfolio).model_dump(), + total_value=float(portfolio.cash_balance) + float(portfolio.asset_quantity) * float(portfolio.price), + ) + for portfolio in portfolios + ] + + def get_portfolio_status(self, *, portfolio_id: str, recent_trades: int = 5) -> PortfolioDetailResponse: + status_result = get_portfolio_status(self._repository, portfolio_id, recent_trades=recent_trades) + return _to_detail_response(status_result) + + def create_portfolio(self, payload: CreatePortfolioRequest) -> PortfolioDetailResponse: + strategy_set = self._repository.get_or_create_strategy_set( + payload.strategy_name, + json.dumps(payload.strategy_params, sort_keys=True, ensure_ascii=True), + ) + portfolio = self._repository.create_portfolio( + payload.portfolio_id, + broker=payload.broker, + status=payload.status, + symbol=payload.symbol, + strategy_name=payload.strategy_name, + strategy_params=payload.strategy_params, + cash_balance=payload.cash_balance, + asset_quantity=payload.asset_quantity, + asset_price=payload.asset_price, + profit_buffer=payload.profit_buffer, + ) + portfolio = self._repository.update_portfolio( + payload.portfolio_id, + current_strategy_set_id=strategy_set.id, + ) + return self.get_portfolio_status(portfolio_id=portfolio.portfolio_id, recent_trades=5) + + def update_portfolio(self, portfolio_id: str, payload: UpdatePortfolioRequest) -> PortfolioDetailResponse: + fields = payload.model_dump(exclude_none=True) + if "strategy_name" in fields or "strategy_params" in fields: + current = self._repository.get(portfolio_id) + strategy_name = fields.get("strategy_name", current.strategy_name) + strategy_params = fields.get("strategy_params", current.strategy_params) + strategy_set = self._repository.get_or_create_strategy_set( + strategy_name, + json.dumps(strategy_params, sort_keys=True, ensure_ascii=True), + ) + fields["current_strategy_set_id"] = strategy_set.id + self._repository.update_portfolio(portfolio_id, **fields) + return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5) + + def buy(self, portfolio_id: str, payload: BuyRequest) -> PortfolioDetailResponse: + apply_buy( + self._repository, + portfolio_id, + quantity=payload.quantity, + price=payload.price, + fee=payload.fee, + tax=payload.tax, + note=payload.note, + ) + return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5) + + def sell(self, portfolio_id: str, payload: SellRequest) -> PortfolioDetailResponse: + apply_sell( + self._repository, + portfolio_id, + quantity=payload.quantity, + price=payload.price, + fee=payload.fee, + tax=payload.tax, + note=payload.note, + ) + return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5) + + def invest(self, portfolio_id: str, payload: InvestRequest) -> PortfolioDetailResponse: + apply_invest( + self._repository, + portfolio_id, + movement_type=payload.movement_type, + amount=payload.amount, + note=payload.note, + ) + return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5) + + def transfer(self, portfolio_id: str, payload: TransferRequest) -> PortfolioDetailResponse: + apply_transfer( + self._repository, + portfolio_id, + direction=payload.direction, + amount=payload.amount, + note=payload.note, + ) + return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5) + + def reset(self, portfolio_id: str, payload: ResetPortfolioRequest) -> PortfolioDetailResponse: + infer_symbol = payload.infer_symbol or _infer_symbol(portfolio_id) + apply_reset(self._repository, portfolio_id, infer_symbol=lambda _: infer_symbol) + return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5) + + def delete(self, portfolio_id: str, *, keep_ledger: bool = False) -> None: + apply_delete(self._repository, portfolio_id, clear_ledger=not keep_ledger) + + +def get_portfolio_service( + repository: PortfolioRepository = Depends(get_portfolio_repository), +) -> PortfolioService: + return PortfolioService(repository) diff --git a/migrations/env.py b/migrations/env.py new file mode 100644 index 0000000..6f4d92a --- /dev/null +++ b/migrations/env.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import engine_from_config, pool + +config = context.config + +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = None + + +def run_migrations_offline() -> None: + url = config.get_main_option("sqlalchemy.url") + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + connectable = engine_from_config( + config.get_section(config.config_ini_section, {}), + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata) + + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() \ No newline at end of file diff --git a/migrations/script.py.mako b/migrations/script.py.mako new file mode 100644 index 0000000..6c9eca9 --- /dev/null +++ b/migrations/script.py.mako @@ -0,0 +1,20 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} \ No newline at end of file diff --git a/migrations/versions/20260403_0001_initial_portfolio_schema.py b/migrations/versions/20260403_0001_initial_portfolio_schema.py new file mode 100644 index 0000000..d0c7d33 --- /dev/null +++ b/migrations/versions/20260403_0001_initial_portfolio_schema.py @@ -0,0 +1,90 @@ +"""initial portfolio schema + +Revision ID: 20260403_0001 +Revises: +Create Date: 2026-04-03 00:00:00 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +revision = "20260403_0001" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "strategy_sets", + sa.Column("id", sa.Text(), primary_key=True), + sa.Column("strategy_name", sa.Text(), nullable=False), + sa.Column("strategy_config", sa.Text(), nullable=False, server_default=""), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("NOW()")), + sa.UniqueConstraint("strategy_name", "strategy_config", name="uq_strategy_sets_name_config"), + ) + + op.create_table( + "portfolios", + sa.Column("portfolio_id", sa.Text(), primary_key=True), + sa.Column("broker", sa.Text(), nullable=False, server_default="kis"), + sa.Column("status", sa.Text(), nullable=False, server_default="enabled"), + sa.Column("symbol", sa.Text(), nullable=False), + sa.Column("strategy_name", sa.Text(), nullable=False, server_default=""), + sa.Column("strategy_params", postgresql.JSONB(astext_type=sa.Text()), nullable=False, server_default=sa.text("'{}'::jsonb")), + sa.Column("cash_balance", sa.Float(), nullable=False), + sa.Column("asset_quantity", sa.Float(), nullable=False), + sa.Column("asset_price", sa.Float(), nullable=False, server_default="0"), + sa.Column("profit_buffer", sa.Float(), nullable=False, server_default="0"), + sa.Column("current_strategy_set_id", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("NOW()")), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("NOW()")), + sa.Column("last_trade_at", sa.DateTime(timezone=True), nullable=True), + ) + + op.create_table( + "portfolio_ledger", + sa.Column("portfolio_id", sa.Text(), nullable=False), + sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("action", sa.Text(), nullable=False), + sa.Column("price", sa.Float(), nullable=False), + sa.Column("quantity", sa.Float(), nullable=False), + sa.Column("fee_amount", sa.Float(), nullable=False, server_default="0"), + sa.Column("tax_amount", sa.Float(), nullable=False, server_default="0"), + sa.Column("cash_balance_after", sa.Float(), nullable=False), + sa.Column("asset_quantity_after", sa.Float(), nullable=False), + sa.Column("total_value_after", sa.Float(), nullable=False), + sa.Column("profit_buffer_after", sa.Float(), nullable=False, server_default="0"), + sa.Column("strategy_set_id", sa.Text(), nullable=False, server_default="legacy"), + sa.Column("note", sa.Text(), nullable=False, server_default=""), + ) + + op.create_index( + "idx_portfolio_ledger_portfolio_time", + "portfolio_ledger", + ["portfolio_id", sa.text("recorded_at DESC")], + unique=False, + ) + op.create_index( + "idx_portfolios_broker_status_updated", + "portfolios", + ["broker", "status", sa.text("updated_at DESC")], + unique=False, + ) + + bind = op.get_bind() + has_timescaledb = bind.execute(sa.text("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'timescaledb')")).scalar() + if has_timescaledb: + bind.execute(sa.text("SELECT create_hypertable('portfolio_ledger', 'recorded_at', if_not_exists => TRUE)")) + + +def downgrade() -> None: + op.drop_index("idx_portfolios_broker_status_updated", table_name="portfolios") + op.drop_index("idx_portfolio_ledger_portfolio_time", table_name="portfolio_ledger") + op.drop_table("portfolio_ledger") + op.drop_table("portfolios") + op.drop_table("strategy_sets") \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..595366c --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,40 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "portfoliohub" +version = "0.1.0" +description = "PortfolioHub API for portfolio management and ledger access" +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "alembic>=1.14.0,<2.0.0", + "fastapi>=0.115.0,<1.0.0", + "pydantic-settings>=2.4.0,<3.0.0", + "psycopg[binary]>=3.2.0,<4.0.0", + "python-dotenv>=1.0.0,<2.0.0", + "rich>=13.9.0,<14.0.0", + "typer>=0.12.5,<1.0.0", + "uvicorn[standard]>=0.30.0,<1.0.0", +] + +[project.scripts] +portfoliohub = "app.cli.main:main" + +[project.optional-dependencies] +dev = [ + "httpx>=0.27.0,<1.0.0", + "pytest>=8.3.0,<9.0.0", +] + +[tool.pytest.ini_options] +addopts = ["tests"] +testpaths = ["tests"] +norecursedirs = ["temp", ".git", ".venv"] + +[tool.setuptools] +package-dir = {"" = "."} + +[tool.setuptools.packages.find] +include = ["app*"] diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..d2aabea --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,27 @@ +from fastapi.testclient import TestClient + +from app.core.config import get_settings +from app.main import app + + +def test_healthcheck() -> None: + client = TestClient(app) + + response = client.get("/health") + + assert response.status_code == 200 + assert response.json() == {"status": "ok"} + + +def test_docs_are_exposed() -> None: + client = TestClient(app) + + response = client.get("/docs") + + assert response.status_code == 200 + + +def test_settings_default_origin() -> None: + settings = get_settings() + + assert settings.allowed_origins == ["http://localhost:3000"] diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..9b9a41b --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +from fastapi.testclient import TestClient + +from app.core.config import get_settings +from app.main import app + + +def _client_with_tokens() -> TestClient: + get_settings.cache_clear() + app.dependency_overrides.clear() + return TestClient(app) + + +def test_read_route_requires_token(monkeypatch) -> None: + monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token") + get_settings.cache_clear() + client = _client_with_tokens() + + response = client.get("/portfolios") + + assert response.status_code == 401 + assert response.json()["detail"] == "Missing bearer token" + + +def test_read_route_accepts_read_token(monkeypatch) -> None: + monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token") + get_settings.cache_clear() + client = _client_with_tokens() + + response = client.get("/portfolios", headers={"Authorization": "Bearer read-token"}) + + assert response.status_code == 501 + assert response.json()["detail"] == "Portfolio persistence is not implemented yet" + + +def test_trade_route_rejects_read_token(monkeypatch) -> None: + monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token") + monkeypatch.setenv("PORTFOLIOHUB_TRADE_WRITE_TOKENS", "trade-token") + get_settings.cache_clear() + client = _client_with_tokens() + + response = client.post( + "/portfolios/p1/buy", + headers={"Authorization": "Bearer read-token"}, + json={"quantity": 1, "price": 1000, "fee": 0, "tax": 0, "note": ""}, + ) + + assert response.status_code == 403 + assert response.json()["detail"] == "Insufficient permission" + + +def test_admin_route_accepts_admin_token(monkeypatch) -> None: + monkeypatch.setenv("PORTFOLIOHUB_ADMIN_TOKENS", "admin-token") + get_settings.cache_clear() + client = _client_with_tokens() + + response = client.delete("/portfolios/p1", headers={"Authorization": "Bearer admin-token"}) + + assert response.status_code == 501 + assert response.json()["detail"] == "Portfolio persistence is not implemented yet" diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..0c02766 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,386 @@ +from __future__ import annotations + +import json +from datetime import datetime, timezone + +import app.cli.commands.health as health_commands +from typer.testing import CliRunner + +from app.cli.main import create_app +from app.core.config import Settings +from app.domain.models import LedgerEntry, Portfolio, StrategySet + + +class FakeRepo: + def __init__(self, portfolio: Portfolio | None = None) -> None: + self._portfolio = portfolio + self.entries: list[LedgerEntry] = [] + self.strategy_set = StrategySet(id="set-1", strategy_name="target_value", strategy_config="{}") + self.closed = False + + def close(self) -> None: + self.closed = True + + def ensure_schema(self) -> None: + return None + + def ping(self) -> None: + return None + + def get(self, portfolio_id: str) -> Portfolio: + if self._portfolio is None or self._portfolio.portfolio_id != portfolio_id: + raise ValueError("포트폴리오를 찾을 수 없습니다") + return self._portfolio + + def get_or_create( + self, + portfolio_id: str, + symbol: str, + *, + broker: str = "kis", + status: str = "enabled", + strategy_name: str = "", + strategy_params: dict | None = None, + initial_cash: float = 0.0, + ) -> Portfolio: + if self._portfolio is None: + self._portfolio = Portfolio( + portfolio_id=portfolio_id, + broker=broker, + status=status, + symbol=symbol, + strategy_name=strategy_name, + strategy_params=strategy_params or {}, + cash_balance=initial_cash, + ) + return self._portfolio + + def create_portfolio( + self, + portfolio_id: str, + *, + broker: str, + status: str, + symbol: str, + strategy_name: str, + strategy_params: dict, + cash_balance: float, + asset_quantity: float, + asset_price: float, + profit_buffer: float, + ) -> Portfolio: + self._portfolio = Portfolio( + portfolio_id=portfolio_id, + broker=broker, + status=status, + symbol=symbol, + strategy_name=strategy_name, + strategy_params=strategy_params, + cash_balance=cash_balance, + asset_quantity=asset_quantity, + price=asset_price, + profit_buffer=profit_buffer, + ) + return self._portfolio + + def update_portfolio(self, portfolio_id: str, **fields: object) -> Portfolio: + portfolio = self.get(portfolio_id) + for key, value in fields.items(): + if value is not None: + setattr(portfolio, key, value) + return portfolio + + def list_portfolios(self, *, broker: str | None = None, status: str | None = None) -> list[Portfolio]: + if self._portfolio is None: + return [] + if broker is not None and self._portfolio.broker != broker: + return [] + if status is not None and self._portfolio.status != status: + return [] + return [self._portfolio] + + def get_or_create_strategy_set(self, strategy_name: str, strategy_config: str) -> StrategySet: + self.strategy_set = StrategySet(id="set-1", strategy_name=strategy_name, strategy_config=strategy_config) + return self.strategy_set + + def get_strategy_set(self, id: str) -> StrategySet | None: + return self.strategy_set if self.strategy_set.id == id else None + + 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._portfolio = portfolio + self.entries.insert( + 0, + LedgerEntry( + portfolio_id=portfolio.portfolio_id, + recorded_at=datetime.now(timezone.utc), + action=action, + price=trade_price, + quantity=quantity, + fee_amount=fee_amount, + tax_amount=tax_amount, + cash_balance_after=portfolio.cash_balance, + asset_quantity_after=portfolio.asset_quantity, + total_value_after=total_value_after, + profit_buffer_after=portfolio.profit_buffer, + strategy_set_id=strategy_set_id, + note=note, + ), + ) + + def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]: + return self.entries[:limit] + + def reset(self, portfolio_id: str) -> Portfolio: + portfolio = self.get(portfolio_id) + portfolio.cash_balance = 0.0 + portfolio.asset_quantity = 0.0 + portfolio.price = 0.0 + portfolio.profit_buffer = 0.0 + portfolio.current_strategy_set_id = None + self.entries = [] + return portfolio + + def adjust_buffer(self, portfolio_id: str, amount: float, action: str, note: str = "") -> Portfolio: + portfolio = self.get(portfolio_id) + if action == "INVEST": + portfolio.profit_buffer += amount + else: + portfolio.profit_buffer -= amount + return portfolio + + def transfer_buffer(self, portfolio_id: str, amount: float, direction: str, note: str = "") -> Portfolio: + portfolio = self.get(portfolio_id) + if direction == "to-buffer": + portfolio.cash_balance -= amount + portfolio.profit_buffer += amount + else: + portfolio.cash_balance += amount + portfolio.profit_buffer -= amount + return portfolio + + def delete(self, portfolio_id: str, *, clear_ledger: bool = True) -> None: + self._portfolio = None + if clear_ledger: + self.entries = [] + + +def _settings() -> Settings: + return Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000") + + +def build_runner(repo: FakeRepo) -> CliRunner: + app = create_app(settings_factory=_settings, repository_factory=lambda settings: repo) + runner = CliRunner() + runner.app = app + return runner + + +def _portfolio() -> Portfolio: + return Portfolio( + portfolio_id="005930:tv", + broker="kis", + status="enabled", + symbol="005930", + strategy_name="target_value", + strategy_params={"target_asset_ratio": 0.7}, + cash_balance=1_000_000.0, + asset_quantity=1.0, + price=70_000.0, + profit_buffer=50_000.0, + current_strategy_set_id="set-1", + ) + + +def test_cli_list_renders_table() -> None: + repo = FakeRepo(_portfolio()) + runner = build_runner(repo) + + result = runner.invoke(runner.app, ["portfolio", "list"]) + + assert result.exit_code == 0 + assert "005930:tv" in result.stdout + assert "Portfolios" in result.stdout + + +def test_cli_status_supports_json_output() -> None: + repo = FakeRepo(_portfolio()) + runner = build_runner(repo) + + result = runner.invoke(runner.app, ["--format", "json", "portfolio", "status", "005930:tv"]) + + assert result.exit_code == 0 + payload = json.loads(result.stdout) + assert payload["portfolio"]["portfolio_id"] == "005930:tv" + + +def test_cli_create_buy_and_delete_flow() -> None: + repo = FakeRepo() + runner = build_runner(repo) + + create_result = runner.invoke( + runner.app, + [ + "portfolio", + "create", + "005930:tv", + "--symbol", + "005930", + "--cash-balance", + "1000000", + "--strategy-params", + '{"target_asset_ratio": 0.5}', + "--force", + ], + ) + buy_result = runner.invoke( + runner.app, + [ + "portfolio", + "buy", + "005930:tv", + "--quantity", + "2", + "--price", + "50000", + "--fee", + "100", + "--force", + ], + ) + assert repo.entries[0].action == "BUY" + delete_result = runner.invoke(runner.app, ["portfolio", "delete", "005930:tv", "--keep-ledger", "--force"]) + + assert create_result.exit_code == 0 + assert buy_result.exit_code == 0 + assert delete_result.exit_code == 0 + assert "Deleted portfolio: 005930:tv" in delete_result.stdout + + +def test_cli_write_command_prompts_before_create() -> None: + repo = FakeRepo() + runner = build_runner(repo) + + result = runner.invoke( + runner.app, + [ + "portfolio", + "create", + "005930:tv", + "--symbol", + "005930", + "--cash-balance", + "1000000", + ], + input="y\n", + ) + + assert result.exit_code == 0 + assert "Create portfolio 005930:tv for symbol 005930?" in result.stdout + + +def test_cli_write_command_can_abort() -> None: + repo = FakeRepo() + runner = build_runner(repo) + + result = runner.invoke( + runner.app, + [ + "portfolio", + "create", + "005930:tv", + "--symbol", + "005930", + "--cash-balance", + "1000000", + ], + input="n\n", + ) + + assert result.exit_code == 0 + assert repo._portfolio is None + + +def test_cli_health_check_reports_repository() -> None: + repo = FakeRepo(_portfolio()) + runner = build_runner(repo) + + result = runner.invoke(runner.app, ["health-check"]) + + assert result.exit_code == 0 + assert "repository: FakeRepo" in result.stdout + + +def test_cli_init_db_reports_created(monkeypatch) -> None: + repo = FakeRepo(_portfolio()) + runner = build_runner(repo) + + def fake_initialize_database(*, settings, maintenance_db: str = "postgres") -> dict[str, object]: + return { + "database": settings.active_db_name, + "host": settings.active_db_host, + "port": settings.active_db_port, + "created": True, + "schema_initialized": True, + "migration_applied": "head", + } + + monkeypatch.setattr(health_commands, "initialize_database", fake_initialize_database) + + result = runner.invoke(runner.app, ["init-db"]) + + assert result.exit_code == 0 + assert "created: True" in result.stdout + assert "schema_initialized: True" in result.stdout + assert "migration_applied: head" in result.stdout + + +def test_cli_init_db_supports_json(monkeypatch) -> None: + repo = FakeRepo(_portfolio()) + runner = build_runner(repo) + + def fake_initialize_database(*, settings, maintenance_db: str = "postgres") -> dict[str, object]: + return { + "database": "portfoliohub", + "host": "localhost", + "port": 5432, + "created": False, + "schema_initialized": True, + "migration_applied": "head", + } + + monkeypatch.setattr(health_commands, "initialize_database", fake_initialize_database) + + result = runner.invoke(runner.app, ["--format", "json", "init-db"]) + + assert result.exit_code == 0 + payload = json.loads(result.stdout) + assert payload["database"] == "portfoliohub" + assert payload["schema_initialized"] is True + + +def test_cli_migrate_db_reports_head(monkeypatch) -> None: + repo = FakeRepo(_portfolio()) + runner = build_runner(repo) + + def fake_bootstrap_or_upgrade_to_head(settings) -> None: + return None + + monkeypatch.setattr(health_commands, "bootstrap_or_upgrade_to_head", fake_bootstrap_or_upgrade_to_head) + + result = runner.invoke(runner.app, ["migrate-db"]) + + assert result.exit_code == 0 + assert "migration_applied: head" in result.stdout \ No newline at end of file diff --git a/tests/test_domain_services.py b/tests/test_domain_services.py new file mode 100644 index 0000000..3468fda --- /dev/null +++ b/tests/test_domain_services.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from app.domain.models import LedgerEntry, Portfolio, StrategySet +from app.domain.services import apply_buy, apply_invest, apply_reset, apply_sell, apply_transfer, get_portfolio_status + + +class FakeRepo: + def __init__(self, portfolio: Portfolio | None = None) -> None: + self._portfolio = portfolio + self.entries: list[LedgerEntry] = [] + self.strategy_set = StrategySet(id="set-1", strategy_name="target_value", strategy_config="{}") + self.last_delete: tuple[str, bool] | None = None + + def get(self, portfolio_id: str) -> Portfolio: + if self._portfolio is None or self._portfolio.portfolio_id != portfolio_id: + raise ValueError("포트폴리오를 찾을 수 없습니다") + return self._portfolio + + def get_or_create( + self, + portfolio_id: str, + symbol: str, + *, + broker: str = "kis", + status: str = "enabled", + strategy_name: str = "", + strategy_params: dict | None = None, + initial_cash: float = 0.0, + ) -> Portfolio: + self._portfolio = Portfolio( + portfolio_id=portfolio_id, + broker=broker, + status=status, + symbol=symbol, + strategy_name=strategy_name, + strategy_params=strategy_params or {}, + cash_balance=initial_cash, + ) + return self._portfolio + + def create_portfolio(self, *args, **kwargs): + raise NotImplementedError() + + def update_portfolio(self, portfolio_id: str, **fields: object) -> Portfolio: + portfolio = self.get(portfolio_id) + for key, value in fields.items(): + if value is not None and hasattr(portfolio, key): + setattr(portfolio, key, value) + return portfolio + + def list_portfolios(self, *, broker: str | None = None, status: str | None = None) -> list[Portfolio]: + if self._portfolio is None: + return [] + return [self._portfolio] + + def get_strategy_set(self, id: str) -> StrategySet | None: + return self.strategy_set if id == self.strategy_set.id else None + + def get_or_create_strategy_set(self, strategy_name: str, strategy_config: str) -> StrategySet: + self.strategy_set = StrategySet(id="set-1", strategy_name=strategy_name, strategy_config=strategy_config) + return self.strategy_set + + 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._portfolio = portfolio + self.entries.insert( + 0, + LedgerEntry( + portfolio_id=portfolio.portfolio_id, + recorded_at=datetime.now(timezone.utc), + action=action, + price=trade_price, + quantity=quantity, + cash_balance_after=portfolio.cash_balance, + asset_quantity_after=portfolio.asset_quantity, + total_value_after=total_value_after, + fee_amount=fee_amount, + tax_amount=tax_amount, + profit_buffer_after=portfolio.profit_buffer, + strategy_set_id=strategy_set_id, + note=note, + ), + ) + + def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]: + return self.entries[:limit] + + def reset(self, portfolio_id: str) -> Portfolio: + portfolio = self.get(portfolio_id) + portfolio.cash_balance = 0.0 + portfolio.asset_quantity = 0.0 + portfolio.price = 0.0 + portfolio.profit_buffer = 0.0 + portfolio.current_strategy_set_id = None + self.entries = [] + return portfolio + + def adjust_buffer(self, portfolio_id: str, amount: float, action: str, note: str = "") -> Portfolio: + portfolio = self.get(portfolio_id) + if action == "INVEST": + portfolio.profit_buffer += amount + else: + if portfolio.profit_buffer < amount: + raise ValueError("수익버퍼 잔액이 부족합니다") + portfolio.profit_buffer -= amount + return portfolio + + def transfer_buffer(self, portfolio_id: str, amount: float, direction: str, note: str = "") -> Portfolio: + portfolio = self.get(portfolio_id) + if direction == "to-buffer": + if portfolio.cash_balance < amount: + raise ValueError("이체 금액이 현금 잔고를 초과합니다") + portfolio.cash_balance -= amount + portfolio.profit_buffer += amount + else: + if portfolio.profit_buffer < amount: + raise ValueError("이체 금액이 수익버퍼 잔고를 초과합니다") + portfolio.cash_balance += amount + portfolio.profit_buffer -= amount + return portfolio + + def delete(self, portfolio_id: str, clear_ledger: bool = True) -> None: + self.last_delete = (portfolio_id, clear_ledger) + self._portfolio = None + + +def _portfolio() -> Portfolio: + return Portfolio( + portfolio_id="p1", + broker="kis", + status="enabled", + symbol="005930", + strategy_name="target_value", + strategy_params={"target_asset_ratio": 0.7}, + cash_balance=1_000_000.0, + asset_quantity=0.0, + price=70_000.0, + profit_buffer=0.0, + current_strategy_set_id="set-1", + ) + + +def test_apply_buy_updates_cash_and_quantity() -> None: + repo = FakeRepo(_portfolio()) + + updated = apply_buy(repo, "p1", quantity=2, price=50_000, fee=100, tax=0, note="buy") + + assert updated.cash_balance == pytest.approx(899_900.0) + assert updated.asset_quantity == pytest.approx(2.0) + assert repo.entries[0].action == "BUY" + + +def test_apply_sell_updates_cash_and_quantity() -> None: + portfolio = _portfolio() + portfolio.asset_quantity = 3.0 + repo = FakeRepo(portfolio) + + updated = apply_sell(repo, "p1", quantity=1, price=80_000, fee=100, tax=50, note="sell") + + assert updated.cash_balance == pytest.approx(1_079_850.0) + assert updated.asset_quantity == pytest.approx(2.0) + assert repo.entries[0].action == "SELL" + + +def test_apply_invest_deposit_updates_buffer() -> None: + repo = FakeRepo(_portfolio()) + + updated = apply_invest(repo, "p1", movement_type="deposit", amount=50_000, note="deposit") + + assert updated.profit_buffer == pytest.approx(50_000.0) + + +def test_apply_transfer_to_buffer_moves_cash() -> None: + repo = FakeRepo(_portfolio()) + + updated = apply_transfer(repo, "p1", direction="to-buffer", amount=100_000, note="shift") + + assert updated.cash_balance == pytest.approx(900_000.0) + assert updated.profit_buffer == pytest.approx(100_000.0) + + +def test_apply_reset_creates_when_missing() -> None: + repo = FakeRepo(None) + + updated, created = apply_reset(repo, "005930:tv", infer_symbol=lambda portfolio_id: portfolio_id.split(":")[0]) + + assert created is True + assert updated.symbol == "005930" + + +def test_get_portfolio_status_includes_strategy_set_and_entries() -> None: + repo = FakeRepo(_portfolio()) + apply_buy(repo, "p1", quantity=1, price=50_000) + + result = get_portfolio_status(repo, "p1", recent_trades=5) + + assert result.portfolio.portfolio_id == "p1" + assert len(result.recent_entries) == 1 + assert result.strategy_set is not None