try:
import pandas_ta as ta
except ImportError:
import os
os.system('pip install pandas_ta')
import pandas_ta as ta
import yfinance as yf
import pandas as pd
import requests
import json
import time
from datetime import datetime
import pytz
import warnings
warnings.filterwarnings('ignore', category=FutureWarning)
# ==========================================
# 🚨 사용자 정보 설정
# ==========================================
BOT_TOKEN = "8753917672:AAF1R00WdzDL-g71IE62SLRH-JzXLMZmD1M"
CHAT_ID = "8870518982"
KIS_APP_KEY = "PSNs2nTMTDtLfYE7FaWzAelDx8lJeyg9IjWo"
KIS_APP_SECRET = "Eye5K1BIrS8jpw7GrbQiaHOO/piXmh6WqD5l2U4b3FihOTMMnpJ5yZTmkeefh7JCP+oJ/oWruP/ilMHbi5EbJikdV2TqzNdyWRzN45ib8O9OuZsLtee0qq8+BjtIFNpS8K7/O7a7Cr8NZ6Y1ZBs5FANArHo0+mx1fIzie/j3RLNexEb7b4c="
KIS_ACCESS_TOKEN = ""
last_update_id = 0
alert_history = {}
PREV_CLOSE_MEMORY = {}
KR_STOCKS = {'005930.KS':'삼성전자', '000660.KS':'SK하이닉스'}
WATCH_LIST = {'SOXL':'SOXL(롱)', 'SOXS':'SOXS(숏)', 'NVDA':'엔비디아', 'TSM':'TSMC', 'AVGO':'브로드컴'}
def send_telegram(message):
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
payload = {"chat_id": CHAT_ID, "text": message}
try: requests.post(url, json=payload)
except: pass
def auth_kis_api():
global KIS_ACCESS_TOKEN
url = "https://openapi.koreainvestment.com:9443/oauth2/tokenP"
headers = {"content-type": "application/json"}
body = {"grant_type": "client_credentials", "appkey": KIS_APP_KEY, "appsecret": KIS_APP_SECRET}
try:
res = requests.post(url, headers=headers, data=json.dumps(body))
if res.status_code == 200:
KIS_ACCESS_TOKEN = res.json().get("access_token")
return True
return False
except: return False
def get_cached_prev_close(ticker):
if ticker in PREV_CLOSE_MEMORY:
return PREV_CLOSE_MEMORY[ticker]
try:
prev_close = float(yf.Ticker(ticker).fast_info.previous_close)
PREV_CLOSE_MEMORY[ticker] = prev_close
return prev_close
except: return None
# 🔥 [핵심 복구] KIS(한국투자증권) 실시간 데이터 엔진 롤백! (HTS와 오차 0%)
def get_kis_realtime_price(ticker, is_us=False):
if not is_us:
url = "https://openapi.koreainvestment.com:9443/uapi/domestic-stock/v1/quotations/inquire-price"
headers = {
"content-type": "application/json; charset=utf-8",
"authorization": f"Bearer {KIS_ACCESS_TOKEN}",
"appkey": KIS_APP_KEY, "appsecret": KIS_APP_SECRET, "tr_id": "FHKST01010100"
}
params = {"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": ticker.split('.')[0]}
try:
res = requests.get(url, headers=headers, params=params, timeout=3)
if res.status_code == 200:
data = res.json().get('output', {})
return float(data.get('stck_prpr', 0)), float(data.get('prdy_ctrt', 0))
except: pass
else:
url = "https://openapi.koreainvestment.com:9443/uapi/overseas-price/v1/quotations/price-detail"
headers = {
"content-type": "application/json; charset=utf-8",
"authorization": f"Bearer {KIS_ACCESS_TOKEN}",
"appkey": KIS_APP_KEY, "appsecret": KIS_APP_SECRET, "tr_id": "HHDFS76200200"
}
for excd in ["NAS", "NYS", "AMS"]:
params = {"AUTH": "", "EXCD": excd, "SYMB": ticker}
try:
res = requests.get(url, headers=headers, params=params, timeout=3)
if res.status_code == 200 and res.json().get('rt_cd') == '0':
data = res.json().get('output', {})
p = float(data.get('last', 0))
c = float(data.get('rate', 0))
return p, c
except: pass
return 0.0, 0.0
def get_us_price_with_fallback(ticker):
p, c = get_kis_realtime_price(ticker, is_us=True)
status_tag = ""
prev_close = get_cached_prev_close(ticker)
if p == 0.0:
try:
df_1m = yf.download(ticker, period='1d', interval='1m', prepost=True, progress=False, auto_adjust=False)
if not df_1m.empty:
closes_1m = df_1m['Close'][ticker] if isinstance(df_1m.columns, pd.MultiIndex) else df_1m['Close']
p = float(closes_1m.dropna().iloc[-1])
if prev_close: c = ((p - prev_close) / prev_close) * 100
status_tag = " 🔄(야후 백업)"
else:
status_tag = " 💤(수신대기)"
except: status_tag = " 💤(수신대기)"
elif p != 0.0 and c == 0.0 and prev_close:
c = ((p - prev_close) / prev_close) * 100
return p, c, status_tag
def get_index_data(ticker):
try:
prev_close = get_cached_prev_close(ticker)
current_close = float(yf.Ticker(ticker).fast_info.last_price)
if prev_close and current_close:
change = ((current_close - prev_close) / prev_close) * 100
return current_close, change
except: pass
return 0.0, 0.0
def get_ta_indicators(ticker):
try:
df = yf.download(ticker, period='2mo', interval='1d', progress=False, auto_adjust=False)
if df.empty: return 50.0, 0.0, 0.0
close_col = df['Close'][ticker] if isinstance(df.columns, pd.MultiIndex) else df['Close']
df_ta = pd.DataFrame({'Close': close_col.dropna()})
rsi_series = ta.rsi(df_ta['Close'], length=14)
bbands = ta.bbands(df_ta['Close'], length=20, std=2)
rsi = float(rsi_series.iloc[-1]) if not pd.isna(rsi_series.iloc[-1]) else 50.0
bb_low = float(bbands.iloc[-1, 0]) if bbands is not None and not pd.isna(bbands.iloc[-1, 0]) else 0.0
bb_high = float(bbands.iloc[-1, 2]) if bbands is not None and not pd.isna(bbands.iloc[-1, 2]) else 0.0
return rsi, bb_low, bb_high
except: return 50.0, 0.0, 0.0
# 🔥 스마트 세력 거래량 감지는 유지
def detect_smart_money(ticker):
try:
df = yf.Ticker(ticker).history(period='1d', interval='1m', prepost=True)
if len(df) < 5: return 0.0, 0.0
recent_avg_vol = df['Volume'].iloc[-5:-1].mean()
current_vol = df['Volume'].iloc[-1]
price_change = float(df['Close'].iloc[-1] - df['Open'].iloc[-1])
if recent_avg_vol > 0 and current_vol > (recent_avg_vol * 3):
return current_vol, price_change
except: pass
return 0.0, 0.0
def generate_mega_report():
print("📊 [리포트 생성 중] 한투(KIS) 실시간 데이터를 취합합니다...")
try:
tz_korea = pytz.timezone('Asia/Seoul')
now = datetime.now(tz_korea)
ndx_p, ndx_c = get_index_data('^IXIC')
spx_p, spx_c = get_index_data('^GSPC')
qqq_p, qqq_c, qqq_tag = get_us_price_with_fallback('QQQ')
soxl_p, soxl_c, soxl_tag = get_us_price_with_fallback('SOXL')
soxs_p, soxs_c, soxs_tag = get_us_price_with_fallback('SOXS')
nvda_p, nvda_c, nvda_tag = get_us_price_with_fallback('NVDA')
soxl_rsi, soxl_bb_low, soxl_bb_high = get_ta_indicators('SOXL')
report = f"📋 [2SK_Bot Smart Dashboard v13.0]\n⏰ {now.strftime('%Y-%m-%d %H:%M:%S')}\n\n"
report += "🌐 [주요 지수]\n"
report += f"• 나스닥: {ndx_p:,.2f} ({ndx_c:+.2f}%)\n"
report += f"• QQQ (ETF): ${qqq_p:.2f} ({qqq_c:+.2f}%)\n\n"
report += "--- ⚡ 실시간 타점 분석 (HTS 오차 0%) ---\n\n"
report += "🇺🇸 [SOXL/SOXS 포지션]\n"
report += f"• SOXL (롱): ${soxl_p:.2f} ({soxl_c:+.2f}%){soxl_tag}\n"
report += f" └ 📈 RSI: {soxl_rsi:.1f} | 밴드하단: ${soxl_bb_low:.2f} | 밴드상단: ${soxl_bb_high:.2f}\n"
report += f"• SOXS (숏): ${soxs_p:.2f} ({soxs_c:+.2f}%){soxs_tag}\n\n"
report += "🦅 [대장주 선행지표]\n"
report += f"• 엔비디아(NVDA): ${nvda_p:.2f} ({nvda_c:+.2f}%){nvda_tag}\n\n"
report += "🇰🇷 [K-반도체 지표]\n"
kr_trend = 0
for t, name in list(KR_STOCKS.items()):
p, c = get_kis_realtime_price(t, is_us=False)
report += f"• {name}: {int(p):,}원 ({c:+.2f}%)\n"
kr_trend += c
report += "\n🧠 [트레이딩 조언]\n"
if "💤" in soxl_tag:
report += "💡 데이터 수신 대기 중입니다."
else:
vol, vol_price_dir = detect_smart_money('SOXL')
if vol > 0 and vol_price_dir < 0:
report += "🚨 [투매 발생] SOXL에 막대한 거래량을 동반한 매도세가 출회 중입니다. 매수를 멈추고 관망하십시오."
elif vol > 0 and vol_price_dir > 0:
report += "🔥 [수급 폭발] SOXL에 거대한 매수세가 들어왔습니다. 단기 상승 모멘텀이 좋습니다."
elif soxl_c <= -4.0:
report += "⚠️ [하락 변동성] SOXL 하락세가 유지 중입니다. 추가 하락 가능성을 열어두고 숏(SOXS) 방어를 유지하십시오."
elif kr_trend >= 2.0 and soxl_c <= -2.0:
report += "💡 [수급 디커플링] 국장 대비 미장 반도체 수급이 부진합니다. 방향성 확인 후 진입을 권장합니다."
elif nvda_c <= -2.0:
report += "⚠️ [섹터 약세] 엔비디아 주도 하락이 진행 중입니다. 섣부른 물타기를 금지하십시오."
elif soxl_rsi <= 35:
report += "🟢 [기술적 반등권] RSI가 과매도권입니다. 보수적 분할 매수 접근이 유리합니다."
elif soxl_rsi >= 65:
report += "⚠️ [과매수 구간] 차익 실현 매물이 나올 수 있는 고점 징후가 있습니다."
elif nvda_c >= 2.0:
report += "🟢 [섹터 강세] 대장주 수급이 견조합니다. 매수 포지션(SOXL) 유지에 긍정적입니다."
else:
report += "⚖️ [방향성 탐색] 현재 뚜렷한 거래량 분출이나 추세가 없습니다. 관망을 권장합니다."
send_telegram(report)
print("✅ 리포트 발송 완료!")
except Exception as e: print(f"⚠️ 리포트 에러: {e}")
def run_smart_radar():
alerts_this_cycle = []
current_time = time.time()
for ticker, name in WATCH_LIST.items():
vol, price_dir = detect_smart_money(ticker)
if vol > 0:
last_vol_alert = alert_history.get(f"{ticker}_vol_time", 0)
if current_time - last_vol_alert >= 1200: # 20분 쿨다운
action = "매수세(펌핑)" if price_dir > 0 else "투매 물량(덤핑)"
alerts_this_cycle.append(f"• {name}: 🚨 최근 1분 거래량 300% 이상 폭발! ({action} 진행 중)")
alert_history[f"{ticker}_vol_time"] = current_time
if alerts_this_cycle:
msg = "🚨 [스마트 머니(세력) 수급 이상 감지]\n\n"
msg += "\n".join(alerts_this_cycle)
msg += "\n\n👉 대량 수급이 쏟아졌습니다. 즉시 HTS 차트를 확인하십시오!"
send_telegram(msg)
# ==========================================
# 🚀 메인 루프
# ==========================================
print("📡 [2SK_Bot v13.0] KIS API 복구 및 선행 수급 레이더 가동...")
if auth_kis_api():
print("✅ 한투 실시간 기관망 접속 완료!")
send_telegram("🚀 [가격/시세 오류 완벽 픽스]\n\n사장님! 사장님의 HTS와 가격이 완벽히 똑같도록 메인 엔진을 다시 한국투자증권(KIS) API로 롤백했습니다. 가격은 정확하게, 폭락 경고는 '거래량 감지'로 한 발 더 빠르게 쏴드리겠습니다!")
try: requests.get(f"https://api.telegram.org/bot{BOT_TOKEN}/getUpdates?offset=-1")
except: pass
last_analyze_time = time.time()
last_print_time = time.time()
while True:
try:
url = f"https://api.telegram.org/bot{BOT_TOKEN}/getUpdates?offset={last_update_id}&timeout=1"
res = requests.get(url)
if res.status_code == 200:
for item in res.json().get('result', []):
last_update_id = item['update_id'] + 1
msg_text = item.get('message', {}).get('text', '')
if msg_text in ['보고', '상황', '/now', 'ㅂ', 'q']:
generate_mega_report()
except: pass
current_time = time.time()
# 거래량 선행 지표 감시
if current_time - last_analyze_time >= 5:
run_smart_radar()
last_analyze_time = current_time
if current_time - last_print_time >= 15:
now_str = datetime.now(pytz.timezone('Asia/Seoul')).strftime('%H:%M:%S')
print(f"[{now_str}] ⚡ v13.0 KIS망 + 선행 거래량 감시 작동 중...")
last_print_time = current_time
time.sleep(1)