vcp_debug.py

# -*- coding: utf-8 -*-
"""
VCP (Volatility Contraction Pattern) - 미너비니 기준
TICKER 하나만 바꾸면 됨!

돌파 조건 (모두 충족해야):
1. 종가가 수축구간 고가를 돌파
2. 당일 가격 상승률 3% 이상 (전일 종가 대비)
3. 당일 양봉 (종가 > 시가)
4. 거래량 20일MA × 2.0 이상 폭발
"""

import sys, ssl
import pandas as pd
import numpy as np
import FinanceDataReader as fdr
from datetime import datetime, timedelta, timezone
_KST = timezone(timedelta(hours=9))
def _now_kst(): return datetime.now(_KST)

ssl._create_default_https_context = ssl._create_unverified_context
if sys.platform == 'win32':
    sys.stdout.reconfigure(encoding='utf-8')

# =============================================
# [여기만 바꾸면 됨]
# =============================================
TICKER = '222080'

# =============================================
# 설정
# =============================================
VOL_MA_PERIOD      = 20
VOL_CONTRACT_THR   = 0.8    # 수축 기준 (20일MA × 0.8 이하)
VOL_EXPLODE_THR    = 2.0    # 폭발 기준 (20일MA × 2.0 이상)
MIN_PRICE_GAIN     = 0.03   # 돌파일 최소 상승률 (3%)
MIN_CONTRACT_DAYS  = 2      # 최소 수축 일수
EXPLODE_WINDOW     = 5      # 수축 후 N일 내 돌파 확인
DATA_START   = (_now_kst() - timedelta(days=400)).strftime('%Y-%m-%d')
ANALYZE_FROM = (_now_kst() - timedelta(days=365)).strftime('%Y-%m-%d')
ANALYZE_TO   = _now_kst().strftime('%Y-%m-%d')   # 오늘(KST) — 매 실행시 최신

# =============================================
# 데이터 로드
# =============================================
print("=" * 65)
print(f"VCP 분석: {TICKER} | 기간: {ANALYZE_FROM} ~ {ANALYZE_TO}")
print("=" * 65)

df = fdr.DataReader(TICKER, DATA_START, ANALYZE_TO)
if df is None or len(df) == 0:
    print("데이터 없음"); sys.exit()

df['vol_ma20']     = df['Volume'].rolling(VOL_MA_PERIOD).mean()
df['vol_contract'] = df['Volume'] < df['vol_ma20'] * VOL_CONTRACT_THR
df['price_range']  = df['High'] - df['Low']
df['range_ma10']   = df['price_range'].rolling(10).mean()
df['prev_close']   = df['Close'].shift(1)

print(f"데이터: {len(df)}일 로드 완료\n")

# =============================================
# VCP 감지
# =============================================
vcps = []
i = VOL_MA_PERIOD

while i < len(df) - 1:
    if not df['vol_contract'].iloc[i]:
        i += 1; continue

    # 수축 구간 찾기
    start = i
    while start > VOL_MA_PERIOD and df['vol_contract'].iloc[start - 1]:
        start -= 1
    end = i
    while end < len(df) - 1 and df['vol_contract'].iloc[end + 1]:
        end += 1

    days = end - start + 1
    if days < MIN_CONTRACT_DAYS:
        i = end + 1; continue

    zone       = df.iloc[start:end+1]
    zone_low   = float(zone['Low'].min())
    zone_high  = float(zone['High'].max())
    zone_vol   = float(zone['Volume'].mean())
    vol_ma     = float(df['vol_ma20'].iloc[end])
    vol_ratio  = round(zone_vol / vol_ma, 2) if vol_ma > 0 else 0

    # 가격 수축 확인
    pre_range_ma   = float(df['range_ma10'].iloc[max(0, start-1)])
    zone_range     = float(zone['price_range'].mean())
    is_contracting = (zone_range < pre_range_ma) if pre_range_ma > 0 else False

    # 돌파 확인 (4가지 조건 모두)
    breakout       = False
    breakout_date  = None
    breakout_price = None
    breakout_vol_r = None
    breakout_gain  = None
    fail_reason    = None

    for j in range(end + 1, min(end + EXPLODE_WINDOW + 1, len(df))):
        close_j    = float(df['Close'].iloc[j])
        open_j     = float(df['Open'].iloc[j])
        vol_j      = float(df['Volume'].iloc[j])
        vol_ma_j   = float(df['vol_ma20'].iloc[j])
        prev_close = float(df['prev_close'].iloc[j])

        if prev_close == 0 or pd.isna(prev_close):
            continue

        price_gain  = (close_j - prev_close) / prev_close  # 전일 대비 상승률
        is_bullish  = close_j > open_j                     # 양봉
        price_break = close_j > zone_high                  # 구간 고가 돌파
        vol_break   = vol_j > vol_ma_j * VOL_EXPLODE_THR  # 거래량 폭발
        gain_ok     = price_gain >= MIN_PRICE_GAIN         # 3% 이상 상승

        if price_break and vol_break and gain_ok and is_bullish:
            breakout       = True
            breakout_date  = str(df.index[j])[:10]
            breakout_price = close_j
            breakout_vol_r = round(vol_j / vol_ma_j, 2)
            breakout_gain  = round(price_gain * 100, 1)
            break
        elif price_break:
            # 가격은 돌파했지만 다른 조건 미충족 → 이유 기록
            reasons = []
            if not vol_break:
                reasons.append(f"거래량 부족({round(vol_j/vol_ma_j,1)}x, 필요 {VOL_EXPLODE_THR}x)")
            if not gain_ok:
                reasons.append(f"상승률 부족({price_gain*100:+.1f}%, 필요 {MIN_PRICE_GAIN*100:.0f}%+)")
            if not is_bullish:
                reasons.append("음봉")
            fail_reason = str(df.index[j])[:10] + " 가격돌파했지만: " + ", ".join(reasons)

    vcp = {
        'start_date':     str(df.index[start])[:10],
        'end_date':       str(df.index[end])[:10],
        'days':           days,
        'zone_low':       round(zone_low, 0),
        'zone_high':      round(zone_high, 0),
        'vol_ratio':      vol_ratio,
        'is_contracting': is_contracting,
        'zone_range':     round(zone_range, 0),
        'pre_range_ma':   round(pre_range_ma, 0),
        'confirmed':      breakout,
        'breakout_date':  breakout_date,
        'breakout_price': breakout_price,
        'breakout_vol_r': breakout_vol_r,
        'breakout_gain':  breakout_gain,
        'fail_reason':    fail_reason,
    }
    vcps.append(vcp)
    i = end + 1

# 분석 기간 필터
vcps = [v for v in vcps if v['end_date'] >= ANALYZE_FROM]

# =============================================
# 결과 출력
# =============================================
print(f"감지된 VCP: {len(vcps)}개\n")

if len(vcps) == 0:
    print("VCP 패턴 없음"); sys.exit()

for idx, v in enumerate(vcps, 1):
    status = "[확정]  " if v['confirmed'] else "[미확정]"
    print(f"{'='*65}")
    print(f"VCP #{idx} {status} {v['start_date']} ~ {v['end_date']} ({v['days']}일)")
    print(f"{'='*65}")
    print(f"  수축구간:  {v['zone_low']:,.0f} ~ {v['zone_high']:,.0f}원")
    print(f"  거래량:    20일MA 대비 {v['vol_ratio']*100:.0f}% (기준 {VOL_CONTRACT_THR*100:.0f}% 이하)")

    if v['pre_range_ma'] > 0:
        shrink = (1 - v['zone_range'] / v['pre_range_ma']) * 100
        mark = 'O' if v['is_contracting'] else 'X'
        print(f"  가격수축:  {mark} (구간 {v['zone_range']:,.0f} vs 직전MA {v['pre_range_ma']:,.0f}, {shrink:+.0f}%)")

    if v['confirmed']:
        print(f"  돌파:      {v['breakout_date']} | "
              f"종가 {v['breakout_price']:,.0f}원 | "
              f"상승률 +{v['breakout_gain']}% | "
              f"거래량 MA×{v['breakout_vol_r']} | 양봉")
    elif v['fail_reason']:
        print(f"  돌파시도:  {v['fail_reason']}")
    else:
        print(f"  돌파:      없음")

    print(f"\n  [판단 이유]")
    print(f"    - 거래량 {v['days']}일 연속 20일MA×{VOL_CONTRACT_THR} 이하 수축")
    print(f"    - 가격수축: {'확인' if v['is_contracting'] else '미확인'}")
    if v['confirmed']:
        print(f"    - 구간고가({v['zone_high']:,.0f}) 돌파 + 상승률 {v['breakout_gain']}% + 거래량 MA×{v['breakout_vol_r']} + 양봉 → 확정")
    else:
        print(f"    - 4가지 돌파조건 미충족 (가격돌파+3%상승+거래량2x+양봉)")
    print()

# 요약
confirmed = [v for v in vcps if v['confirmed']]
pending   = [v for v in vcps if not v['confirmed']]

print("=" * 65)
print("최종 요약")
print("=" * 65)
print(f"전체: {len(vcps)}개 | 확정: {len(confirmed)}개 | 미확정: {len(pending)}개")

if confirmed:
    print("\n[확정 VCP]")
    for v in confirmed:
        print(f"  {v['start_date']}~{v['end_date']} | "
              f"구간 {v['zone_low']:,.0f}~{v['zone_high']:,.0f} | "
              f"돌파 {v['breakout_date']} +{v['breakout_gain']}% MA×{v['breakout_vol_r']}")

if pending:
    print("\n[미확정 VCP]")
    for v in pending:
        print(f"  {v['start_date']}~{v['end_date']} | "
              f"구간 {v['zone_low']:,.0f}~{v['zone_high']:,.0f}")