다른 명령
- ✅ Telegram 알림 기능이 추가된 최종 버전**입니다.
- 1. 먼저 Telegram Bot 설정 방법 (필수)
1. Telegram에서 **@BotFather** 검색 → `/newbot` 명령어로 봇 생성 2. Bot Token 복사 (예: `123456789:AAH...`) 3. 봇을 본인 채팅방에 추가 후, 아래 링크로 Chat ID 확인:
``` https://api.telegram.org/bot[YOUR_BOT_TOKEN]/getUpdates ```
4. Chat ID 복사 (보통 `-`로 시작하는 숫자)
- 2. 완전한 코드 (pandas_ta + Telegram 알림)
```python import FinanceDataReader as fdr import pandas as pd import pandas_ta as ta from datetime import datetime, timedelta import requests import warnings warnings.filterwarnings('ignore')
- ==================== 설정 ====================
- ==================== Telegram 설정 ====================
TELEGRAM_BOT_TOKEN = "YOUR_BOT_TOKEN_HERE" # ← 여기에 Bot Token 입력 TELEGRAM_CHAT_ID = "YOUR_CHAT_ID_HERE" # ← 여기에 Chat ID 입력
RSI_THRESHOLD = 30 BB_PERIOD = 20 BB_STD = 2.0 MA_SHORT = 20 MA_LONG = 60
MONTH1_THRESHOLD = -15 # % MONTH3_THRESHOLD = -20 # %
- ==================== Telegram 메시지 전송 함수 ====================
def send_telegram_message(message):
if not TELEGRAM_BOT_TOKEN or TELEGRAM_BOT_TOKEN == "YOUR_BOT_TOKEN_HERE":
print("⚠️ Telegram 설정이 완료되지 않았습니다.")
return False
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
"chat_id": TELEGRAM_CHAT_ID,
"text": message,
"parse_mode": "Markdown"
}
try:
response = requests.post(url, json=payload, timeout=10)
if response.status_code == 200:
print("📤 Telegram 알림 전송 완료")
return True
else:
print(f"❌ Telegram 전송 실패: {response.text}")
return False
except Exception as e:
print(f"❌ Telegram 전송 오류: {e}")
return False
- ==================== Top 100 추출 ====================
def get_kospi_top100():
kospi = fdr.StockListing('KOSPI')
if 'Marcap' in kospi.columns:
top100 = kospi.nlargest(100, 'Marcap').reset_index(drop=True)
else:
top100 = kospi.head(100).reset_index(drop=True)
print(f"✅ KOSPI 시가총액 Top 100 로드 완료: {len(top100)}개")
return top100'Symbol', 'Name'
- ==================== 지표 계산 ====================
def calculate_indicators(df):
if len(df) < 100:
return None
close = df['Close']
high = df['High']
low = df['Low']
df = df.copy()
df['RSI'] = ta.rsi(close, length=14)
bb = ta.bbands(close, length=BB_PERIOD, std=BB_STD)
df = pd.concat([df, bb], axis=1)
df['MA20'] = ta.sma(close, length=MA_SHORT)
df['MA60'] = ta.sma(close, length=MA_LONG)
latest = df.iloc[-1]
return_1m = (latest['Close'] / df['Close'].iloc[-22] - 1) * 100 if len(df) >= 22 else None
return_3m = (latest['Close'] / df['Close'].iloc[-66] - 1) * 100 if len(df) >= 66 else None
return {
'RSI': latest['RSI'],
'BB_Lower': latest[f'BBL_{BB_PERIOD}_{BB_STD:.1f}'],
'Close': latest['Close'],
'MA20': latest['MA20'],
'MA60': latest['MA60'],
'Return_1M': return_1m,
'Return_3M': return_3m
}
- ==================== 메인 함수 ====================
def monitor_stocks():
top100 = get_kospi_top100()
results = []
end_date = datetime.today().strftime('%Y-%m-%d')
start_date = (datetime.today() - timedelta(days=500)).strftime('%Y-%m-%d')
print("🔄 기술적 지표 계산 중...")
for idx, row in top100.iterrows():
try:
symbol = row['Symbol']
name = row['Name']
df = fdr.DataReader(symbol, start_date, end_date)
if df.empty or len(df) < 80:
continue
ind = calculate_indicators(df)
if ind is None or pd.isna(ind['RSI']):
continue
conditions_met = []
if ind['RSI'] <= RSI_THRESHOLD:
conditions_met.append(f"RSI({ind['RSI']:.1f})")
if ind['Close'] <= ind['BB_Lower'] * 1.01:
conditions_met.append("BB Lower")
ma20_dev = (ind['Close'] / ind['MA20'] - 1) * 100
ma60_dev = (ind['Close'] / ind['MA60'] - 1) * 100
if ma20_dev <= -15 or ma60_dev <= -25:
conditions_met.append(f"MA괴리({ma20_dev:.1f}%)")
if (ind['Return_1M'] and ind['Return_1M'] <= MONTH1_THRESHOLD) or \
(ind['Return_3M'] and ind['Return_3M'] <= MONTH3_THRESHOLD):
conditions_met.append(f"하락({ind['Return_1M']:.1f}%/{ind['Return_3M']:.1f}%)")
if conditions_met:
results.append({
'순위': idx + 1,
'종목코드': symbol,
'종목명': name,
'현재가': int(ind['Close']),
'RSI': round(ind['RSI'], 1),
'조건': " | ".join(conditions_met),
'1개월': f"{ind['Return_1M']:.1f}%" if ind['Return_1M'] else "-",
'3개월': f"{ind['Return_3M']:.1f}%" if ind['Return_3M'] else "-"
})
except:
continue
# 결과 처리
if results:
df_result = pd.DataFrame(results)
print("\n" + "="*140)
print(f"🔥 KOSPI Top 100 조건 충족 종목 ({len(df_result)}개) - {datetime.today().strftime('%Y-%m-%d %H:%M')}")
print("="*140)
print(df_result.to_string(index=False))
# Telegram 메시지 만들기
msg = f"🔥 *KOSPI Top 100 고급지표 알림*\n"
msg += f"📅 {datetime.today().strftime('%Y-%m-%d %H:%M')}\n"
msg += f"📊 조건 충족: {len(df_result)}개\n\n"
for _, row in df_result.iterrows():
msg += f"• *{row['종목명']}* ({row['종목코드']})\n"
msg += f" 현재가: {row['현재가']:,}원 | RSI: {row['RSI']}\n"
msg += f" 조건: {row['조건']}\n\n"
send_telegram_message(msg)
else:
print("✅ 오늘은 조건을 충족하는 대형주가 없습니다.")
# send_telegram_message("✅ 오늘 KOSPI Top 100 중 조건 충족 종목이 없습니다.")
if __name__ == "__main__":
monitor_stocks()
```
- 사용법
1. `TELEGRAM_BOT_TOKEN`과 `TELEGRAM_CHAT_ID`를 실제 값으로 바꾸기 2. 파일 저장 후 `python kospi_monitor.py` 실행
원하시면 **매일 자동 실행**을 위한 스케줄러(crond / Windows Task Scheduler) 설정 방법도 알려드릴 수 있어요!