# -*- coding: utf-8 -*-
import os
import pandas as pd
import requests
import feedparser
from datetime import datetime, timedelta
from dotenv import load_dotenv
from pykrx import stock
# ===============================
# ENV
# ===============================
load_dotenv()
TOKEN = os.getenv(“TELEGRAM_TOKEN”)
CHAT_ID = os.getenv(“CHAT_ID”)
TOP_N = int(os.getenv(“TOP_N”, 60))
PERIOD_DAYS = int(os.getenv(“PERIOD_DAYS”, 120))
# ===============================
# 텔레그램
# ===============================
def send_telegram(msg):
url = f”https://api.telegram.org/bot{TOKEN}/sendMessage”
requests.post(url, data={“chat_id”: CHAT_ID, “text”: msg,”parse_mode” : “HTML”})
# ===============================
# RSI
# ===============================
def get_rsi(df, period=14):
delta = df[‘Close’].diff()
up = delta.clip(lower=0)
down = -delta.clip(upper=0)
ma_up = up.rolling(period).mean()
ma_down = down.rolling(period).mean()
rs = ma_up / ma_down
return 100 – (100 / (1 + rs))
# ===============================
# 뉴스 감성
# ===============================
def get_news_sentiment(keyword):
url = f”https://news.google.com/rss/search?q={keyword}&hl=ko&gl=KR&ceid=KR:ko”
news = feedparser.parse(url)
positive = [“호재”, “성장”, “상승”, “수혜”, “급등”, “계약”, “실적”]
negative = [“악재”, “하락”, “적자”, “리스크”, “급락”, “소송”, “규제”]
score = 0
for entry in news.entries[:5]:
title = entry.title
for p in positive:
if p in title: score += 1
for n in negative:
if n in title: score -= 1
return score
# ===============================
# 뉴스 TOP10
# ==============================
def get_top_news():
url = “https://news.google.com/rss?hl=ko&gl=KR&ceid=KR:ko”
news = feedparser.parse(url)
result = []
seen = set()
for entry in news.entries:
if entry.title not in seen:
result.append((entry.title, entry.link))
seen.add(entry.title)
if len(result) >= 10:
break
return result
# ===============================
# 티커
# ===============================
TICKERS = {
“삼성전자”: “005930”,
“SK하이닉스”: “000660”,
“KODEX미국S&P500”: “379800”,
“TIGER200”: “102110”,
“TiGER반도체TOP10”: “396500”,
“KT”: “030200”,
“TIGER미국나스닥100”: “133690”,
“KODEX반도체”: “091150”
}
# ===============================
# 날짜
# ===============================
end = datetime.today()
start = end – timedelta(days=PERIOD_DAYS * 5)
start_str = start.strftime(“%Y%m%d”)
end_str = end.strftime(“%Y%m%d”)
# ===============================
# 코스피 (ETF 대체)
# ===============================
kospi = stock.get_market_ohlcv_by_date(start_str, end_str, “069500”)
kospi.rename(columns={“종가”: “Close”}, inplace=True)
if kospi.empty or len(kospi) < PERIOD_DAYS:
send_telegram(“❌ 코스피 데이터 오류”)
exit()
kospi_return = (kospi[‘Close’].iloc[-1] / kospi[‘Close’].iloc[-PERIOD_DAYS]) – 1
# ===============================
# 종목 분석
# ===============================
result = []
for name, ticker in TICKERS.items():
try:
df = stock.get_market_ohlcv_by_date(start_str, end_str, ticker)
if df.empty or len(df) < PERIOD_DAYS:
continue
df.rename(columns={
“시가”: “Open”,
“고가”: “High”,
“저가”: “Low”,
“종가”: “Close”,
“거래량”: “Volume”
}, inplace=True)
# 현재가 / 전일 대비
price_now = df[‘Close’].iloc[-1]
price_prev = df[‘Close’].iloc[-2]
diff = price_now – price_prev
diff_pct = (diff / price_prev) * 100
# 수익률
stock_return = (price_now / df[‘Close’].iloc[-PERIOD_DAYS]) – 1
excess = stock_return – kospi_return
# RSI
df[‘RSI’] = get_rsi(df)
rsi = df[‘RSI’].iloc[-1]
# 이동평균
ma20 = df[‘Close’].rolling(20).mean().iloc[-1]
# 거래량
vol = df[‘Volume’].iloc[-1]
vol_avg = df[‘Volume’].rolling(20).mean().iloc[-1]
# 뉴스 점수
news_score = get_news_sentiment(name)
# ===============================
# 매수 점수
# ===============================
buy_score = 0
if excess > 0: buy_score += 2
if rsi > 50: buy_score += 2
if price_now > ma20: buy_score += 2
if vol > vol_avg: buy_score += 2
buy_score += news_score
# ===============================
# 매도 점수
# ===============================
sell_score = 0
if rsi > 70: sell_score += 2
if price_now < ma20: sell_score += 2
if diff < 0: sell_score += 1
if excess < 0: sell_score += 1
# ===============================
# 신호 결정
# ===============================
if sell_score >= 4:
signal = “🔻매도”
elif buy_score >= 6:
signal = “🔥매수”
elif buy_score >= 4:
signal = “🟡관망”
else:
signal = “❌회피”
result.append([
name, stock_return, excess, rsi,
buy_score, sell_score, signal, news_score,
price_now, diff, diff_pct
])
except Exception as e:
print(f”{name} 오류:”, e)
# ===============================
# 정렬
# ===============================
df_result = pd.DataFrame(
result,
columns=[
“종목”, “수익률”, “초과수익”, “RSI”,
“매수점수”, “매도점수”, “신호”, “뉴스점수”,
“현재가”, “등락”, “등락률”
]
)
if df_result.empty:
send_telegram(“❌ 결과 없음”)
exit()
df_result = df_result.sort_values(by=”매수점수”, ascending=False)
topN = df_result.head(TOP_N)
# ===============================
# 메시지 생성
# ===============================
msg = f”📊 [AI 종목 스캐너]\n”
msg += f”KODEX200 기준 {PERIOD_DAYS}일 수익률: {kospi_return:.2%}\n\n”
for _, row in topN.iterrows():
arrow = “🔺” if row[‘등락’] > 0 else “🔻” if row[‘등락’] < 0 else “➖”
msg += (
f”{row[‘종목’]} | {row[‘신호’]}\n”
f”현재가: {row[‘현재가’]:,}원 ({arrow}{row[‘등락’]:+,.0f} / {row[‘등락률’]:+.2f}%)\n”
f”수익률: {row[‘수익률’]:.2%} / 초과: {row[‘초과수익’]:.2%}\n”
f”RSI: {row[‘RSI’]:.1f} | 매수:{row[‘매수점수’]} / 매도:{row[‘매도점수’]} | 뉴스:{row[‘뉴스점수’]}\n\n”
)
# ===============================
# 뉴스 TOP10
# ===============================
msg += “📰 [실시간 뉴스 TOP10]\n\n”
for i, (title, link) in enumerate(get_top_news(), 1):
msg += f'{i}. <a href=”{link}”>”{title}</a>\n\n’
# ===============================
# 전송
# ===============================
send_telegram(msg)
print(“✅ 전송 완료”)
Leave a Reply