166 lines
4.7 KiB
Python
166 lines
4.7 KiB
Python
"""Portfolio repository contract."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Protocol
|
|
|
|
from app.domain.models import AuthToken, BotPrincipal, LedgerEntry, Portfolio, PortfolioComment, StrategySet, TokenPrincipal, User
|
|
|
|
|
|
class PortfolioRepository(Protocol):
|
|
def ping(self) -> None: ...
|
|
|
|
def get_user(self, user_id: str) -> User: ...
|
|
|
|
def get_user_by_username(self, username: str) -> User | None: ...
|
|
|
|
def list_users(self) -> list[User]: ...
|
|
|
|
def create_user(
|
|
self,
|
|
username: str,
|
|
*,
|
|
email: str | None = None,
|
|
password_hash: str | None = None,
|
|
permission: str = "read",
|
|
is_active: bool = True,
|
|
) -> User: ...
|
|
|
|
def update_user(
|
|
self,
|
|
user_id: str,
|
|
*,
|
|
username: str | None = None,
|
|
email: str | None = None,
|
|
password_hash: str | None = None,
|
|
permission: str | None = None,
|
|
is_active: bool | None = None,
|
|
) -> User: ...
|
|
|
|
def delete_user(self, user_id: str) -> None: ...
|
|
|
|
def update_user_last_login(self, user_id: str) -> None: ...
|
|
|
|
def create_auth_token(
|
|
self,
|
|
user_id: str,
|
|
*,
|
|
name: str,
|
|
token_hash: str,
|
|
token_prefix: str,
|
|
permission: str,
|
|
token_type: str,
|
|
bot_principal_id: str | None = None,
|
|
expires_at: datetime | None = None,
|
|
) -> AuthToken: ...
|
|
|
|
def list_auth_tokens(self, user_id: str) -> list[AuthToken]: ...
|
|
|
|
def revoke_auth_token(self, user_id: str, token_id: str) -> None: ...
|
|
|
|
def create_bot_principal(
|
|
self,
|
|
*,
|
|
name: str,
|
|
permission: str,
|
|
description: str | None = None,
|
|
) -> BotPrincipal: ...
|
|
|
|
def get_bot_principal(self, bot_principal_id: str) -> BotPrincipal: ...
|
|
|
|
def list_bot_principals(self, *, is_active: bool | None = None) -> list[BotPrincipal]: ...
|
|
|
|
def deactivate_bot_principal(self, bot_principal_id: str) -> BotPrincipal: ...
|
|
|
|
def list_bot_auth_tokens(self, bot_principal_id: str) -> list[AuthToken]: ...
|
|
|
|
def revoke_bot_auth_token(self, bot_principal_id: str, token_id: str) -> None: ...
|
|
|
|
def get_token_principal(self, token_hash: str) -> TokenPrincipal | None: ...
|
|
|
|
def get(self, portfolio_id: str) -> Portfolio: ...
|
|
|
|
def get_or_create(
|
|
self,
|
|
portfolio_id: str,
|
|
symbol: str,
|
|
*,
|
|
name: str | None = None,
|
|
broker: str = "kis",
|
|
status: str = "enabled",
|
|
active_strategy_id: str | None = None,
|
|
avabile_strategies: list[str] | None = None,
|
|
owner_user_id: str | None = None,
|
|
initial_cash: float = 0.0,
|
|
) -> Portfolio: ...
|
|
|
|
def create_portfolio(
|
|
self,
|
|
portfolio_id: str,
|
|
*,
|
|
owner_user_id: str,
|
|
broker: str,
|
|
status: str,
|
|
name: str,
|
|
symbol: str,
|
|
cash_balance: float,
|
|
asset_quantity: float,
|
|
asset_price: float,
|
|
profit_buffer: float,
|
|
active_strategy_id: str | None,
|
|
avabile_strategies: list[str] | None,
|
|
) -> Portfolio: ...
|
|
|
|
def update_portfolio(self, portfolio_id: str, **fields: object) -> Portfolio: ...
|
|
|
|
def list_portfolios(
|
|
self,
|
|
*,
|
|
broker: str | None = None,
|
|
status: str | None = None,
|
|
owner_user_id: str | None = None,
|
|
) -> list[Portfolio]: ...
|
|
|
|
def get_or_create_strategy_set(self, name: str, config: str) -> StrategySet: ...
|
|
|
|
def get_strategy_set(self, id: str) -> StrategySet | None: ...
|
|
|
|
def list_strategy_sets(self) -> list[StrategySet]: ...
|
|
|
|
def delete_strategy_set(self, id: str) -> 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: ...
|
|
|
|
def create_comment(self, portfolio_id: str, *, content: str) -> PortfolioComment: ...
|
|
|
|
def list_comments(self, portfolio_id: str) -> list[PortfolioComment]: ...
|
|
|
|
def update_comment(self, portfolio_id: str, comment_id: str, *, content: str) -> PortfolioComment: ...
|
|
|
|
def delete_comment(self, portfolio_id: str, comment_id: str) -> None: ...
|
|
|