모멘텀-수급 동조 와해 (momentumFlowSplit v2 변형)

공개 호출 방식

import dartlab
import polars as pl
import statistics

codes = ["005930", "000660", "035420", "051910", "207940"]

def pearson(a, b):
    if len(a) < 5 or len(a) != len(b):
        return None
    ma, mb = statistics.mean(a), statistics.mean(b)
    num = sum((x-ma)*(y-mb) for x, y in zip(a, b))
    da = sum((x-ma)**2 for x in a) ** 0.5
    db = sum((y-mb)**2 for y in b) ** 0.5
    return num / (da*db) if da*db else None

rows = []
for code in codes:
    try:
        px = dartlab.gather("price", code).head(80).to_dicts()
        fl = dartlab.gather("flow", code).head(80).to_dicts()
    except Exception:
        px, fl = [], []
    if not px or not fl:
        rows.append({"code": code, "corr60": None, "corr20": None, "decouple": None, "sampleN": 0})
        continue
    px.sort(key=lambda r: str(r.get("date") or r.get("tradeDate")))
    fl.sort(key=lambda r: str(r.get("date") or r.get("tradeDate")))
    flow_by_date = {str(r.get("date") or r.get("tradeDate"))[:10]: (
        float(r.get("foreignNet") or 0) + float(r.get("institutionNet") or 0)
    ) for r in fl}
    closes = []
    flows = []
    prev_close = None
    for p in px:
        d = str(p.get("date") or p.get("tradeDate"))[:10]
        close = float(p.get("close") or p.get("closePrice") or 0)
        if prev_close and prev_close > 0 and d in flow_by_date:
            closes.append((close / prev_close) - 1)
            flows.append(flow_by_date[d])
        prev_close = close
    if len(closes) < 60:
        rows.append({"code": code, "corr60": None, "corr20": None, "decouple": None, "sampleN": len(closes)})
        continue
    corr60 = pearson(closes[-60:-20], flows[-60:-20])
    corr20 = pearson(closes[-20:], flows[-20:])
    decouple = (corr60 - corr20) if (corr60 is not None and corr20 is not None) else None
    rows.append({
        "code": code,
        "corr60": corr60,
        "corr20": corr20,
        "decouple": decouple,
        "sampleN": len(closes),
    })

table = pl.DataFrame(rows)
decouples = [v for v in table["decouple"].to_list() if v is not None]
crossSectionStd = statistics.stdev(decouples) if len(decouples) >= 3 else None

emit_result(
    table=table,
    values={"universe": len(codes), "crossSectionStd": crossSectionStd},
    date=None,
    sources=["dartlab://gather/price", "dartlab://gather/flow"],
)

호출 동작

1. 결론 도출

가격-수급 상관 decouple 단정 (corr60 - corr20). 양수 = 상관 와해 (수급 이탈), 음수 = 상관 강화. 예: “corr60=+0.45 vs corr20=+0.12 → decouple=+0.33 (상관 와해) — 가격 상승 vs 수급 정체 신호.”

2. 핵심 근거 수집

• 종목 가격 시계열 60+20 거래일 (Company.gather(‘price’))
• 수급 시계열 (외인 + 기관 순매수, Company.gather(‘flow’))
• Pearson 상관 계산 — 이전 60d (최근 20d 제외) + 최근 20d 분리

3. 메커니즘 분석

가격 일변동률 + 수급 (외인+기관) 60+20 거래일
   ↓
이전 60일 (latest -80 ~ -20) Pearson 상관 = corr60
   ↓
최근 20일 Pearson 상관 = corr20
   ↓
decouple = corr60 - corr20
   > +0.2  → 상관 와해 (수급 vs 가격 이격 강함)
   ±0.2    → 상관 유지 (정상)
   < -0.2  → 상관 강화 (수급-가격 동행 강화)

decouple 양수 + 가격 상승 → 수급 따라가지 않는 가격 (skeptical buying). decouple 음수 + 가격 하락 → 수급 일관 매도 (capitulation 위험).

4. 반례·한계

• 거래량 적은 종목 (volume < average) 의 수급 데이터 노이즈 큼.
• 시장 전체 동시 충격 (지수 변동) 보정 없음.
• Pearson 상관은 선형 관계만 — 비선형 (jump · gap) 신호 포착 X.
• 60d/20d 짧으면 표본 부족, 길면 regime shift 묻혀.

5. 후속 모니터링

• decouple 양수 지속: recipes.sentiment.flowImbalance 로 외인/기관 imbalance 확인.
• decouple 음수 + 가격 하락: recipes.sentiment.retailFlowReversal 로 개인 vs 스마트머니 발산 확인.
• corr60 절대값 < 0.2 종목: 가격-수급 비동조 — 다른 신호 source (공시·뉴스) cross-check.

대표 반환 형태

column	의미
code	종목코드
corr60	이전 60 거래일 (최근 20일 제외) 가격-수급 Pearson
corr20	최근 20 거래일 가격-수급 Pearson
decouple	corr60 - corr20
sampleN	join 성공 표본 수

연계 절차

1. recipes.technical.momentumFlowDivergence - 직접 corr 만 (v1 차분 없이).
2. recipes.technical.momentumFlowSplit - v1 spread 와 비교.
3. recipes.sentiment.flowImbalance - 와해 시점 수급 imbalance 확인.

기본 검증

• sampleN < 60 인 종목은 corr60 결론 X.
• corr60 / corr20 둘 다 |값| < 0.1 이면 상관 자체가 없는 상황 — decouple 신호도 의미 없음.
• decouple 단일값 → 매수/매도 라벨 변환 금지.