QQuery / bollinger_count_debug.py
glorifiedjx's picture
upto date!
124168e verified
Raw
History Blame Contribute Delete
4.68 kB
"""
์‹ค๋ฆฌ์ฝ˜ํˆฌ ๋ณผ๋ฆฐ์ €๋ฐด๋“œ ์นด์šดํŒ… ๋””๋ฒ„๊น… v5
ํ™•์ • ๊ทœ์น™:
1. ์‹œ์ดˆ๊ฐ€ > 240์ผ์„  โ† ์‹ ๊ทœ ์ถ”๊ฐ€
2. ์ข…๊ฐ€ > BB ์ƒ๋‹จ ร— 1.03
3. 60์ผ์„  ์•„๋ž˜ ๋‹ค๋…€์˜จ ํ›„: ์‹œ์ดˆ๊ฐ€ > ์ด์ „ ํ”ผํฌ ์ข…๊ฐ€ (์ถ”๊ฐ€ ์กฐ๊ฑด)
4. 240์ผ์„  ์•„๋ž˜ โ†’ ์™„์ „ ๋ฆฌ์…‹
5. 60์ผ์„  ์•„๋ž˜ โ†’ ์นด์šดํŠธ ์œ ์ง€
"""
import pandas as pd
import numpy as np
import FinanceDataReader as fdr
import warnings
from datetime import datetime, timezone, timedelta
warnings.filterwarnings('ignore')
_KST = timezone(timedelta(hours=9))
TICKER = '420770'
DEBUG_FROM = '2025-05-02'
DEBUG_TO = datetime.now(_KST).strftime('%Y-%m-%d') # ์˜ค๋Š˜(KST) โ€” ๋งค ์‹คํ–‰์‹œ ์ตœ์‹ (์žฅ์ค‘์ด๋ฉด ํ˜„์žฌ๊ฐ€)
BB_WINDOW = 20
BB_STD = 2
MA_240 = 240
MA_60 = 60
BB_THRESHOLD = 1.030
print(f"({TICKER}) ๋ฐ์ดํ„ฐ ๋กœ๋“œ ์ค‘...")
df = fdr.DataReader(TICKER, '2022-01-01', DEBUG_TO)
print(f"๋ฐ์ดํ„ฐ: {len(df)}์ผ์น˜\n")
df['ma20'] = df['Close'].rolling(BB_WINDOW).mean()
df['std20'] = df['Close'].rolling(BB_WINDOW).std()
df['bb_upper'] = df['ma20'] + BB_STD * df['std20']
df['bb_lower'] = df['ma20'] - BB_STD * df['std20']
df['ma60'] = df['Close'].rolling(MA_60).mean()
df['ma240'] = df['Close'].rolling(MA_240).mean()
count = 0
peak_close = 0
was_below_60 = False
counts = []
peak_list = []
flags = []
for i in range(len(df)):
c = df['Close'].iloc[i]
o = df['Open'].iloc[i]
u = df['bb_upper'].iloc[i]
m60 = df['ma60'].iloc[i]
m240 = df['ma240'].iloc[i]
if pd.isna(m240) or pd.isna(u) or pd.isna(m60):
counts.append(count)
peak_list.append(peak_close)
flags.append('๋ฐ์ดํ„ฐ๋ถ€์กฑ')
continue
if c < m240:
count = 0
peak_close = 0
was_below_60 = False
flag = '240์„ ์ดํƒˆโ†’๋ฆฌ์…‹'
elif c < m60:
was_below_60 = True
flag = f'60์„ ์ดํƒˆโ†’์œ ์ง€(count={count})'
else:
# ํ•ต์‹ฌ ์กฐ๊ฑด๋“ค
open_above_240 = (o > m240) # โ† ์‹ ๊ทœ: ์‹œ์ดˆ๊ฐ€ > 240์ผ์„ 
clearly_above = (c > u * BB_THRESHOLD) # ์ข…๊ฐ€ > BB์ƒ๋‹จ ร— 1.020
if not open_above_240:
flag = f'์‹œ์ดˆ๊ฐ€({o:,.0f})<240์„ ({m240:,.0f})โ†’๋ฌดํšจ'
elif clearly_above:
if was_below_60:
if o > peak_close:
count += 1
peak_close = c
was_below_60 = False
flag = f'+{count} (60์„ ๋ณต๊ท€,์‹œ์ดˆ๊ฐ€>{peak_close:,.0f})'
else:
flag = f'BB๋ŒํŒŒbut์‹œ์ดˆ๊ฐ€({o:,.0f})โ‰คํ”ผํฌ({peak_close:,.0f})โ†’์Šคํ‚ต'
else:
count += 1
peak_close = max(peak_close, c)
flag = f'+{count}'
else:
pct = (c / u - 1) * 100
flag = f'์ข…๊ฐ€๋ฏธ๋‹ฌ(BB๋Œ€๋น„{pct:+.1f}%)'
counts.append(count)
peak_list.append(peak_close)
flags.append(flag)
df['bb_count'] = counts
df['peak_close'] = peak_list
df['flag'] = flags
mask = (df.index >= DEBUG_FROM) & (df.index <= DEBUG_TO)
debug_df = df[mask].copy()
print("=" * 115)
print(f"๋‚ ์งœ๋ณ„ ์นด์šดํŒ… ({DEBUG_FROM} ~ {DEBUG_TO})")
print("=" * 115)
print(f"{'๋‚ ์งœ':<12} {'์ข…๊ฐ€':>7} {'์‹œ๊ฐ€':>7} {'BB์ƒ๋‹จ':>8} {'BBร—1.03':>9} "
f"{'60MA':>7} {'240MA':>7} {'ํ”ผํฌ':>8} {'์นด์šดํŠธ':>6} ํŒ์ •")
print("-" * 115)
prev_count = 0
for date, row in debug_df.iterrows():
cnt = int(row['bb_count'])
flag = row['flag']
is_notable = (cnt != prev_count or
'๋ฆฌ์…‹' in flag or '์ดํƒˆ' in flag or
'๋ณต๊ท€' in flag or '์Šคํ‚ต' in flag or
'๋ฌดํšจ' in flag)
if is_notable:
mark = ' โ—€' if cnt > prev_count else ''
print(f"{str(date)[:10]:<12} "
f"{row['Close']:>7,.0f} "
f"{row['Open']:>7,.0f} "
f"{row['bb_upper']:>8,.0f} "
f"{row['bb_upper']*BB_THRESHOLD:>9,.0f} "
f"{row['ma60']:>7,.0f} "
f"{row['ma240']:>7,.0f} "
f"{row['peak_close']:>8,.0f} "
f"{cnt:>6} {flag}{mark}")
prev_count = cnt
print("\n" + "=" * 60)
print("์นด์šดํŠธ ์ฆ๊ฐ€ํ•œ ๋‚ ๋“ค ์š”์•ฝ")
print("=" * 60)
prev = 0
for date, row in debug_df.iterrows():
cnt = int(row['bb_count'])
if cnt > prev:
print(f" {str(date)[:10]}: {row['flag']}"
f" (์ข…๊ฐ€:{row['Close']:,.0f}, ์‹œ๊ฐ€:{row['Open']:,.0f}, "
f"BBร—1.03:{row['bb_upper']*BB_THRESHOLD:,.0f}, "
f"240MA:{row['ma240']:,.0f})")
prev = cnt
print(f"\n์ตœ๋Œ€ ์นด์šดํŠธ: {debug_df['bb_count'].max()}")