RSI + 볼린저 밴드 동시 cluster (이중 oversold/overbought)

공개 호출 방식

import dartlab
import polars as pl
import statistics

target = "005930"

try:
    px = dartlab.gather("price", target).head(80).to_dicts()
except Exception:
    px = []
px.sort(key=lambda r: str(r.get("date") or r.get("tradeDate")))
closes = [(r.get("date") or r.get("tradeDate"), float(r.get("close") or 0)) for r in px if r.get("close")]

if len(closes) < 30:
    table = pl.DataFrame(schema={"date": pl.Utf8, "close": pl.Float64, "rsi": pl.Float64,
                                  "bbUpper": pl.Float64, "bbLower": pl.Float64, "signal": pl.Utf8})
else:
    # RSI(14)
    rsi_series = [None]*14
    gains, losses = [], []
    for i in range(1, len(closes)):
        diff = closes[i][1] - closes[i-1][1]
        gains.append(max(diff, 0))
        losses.append(max(-diff, 0))
        if i >= 14:
            avg_gain = statistics.mean(gains[-14:])
            avg_loss = statistics.mean(losses[-14:])
            rs = avg_gain / avg_loss if avg_loss else 0
            rsi = 100 - (100 / (1 + rs)) if avg_loss else 100
            rsi_series.append(rsi)
    # Bollinger(20, 2)
    bb_upper = [None]*20
    bb_lower = [None]*20
    for i in range(20, len(closes)):
        window = [c[1] for c in closes[i-20:i]]
        mu = statistics.mean(window)
        sd = statistics.stdev(window)
        bb_upper.append(mu + 2*sd)
        bb_lower.append(mu - 2*sd)
    rows = []
    for i, (d, c_val) in enumerate(closes):
        if i < 20 or rsi_series[i] is None: continue
        rsi = rsi_series[i]
        bu = bb_upper[i]
        bl = bb_lower[i]
        signal = "oversoldCluster" if (rsi < 30 and c_val < bl) else 
                 "overboughtCluster" if (rsi > 70 and c_val > bu) else "normal"
        if signal != "normal":
            rows.append({"date": d, "close": c_val, "rsi": rsi, "bbUpper": bu, "bbLower": bl, "signal": signal})
    table = pl.DataFrame(rows) if rows else pl.DataFrame(
        schema={"date": pl.Utf8, "close": pl.Float64, "rsi": pl.Float64,
                "bbUpper": pl.Float64, "bbLower": pl.Float64, "signal": pl.Utf8})

emit_result(
    table=table,
    values={"clusterCount": table.height},
    date=str(closes[-1][0]) if closes else None,
    sources=["dartlab://gather/price"],
)

호출 동작

1. 결론 도출

RSI + 볼린저 동시 cluster 단정 (oversoldCluster / overboughtCluster / 무신호). 예: “최근 60 거래일 oversold cluster 3 건 (RSI<30 + close<lower), overbought 0 건 — 이중 oversold 신호 활성.”

2. 핵심 근거 수집

• 가격 row 60+ 거래일 (Company.gather(‘price’) high/low/close)
• RSI(14) — 14일 평균 상승 / (상승+하락) × 100
• 볼린저 밴드(20, 2σ) — 20일 이동평균 ± 2 × 20일 표준편차

3. 메커니즘 분석

60거래일 close → RSI(14) 시계열 + 볼린저(20, 2σ) upper/lower band
   ↓
각 row 에서 두 조건 동시 점검
   RSI<30 AND close<lower → oversoldCluster (이중 oversold)
   RSI>70 AND close>upper → overboughtCluster (이중 overbought)
   둘 다 아님              → 무신호

단일 oscillator 함정 회피 — RSI 또는 볼린저 단독으로는 흔한 false positive. 두 정의 동시 충족만 신호로 사용. cluster row 수 ↑ 또는 연속 발생 = 신호 강도 ↑.

4. 반례·한계

• 강한 추세 (uptrend) 중에는 RSI > 70 + close > upper 가 정상 — overbought 신호 false positive.
• 변동성 낮은 종목 (zero band width) 의 볼린저 cluster 신뢰도 낮음.
• 60일 미만 history 시 cluster 측정 불가.
• 단일 cluster 만으로 매수/매도 단정 금지 — 다른 신호 (수급 · 가격 모멘텀) cross-check.

5. 후속 모니터링

• oversoldCluster 발생 직후: recipes.technical.priceVolumeZScore 로 거래량 동행 확인 (오버솔드 + 거래량 폭증 = 저점 매수 후보 강함).
• overboughtCluster 지속 (3 row+): recipes.technical.atrRegimeShift 로 변동성 확대 확인 (고점 신호).
• 신호 없는 종목 (cluster 0): 60d 추세 우세 — recipes.technical.movingAverageConfluence 로 추세 합의 확인.

대표 반환 형태

column	의미
date	거래일
close	종가
rsi	RSI(14)
bbUpper / bbLower	볼린저 밴드
signal	oversoldCluster / overboughtCluster

연계 절차

1. recipes.technical.momentumFlowDivergence - cluster 시점 수급 정합.
2. recipes.sentiment.flowImbalance - cluster 시점 imbalance.

기본 검증

• 거래일 < 30 이면 결론 X.
• 갭상승·하한가 등 단일 day noise 는 cluster 처리 X.
• 신호 단독 매매 결정 X — 펀더멘털 결합 후.