내부자 매수/매도 cluster + 가격 lag

공개 호출 방식

import dartlab
import polars as pl
from datetime import datetime, timedelta

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

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

def parseDate(v):
    if isinstance(v, datetime):
        return v.date()
    if v is None:
        return None
    s = str(v)[:10].replace(".", "-").replace("/", "-")
    try:
        return datetime.strptime(s, "%Y-%m-%d").date()
    except Exception:
        return None

insider_rows = rows(c.gather("insider"), limit=200)
price_rows = rows(c.gather("price"), limit=200)

# date → price 시계열
price_by_date = {}
for p in price_rows:
    d = parseDate(p.get("date") or p.get("tradeDate"))
    if d:
        price_by_date[d] = float(p.get("close") or p.get("closePrice") or 0)

# insider events sorted
events = []
for r in insider_rows:
    d = parseDate(r.get("date") or r.get("tradeDate") or r.get("filedAt"))
    if not d:
        continue
    # insider gather 스키마: tradeType(buy/sell 또는 코드), changeShares(부호로 buy/sell 추론 fallback)
    tt = str(r.get("tradeType") or r.get("direction") or "").lower()
    if tt.startswith("b") or tt == "1" or tt == "buy":
        direction = "buy"
    elif tt.startswith("s") or tt == "0" or tt == "sell":
        direction = "sell"
    else:
        try:
            direction = "buy" if float(r.get("changeShares") or 0) > 0 else "sell"
        except Exception:
            direction = "sell"
    person = r.get("name") or r.get("person") or r.get("filer") or "?"
    events.append({"date": d, "direction": direction, "person": person})

events.sort(key=lambda x: x["date"])

WINDOW = timedelta(days=180)
clusters = []
for i, e in enumerate(events):
    window_start = e["date"] - WINDOW
    same_dir = [x for x in events[:i+1] if x["direction"] == e["direction"] and x["date"] >= window_start]
    persons = {x["person"] for x in same_dir}
    if len(persons) >= 3:
        # 직전 30 거래일 가격 변동
        before = [p for d, p in price_by_date.items() if (e["date"] - timedelta(days=45)) <= d < e["date"]]
        price_chg = (before[-1] / before[0] - 1) if len(before) >= 2 and before[0] > 0 else None
        clusters.append({
            "date": str(e["date"]),
            "direction": e["direction"],
            "personsInWindow": len(persons),
            "pricePctBefore30d": price_chg,
            "latestPerson": e["person"],
        })

_cluster_schema = {
    "date": pl.Utf8,
    "direction": pl.Utf8,
    "personsInWindow": pl.Int64,
    "pricePctBefore30d": pl.Float64,
    "latestPerson": pl.Utf8,
}
table = pl.DataFrame(clusters, schema=_cluster_schema, infer_schema_length=None) if clusters else pl.DataFrame(schema=_cluster_schema)

buy_n = int((table["direction"] == "buy").sum()) if table.height else 0
sell_n = int((table["direction"] == "sell").sum()) if table.height else 0

if table.height == 0:
    # cluster 미감지 — 가장 최근 insider event 날짜를 placeholder 로 emit (date 보장).
    if events:
        latest_event_date = str(events[-1]["date"])
        table = [{"direction": "no_cluster", "date": latest_event_date, "insiderEventsScanned": len(events)}]
    else:
        latest_event_date = None
        table = [{"direction": "no_insider_data", "date": None}]
else:
    latest_event_date = str(table["date"].max())

emit_result(
    table=table,
    values={
        "clusters": (table.height if hasattr(table, "height") else len(table)),
        "buyClusters": buy_n,
        "sellClusters": sell_n,
    },
    date=latest_event_date,
    sources=["dartlab://gather/insider", "dartlab://gather/price"],
)

호출 동작

1. 결론 도출

내부자 cluster 형성 여부 + 방향 (buy/sell) + 직전 30 거래일 가격 lag 단정. 예: “최근 180 일 buy cluster N 건, sell cluster M 건. 직전 cluster 시점 30 거래일 가격 +X% — 가격 vs 내부자 방향 정합 여부 판정.”

2. 핵심 근거 수집

• 내부자 거래 row 200 건 (Company.gather(‘insider’))
• 가격 row 200 건 (Company.gather(‘price’))
• 거래 방향 추론: tradeType 직접 또는 changeShares 부호 fallback

3. 메커니즘 분석

insider events 시간순 정렬 → 각 시점 t 기준 직전 180일 window 안 같은 방향 거래자 N 명
→ N ≥ 3 → cluster 형성 (date=t, direction=buy|sell)
→ 동시 직전 30 거래일 누적 가격 변동 (pricePctBefore30d) 계산
→ buy cluster + 가격 하락 = 저점 매수 신호 / sell cluster + 가격 상승 = 고점 매도 신호

cluster 안 personsInWindow 가 크고 같은 방향이 일관 → 집단 timing 신뢰도 ↑.

4. 반례·한계

• 자사주 취득·임원 stock option 행사·상속·분할 등 의무 거래 가 cluster 에 섞이면 sentiment 신호 X.
• 단일 거래자가 분할 매매 N 회 → 같은 사람이지만 cluster 로 잘못 분류 (현재 person 명 dedup 없음).
• 가격 lag 만으로 인과 단정 금지 — 시장 전체 동시 변동 보정 없음.
• 180 day window 너무 길면 무관 이벤트 결합, 짧으면 cluster 미감지.

5. 후속 모니터링

• 직전 cluster 후 30일 가격 vs 60일 가격 차로 cluster 적중률 추적 (recipes.sentiment.foreignBuyMomentum 와 cross-check).
• buy cluster + sell cluster 동시 발현 시 recipes.sentiment.flowImbalance 로 수급 imbalance 확인.
• cluster 시점 ±5일 안 공시 동행 여부 recipes.fundamental.disclosure.eventRadar.eventInbox.

대표 반환 형태

column	의미
date	cluster 형성 시점 (최신 거래일)
direction	buy / sell
personsInWindow	180 일 안 같은 방향 거래자 수
pricePctBefore30d	직전 30 거래일 누적 가격 변동
latestPerson	마지막 거래자 이름

연계 절차

1. recipes.sentiment.flowImbalance - cluster 시점에 외인/기관 수급도 같이 본다.
2. recipes.fundamental.disclosure.insiderEarningsLeading - cluster 가 분기 EPS surprise 와 양의 IC 인지 (별 절차).

기본 검증

• ≥3 명 기준은 본 recipe 의 정의 — 단일 거래는 cluster 아님.
• pricePctBefore30d 가 cluster 형성 원인 이라는 인과 단정 금지.
• 180 day window 가 너무 길면 무관 사건 묶일 수 있음 — 답변에 window 값 명시.