zerohertzLib.quant.koreainvestment ¶

def __call__(self) -> int:
    """현재 보유 금액을 반환

    Returns:
        현재 보유 금액
    """
    return self.balance["cash"]

def __contains__(self, item: Any) -> bool:
    """보유 종목 여부를 확인

    Args:
        item: 보유 여부를 판단할 종목명

    Returns:
        입력 종목명의 보유 여부
    """
    return item in self.balance["stock"]

def __getitem__(self, idx: int) -> list[int | float | str]:
    """Index에 따른 주식 정보를 반환

    Args:
        idx: Index

    Returns:
        Index에 따른 주식의 매수 시점과 현재의 정보
    """
    return self.balance["stock"][self.symbols[idx]]

def __len__(self) -> int:
    """보유 주식 종류의 개수를 반환

    Returns:
        보유 주식 종류의 수
    """
    return len(self.balance["stock"])

def _exchange(self) -> float:
    """USD/KRW의 현재 시세를 조회

    Returns:
        USD/KRW의 현재 시세
    """
    now = datetime.now()
    data = fdr.DataReader("USD/KRW", now - timedelta(days=10))
    return float(data.Close[-1])

def barv(self) -> str:
    """현재 보유 종목의 이익과 손실을 bar chart로 시각화

    Returns:
        저장된 graph의 절대 경로

    Examples:
        >>> balance.barv()
    """
    if self.kor:
        dim = "₩"
    else:
        dim = "$"
    data = {}
    for value in self:
        data[value[0]] = value[5]
    figure((30, 10))
    barv(
        data,
        xlab="",
        ylab=f"Profit and Loss (P&L) [{dim}]",
        title="",
        dim="",
        dimsize=16,
    )
    plt.gca().yaxis.set_major_formatter(
        ticker.FuncFormatter(lambda x, p: format(int(x), ","))
    )
    return savefig("ProfitLoss", 100)

def bought_symbols(self) -> list[str]:
    """보유 주식의 종목 code return

    Returns:
        보유 주식의 종목 code들

    Examples:
        >>> balance.bought_symbols():
        ['066570']
    """
    return [value[0] for _, value in self.items()]

def items(self) -> ItemsView[str, list[int | float | str]]:
    """보유 주식의 반복문 사용을 위한 method

    Returns:
        보유 종목 code와 그에 따른 정보들

    Examples:
        >>> for k, v in balance.items():
        >>>     print(k, v)
    """
    return self.balance["stock"].items()

def merge(self, balance: "Balance") -> None:
    """현재 계좌와 입력 계좌의 정보를 병합하는 function

    Args:
        balance: 병합될 계좌 정보

    Returns:
        현재 계좌에 정보 update

    Examples:
        >>> balance_1.merge(balance_2)
    """
    merged_balance = copy.deepcopy(balance.balance)
    if self.kor != balance.kor:
        exchange = self._exchange()
        if not self.kor:
            exchange = 1 / exchange
        for key, value in balance.items():
            merged_balance["stock"][key][1] = value[1] * exchange
            merged_balance["stock"][key][2] = value[2] * exchange
            merged_balance["stock"][key][-1] = value[-1] * exchange
        merged_balance["cash"] = balance.balance["cash"] * exchange
    for key, value in merged_balance["stock"].items():
        if key in self:
            (
                _merged_code,
                _merged_buy_price,
                _merged_present_price,
                _merged_cnt,
                _merged_pandl_per,
                _merged_pandl_abs,
            ) = self.balance["stock"][key]
            (
                _tmp_code,
                _tmp_buy_price,
                _tmp_present_price,
                _tmp_cnt,
                _tmp_pandl_per,
                _tmp_pandl_abs,
            ) = value
            assert _merged_code == _tmp_code
            _merged_buy_price = (
                _merged_buy_price * _merged_cnt + _tmp_buy_price * _tmp_cnt
            ) / (_merged_cnt + _tmp_cnt)
            _merged_present_price = (_merged_present_price + _tmp_present_price) / 2
            _merged_cnt += _tmp_cnt
            _merged_pandl_abs = (
                _merged_present_price - _merged_buy_price
            ) * _merged_cnt
            _merged_pandl_per = (
                (_merged_present_price - _merged_buy_price)
                / _merged_buy_price
                * 100
            )
            self.balance["stock"][key] = [
                _merged_code,
                _merged_buy_price,
                _merged_present_price,
                _merged_cnt,
                _merged_pandl_per,
                _merged_pandl_abs,
            ]
        else:
            self.symbols.append(key)
            self.balance["stock"][key] = value
    self.balance["cash"] += merged_balance["cash"]
    self.balance["stock"] = dict(
        sorted(
            self.balance["stock"].items(),
            key=lambda item: item[1][1] * item[1][3],
            reverse=True,
        )
    )
    self.symbols = list(self.balance["stock"].keys())

def pie(self) -> str:
    """현재 보유 종목을 pie chart로 시각화

    Returns:
        저장된 graph의 절대 경로

    Examples:
        >>> balance.pie()
    """
    if self() == 0:
        return None
    if self.kor:
        dim = "₩"
    else:
        dim = "$"
    data = defaultdict(float)
    data["Cash"] = 0
    for name, value in self.items():
        _, purchase, _, quantity, _, _ = value
        data[name] = purchase * quantity
    cash = self() - sum(data.values())
    data["Cash"] = max(data["Cash"], cash)
    return pie(data, dim, title="Portfolio", dpi=100, int_label=self.kor)

def table(self) -> str:
    """현재 계좌의 상태를 image로 저장

    Returns:
        저장된 image의 절대 경로

    Examples:
        >>> balance.table()
    """
    if self() == 0:
        return None
    if self.kor:
        col = [
            "Purchase Price [₩]",
            "Current Price [₩]",
            "Quantity",
            "Profit and Loss (P&L) [%]",
            "Profit and Loss (P&L) [₩]",
        ]
    else:
        col = [
            "Purchase Price [$]",
            "Current Price [$]",
            "Quantity",
            "Profit and Loss (P&L) [%]",
            "Profit and Loss (P&L) [$]",
        ]
    row = []
    data = []
    purchase_total = 0
    current_total = 0
    for name, value in self.items():
        _, purchase, current, quantity, pandl_per, pandl_abs = value
        row.append(name)
        data.append(
            [
                _cash2str(purchase, self.kor),
                _cash2str(current, self.kor),
                quantity,
                f"{pandl_per:.2f}%",
                _cash2str(pandl_abs, self.kor),
            ]
        )
        purchase_total += purchase * quantity
        current_total += current * quantity
    row.append("TOTAL")
    if purchase_total == 0:
        pandl_total = 0
    else:
        pandl_total = (current_total - purchase_total) / purchase_total * 100
    data.append(
        [
            _cash2str(purchase_total, self.kor),
            _cash2str(current_total, self.kor),
            "-",
            f"{pandl_total:.2f}%",
            f"{_cash2str(current_total - purchase_total, self.kor)}\n\n{_cash2str(self(), self.kor)}",
        ]
    )
    return table(
        data,
        col,
        row,
        title="balance",
        figsize=(16, int(1.2 * len(row))),
        dpi=100,
    )

def _get_data(self, symbol: str) -> tuple[str, pd.DataFrame]:
    response = self.get_ohlcv(symbol, start_day=self.start_day, kor=self.kor)
    title, data = self.response2ohlcv(response)
    return title, data

def sell(self) -> None:
    """매도 signals 탐색

    한국투자증권의 잔고와 주식 보유 상황을 image로 변환하여 slack으로 전송 및 보유 중인 주식에 대해 매도 signals 탐색
    """
    path_balance, path_portfolio = self.table(), self.pie()
    if path_balance is None:
        self.bot.message("Balance: NULL", True)
        return None
    response = self.bot.message("> :bank: Balance")
    thread_id = self.bot.get_thread_id(response)
    self.bot.file(path_balance, thread_id=thread_id)
    self.bot.file(path_portfolio, thread_id=thread_id)
    self._inference(self.symbols_bought, "Sell")
    return None