zerohertzLib.quant ¶

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 _experiments(
    self,
    method: Callable[[Any], pd.DataFrame],
    exps: list[list[Any]],
) -> dict[str, list[Any]]:
    return experiments(
        self.title,
        self.data,
        method,
        exps,
        ohlc=self.ohlc,
        vis=self.vis,
        report=self.report,
    )

def bollinger_bands(
    self, exps: list[list[Any]] | None = None
) -> dict[str, list[Any]]:
    """Bollinger bands 전략 실험

    Args:
        exps: 전략 function에 입력될 변수들의 범위

    Returns:
        손실 거래 비율에 따른 수익률, `signals`, parameters
    """
    if exps is None:
        exps = self.exps_bollinger_bands
    return self._experiments(bollinger_bands, exps)

def macd(self, exps: list[list[Any]] | None = None) -> dict[str, list[Any]]:
    """MACD 전략 실험

    Args:
        exps: 전략 function에 입력될 변수들의 범위

    Returns:
        손실 거래 비율에 따른 수익률, `signals`, parameters
    """
    if exps is None:
        exps = self.exps_macd
    return self._experiments(macd, exps)

def momentum(self, exps: list[list[Any]] | None = None) -> dict[str, list[Any]]:
    """Momentum 전략 실험

    Args:
        exps: 전략 function에 입력될 변수들의 범위

    Returns:
        손실 거래 비율에 따른 수익률, `signals`, parameters
    """
    if exps is None:
        exps = self.exps_momentum
    return self._experiments(momentum, exps)

def moving_average(
    self, exps: list[list[Any]] | None = None
) -> dict[str, list[Any]]:
    """Moving average 전략 실험

    Args:
        exps: 전략 function에 입력될 변수들의 범위

    Returns:
        손실 거래 비율에 따른 수익률, `signals`, parameters
    """
    if exps is None:
        exps = self.exps_moving_average
    return self._experiments(moving_average, exps)

def rsi(self, exps: list[list[Any]] | None = None) -> dict[str, list[Any]]:
    """RSI 전략 실험

    Args:
        exps: 전략 function에 입력될 변수들의 범위

    Returns:
        손실 거래 비율에 따른 수익률, `signals`, parameters
    """
    if exps is None:
        exps = self.exps_rsi
    return self._experiments(rsi, exps)

def __call__(self, day: str | int = -1) -> dict[str, Any]:
    """
    입력된 날짜에 대해 분석 정보 return

    Args:
        day: 분석할 날짜

    Returns:
        각 전략에 따른 분석 정보 및 결론
    """
    if self.total_cnt < 1:
        return {"position": "NULL"}
    if day != -1 and "-" not in day:
        day = day[:4] + "-" + day[4:6] + "-" + day[6:8]
    possibility = defaultdict(list)
    for key in self.methods:
        possibility[key] = [
            self.signals[key].iloc[day],
            self.signals[key].iloc[day] / self.methods_cnt[key] * 100,
        ]
    possibility["logic"] = self.signals["logic"].iloc[day]
    possibility["total"] = [
        self.signals["signals"].iloc[day],
        self.signals["signals"].iloc[day] / self.total_cnt * 100,
    ]
    if 0 < possibility["logic"]:
        possibility["position"] = "Buy"
    elif 0 > possibility["logic"]:
        possibility["position"] = "Sell"
    elif self.threshold_buy <= possibility["total"][0]:
        possibility["position"] = "Buy"
        possibility["logic"] = 1
    elif self.threshold_sell >= possibility["total"][0]:
        possibility["position"] = "Sell"
        possibility["logic"] = -1
    else:
        possibility["position"] = "None"
    return possibility