공매도 잔고 변화율 모멘텀

공개 호출 방식

import dartlab
import polars as pl
import statistics

target = "005930"
c = dartlab.Company(target)

def rows(value, limit=120):
    if hasattr(value, "head") and hasattr(value, "to_dicts"):
        return value.head(limit).to_dicts()
    if isinstance(value, list):
        return value[:limit]
    return []

short_rows = rows(c.gather("shortBalance"), limit=120)

series = []
for r in short_rows:
    bal = r.get("shortBalance") or r.get("short_balance") or r.get("balanceShares")
    try:
        bal = float(bal) if bal is not None else None
    except (TypeError, ValueError):
        bal = None
    series.append({
        "date": r.get("date") or r.get("tradeDate"),
        "balance": bal,
    })

series.sort(key=lambda x: str(x["date"] or ""))
prev = None
for s in series:
    if prev is not None and s["balance"] is not None and prev > 0:
        s["changePct"] = (s["balance"] - prev) / prev
    else:
        s["changePct"] = None
    if s["balance"] is not None:
        prev = s["balance"]

WINDOW = 20
changes = [s["changePct"] for s in series if s["changePct"] is not None]
for idx, s in enumerate(series):
    if s["changePct"] is None:
        s["z"] = None
        continue
    earlier = [c for c in changes[:idx] if c is not None][-WINDOW:]
    if len(earlier) < 5:
        s["z"] = None
        continue
    mu = statistics.mean(earlier)
    sd = statistics.stdev(earlier) if len(earlier) > 1 else 0
    s["z"] = (s["changePct"] - mu) / sd if sd > 0 else None

table = pl.DataFrame(series) if series else pl.DataFrame(
    schema={"date": pl.Utf8, "balance": pl.Float64, "changePct": pl.Float64, "z": pl.Float64}
)

surge = int((table["z"].fill_null(0) >= 2).sum()) if table.height else 0
covering = int((table["z"].fill_null(0) <= -2).sum()) if table.height else 0

emit_result(
    table=table,
    values={"rows": table.height, "surgeCount": surge, "coveringCount": covering},
    date=str(table["date"].max()) if table.height else None,
    sources=["dartlab://gather/shortBalance"],
)

호출 동작

일별 공매도 잔고 시계열에서 직전일 대비 변화율 산출 후 20 거래일 rolling z-score. z ≥ 2 = 잔고 급증, z ≤ -2 = covering 단서. 잔고 absolute 크기 자체 는 결론 근거 아님.

대표 반환 형태

column	의미
date	거래일
balance	공매도 잔고 (주)
changePct	직전일 대비 변화율
z	20 거래일 변화율 z-score

연계 절차

1. recipes.sentiment.flowImbalance - 외국인/기관 수급과 같이 보면 방향 단서.
2. recipes.fundamental.quality.forensics.controllingPowerJudgment - 잔고 급증 시 의결권 분쟁 의심.

기본 검증

• 거래일 < 30 이면 z-score 결론 X.
• 공매도 일변동량 (volume) 과 잔고 (stock) 혼동 금지.
• short covering 신호를 “상승 시그널” 로 라벨링 금지 — covering 은 기존 약세 베팅 해제 의 정량 사실일 뿐.