Bithumb/deepcoin/analysis/general_analysis_chart.py

"""
general_analysis 차트 유형 (캔들·선·바·Renko·P&F).
"""

from __future__ import annotations

import numpy as np
import pandas as pd

from deepcoin.analysis.general_analysis_core import ga_col


def _renko_direction_series(close: pd.Series, brick: pd.Series) -> pd.Series:
    """ATR 기반 브릭 크기로 Renko 방향 (+1/-1/0) 시계열."""
    n = len(close)
    direction = pd.Series(0, index=close.index, dtype=int)
    if n < 2:
        return direction

    price = float(close.iloc[0])
    for i in range(1, n):
        b = float(brick.iloc[i]) if not np.isnan(brick.iloc[i]) else float(close.diff().abs().median())
        if b < 1e-9:
            b = 1e-9
        c = float(close.iloc[i])
        if c >= price + b:
            steps = int((c - price) // b)
            direction.iloc[i] = 1
            price += steps * b
        elif c <= price - b:
            steps = int((price - c) // b)
            direction.iloc[i] = -1
            price -= steps * b
    return direction


def general_analysis_apply_chart_bars(df: pd.DataFrame) -> pd.DataFrame:
    """
    봉 단위 차트 파생 컬럼 (선 기울기, Renko, P&F).

    Args:
        df: OHLCV (+ ga_atr_14 권장).

    Returns:
        ga_chart_* 시계열 컬럼 추가.
    """
    out = df.copy()
    c = out["Close"].astype(float)
    h = out["High"].astype(float)
    l = out["Low"].astype(float)

    out[ga_col("chart_line_slope_1")] = c.diff()
    out[ga_col("chart_bar_range_pct")] = (h - l) / c.replace(0, np.nan) * 100

    from config import GA_ATR_PERIOD

    brick = (
        out[ga_col("atr_14")]
        if ga_col("atr_14") in out.columns
        else (h - l).rolling(GA_ATR_PERIOD).mean()
    )
    brick = brick.fillna(c.diff().abs().rolling(20).median()).replace(0, np.nan).bfill()
    renko_dir = _renko_direction_series(c, brick)
    out[ga_col("chart_renko_dir")] = renko_dir
    out[ga_col("chart_renko_up")] = (renko_dir == 1).astype(int)

    box = brick.fillna(1.0)
    pnf = pd.Series(0, index=out.index, dtype=int)
    col = 0
    for i in range(1, len(c)):
        b = float(box.iloc[i])
        move = c.iloc[i] - c.iloc[i - 1]
        if move >= b:
            col += 1
            pnf.iloc[i] = 1
        elif move <= -b:
            col -= 1
            pnf.iloc[i] = -1
    out[ga_col("chart_pnf_col")] = pnf

    if "Volume" in out.columns:
        v = out["Volume"].astype(float)
        out[ga_col("chart_vol_spike")] = (v > v.rolling(20).mean() * 1.8).astype(int)

    if ga_col("ha_trend_up") in out.columns:
        out[ga_col("chart_ha_trend")] = out[ga_col("ha_trend_up")]
    elif ga_col("ha_bull") in out.columns:
        out[ga_col("chart_ha_trend")] = out[ga_col("ha_bull")]

    return out


def general_analysis_chart_metrics(df: pd.DataFrame) -> dict[str, object]:
    """
    lookback 구간 차트 요약 (마지막 봉 스냅샷용).

    Returns:
        ga_chart_* dict.
    """
    res: dict[str, object] = {
        "chart_type_candle": 1,
        "chart_line_slope": 0.0,
        "chart_bar_range_pct": 0.0,
        "chart_ha_trend": 0,
        "chart_renko_brick_up_ratio": 0.5,
        "chart_renko_dir": 0,
        "chart_pnf_col": 0,
        "chart_vol_spike": 0,
    }
    if df is None or len(df) < 5:
        return {ga_col(k): v for k, v in res.items()}

    row = df.iloc[-1]
    c = df["Close"].astype(float)
    res["chart_line_slope"] = float((c.iloc[-1] - c.iloc[0]) / max(len(c) - 1, 1))
    res["chart_bar_range_pct"] = float(
        (df["High"].iloc[-1] - df["Low"].iloc[-1]) / max(c.iloc[-1], 1e-9) * 100
    )

    if ga_col("chart_renko_dir") in df.columns:
        rd = df[ga_col("chart_renko_dir")].astype(float)
        up = (rd == 1).sum()
        down = (rd == -1).sum()
        res["chart_renko_brick_up_ratio"] = round(up / max(up + down, 1), 3)
        res["chart_renko_dir"] = int(rd.iloc[-1])
    else:
        diff = c.diff().fillna(0)
        up = (diff > 0).sum()
        down = (diff < 0).sum()
        res["chart_renko_brick_up_ratio"] = round(up / max(up + down, 1), 3)

    if ga_col("chart_pnf_col") in df.columns:
        res["chart_pnf_col"] = int(df[ga_col("chart_pnf_col")].iloc[-1])
    if ga_col("chart_ha_trend") in df.columns:
        res["chart_ha_trend"] = int(row[ga_col("chart_ha_trend")])
    elif ga_col("ha_trend_up") in df.columns:
        res["chart_ha_trend"] = int(row[ga_col("ha_trend_up")])
    if ga_col("chart_vol_spike") in df.columns:
        res["chart_vol_spike"] = int(row[ga_col("chart_vol_spike")])
    elif "Volume" in df.columns:
        v = df["Volume"].astype(float)
        res["chart_vol_spike"] = int(v.iloc[-1] > v.iloc[-20:].mean() * 1.8)

    return {ga_col(k): v for k, v in res.items()}


def general_analysis_chart_columns() -> list[str]:
    return [
        "chart_type_candle",
        "chart_line_slope",
        "chart_line_slope_1",
        "chart_bar_range_pct",
        "chart_ha_trend",
        "chart_renko_brick_up_ratio",
        "chart_renko_dir",
        "chart_renko_up",
        "chart_pnf_col",
        "chart_vol_spike",
    ]