외국인·기관·개인 순매수 imbalance z-score

공개 호출 방식

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 []

flow_rows = rows(c.gather("flow"), limit=120)

series = []
for r in flow_rows:
    f = float(r.get("foreignNet") or r.get("foreign_net") or 0)
    i = float(r.get("institutionNet") or r.get("inst_net") or 0)
    p = float(r.get("individualNet") or r.get("indiv_net") or 0)
    series.append({
        "date": r.get("date") or r.get("tradeDate"),
        "foreign": f,
        "inst": i,
        "indiv": p,
        "imbalance": (f + i) - p,
    })

series.sort(key=lambda x: str(x["date"] or ""))
imbal = [s["imbalance"] for s in series]
WINDOW = 20
for idx, s in enumerate(series):
    if idx < WINDOW:
        s["z"] = None
        continue
    window = imbal[idx-WINDOW:idx]
    mu = statistics.mean(window)
    sd = statistics.stdev(window) if len(window) > 1 else 0
    s["z"] = (s["imbalance"] - mu) / sd if sd > 0 else None

table = pl.DataFrame(series) if series else pl.DataFrame(
    schema={"date": pl.Utf8, "foreign": pl.Float64, "inst": pl.Float64,
            "indiv": pl.Float64, "imbalance": pl.Float64, "z": pl.Float64}
)

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

emit_result(
    table=table,
    values={"rows": table.height, "extremePos": extreme_pos, "extremeNeg": extreme_neg},
    date=str(table["date"].max()) if table.height else None,
    sources=["dartlab://gather/flow"],
)

호출 동작

1. 결론 도출

외인+기관 vs 개인 imbalance z-score cluster 단정. 예: “20거래일 imbalance row 중 z ≥ 2 인 day 5건 (모두 양수 — 외인+기관 매수 우세), z ≤ -2 인 day 1건 → 스마트머니 매수 cluster 강함.”

2. 핵심 근거 수집

• 일별 외인·기관·개인 순매수 row (Company.gather(‘flow’))
• imbalance = (foreignNet + institutionNet) - retailNet
• 직전 20거래일 rolling mean + stdev → z-score

3. 메커니즘 분석

일별 imbalance 시계열 60+거래일
   ↓
20거래일 rolling mean/stdev → z = (imb[t] - rolling_mean) / rolling_std
   ↓
z ≥ +2  → 스마트머니 매수 cluster 후보 (외인+기관 ↑, 개인 ↓ 또는 매수 동기 약)
z ≤ -2  → 스마트머니 매도 cluster 후보 (외인+기관 ↓, 개인 ↑)
±2 안   → 정상 imbalance

z ≥ 2 row 다수 발생 + 가격 상승 = 스마트머니 매수 신호 강함. z ≤ -2 + 가격 상승 = 개인 매수 추격 (역사적 고점 후보).

4. 반례·한계

• 20거래일 rolling baseline 짧음 — regime shift 시 z-score 후행.
• 외인 vs 기관 분리 안 함 — 외인 매도 + 기관 매수 동시면 imbalance 작아 보임.
• single-day spike (대규모 block deal) 영향 큼.
• 분기 조정 (MSCI 리밸런싱 · 배당락) 시점 noise.

5. 후속 모니터링

• z ≥ 2 cluster 다수: recipes.sentiment.foreignBuyMomentum 으로 외인 가속도 cross-check.
• z ≤ -2 + 가격 상승: recipes.sentiment.retailFlowReversal 로 개인 매수 추격 신호 확인.
• z 부호 빈번 전환 (±2 사이 oscillation): recipes.industry.sectorFlowConcentration 으로 자금 집중도 확인.

대표 반환 형태

column	의미
date	거래일
foreign	외국인 순매수
inst	기관 순매수
indiv	개인 순매수
imbalance	(외인+기관) − 개인
z	20 거래일 rolling z-score

연계 절차

1. recipes.sentiment.shortBalanceMomentum - 공매도 잔고와 같이 보면 매도 측 sentiment 확인.
2. recipes.sentiment.insiderClusterTiming - imbalance 와 내부자 cluster 시점 비교.

기본 검증

• 거래일 < 30 이면 z-score 결론 X, coverage 한계 만 명시.
• z-score 단일값 → sentiment 라벨 변환 금지.
• MSCI 리밸런싱·IPO 록업 해제 등 정기 이벤트 와 겹친 row 는 한계로 분리.