upto date!

vcp_debug.py

@@ -1,24 +1,22 @@
 # -*- coding: utf-8 -*-
 """
-VCP 돌파 감지 - 네 정의 버전 ('조용한 N일 -> 거래량+가격 폭발')
+VCP (Volatility Contraction Pattern) - 미너비니 기준
 TICKER 하나만 바꾸면 됨!
 
-[ 돌파 조건 (모두 충족) ]
-1. 폭발일 거래량 >= 직전 CONTRACT_DAYS 일 거래량 중 최대 × VOL_DRYUP_SURGE
-   (= 조용했다가 거래량이 확 터짐. '거래량 20일선'이 아니라 '직전 조용한 날 대비'라서
-      평소 거래량이 적은 종목/구간에서도 잡히고, 폭발 후 평균이 튀는 문제도 없음.)
-2. 폭발일 종가 > 직전 BASE_LOOKBACK일 고가  (가격 돌파)
-3. 폭발일 상승률(전일 종가 대비) >= MIN_PRICE_GAIN
-4. 폭발일 양봉 (종가 > 시가)
-
-★ 이 파일의 '감지 설정' 블록은 vcp_sell_debug.py 와 똑같이 유지할 것!
+돌파 조건 (모두 충족해야):
+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
+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':
@@ -27,137 +25,197 @@ if sys.platform == 'win32':
 # =============================================
 # [여기만 바꾸면 됨]
 # =============================================
-TICKER = '007340'
+TICKER = '222080'
 
 # =============================================
-# 감지 설정  (★ vcp_sell_debug.py 와 똑같이!)
+# 설정
 # =============================================
-CONTRACT_DAYS    = 2      # 폭발 직전 '조용한 날' 수
-# VOL_DRYUP_SURGE  = 6.0    # 폭발일 거래량 ÷ 직전 조용한 날 평균 (조용했다가 몇 배로 터졌나)
-VOL_DRYUP_SURGE  = 5.0    # 폭발일 거래량 ÷ 직전 조용한 날 평균 (조용했다가 몇 배로 터졌나)
-MIN_PRICE_GAIN   = 0.13   # 폭발일 상승률 (전일 종가 대비)
-BASE_LOOKBACK    = 10     # 폭발일 종가가 '직전 N일 고가'를 넘어야 (가격 돌파 기준)
-
-# DIAGNOSE_DATE    = '2025-12-16'   # 예: '2025-07-08' -> 그 날이 왜 잡혔나/안 잡혔나 직접 판정
-DIAGNOSE_DATE    = None
+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) — 매 실행시 최신
 
 # =============================================
-# 분석 기간 (더 과거까지 보려면 ANALYZE_DAYS만)
+# 데이터 로드
 # =============================================
-ANALYZE_DAYS = 365        # 며칠 전까지의 돌파를 볼지 (730=2년, 1095=3년)
-WARMUP_DAYS  = 40         # 고가 lookback 워밍업용 추가 로드 (보통 그대로)
-DATA_START   = (datetime.today() - timedelta(days=ANALYZE_DAYS + WARMUP_DAYS)).strftime('%Y-%m-%d')
-ANALYZE_FROM = (datetime.today() - timedelta(days=ANALYZE_DAYS)).strftime('%Y-%m-%d')
-ANALYZE_TO   = datetime.today().strftime('%Y-%m-%d')
+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()
 
-# =============================================
-# 돌파 감지 (★ vcp_sell_debug.py 와 동일 로직)
-# =============================================
-def find_vcp_breakouts(df):
-    O = df['Open'].values.astype(float);  H = df['High'].values.astype(float)
-    L = df['Low'].values.astype(float);   C = df['Close'].values.astype(float)
-    V = df['Volume'].values.astype(float)
-    out = []
-    for t in range(BASE_LOOKBACK, len(df)):
-        quiet_max = V[t - CONTRACT_DAYS:t].max()   # 직전 조용한 날들 중 '최대' 거래량
-        if quiet_max <= 0 or C[t - 1] <= 0:
-            continue
-        surge     = V[t] / quiet_max   # 직전 며칠 '각각'의 N배 (하나라도 크면 탈락 = 폭발 다음날 방지)
-        base_high = H[t - BASE_LOOKBACK:t].max()
-        gain      = C[t] / C[t - 1] - 1
-        if (surge >= VOL_DRYUP_SURGE and C[t] > base_high
-                and gain >= MIN_PRICE_GAIN and C[t] > O[t]):
-            out.append({
-                'i': t, 'date': str(df.index[t])[:10],
-                'close': C[t], 'open': O[t], 'low': L[t],
-                'gain': round(gain * 100, 1), 'surge': round(surge, 2),
-                'base_high': base_high, 'base_low': float(L[t - BASE_LOOKBACK:t].min()),
-            })
-    return [b for b in out if b['date'] >= ANALYZE_FROM]
+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 감지
 # =============================================
-def run_diagnose(df, date_str):
-    O = df['Open'].values.astype(float);  H = df['High'].values.astype(float)
-    C = df['Close'].values.astype(float); V = df['Volume'].values.astype(float)
-    idx = [str(d)[:10] for d in df.index]
-    pos = next((k for k, d in enumerate(idx) if d >= date_str), len(df) - 1)
-    lo  = max(BASE_LOOKBACK, pos - 12); hi = min(len(df), pos + 4)
-    bh_label = f"{BASE_LOOKBACK}일고가"
-
-    print("\n" + "=" * 92)
-    print(f"[진단] {date_str} 주변 일별  (배수 = 그날 거래량 / 직전 {CONTRACT_DAYS}일 '최대' 거래량)")
-    print("=" * 92)
-    print(f"{'날짜':<11}{'종가':>8}{'시가':>8}{'거래량':>12}{'직전최대':>12}{'배수':>7}"
-          f"{'대비%':>7}{'양봉':>5}{bh_label:>9}{'고가돌파':>7}")
-    print("-" * 92)
-    for k in range(lo, hi):
-        qa  = V[k - CONTRACT_DAYS:k].max()
-        sg  = (V[k] / qa) if qa > 0 else 0
-        bh  = H[k - BASE_LOOKBACK:k].max()
-        gn  = (C[k] / C[k - 1] - 1) * 100 if C[k - 1] > 0 else 0
-        bull = 'O' if C[k] > O[k] else ''
-        pbk  = 'O' if C[k] > bh else ''
-        mark = '  <==' if idx[k] == date_str else ''
-        print(f"{idx[k]:<11}{C[k]:>8,.0f}{O[k]:>8,.0f}{V[k]:>12,.0f}{qa:>12,.0f}{sg:>6.1f}x"
-              f"{gn:>+6.1f}%{bull:>5}{bh:>9,.0f}{pbk:>7}{mark}")
-
-    # 판정
-    print("\n  [판정]")
-    target = idx[pos]
-    if target != date_str:
-        print(f"  (참고: {date_str}은 거래일 아님 -> 가장 가까운 {target} 기준)")
-    qa = V[pos - CONTRACT_DAYS:pos].max()
-    sg = (V[pos] / qa) if qa > 0 else 0
-    bh = H[pos - BASE_LOOKBACK:pos].max()
-    gn = (C[pos] / C[pos - 1] - 1) * 100 if C[pos - 1] > 0 else 0
-    c1 = sg >= VOL_DRYUP_SURGE; c2 = C[pos] > bh
-    c3 = gn >= MIN_PRICE_GAIN * 100; c4 = C[pos] > O[pos]
-    print(f"    거래량 배수 {sg:.1f}x >= {VOL_DRYUP_SURGE}?        {'O' if c1 else 'X'}")
-    print(f"    종가 {C[pos]:,.0f} > {BASE_LOOKBACK}일고가 {bh:,.0f}?   {'O' if c2 else 'X'}")
-    print(f"    상승률 {gn:+.1f}% >= {MIN_PRICE_GAIN*100:.0f}%?         {'O' if c3 else 'X'}")
-    print(f"    양봉 (종가 {C[pos]:,.0f} > 시가 {O[pos]:,.0f})?    {'O' if c4 else 'X'}")
-    if c1 and c2 and c3 and c4:
-        print(f"  => 4조건 통과 -> 잡혀야 함.")
-    else:
-        fails = [n for n, ok in [(f'거래량 배수 부족({sg:.1f}x)', c1), ('가격이 고가 못 넘음', c2),
-                                 (f'상승률 부족({gn:+.1f}%)', c3), ('음봉', c4)] if not ok]
-        print(f"  => 안 잡힘. 이유: {', '.join(fails)}")
-        if not c1:
-            print(f"     (VOL_DRYUP_SURGE를 {sg:.1f} 밑으로 낮추면 잡혀)")
+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("=" * 70)
-print(f"VCP 돌파 감지: {TICKER} | 기간: {ANALYZE_FROM} ~ {ANALYZE_TO}")
-print(f"조건: 직전{CONTRACT_DAYS}일 대비 거래량 {VOL_DRYUP_SURGE}배+ & {BASE_LOOKBACK}일고가 돌파 "
-      f"& 상승 +{MIN_PRICE_GAIN*100:.0f}%+ & 양봉")
-print("=" * 70)
-
-df = fdr.DataReader(TICKER, DATA_START, ANALYZE_TO)
-if df is None or len(df) == 0:
-    print("데이터 없음"); sys.exit()
-print(f"데이터: {len(df)}일 로드 완료\n")
-
-bos = find_vcp_breakouts(df)
-print(f"감지된 VCP 돌파: {len(bos)}개\n")
+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"  돌파:      없음")
 
-for n, b in enumerate(bos, 1):
-    print("=" * 70)
-    print(f"VCP #{n}  돌파 {b['date']} @ {b['close']:,.0f}원")
-    print(f"  상승률 +{b['gain']}% | 거래량 {b['surge']}배(직전{CONTRACT_DAYS}일의 최대 기준) | "
-          f"{BASE_LOOKBACK}일고가({b['base_high']:,.0f}) 돌파 | 양봉")
+    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()
 
-print("=" * 70)
-print(f"요약: 총 {len(bos)}개 돌파")
-for b in bos:
-    print(f"  {b['date']} | {b['close']:,.0f}원 | +{b['gain']}% | 거래량 {b['surge']}배")
-
-if DIAGNOSE_DATE:
-    run_diagnose(df, DIAGNOSE_DATE)
+# 요약
+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}")